凌晨两点,我的智能客服 Agent 又一次陷入了死循环——它不断地调用搜索工具、查询数据库、又调用搜索工具,却始终无法给出最终答案。日志里充斥着这样的错误:

ConnectionError: HTTPSConnectionPool(host='api.holysheep.ai', port=443): Max retries exceeded
RuntimeError: Maximum recursion depth exceeded in __pure_call__
ValueError: Tool result parsing failed after 3 attempts

这不是 API 的问题,也不是网络的问题。问题出在我最初设计的 Agent 架构上——我没有正确区分「规划」与「执行」这两个截然不同的阶段。如果你也在开发复杂的 AI Agent,这篇文章将帮你彻底解决这个问题。

为什么 AI Agent 需要规划与执行分离

在我早期的一个项目里,我设计了一个多步骤数据分析师 Agent。它需要:理解用户问题 → 分解任务 → 选择工具 → 执行 → 验证结果 → 汇总报告。最初我用一个统一的 ReAct 循环来处理所有这些步骤,结果导致两个严重问题:

解决方案是将 Agent 拆分为独立的「规划器」和「执行器」,这正是我在本文要详细介绍的两种主流模式:ReAct 模式Plan 模式

ReAct 模式:边想边做的反应式执行

核心原理

ReAct(Reasoning + Acting)模式将推理和执行交织在一起。Agent 在每一步都会:思考当前状态 → 选择一个动作 → 执行并观察结果 → 继续下一轮循环。这种模式适合动态环境,工具返回结果不可预测的场景。

API 设计实现

import httpx
import json
from typing import List, Dict, Any

class ReActAgent:
    def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
        self.client = httpx.Client(timeout=120.0)
        self.base_url = base_url
        self.api_key = api_key
        self.max_iterations = 15
        
    def chat(self, messages: List[Dict], tools: List[Dict]) -> Dict[str, Any]:
        """ReAct 模式:带工具调用的对话"""
        system_prompt = """你是一个智能助手。在每一步推理中,你需要:
1. 思考:分析当前状态和目标
2. 行动:从可用工具中选择一个执行
3. 观察:记录工具返回结果
当任务完成时,用自然语言返回最终答案。"""
        
        full_messages = [{"role": "system", "content": system_prompt}] + messages
        
        for iteration in range(self.max_iterations):
            response = self._call_llm(full_messages, tools)
            
            # 检查是否完成
            if response.get("finish_reason") == "stop":
                return {"status": "success", "result": response}
            
            # 处理工具调用
            tool_calls = response.get("tool_calls", [])
            if not tool_calls:
                break
                
            # 执行工具并收集结果
            tool_results = []
            for call in tool_calls:
                result = self._execute_tool(call)
                tool_results.append(result)
                full_messages.append({
                    "role": "tool",
                    "tool_call_id": call["id"],
                    "content": json.dumps(result, ensure_ascii=False)
                })
                
        return {"status": "completed", "messages": full_messages}
    
    def _call_llm(self, messages: List[Dict], tools: List[Dict]) -> Dict:
        """调用 HolySheep API"""
        response = self.client.post(
            f"{self.base_url}/chat/completions",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": "gpt-4.1",
                "messages": messages,
                "tools": tools,
                "temperature": 0.3
            }
        )
        
        if response.status_code == 401:
            raise PermissionError("API Key 无效或已过期,请检查 https://www.holysheep.ai/settings")
        elif response.status_code == 429:
            raise TimeoutError("请求频率超限,请稍后重试或升级套餐")
        
        return response.json()["choices"][0]["message"]
    
    def _execute_tool(self, tool_call: Dict) -> Dict:
        """模拟工具执行(实际项目中替换为真实工具调用)"""
        return {"status": "executed", "tool": tool_call["function"]["name"]}

ReAct 适用场景

Plan 模式:先规划后执行的两阶段架构

核心原理

Plan 模式将 Agent 行为分为两个独立阶段:

  1. 规划阶段:Agent 先理解任务目标,生成完整的执行计划(包含步骤序列、依赖关系、备选方案)
  2. 执行阶段:严格按照计划执行,可并行处理无依赖的步骤

我在开发企业知识库问答系统时采用了这种架构,将响应时间从平均 8 秒降低到了 2.5 秒,Token 消耗减少了 65%。

API 设计实现

import httpx
import json
from typing import List, Dict, Any
from concurrent.futures import ThreadPoolExecutor, as_completed

class PlanModeAgent:
    def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
        self.client = httpx.Client(timeout=180.0)
        self.base_url = base_url
        self.api_key = api_key
        
    def execute(self, user_query: str) -> Dict[str, Any]:
        """Plan 模式完整流程"""
        
        # 第一阶段:生成执行计划
        plan = self._generate_plan(user_query)
        print(f"[计划] 任务分解为 {len(plan['steps'])} 个步骤")
        
        # 第二阶段:按依赖关系执行计划
        results = self._execute_plan(plan)
        
        # 第三阶段:整合结果
        return self._synthesize_results(user_query, plan, results)
    
    def _generate_plan(self, query: str) -> Dict[str, Any]:
        """调用规划器生成执行计划"""
        
        planning_prompt = f"""用户需求:{query}

请将这个任务分解为可执行的步骤,输出 JSON 格式:
{{
    "steps": [
        {{
            "id": 1,
            "description": "步骤描述",
            "tool": "工具名",
            "dependencies": [],
            "estimated_time": "预估时间"
        }}
    ],
    "parallel_groups": [[1,2], [3]],  // 可并行的步骤分组
    "final_answer_format": "最终答案格式要求"
}}"""
        
        response = self.client.post(
            f"{self.base_url}/chat/completions",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": "claude-sonnet-