11 · Agent 工程原理与设计模式

本章目标：从"会用框架"上升到"会设计"——把 Agent 拆解成记忆、规划、工具、反思四块能力，并给出主流设计模式的选择指南。

本章你将学到

1. Agent 的四大核心能力拆解，以及每一块的工程实现；
2. 记忆的分类（短期/长期/事实记忆）与正确选型；
3. 主流设计模式（ReAct、Plan-and-Execute、Reflection、Multi-Agent…）的适用场景；
4. 关键判断：什么时候根本不需要 Agent。

前置知识

• 已完成第 10 章，会搭图、会用 interrupt。

11.1 Agent = 感知 + 记忆 + 规划 + 反思

任何 Agent 都可以拆成四块能力：

         ┌────────────────────────────┐
         │          Agent            │
         │  ┌────┐ ┌────┐ ┌────┐ ┌────┐
         │  │感知 │ │记忆 │ │规划 │ │反思 │
         │  └────┘ └────┘ └────┘ └────┘
         └────────────────────────────┘
感知：用工具/检索获得外部世界的信息（第 8、10 章已做）
记忆：记住对话与事实（本节）
规划：把大任务拆成步骤（11.3）
反思：检查自己的输出并改进（11.4）

前两章我们一直在做"感知"（工具调用、RAG 检索）。这一章补齐剩下的三块，并教你如何组合。

11.2 记忆：三类，别混用

Agent 的"记忆"在工程上是三类完全不同的存储，搞混是常见的架构错误。

11.2.1 短期记忆：上下文窗口管理

短期记忆的容量 = 模型上下文窗口。要管理的不只是"存",更是"放得下"：

• 滑动窗口：只保留最近 N 轮消息；
• 摘要压缩：对较早的历史做一轮摘要，替换原始消息（省 token、不丢主线）；
• 一个简易的"窗口 + 摘要"实现：

from langchain_core.messages import AIMessage, HumanMessage, SystemMessage

def trim_history(history: list, max_messages: int = 20) -> list:
    """超长时截断：保留系统提示与最近消息。"""
    if len(history) <= max_messages:
        return history
    return [SystemMessage(content="以下是早前对话的摘要：……（对话已被压缩）")] + history[-(max_messages - 1):]

注意：很多"客户说了一句，模型全忘了"的 bug，本质是没把短期记忆接上（忘了传历史），而不是模型能力问题。

11.2.2 长期记忆：语义检索

跨会话记得"用户偏好"（"常飞北京-上海"、"喜欢靠窗"），需要长期记忆。实现方式与 RAG 同构：把记忆写成向量存库，按当前对话检索相关记忆拼进上下文。

# 伪代码：长期记忆 = 小型 RAG
memories = vectorstore.similarity_search(f"用户 {user_id} 的偏好", k=5)
# 把检索到的记忆拼进 System 提示，再走 Agent

成本与权衡：每次对话都要多一次向量检索与拼接，但换来跨会话一致性。何时值得做：面向重复访问的 toC 产品（如会员服务）；一次性的 toB 工具场景可以先不做。

11.2.3 事实记忆：业务系统说了算

订单号、航班号、身份信息这类"必须精确"的事实，不要让模型"记住"——让模型通过工具从业务数据库读取（第 8 章的工具就是这么做的）。模型负责"决定去查"，你的系统负责"保证准确"。这条边界是 Agent 应用准确性的底线。

11.3 规划：把大任务拆成步骤

复杂任务（"帮我做一份下季度航线调整报告"）一次让模型输出往往又长又糊。规划（Planning）让它先列出步骤，再逐步执行。

11.3.1 Plan-and-Execute 模式

计划节点：把大任务拆成有序子任务
   ↓
执行循环：逐个执行子任务（每步可用工具/RAG）
   ↓
若有意外 → 重新计划(Re-plan)，否则汇总输出

与 ReAct（边想边做）的区别：Plan-and-Execute 先给全局计划再执行，更适合步骤明确、可并行的任务；ReAct 适合探索性任务。

11.3.2 一个最小"计划-执行"示例

"""Plan-and-Execute 最小实现（LangGraph）。"""
import os
from typing import Annotated, TypedDict

from dotenv import load_dotenv
from langchain_core.messages import AIMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages

load_dotenv()

model = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    temperature=0,
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)


class State(TypedDict):
    task: str
    plan: list[str]        # 已生成的子任务列表
    current_step: int
    results: list[str]     # 每步的执行结果
    messages: Annotated[list, add_messages]


def planner_node(state: State) -> dict:
    """计划：把大任务拆成 3 步。"""
    resp = model.with_structured_output(
        type("Plan", (), {"__annotations__": {"steps": list[str]}})
    ).invoke(f"把任务拆成最多3个步骤：{state['task']}")
    return {"plan": resp.steps, "current_step": 0, "results": []}


