第5章 AI Agent:让模型自己干活

前面几章,模型一直是个"问答机器":你提问,它回答,一轮结束。这一章,我们让它"自己干活" - 面对一个任务,它自己判断该做什么、自己调用工具、自己检查结果,直到把事情办完。这就是 AI Agent(智能体)。我们会先拆解 Agent 的四要素,再讲最经典的 ReAct 模式,最后动手写一个几十行、真正能干活的最小 Agent。学完这一章,你会看穿所有 Agent 框架(LangChain、AutoGen 等)背后的那套循环 - 因为框架不神秘,循环才是。

5.1 Agent 四要素:模型、规划、工具、记忆

先看一个具体场景。你让 ChatGPT 对比三款手机的到手价并给出购买建议,它通常回答:"抱歉,我无法访问实时网页信息。"模型不缺知识,缺的是"手" - 它只能说话,不能做事。Agent 要解决的正是这个问题。

Agent 的定义可以浓缩成一句话:以模型为大脑,通过规划、工具、记忆三件装备,自主完成任务的程序。四个要素缺一不可,如图 5-1 所示。

图 5-1 Agent 四要素:模型是大脑,规划、工具、记忆是它的三件装备

逐个拆开看。模型是大脑,负责理解任务、推理决策 - "下一步做什么"由它说了算,这是第 2 章调 API 的能力,Agent 只是把它放进循环里反复使用。规划是调度:把"写一份季度报告"拆成"查数据、算指标、写正文、排版"四步,再安排先后顺序(5.5 详述)。工具是模型的手脚:搜索、计算器、数据库、代码执行,把"决定"变成"动作" - 模型说"我要查天气",真正联网的是你的函数(5.3)。记忆是笔记本:短期记忆就是上下文窗口,长期记忆是向量库,让 Agent 记得住"你是谁、干到哪了"(5.4)。

"ChatGPT 是对话,Agent 是做事",区别不在模型,而在程序结构。一次 API 调用是"回答一个问题";Agent 是一个 while 循环 - 围绕一个任务,反复"思考→行动→观察",直到任务完成。模型还是那个模型,但程序从"一问一答"变成了"自主闭环"。

重要

记住这个公式
Agent = 模型(大脑)+ 规划(调度)+ 工具(手脚)+ 记忆(笔记本)。没有工具,模型只会说不会做;没有记忆,它每轮都像第一次见面;没有规划,复杂任务一步到不了。

提示

类比:Agent 像一个实习生
脑子好使(模型)、会列工作计划(规划)、有手有脚能跑腿(工具)、随身带小本子记下老板的偏好(记忆)。你只交代一句"把这件事办完",剩下的他自主推进,关键节点回来向你汇报。

5.2 ReAct:思考、行动、观察

上一节说 Agent 是一个循环,这一节讲清楚循环怎么转。先回答一个"为什么":为什么不能一次调用解决?因为一次调用只能"想",不能"做" - 模型算出"需要查天气",但查天气这个动作必须由你的代码完成,而执行结果又要送回模型,它才能继续推理。这个"想-做-看-再想"的交替,就是 ReAct。

ReAct 由论文《ReAct: Synergizing Reasoning and Acting in Language Models》(2022)提出,名字是 Reasoning(推理)与 Acting(行动)的合成。它的核心洞察是:把思考说出来,把行动做出来,再把结果看回来 - 每一步都用真实世界的反馈修正思考,而不是让模型闷头瞎想。注意,ReAct 不是一个新模型,而是一种组织提示与循环的方式,任何能对话的模型都能跑,2.4 节的函数调用是它的天然载体。

循环分三步,如图 5-2 所示:

图 5-2 ReAct 循环:思考 → 行动 → 观察 → 再思考,直到输出最终答案(全章核心图)

第一步 Thought(思考):模型用自然语言写出推理,例如"用户要对比两地天气,我需要先分别查询北京和上海"。第二步 Action(行动):模型输出要调用的工具和参数,例如 get_weather("北京") - 注意,模型只负责"决定",真正执行的是你的代码。第三步 Observation(观察):程序执行工具,把结果作为一条消息放回对话上下文,模型"看到"结果后继续思考,进入下一轮循环。如此往复,直到模型认为任务完成,输出最终答案(代码里我们约定 FINAL 前缀)。

整个循环里,模型始终是"决策者",你的代码是"执行者"和"搬运工" - 把工具结果搬回上下文,是 Agent 程序最核心的职责。什么时候停?不要等模型"说完了"就停,要设显式的退出条件:模型输出 FINAL,或超过最大步数。这两条,我们在 5.6 的代码里都会见到。

注记

