凌晨两点,我的智能客服 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 循环来处理所有这些步骤,结果导致两个严重问题:
- 执行效率低下:每一步都需要调用 LLM 做推理,即使某些步骤可以并行执行
- Token 成本失控:一个简单的数据分析请求产生了超过 50 万 Token 的消耗
- 错误累积:早期步骤的错误会级联传播,最终结果完全偏离用户意图
解决方案是将 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 行为分为两个独立阶段:
- 规划阶段:Agent 先理解任务目标,生成完整的执行计划(包含步骤序列、依赖关系、备选方案)
- 执行阶段:严格按照计划执行,可并行处理无依赖的步骤
我在开发企业知识库问答系统时采用了这种架构,将响应时间从平均 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-