def worker_node(state: State) -> dict:
    """执行：完成当前一步。"""
    step = state["plan"][state["current_step"]]
    reply = model.invoke(f"完成这一步并简短汇报：{step}")
    return {
        "results": state["results"] + [reply.content],
        "current_step": state["current_step"] + 1,
    }


def route(state: State) -> str:
    return "worker" if state["current_step"] < len(state["plan"]) else "END"


def summarizer_node(state: State) -> dict:
    """汇总：把各步结果整合成最终答案。"""
    body = "\n".join(f"- {r}" for r in state["results"])
    reply = model.invoke(f"基于以下执行结果给出最终报告：\n{body}")
    return {"messages": [AIMessage(content=reply.content)]}


builder = StateGraph(State)
builder.add_node("planner", planner_node)
builder.add_node("worker", worker_node)
builder.add_node("summarizer", summarizer_node)
builder.add_edge(START, "planner")
builder.add_conditional_edges("planner", lambda s: "worker")
builder.add_conditional_edges("worker", route, {"worker": "worker", "END": "summarizer"})
builder.add_edge("summarizer", END)

graph = builder.compile()

result = graph.invoke({"task": "分析近三月航班准点率变化，并给出三条改进建议"})
print(result["messages"][-1].content)

简化说明：真实实现里每步可能调用工具/检索（把 worker 换成工具循环即可）。这个骨架展示了"计划 → 循环执行 → 汇总"的控制流。

11.4 反思：让 Agent 自己检查自己

模型一次生成的答案可能有漏洞。反思（Reflection）模式让 Agent 用第二遍审视第一遍的输出：

11.4.1 Self-Refine

生成(Generator) → 批评(Critic)：找出漏洞 → 改写(Refine)：按批评改进 → 循环 N 次

"""Self-Refine 最小实现：生成 → 批评 → 改写。"""
from langchain_core.messages import HumanMessage

def generate(task: str) -> str:
    return model.invoke([HumanMessage(content=f"写一段关于X的说明。{task}")]).content

def critique(text: str) -> str:
    return model.invoke([HumanMessage(
        content=f"指出这段文字的准确性问题：\n{text}"
    )]).content

def refine(task: str, text: str, criticism: str) -> str:
    return model.invoke([HumanMessage(
        content=f"根据批评改进，回答：\n【原文】{text}\n【批评】{criticism}\n【任务】{task}"
    )]).content

def self_refine(task: str, rounds: int = 2) -> str:
    text = generate(task)
    for _ in range(rounds):
        c = critique(text)
        text = refine(task, text, c)
    return text

代价：生成 + 2×N 次额外调用，成本和延迟翻倍。适用：正确性要求高、输出不赶时间的场景（如报告生成、代码审查）。对话式客服一般不值得上反思——等得不耐烦。

11.4.2 ReAct 的"反思"本质

其实第 10 章的 ReAct 循环已经隐含了反思：模型每看到一次工具结果，都要判断"这个结果够不够回答"。你可以在 System 里强化这一点：

调用工具得到结果后，先自检：
- 结果能否回答用户问题？
- 是否需要再调一次工具 / 换个查询？
- 结果与已掌握的信息是否有冲突？
确认后给出最终回答。

这比单独的 Self-Refine 便宜得多——它把"检查"融进了决策循环。

11.5 Agent 设计模式选型总表

一句话选型：从"最简单的能跑"开始——单模型 + 工具循环 → 需要记忆与分支就上 LangGraph ReAct → 任务复杂才考虑 Plan-and-Execute → 只有高正确性要求才上 Reflection；Multi-Agent 是最后的手段，不是加分项。

11.6 反套路：什么时候不需要 Agent

老工程师的优势是知道不做比做更值。以下场景，一个普通 RAG 链就够，别上 Agent：

1. 知识问答边界清晰：问题都是"查政策、查文档"，单次检索即可回答——固定 RAG 链更快、更便宜、更可控（第 8.1 节那种）；
2. 流程完全确定：步骤和判断都是写死的，用 Workflow/状态机（第 9 章那套图，但节点是纯代码，不调 LLM）；
3. 数据精确优先：涉及金额、时间、身份，让模型"决定"不如让代码"决定"；
4. QPS 高、成本敏感：Agent 一轮问答多次模型调用，成本是单次问答的数倍。

判断标准一句话：需要"根据情况自主决定下一步"才需要 Agent；否则用确定性代码。 Agent 是"用成本换灵活性"，别让灵活性变成没必要的税。

11.7 小结

• Agent = 感知 + 记忆 + 规划 + 反思；感知（工具/RAG）前几章已实现，本章补齐其余三块；
• 记忆分三类：短期（Checkpointer）、长期（向量语义记忆）、事实（业务数据库）——别用模型"记"必须精确的事实；
• 规划（Plan-and-Execute）适合多步骤任务；反思（Self-Refine）用调用次数换准确率，要算成本账；
• 设计模式从简到繁有清晰梯度，选型要对着"场景 vs 成本"表逐行比；
• 最大的工程智慧是知道什么时候不用 Agent。

下一章把这些系统部署到生产：第 12 章。