Observation 在 API 层怎么实现
把工具返回的字符串作为一条新消息追加进 messages - 有的 SDK 用 role:"tool"(第 2.4 节),有的直接塞一条 user 消息。形式不重要,重要的是结果必须由你的代码回填,而不是让模型自己脑补

5.3 工具调用实战

ReAct 的 Action 层,落到 API 上就是第 2.4 节的函数调用:把函数声明传给模型(tools 参数),模型返回要调用的函数名和参数(tool_calls),你执行后把结果以 role:"tool" 回传。第 2 章你用它做"一句话查天气",这一章把它升级成 Agent 的"手" - 同一套机制,放进循环就是工具调用。

先建立一个正确认知:工具不是模型"会"的。模型不联网、不算数、不读写文件,它只会"读 JSON、选参数"。真正的能力在你的函数里,模型的价值是理解意图、决定用哪个工具、填对参数。所以工具层要做好两件事:声明清楚(让模型知道有什么、怎么用),执行可靠(让真实世界安全地完成动作)。常用工具按用途大致分几类:

工具 干什么用 典型实现
网页/知识库搜索 查实时信息、公司资料 搜索 API,或第 3 章的 RAG 检索
计算器 精确计算、单位换算 白名单表达式 evaldecimal
代码执行 跑 Python、算数据、画图 沙箱环境(容器/受限解释器)
文件读写 生成文档、整理数据 文件系统 + 路径白名单校验
数据库查询 查结构化业务数据 SQL 封装 + 参数化查询
第三方 API 发消息、下单、支付 对应服务 SDK(危险,见 5.7)

下面是一份典型的工具注册表。声明部分用第 2 章讲过、各家 SDK 都兼容的 JSON Schema 格式;执行部分是一个"名字→函数"的字典:

# ===== 工具注册表:声明 + 执行 =====
def get_weather(city: str, date: str = "今天") -> str:
    """查天气(示例:真实项目换成天气 API)"""
    return f"{city} {date}:多云,22~28℃,微风"

def calc(expr: str) -> str:
    """安全计算器:只允许数字和四则运算"""
    import re
    if not re.fullmatch(r"[\d+\-*/(). ]+", expr):   # 白名单校验
        return "错误:表达式包含非法字符"
    return str(eval(expr))                          # 仅限白名单表达式

# 声明部分:把函数"翻译"成模型能懂的 JSON Schema(第 2.4 节)
TOOLS = [
    {"type": "function", "function": {
        "name": "get_weather",
        "description": "查询指定城市某天的天气",
        "parameters": {"type": "object",
                       "properties": {
                           "city": {"type": "string", "description": "城市名"},
                           "date": {"type": "string", "description": "日期,默认今天"}},
                       "required": ["city"]}}},
    {"type": "function", "function": {
        "name": "calc",
        "description": "计算四则运算表达式",
        "parameters": {"type": "object",
                       "properties": {
                           "expr": {"type": "string", "description": "如 2*3+4"}},
                       "required": ["expr"]}}},
]

# 执行部分:名字 -> 真实函数,统一入口
TOOL_IMPL = {"get_weather": get_weather, "calc": calc}

def run_tool(name: str, args: dict) -> str:
    """执行工具:参数校验 + 调用 + 统一转成字符串返回"""
    if name not in TOOL_IMPL:
        return f"错误:没有名为 {name} 的工具"
    try:
        return str(TOOL_IMPL[name](**args))
    except Exception as e:
        return f"工具执行失败:{e}"

模型返回 tool_calls 之后,你的程序要把"决定"变成"动作",并把结果回填,好让模型在下一轮看到:

# ===== 模型返回 tool_calls 后:执行并回填 =====
def call_llm(messages, tools=None):
    """调用大模型,返回(回答文本, 工具调用列表)"""
    resp = client.chat.completions.create(
        model="gpt-4o-mini", messages=messages, tools=tools)
    msg = resp.choices[0].message
    return msg.content, getattr(msg, "tool_calls", None)

def handle_tool_calls(tool_calls, messages):
    """把模型的工具调用真正执行,结果以 role='tool' 回填"""
    for tc in tool_calls:
        name = tc.function.name
        args = json.loads(tc.function.arguments)   # 参数是 JSON 字符串
        result = run_tool(name, args)              # 执行真实函数
        messages.append({                          # 回填:必须由代码做
            "role": "tool",
            "tool_call_id": tc.id,                 # 用 id 对应本次调用
            "content": result,
        })
    return messages
注记

