构建一个有手有脑的 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 而不是任意图来编排,否则会死循环。

五、上线前的四道保险

  1. 重试与降级:工具调用失败必须指数退避重试,且要有 fallback 工具,绝不直接把异常抛给用户。
  2. 护栏(Guardrail):对模型的工具参数做白名单/范围校验,对输出做敏感词与格式校验。
  3. 可观测性:每一步的 thought、action、observation、token 用量都要落日志,用 trace id 串联,否则出问题完全无法复盘。
  4. 成本闸门:给单次会话设 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 真正的难点从来不在模型,而在工程约束:状态怎么管、工具怎么安全调、失败怎么兜。把上面这套骨架跑通,你手里的就不再是"玩具",而是一个能在生产线上扛事的智能体。

点赞(0) 打赏

评论列表 共有 0 条评论

暂无评论
立即
投稿

微信公众账号

微信扫一扫加关注

发表
评论
返回
顶部
0.362957s