构建一个有手有脑的 AI Agent:ReAct 推理循环、函数调用与多智能体编排实战
2026 年,Agentic AI 已经从 Demo 走向生产线。但大多数号称"智能体"的实现,本质只是把大模型包了一层 while 循环:把用户问题塞进去,把回答吐出来,最多接一两个写死的 API。真正的生产级 Agent 要解决三件事:会思考(推理闭环)、能动手(工具调用)、可协作(多智能体状态编排)。本文用一个可运行的 Python 实现,把这三件事讲透。
一、ReAct 不是 Prompt 模板,而是一个状态机
ReAct(Reasoning + Acting)的核心是把"思考"和"行动"交错进行。每一轮,模型输出一个 Thought(为什么这么做)和一个 Action(调用哪个工具、传什么参数),执行器拿到结果后作为 Observation 喂回模型,直到模型输出 Finish。
from dataclasses import dataclass
@dataclass
class Step:
thought: str
action: str | None
action_input: dict | None
observation: str | None = None
def parse_react(text: str) -> Step:
# 用分隔符解析模型输出,生产环境建议改用结构化 function calling
thought = text.split("Thought:")[1].split("Action:")[0].strip()
if "Action:" in text:
action_part = text.split("Action:")[1]
action = action_part.split("(")[0].strip()
args = action_part.split("(", 1)[1].rstrip(")").strip()
return Step(thought, action, {"arg": args})
return Step(thought, None, None)
注意:用字符串分隔符解析是教学用的脆弱做法。生产环境务必让模型走 function calling(结构化输出),下一节展开。
二、函数调用:把"工具"变成强类型契约
让模型自由生成文本再正则解析,是 90% 线上 Agent 不稳定的根因。正确做法是把工具定义为 JSON Schema,由模型返回结构化的 tool_calls,由你来控制执行与类型校验。
TOOLS = [{
"type": "function",
"function": {
"name": "search_kb",
"description": "检索内部知识库,返回最相关的 3 条文档",
"parameters": {
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"],
},
},
}]
def run_tool(name: str, args: dict) -> str:
if name == "search_kb":
return kb.search(args["query"], top_k=3) # 真实检索实现
raise ValueError(f"unknown tool: {name}")
关键点:工具的描述(description)是给模型看的 prompt,写得越精确,误调用越少。每个参数都要有类型与示例。
三、记忆不是堆历史,而是分层
Agent 的上下文窗口有限,把所有对话塞进去会迅速爆掉并烧钱。我采用两层记忆:
- 短期记忆:最近 N 轮 + 当前任务的中间步骤,放进 prompt。
- 长期记忆:把历史对话 embedding 后存入向量库,需要时按相关性召回。
def build_context(query, history, vectorstore):
recent = history[-6:] # 短期:仅保留近 6 轮
recalled = vectorstore.search(query, k=3) # 长期:语义召回
return format(recent, recalled)
一个反直觉的优化:长期记忆召回时,优先召回"结论"而非"原文"——把摘要索引进去,召回质量更高、token 更省。
四、多智能体:用"主管-Worker"模式拆分职责
单一 Agent 同时做规划、检索、计算、写作,容易顾此失彼。更稳的架构是 Supervisor 负责拆解与派发,Worker 各司其职:
WORKERS = {"researcher": researcher_agent, "coder": coder_agent, "writer": writer_agent}
def supervisor(task):
plan = llm.plan(task) # 拆成子任务列表
results = {}
for sub in plan:
worker = WORKERS[sub.owner]
results[sub.id] = worker.run(sub, ctx=results) # 子任务结果可互相引用
return writer_agent.synthesize(results)
这里有两个工程坑:(1)Worker 之间的上下文传递——不要把全部中间结果无脑转发,只传必要的"契约字段";(2)循环依赖——用 DAG 而不是任意图来编排,否则会死循环。
五、上线前的四道保险
- 重试与降级:工具调用失败必须指数退避重试,且要有 fallback 工具,绝不直接把异常抛给用户。
- 护栏(Guardrail):对模型的工具参数做白名单/范围校验,对输出做敏感词与格式校验。
- 可观测性:每一步的 thought、action、observation、token 用量都要落日志,用 trace id 串联,否则出问题完全无法复盘。
- 成本闸门:给单次会话设 token 预算上限,超限即中断并提示,避免跑飞。
def guard(call):
def wrap(name, args, budget):
if budget.exceeded():
return "⚠️ 已达成本上限,请简化需求"
capped = {k: v for k, v in args.items() if k in ALLOWED[name]}
return call(name, capped)
return wrap
六、结语
ReAct 解决了"会思考",function calling 解决了"能动手",分层记忆与多智能体解决了"可协作"。但 Agent 真正的难点从来不在模型,而在工程约束:状态怎么管、工具怎么安全调、失败怎么兜。把上面这套骨架跑通,你手里的就不再是"玩具",而是一个能在生产线上扛事的智能体。

发表评论 取消回复