防御性工具
模型填错参数是常态。工具函数要做参数校验(类型、取值范围、白名单),返回结果统一转成字符串。宁可返回"参数不合法",也不要让异常把整个 Agent 崩掉;工具命名用动词开头、语义清晰(search_web 而非 s1)。

5.4 记忆系统:短期与长期

Agent 是循环的:每轮调用只看到当前 messages 里的内容,上一轮结束,上下文不在了,它什么都不记得。可真实任务需要连续性 - "我上次让你整理的表格,这次接着改""我习惯早上看简报"。没有记忆,Agent 每轮都像第一次见面。记忆系统分两层。

短期记忆就是上下文窗口本身messages 里的对话历史、工具结果都在这一层;窗口越大,能记住的越多,但代价是费用上涨、延迟变长(2.6 讲过 token 即成本)。对策是第 2.5 节的老朋友:裁剪 - 只保留最近 N 轮;摘要 - 把旧对话压成一段"到目前为止的进展"。在 Agent 里,这两件事通常由程序自动触发,而不是靠人工整理。

长期记忆要解决"跨会话"的问题,用的正是第 3 章 RAG 的机制:把值得记住的内容(用户偏好、任务结论、完成状态)embedding 后存入向量库,新一轮开始时,把当前问题向量化,检索出最相关的几条旧记忆注入上下文。你可以把长期记忆理解为"关于这个用户的 RAG" - 索引的是过去,服务的是现在。

图 5-3 Agent 的三层记忆:工作记忆随会话消失,长期记忆跨会话存活,程序记忆几乎不变

三层记忆对应不同生命周期:工作记忆活在一次任务的几十分钟里,随会话结束而消失;长期记忆活在向量库里,跨会话存活;程序记忆(工具定义、提示词、流程模板)几乎不变,是 Agent 的"技能包"。写入也有讲究:不要每轮都存 - 那是把噪声塞进数据库。通常在任务结束时做一次"总结式写入":"用户偏好 X、任务 A 已完成、结论 B"。

# ===== 长期记忆:写入向量库,用时检索注入 =====
def save_memory(text: str):
    """任务结束时,把值得记住的事写入长期记忆"""
    vec = embed(text)                        # 第 3 章的 embedding
    memory_db.add(id=str(uuid4()), vector=vec, text=text)

def recall_memory(query: str, k: int = 3) -> str:
    """检索与当前问题最相关的旧记忆,拼成一段文本"""
    vec = embed(query)
    hits = memory_db.search(vec, top_k=k)    # 向量库相似度检索
    return "\n".join(f"[记忆{i+1}] {h.text}" for i, h in enumerate(hits))

# 每轮开始前:把相关记忆注入 system 提示
user_input = "继续整理昨天的表格"
messages = [{"role": "system",
             "content": "你是我的助手。以下是关于我的长期记忆:\n"
                        + recall_memory(user_input)},
            {"role": "user", "content": user_input}]
提示

类比:三层记忆对应人脑
工作记忆是手边的草稿纸,随写随扔;长期记忆是笔记本,重要的事才记;程序记忆是肌肉记忆 - 骑车、打字不用想,工具怎么用,Agent 也不用每次重新学。

5.5 规划:任务分解与反思

工具和记忆解决"能不能做",规划解决"复杂任务怎么做"。一个任务,比如"调研三家云厂商的定价,写一份对比报告并给出推荐",一次思考想不全、一次调用做不完。Agent 必须先把任务拆开,再逐个击破。

最直接的方案是 Plan-and-Execute(先计划后执行):第一步,让模型把任务拆成 3~5 个可独立执行的子任务;第二步,逐个执行,每步都可以调用工具;第三步,汇总结果,回答原任务。它的优点是可控:计划先行、步骤明确,适合流程稳定的任务;代价是计划本身可能拆错。与之互补的是 ReAct 的"边想边做",适合探索式任务 - 路径不明,走一步看一步。成熟的 Agent 常把两者结合:大计划用 Plan-and-Execute,单步细节用 ReAct。

拆出来的子任务要有三个特征:可独立执行(不依赖上一步的临时状态)、结果可被后续步骤使用每步尽量只调一个工具。子任务之间传递的是"结果",不是"过程" - 前一步的结论以文本形式交给下一步。

任务做完了,不一定做对了。反思(Self-Refine)是最后一关:让模型当自己的质检员 - 先产出,再自我评估,发现不足就修正,修正后再评估,直到达标或达到重试上限。它和 7.3 的 LLM-as-a-Judge 是同一族思想:用模型评判模型。整个流程如图 5-4 所示。

图 5-4 规划与反思流程:Plan-and-Execute 拆解执行,Self-Refine 检查修正

