MCP与Agent系统集成：ReAct架构实战

{
  "tool_id": "weather_tool",
  "name": "天气查询工具",
  "description": "查询指定城市和日期的天气信息",
  "input_schema": {
    "type": "object",
    "properties": {
      "city": {
        "type": "string",
        "description": "城市名称"
      },
      "date": {
        "type": "string",
        "format": "date",
        "description": "查询日期"
      }
    },
    "required": ["city", "date"]
  },
  "output_schema": {
    "type": "object",
    "properties": {
      "city": {"type": "string"},
      "date": {"type": "string", "format": "date"},
      "temperature": {"type": "number", "description": "温度（摄氏度）"},
      "humidity": {"type": "number", "description": "湿度（百分比）"},
      "description": {"type": "string", "description": "天气描述"},
      "wind_speed": {"type": "number", "description": "风速（km/h）"}
    }
  }
}

# MCP上下文管理示例
class MCPContextManager:
    def __init__(self):
        self.context_store = {}
    
    def create_context(self, context_id, initial_data=None):
        """创建新的上下文"""
        self.context_store[context_id] = {
            "created_at": datetime.now(),
            "updated_at": datetime.now(),
            "data": initial_data or {}
        }
        return context_id
    
    def update_context(self, context_id, data):
        """更新上下文数据"""
        if context_id in self.context_store:
            self.context_store[context_id]["data"].update(data)
            self.context_store[context_id]["updated_at"] = datetime.now()
    
    def get_context(self, context_id):
        """获取上下文数据"""
        return self.context_store.get(context_id, {}).get("data", {})

# 多Agent协作示例
async def multi_agent_collaboration(mcp_client, task_description):
    """多Agent协作完成任务"""
    # 1. 注册Agent
    await mcp_client.register_agent({
        "agent_id": "travel_planner_agent",
        "name": "旅行规划Agent",
        "capabilities": ["flight_booking", "hotel_booking", "itinerary_planning"]
    })
    
    await mcp_client.register_agent({
        "agent_id": "weather_agent",
        "name": "天气查询Agent",
        "capabilities": ["weather_forecast", "climate_analysis"]
    })
    
    # 2. 分配任务
    task_id = await mcp_client.create_task({
        "description": task_description,
        "required_capabilities": ["flight_booking", "weather_forecast"]
    })
    
    # 3. 监控任务执行
    while True:
        task_status = await mcp_client.get_task_status(task_id)
        if task_status["status"] in ["completed", "failed"]:
            break
        await asyncio.sleep(1)
    
    return task_status["result"]

# 强化学习优化的ReAct推理
class RLReactAgent:
    def __init__(self, llm_client, mcp_client, policy_network):
        self.llm_client = llm_client
        self.mcp_client = mcp_client
        self.policy_network = policy_network
        self.context = []
    
    async def react_with_rl(self, user_query):
        """使用强化学习优化的ReAct推理"""
        self.context.append(f"用户: {user_query}")
        
        for step in range(self.max_steps):
            # 1. 生成状态表示
            state = self.build_state()
            
            # 2. 使用策略网络选择动作
            action_type = self.policy_network.predict(state)
            
            # 3. 执行动作
            if action_type == "reason":
                # 生成推理
                thought = await self.generate_thought()
                self.context.append(f"思考: {thought}")
            else:
                # 调用工具
                action = await self.generate_action()
                observation = await self.execute_action(action)
                self.context.append(f"行动: {action}")
                self.context.append(f"观察: {observation}")
            
            # 4. 检查是否完成任务
            if self.check_task_completed():
                return self.generate_final_answer()
        
        return self.generate_final_answer()

