AI Agent工作流设计:从ReAct到Multi-Agent协作
·
AI Agent工作流设计:从ReAct到Multi-Agent协作
AI Agent(智能体)正从概念验证走向生产部署。如何设计可靠的Agent工作流,确保大语言模型在复杂任务中稳定输出,是当前工程实践的核心挑战。本文将从单Agent的ReAct模式出发,逐步深入到Multi-Agent协作架构,结合代码示例展示完整的设计方法论。
一、单Agent基础:ReAct推理框架
ReAct(Reasoning + Acting)是构建AI Agent最经典的框架,它通过"思考→行动→观察"的循环,让LLM具备工具使用和环境交互能力。
1.1 ReAct核心循环
from typing import List, Dict, Any
class ReActAgent:
def __init__(self, llm, tools: List[Dict]):
self.llm = llm
self.tools = {t['name']: t['function'] for t in tools}
self.tool_descriptions = self._format_tool_descriptions(tools)
def run(self, query: str, max_steps: int = 10) -> str:
thought_history = []
action_history = []
for step in range(max_steps):
# 构建Prompt:包含任务、工具说明、历史轨迹
prompt = self._build_prompt(query, thought_history, action_history)
# LLM输出推理与行动决策
response = self.llm.generate(prompt)
# 解析Thought和Action
thought = self._extract_thought(response)
action = self._extract_action(response)
if action['name'] == 'finish':
return action['arguments']['answer']
# 执行工具调用
observation = self.tools[action['name']](**action['arguments'])
thought_history.append(thought)
action_history.append({
'action': action,
'observation': str(observation)[:500] # 截断避免上下文过长
})
return "达到最大步数限制,任务未完成"
def _build_prompt(self, query, thoughts, actions):
template = f"""你是一个智能助手,可以使用以下工具:
{self.tool_descriptions}
请按照以下格式思考并行动:
Thought: [你的推理过程]
Action: [工具名称] [参数JSON]
Observation: [工具返回结果]
任务:{query}
历史轨迹:
"""
for t, a in zip(thoughts, actions):
template += f"Thought: {t}\nAction: {a['action']}\nObservation: {a['observation']}\n"
return template
1.2 ReAct的局限性
ReAct虽然优雅,但在生产环境面临挑战:
| 问题 | 表现 | 影响 | |------|------|------| | 推理漂移 | 多步后偏离原始目标 | 最终答案错误 | | 工具选择错误 | 选错工具或参数格式错误 | 调用失败 | | 循环陷阱 | 重复相同行动 | 无法收敛 | | 上下文膨胀 | 历史轨迹过长 | 成本增加、质量下降 |
二、增强单Agent:规划与记忆
2.1 结构化规划
为缓解推理漂移,引入显式规划模块:
class PlanAndExecuteAgent:
def __init__(self, llm, tools):
self.llm = llm
self.tools = tools
def _create_plan(self, query: str) -> List[str]:
"""先制定执行计划,再按步执行"""
prompt = f"""请为以下任务制定详细的执行计划,每步只使用一个工具。
可用工具:搜索、计算器、代码执行器
任务:{query}
请以JSON数组格式输出步骤:"""
response = self.llm.generate(prompt)
return json.loads(response)
def run(self, query: str) -> str:
plan = self._create_plan(query)
execution_memory = []
for step in 更多推荐


所有评论(0)