# ===== Plan-and-Execute:先计划,后执行 =====
PLAN_PROMPT = """把下面的任务拆成 3~5 个可独立执行的子任务,
每个子任务只做一件事。只输出 JSON 列表:
[{"step": 1, "desc": "子任务说明", "tool": "需要的工具或'无'}]
任务:{task}"""

def plan_and_execute(task: str):
    plan = json.loads(call_llm(PLAN_PROMPT.format(task=task)))  # 1. 出计划
    results = []
    for step in plan:                          # 2. 逐个执行
        r = execute_step(step)                 #    每步可调用工具(ReAct)
        results.append(f"步骤{step['step']}: {r}")
    return call_llm(                           # 3. 汇总回答原任务
        f"根据以下执行结果回答原任务:{task}\n" + "\n".join(results))
注记

反思很贵
评估一次、修正一次,等于多烧几次调用。只在关键产出(报告、邮件、代码)上开反思,高频小任务执行完直接返回。

5.6 动手搭一个 Agent

概念讲了四节,现在动手。目标:写一个真正能干活的最小 Agent - 能查天气、能算数、能组合使用。它不依赖任何 Agent 框架,只有三个零件:一条规定输出格式的 system 提示、一个工具注册表、一个循环调用 API 的主循环。代码不到四十行:

# ===== 最小 ReAct Agent:思考 → 行动 → 观察,循环直到完成 =====
import json

SYSTEM = ("你是一个会使用工具的助手。需要工具时,按以下格式输出:\n"
          "THOUGHT: 你的推理\n"
          "ACTION: 工具名(参数JSON)\n"
          "问题解决后,输出:FINAL: 最终答案")

def run_agent(question: str, max_steps: int = 8):
    # 1. 组装 messages:system 规定格式 + 用户问题
    messages = [{"role": "system", "content": SYSTEM},
                {"role": "user", "content": question}]

    for step in range(max_steps):              # 2. 最大步数:防死循环
        reply = llm(messages)                  # 3. 让模型"想一步"
        messages.append({"role": "assistant", "content": reply})

        if reply.startswith("FINAL"):          # 4. 显式退出条件
            return reply.split(":", 1)[1].strip()

        if reply.startswith("ACTION"):
            # 5. 解析行动指令:ACTION: 工具名(参数JSON)
            _, spec = reply.split(":", 1)
            name, args = spec.split("(", 1)
            args = json.loads(args.rstrip(")"))
            result = run_tool(name.strip(), args)   # 6. 执行真实工具
        else:
            result = "无法解析,请按 THOUGHT/ACTION/FINAL 格式重新输出"

        # 7. 观察回填:结果进上下文,模型才有依据继续想
        messages.append({"role": "user", "content": f"OBSERVATION: {result}"})

    return "超过最大步数,任务未完成"          # 8. 兜底:宁可失败,不要死循环

# 跑一下:"北京和上海今天谁更热?差几度?"
print(run_agent("北京和上海今天谁更热?差几度?"))

逐段看这段代码。第一,SYSTEM 提示词规定了模型输出的三种格式:THOUGHT(想)、ACTION(行动)、FINAL(完成) - 这是 Agent 与普通聊天的唯一区别:模型还是那个模型,只是你要求它"把思考说出来、把行动写出来"。第二,run_agent 里的 for 循环是 Agent 的心脏:每次迭代让模型"想一步",把回答追加进 messages(保持多轮语境),然后解析。第三,解析很简单:ACTION: 工具名(参数JSON),拆开、执行、把结果以 OBSERVATION 塞回上下文 - 模型"看到"结果,下一轮思考就有了依据。第四,两个退出条件:模型输出 FINAL,或者步数用尽 - 宁可失败,不要死循环。

跑起来大概是这样的对话流:用户问"北京和上海今天谁更热?",模型输出 THOUGHT: 需要分别查询两地天气ACTION: get_weather("北京") → 代码执行、回填 OBSERVATION → 模型再 ACTION: get_weather("上海") → 再回填 → 模型输出 FINAL: 北京 28℃,上海 31℃,上海更热。五轮调用,一个任务完成。

自己写一遍,胜过用十个框架。框架封装的正是这个循环 - 等你手写过后,再去看 LangChain 的 AgentExecutor、AutoGen 的对话循环,会发现它们都眼熟:无非是循环、解析、执行、回填、退出。出问题时你也能一眼定位:是模型想错了,还是工具执行错了,还是结果没回填。

这个最小版本是教学骨架,生产环境要做三处升级:用 JSON 格式输出代替文本解析(2.3 结构化输出,更稳);用 tools 参数代替文本 ACTION(2.4 函数调用,参数校验更规范);加记忆与步数之外的熔断机制(5.4、5.7)。骨架不变,升级都是加肉。