{
  "tool_id": "image_analysis_tool",
  "name": "图像分析工具",
  "description": "分析图像内容，提取关键信息",
  "input_schema": {
    "type": "object",
    "properties": {
      "image": {
        "type": "string",
        "format": "binary",
        "description": "图像数据（Base64编码）"
      },
      "analysis_type": {
        "type": "string",
        "enum": ["object_detection", "scene_recognition", "text_extraction"],
        "description": "分析类型"
      }
    },
    "required": ["image", "analysis_type"]
  },
  "output_schema": {
    "type": "object",
    "properties": {
      "result": {
        "type": "array",
        "items": {
          "type": "object",
          "properties": {
            "label": {"type": "string"},
            "confidence": {"type": "number"},
            "bounding_box": {
              "type": "object",
              "properties": {
                "x": {"type": "number"},
                "y": {"type": "number"},
                "width": {"type": "number"},
                "height": {"type": "number"}
              }
            }
          }
        }
      }
    }
  }
}

# 初始化旅行规划Agent
class TravelPlanningAgent:
    def __init__(self, llm_client, mcp_client):
        self.llm_client = llm_client  # LLM客户端
        self.mcp_client = mcp_client  # MCP客户端
        self.context = []  # 上下文历史
        self.max_steps = 20  # 最大推理步骤
    
    async def initialize(self):
        """初始化Agent，获取可用工具列表"""
        self.tools = await self.mcp_client.get_tools()
        self.tool_descriptions = [
            f"{tool['tool_id']}: {tool['description']}" 
            for tool in self.tools
        ]

# ReAct推理循环
async def react_loop(self, user_query):
    """ReAct推理主循环"""
    # 1. 初始化上下文
    self.context.append(f"用户: {user_query}")
    
    for step in range(self.max_steps):
        # 2. 生成推理和行动
        thought_action = await self.generate_thought_action()
        
        # 3. 解析推理和行动
        thought, action = self.parse_thought_action(thought_action)
        
        # 4. 如果是最终回答，返回结果
        if action is None:
            return thought
        
        # 5. 执行行动（调用工具）
        observation = await self.execute_action(action)
        
        # 6. 更新上下文
        self.context.append(f"思考: {thought}")
        self.context.append(f"行动: {action}")
        self.context.append(f"观察: {observation}")
    
    # 超过最大步骤，返回最终答案
    final_answer = await self.generate_final_answer()
    return final_answer

travel_agent/
├── agent/
│   ├── __init__.py
│   ├── travel_agent.py      # 旅行规划Agent核心逻辑
│   ├── react_engine.py       # ReAct推理引擎
│   └── context_manager.py    # 上下文管理
├── clients/
│   ├── __init__.py
│   ├── llm_client.py         # LLM客户端
│   └── mcp_client.py         # MCP客户端
├── tools/
│   ├── __init__.py
│   ├── weather_tool.py       # 天气查询工具
│   ├── flight_tool.py        # 航班查询工具
│   └── hotel_tool.py         # 酒店查询工具
├── config/
│   ├── __init__.py
│   └── config.py             # 配置管理
├── main.py                   # 主程序入口
└── requirements.txt          # 依赖列表

你是一个智能助手，能够根据用户需求执行任务。

可用工具：
{tools}

请按照以下格式生成思考和行动：
思考：[你的思考过程]
行动：工具ID(参数1=值1, 参数2=值2, ...)

或者如果已经有足够信息回答用户问题，直接生成最终答案：
思考：[你的思考过程]
最终答案：[你的最终回答]

上下文历史：
{context}

请生成下一步的思考和行动：

{
  "mcp_server": {
    "host": "0.0.0.0",
    "port": 8000,
    "ssl": {
      "enabled": true,
      "cert_file": "ssl/cert.pem",
      "key_file": "ssl/key.pem"
    },
    "authentication": {
      "enabled": true,
      "methods": ["api_key", "jwt"]
    },
    "rate_limiting": {
      "enabled": true,
      "requests_per_minute": 100
    },
    "monitoring": {
      "enabled": true,
      "metrics_endpoint": "/metrics",
      "logging_level": "INFO"
    }
  }
}