练习

1. 给 11.3 的 Plan-and-Execute 加一个工具（如"查航班准点率"），让 worker 在部分步骤调用它。
2. 用 11.4 的 self_refine 生成一段关于你业务规则的说明，对比"不反思"与"反思两轮"的输出质量与耗时。
3. 把第 8.4 章客服助手改造成"先判断意图，再决定走固定 RAG 还是 Agent 循环"——体会"按需上 Agent"。
4. 列出你手上一个真实需求，判断它属于 11.5 表里哪一行；若你判断不需要 Agent，写清楚理由。

参考答案

<details>
<summary>练习1（Plan-and-Execute 加工具）</summary>

给 11.3 的 worker 装配工具（如"查航班准点率"）：

@tool
def get_punctuality(flight_no: str) -> str:
    """查询航班历史准点率（%）。"""
    ...

# worker 从"纯生成"升级为"可调工具的 ReAct 节点"
worker = create_react_agent(llm, [get_punctuality], prompt=worker_prompt)

改法：计划阶段不变（planner 只出步骤清单），执行阶段的每一步 worker 可以在部分步骤调用工具。验证一条计划，如：

计划: 1. 查 CA1501 状态  2. 查 CA1501 准点率  3. 汇总建议

观察 worker 对第 2 步调用 get_punctuality、其余步骤纯生成——Agent 的自主性体现在"哪一步要查工具、哪一步不用"，由模型判断，而不是计划里写死。若计划里写死了"第 2 步必须调工具"，那就退化成了 Workflow，不是 Agent 了。
</details>

<details>
<summary>练习2（self_refine 对比）</summary>

跑"不反思" vs "反思两轮"，对同一段业务规则说明：

# 不反思：一次生成
plain = generate("退改签规则是什么")

# 反思两轮：生成 → 批评 → 改进 → 批评 → 改进
answer = generate("退改签规则是什么")
for _ in range(2):
    critique = generate(f"对下面的回答做严格审查，指出事实、逻辑、遗漏问题：\n{answer}")
    answer = generate(f"根据审查意见改进回答：\n{answer}\n\n意见：\n{critique}")

对比维度：

• 质量：通常反思版更严谨（补上例外条款、措辞更准确），但事实类内容（金额、时间）反思也补不出来——模型不知道就是不知道；
• 耗时：不反思 1 次调用；反思两轮 = 1 次生成 + 4 次额外调用（2 批评 + 2 改进），耗时与成本约 5 倍；
• 适用判断：反思适合"长文、开放、容错高"的输出（总结、方案、代码评审）；不适合低延迟高并发场景与事实精确场景。

结论：反思是"用调用次数换质量"，先算成本账再决定用不用，这是第 11.5 节选型表的一个具体实例。
</details>

<details>
<summary>练习3（按意图分流：固定 RAG vs Agent 循环）</summary>

把第 8.4 客服助手改造成路由器 + 两分支：

def route(state) -> str:
    intent = classifier.invoke(state["question"])   # 小模型判意图
    if intent == "policy":
        return "rag_chain"      # 固定 RAG：快、便宜、可控（退改签政策等）
    return "agent_loop"         # Agent 循环：需多步工具编排（查航班+算费用+审批）

graph = StateGraph(...)
graph.add_node("rag_chain", rag_branch)      # 8.1 的 RAG 链包一层
graph.add_node("agent_loop", react_branch)   # 8.4 的 ReAct 循环包一层
graph.add_conditional_edges("router", route, {"rag_chain": ..., "agent_loop": ...})

体会（练习的目的）：

• 绝大多数问题走固定 RAG：一次检索一次生成，延迟低、token 省、行为可预测；
• 少数复杂问题才进 Agent 循环：模型自主决定调几个工具，灵活但慢且贵；
• 这就是第 11.5 节"按需上 Agent"：不是所有问题都值得 Agent 的灵活性。生产系统里这样的路由能省下大部分成本。

提示：意图分类用一个小模型（如 qwen-turbo 或本地小模型）就够，别让路由成为新的成本大头。

</details>

<details>
<summary>练习4（真实需求对照选型表）</summary>

以"客服工单自动分派"为例，对照 11.5 的表格逐行判断：

判断：不需要 Agent。理由：分派逻辑完全可以用规则/状态机写死（第 11.6 节），更便宜、更快、零幻觉、可审计。真正值得 Agent 的是"边界模糊 + 步骤不固定"的场景（如"根据工单内容自由检索知识库并编排答复"）。

练习目的：多数企业需求的第一答案都是"不用 Agent"——能把"不需要"的理由写清楚，比会写 ReAct 循环更体现工程判断。
</details>

延伸阅读

• 第 12 章：评估、可观测、安全、成本
• 第 14 章：把这些模式组合进一个完整系统
• ReAct 论文、Plan-and-Solve 论文（如感兴趣）
• LangGraph 官方文档：多 Agent 的 supervisor 模式（0.4.x，以官方文档为准）