重要

最小 Agent 的五个部件
① 规定输出格式的 system 提示;② 工具注册表(名字→函数);③ 循环调用 API;④ 解析模型输出并执行工具;⑤ 显式退出条件(FINAL / 最大步数)。

5.7 Agent 的陷阱与安全

Agent 能"干活",也就能"闯祸" - 而且闯祸的方式比普通聊天程序多得多。普通问答错了顶多答错,Agent 错了可能真的执行了某个动作。这一节盘点五个最常见的坑和对应的工程对策。

第一个坑是死循环。模型反复调用同一个工具(比如一直查天气却不用结果),或者思路绕圈。对策最简单也最有效:最大步数上限 - 5.6 代码里的 max_steps=8 就是这个作用。再进一步,可以检测重复调用:同一个工具、同样的参数连续出现三次,直接中断并提示。

第二个坑是模型编造工具结果。如果工具结果没有回填,模型会"脑补"一个看起来合理的输出,然后基于幻觉继续干活,结果越偏越远。对策是纪律:工具结果必须由你的代码以 role:"tool" 回填,模型在任何情况下都不得"转述"工具输出;关键结论要求模型引用工具返回的原文。

第三个坑是权限过大。Agent 能执行的工具越多,危险面越大 - 删文件、转账、发邮件、下单。工程上把工具分级:只读工具(搜索、查询)放行;有副作用的工具(写入、删除、支付)必须人工确认。宁可多问一句,不可静默执行:

# ===== 危险操作:先确认,再执行 =====
DANGEROUS = {"delete_file", "send_email", "pay_order"}  # 有副作用的工具

def execute_safe(tool_name: str, args: dict) -> str:
    if tool_name in DANGEROUS:
        ok = confirm(f"Agent 想执行 {tool_name}({args}),是否允许?")
        if not ok:
            return "用户已拒绝该操作"
    return run_tool(tool_name, args)

第四个坑是上下文爆炸。每轮都塞全部历史加全部工具结果,几十步下来轻松超窗口。对策:5.4 的裁剪与摘要,加上"工具结果只保留本轮相关" - 大结果(如搜索返回的 50 条)先截断再进上下文。第五个坑是成本失控。一个任务几十次调用,单次不贵,累计可观。对策:步数上限(同死循环)、模型分级(简单步骤用小模型、大模型只在关键决策时出场)、结果缓存(同样的查询不重复付费)。五个坑汇总如下:

陷阱 表现 工程对策
死循环 反复调用同一工具,步数无限 最大步数上限 + 重复调用检测
编造结果 模型"转述"了不存在的工具输出 结果必须代码回填,模型不得转述
权限过大 删文件、转账、发邮件 副作用工具白名单 + 人工确认
上下文爆炸 塞满历史,超窗口、变慢变贵 裁剪 / 摘要 / 工具结果截断
成本失控 一次任务烧掉几十次调用 步数上限、模型分级、结果缓存
注记

还有更隐蔽的威胁
提示注入 - 模型被对话里的恶意内容诱导去调用危险工具 - 比上面五个都隐蔽,第 7.2 节会专门讲。工程上的第一道防线就是本节的"危险操作人工确认":无论模型被怎么诱导,删除、支付这类动作都需要人点头。

重要

本章要点

  1. Agent = 模型(大脑)+ 规划(调度)+ 工具(手脚)+ 记忆(笔记本);"ChatGPT 是对话,Agent 是做事"。
  2. ReAct 循环:思考 → 行动 → 观察 → 再思考,直到输出最终答案;工具结果必须由代码回填。
  3. 工具是模型的"手":声明用 JSON Schema,执行用函数注册表,做好参数校验与防御。
  4. 记忆分三层:短期(上下文,裁剪/摘要)、长期(向量库,检索注入)、程序(工具与技能)。
  5. 复杂任务先规划(Plan-and-Execute 拆子任务),关键任务再反思(Self-Refine 自查重做)。
  6. 最小 Agent 的核心是循环:输出格式约束、循环调用、解析、执行、回填、显式退出。
  7. 五大陷阱:死循环、编造结果、权限过大、上下文爆炸、成本失控 - 对应步数上限、结果回填、人工确认、裁剪摘要、模型分级。
警告

衔接 · 下一站
第 6 章把本章的 Agent 技能装进完整产品 - 6.5 会做一个带 Agent 能力的客服机器人;第 7 章讲安全与评估 - 7.2 的提示注入防线、7.3 的 LLM-as-a-Judge(用模型评判模型),正是 5.5 反思思想的正式化。