#Agent 架构：从能跑的循环到能上线的项目

阅读提示 本笔记的前置阅读是Agent 概念。那篇回答"Agent 是什么、什么时候不该用"，本篇只回答一个问题：决定要做之后，代码该怎么分层。

默认读者已经理解 Agent Loop、Tool Calling、State 与 Memory 的区别，正在用 Python + LangChain / LangGraph 搭 RAG + Agent 项目。

本篇不重复讲工具设计与多 Agent 取舍，那两部分见概念笔记Tools与单 Agent 与多 Agent。

⚠️ = 常见陷阱　🆚 = 对比说明　💡 = 机制或选择建议

#目录

• 一、为什么 while 循环不能直接上线？
• 二、State 才是系统的中心
• 三、如何把循环改写成状态图？
• 四、Router、Planner、Validator 分别在解决什么？
• 五、Context Engine：模型真正读到的是什么？
• 六、Model Gateway：让模型成为可替换的配置
• 七、Middleware：横切能力放在哪一层？
• 八、持久化与人工审批为什么必须成对出现？
• 九、项目结构与落地顺序
• 速查表
• 复习重点

核心概念 生产级 Agent 项目的骨架不是"一个 Agent 类加一堆工具"，而是用 State 做中心、用状态图做编排、把上下文组装 / 模型选择 / 检索 / 横切能力各自关进独立模块。

一条主线贯穿全篇：每个模块只应该知道自己那一层的事。Agent 节点不该知道向量库是 Milvus 还是 Elasticsearch，也不该知道日志写去了哪里。

从上往下看：请求先经过 API 与业务服务层（鉴权、会话、限流），再进入 Orchestrator 编排；编排层的节点向下调用三个引擎取数据和执行动作，引擎再向下访问存储。右侧竖条是横切能力——它们穿过每一层，因此都不写在 Agent 内部。后面每一节都是在展开这张图里的一个方块。

#一、为什么 while 循环不能直接上线？

概念笔记里的最小闭环可以写成十行：

def run(query: str) -> str:
    messages = [{"role": "user", "content": query}]
    while True:
        response = model.invoke(messages)
        if not response.tool_calls:
            return response.content
        messages.append(response)
        messages.extend(execute_tools(response.tool_calls))

这段代码在本地演示完全够用，而且它对机制的表达是准确的：模型只产出决定，工具由外部执行，结果回灌后模型继续判断。

问题在于，它把六件事压在了同一个变量 messages 和同一层缩进里：

每加一件事，循环体就多一层嵌套，很快变成没人敢改的一坨。分层的目的不是好看，而是让这六件事各有归属。

机制补充 LangChain 1.x 的 create_agent 本身就是构建在 LangGraph 之上的一张图，而不是一个 while 循环。也就是说，"把循环换成图"不是本篇的发明，而是官方抽象已经走的路——理解它的动机，比会调它的参数更重要。

#二、State 才是系统的中心

一句话结论：不要围绕 Prompt 设计系统，要围绕 State 设计系统。

概念笔记说清了 State 是什么（当前任务进行到哪里，不等于长期记忆）。工程上要再往前一步：把 State 声明成一个显式的类型，让所有节点只通过它通信。

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
 
class AgentState(TypedDict):
    # 对话主干。add_messages 是 reducer：新消息追加进旧列表；
    # 但如果新消息的 id 和已有消息相同，则是「更新那一条」而不是再追加一条。
    # 这个 by-id 更新语义是流式增量和 RemoveMessage 删除能工作的基础。
    messages: Annotated[list, add_messages]
 
    # 请求身份
    user_id: str
    session_id: str
 
    # 路由与计划
    route: str                  # chat / rag / tool / coding
    plan: list[str]
 
    # 检索与工具的产出
    retrieved_docs: list[dict]
    citations: list[str]
 
    # 执行控制
    iteration: int
    max_iterations: int
    token_spent: int
 
    # 审批
    pending_approval: dict | None
 
    # 结果
    answer: str | None

#为什么"节点读写 State"比"函数互相调用"好？

🆚 两种组织方式的差别不在写法，而在依赖方向：

第三行是关键：**能不能暂停恢复，几乎完全取决于你的中间状态是不是一个可以序列化的普通对象。**这也是第八节人工审批能实现的前提。

⚠️ 在节点里原地修改 State 错误操作： 节点内部直接改传进来的对象。

def agent_node(state: AgentState):
    state["messages"].append(response)   # ❌ 原地修改
    state["iteration"] += 1              # ❌ 同上
    return state

实际结果： 并行分支互相污染；重放同一个检查点时结果和第一次不同；add_messages 这类 reducer 收到已经被改过的列表，出现消息重复。

原因： 图的状态更新走的是差量 + reducer 合并这条路——节点读入 State，返回一份"这次产生了什么"，由框架决定怎么并进去。原地修改绕过了合并逻辑，也破坏了检查点的可重放性。

注意这里说的是状态更新方式，不是"节点必须是纯函数"。框架并不禁止副作用：节点里写库、发请求、调工具都是正常做法，官方文档也明确说节点"就是普通函数，可以是 LLM 调用也可以是普通代码"。真正的要求是重放安全——因为节点可能被重新执行（中断恢复、失败重试），副作用要么幂等，要么带幂等键去重。第八节的 interrupt 会把这一点变成硬约束。

正确做法： 只返回本节点产生的字段，让框架去合并：

def agent_node(state: AgentState) -> dict:
    response = model.invoke(state["messages"])
    return {"messages": [response],                  # reducer 负责追加
            "iteration": state["iteration"] + 1}     # 覆盖式更新

State 里该放什么

• 放：下一个节点做判断需要的东西（进度、计数、待确认动作、检索结果）。
• 不放：能随时重算的派生值、体积很大的原始文件、和本次任务无关的用户档案（那属于长期记忆，用到时再查）。
• 判断标准：这个字段如果丢了，任务能不能从检查点接着跑完？ 不能，才放进 State。

⚠️ TypedDict 的字段不会自动有默认值 错误操作： 像上面那样声明完 AgentState，然后调用时只传一部分字段：graph.invoke({"messages": [...], "user_id": "u1"}, config)。

实际结果： 第一个节点里 state["iteration"] 直接 KeyError: 'iteration'。条件边里的 state["iteration"] >= state["max_iterations"] 同理——而条件边抛异常时的堆栈往往指向框架内部，看不出是自己少传了字段。

原因： TypedDict 只是类型标注，运行时就是个普通 dict，没有构造器、不做校验、更不会填默认值。State 里有哪些 key，完全取决于你传进去什么、以及各节点返回了什么。上面 AgentState 里的 iteration、max_iterations、route、answer 全都是「用之前必须有人先写进去」的字段。

正确做法： 三选一（推荐组合前两条）：

1. 入口统一初始化：写一个 initial_state(query, user_id) 工厂函数，保证每次 invoke 传进去的都是完整 State，别让调用方手拼字典；
2. 读的时候给兜底：条件边等关键路径上用 state.get("iteration", 0)，避免一个字段缺失就让整张图挂掉；
3. 改用带默认值的容器：dataclass（配 field(default=...)）或 Pydantic 模型作为 state schema，把默认值写进类型定义本身——代价是序列化和 reducer 的写法要跟着调整。

#三、如何把循环改写成状态图？

StateGraph（状态图） 的三个基本动作只有：注册节点、连边、按条件选边。

from langgraph.graph import StateGraph, START, END
 
builder = StateGraph(AgentState)
builder.add_node("load_context", load_context)
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_executor)
builder.add_node("validate", validate_node)
 
builder.add_node("give_up", give_up_node)   # 预算耗尽时的兜底出口
 
builder.add_edge(START, "load_context")
builder.add_edge("load_context", "agent")
 
# 条件边：由一个不带副作用的函数决定下一站的名字
def route_after_agent(state: AgentState) -> str:
    if state["iteration"] >= state["max_iterations"]:
        return "give_up"                    # 预算耗尽 → 直接去终止节点
    return "tools" if state["messages"][-1].tool_calls else "validate"
 
def route_after_validate(state: AgentState) -> str:
    if state["answer"] is not None:
        return END                          # 校验通过
    if state["iteration"] >= state["max_iterations"]:
        return "give_up"                    # 校验没过，但已经没预算重试了
    return "agent"                          # 还有预算 → 回去重做
 
builder.add_conditional_edges("agent", route_after_agent, ["tools", "validate", "give_up"])
builder.add_edge("tools", "agent")          # 工具结果回灌，形成循环
builder.add_conditional_edges("validate", route_after_validate, ["agent", "give_up", END])
builder.add_edge("give_up", END)
 
graph = builder.compile(checkpointer=checkpointer)

⚠️ 预算判断必须写在每一条回环边上，否则等于没写 这是最容易写错、且看起来完全正确的一个 bug。上面的代码曾经是这样写的：

def route_after_agent(state) -> str:
    if state["iteration"] >= state["max_iterations"]:
        return "validate"                 # ❌ 以为这样就退出了
    return "tools" if state["messages"][-1].tool_calls else "validate"
 
builder.add_conditional_edges("validate",
    lambda s: "agent" if s["answer"] is None else END, ["agent", END])   # ❌ 这条边没查预算

实际结果： 预算耗尽后走到 validate，如果这次校验没通过（answer 仍是 None），validate 那条边二话不说把它送回 agent；agent 再次发现超预算，又送去 validate……于是在 agent ⇄ validate 之间死循环，直到撞上 LangGraph 的递归上限抛 GraphRecursionError。

表现出来的样子是：一个本该「优雅地放弃」的请求，变成了一次报错，日志里还是个看不懂的递归异常——而 max_iterations 明明设了。

原因： max_iterations 只在一条边上被检查，而回环有两条。任何一条没查预算的回环边，都能把控制流重新送回循环里。递归上限是框架的兜底崩溃保护，不是业务上的停止条件。

正确做法： 两件事一起做——① 预算耗尽时跳到一个专门的终止节点（give_up），而不是跳到另一个还会往回跳的节点；② 每条可能回到 agent 的边，条件函数里都要先查预算。终止节点负责产出「已尽力、未完成」的结构化结果（部分答案 + 失败原因），这比抛异常对上游友好得多。

这段代码属于骨架 为了突出结构，上面省略了 import 细节、节点实现和 checkpointer 的构造，并且 LangGraph 的具体签名会随版本变化。结构是稳定的，参数名请以你安装版本的文档为准。

对照第一节的 while：原来 if not response.tool_calls 那一行，现在是一条具名的条件边；原来隐式的"回到循环开头"，现在是 tools → agent 这条显式的边。收益是所有分支都有名字，因而可以被观测、被中断、被单独测试。

读图顺序：主干自上而下 START → load_context → router → agent → validate → persist → END；两条红色回环是这张图的重点——工具结果回到 agent（继续推理）和校验不通过回到 agent（重做）。右下角的 Checkpointer 不是节点，而是在每个超步边界上自动落盘的机制，第八节会用到它。

⚠️ 只加了循环边，没加退出条件 错误操作： 加上 tools → agent 和 validate → agent 两条回环，但条件函数里只判断"要不要继续"。

实际结果： 模型反复调用同一个工具或反复重写答案，请求挂住，token 与外部 API 费用持续增长；如果工具有写副作用，还会重复产生脏数据。

原因： 图和 while True 一样没有天然终点，只是把无限循环从缩进里挪到了边上。图的执行器通常有递归上限，但那是兜底报错，不是业务上可接受的停止。

正确做法： 每一条回环边的条件函数都必须先查预算再看意图，并且预算耗尽时要跳到一个不会再往回跳的终止节点。只在其中一条边上查预算是无效的——上面那个 danger 框讲的就是这个失效方式。停止条件的完整清单见概念笔记。

#四、Router、Planner、Validator 分别在解决什么？

图里除了 agent 和 tools，最常见的是这三个节点。它们不是必需品，各自解决一个具体问题。

#Router：不让所有请求都走重路径

Router（路由） 在进入 Agent 之前先判断请求类型，把简单请求分流出去。

"你好"              → chat    直接回，不进 Agent 循环
"这份 PDF 讲了什么"  → rag     检索后单轮生成
"帮我改这个函数"     → coding  进完整 Agent 循环

价值是成本与延迟：闲聊走全套循环，意味着为一句"你好"付出多轮模型调用。

⚠️ 用主力大模型做路由 错误操作： 路由节点直接调用和主 Agent 相同的大模型，让它输出一个分类词。

实际结果： 每个请求都多了一次高价调用和几百毫秒延迟。省下来的钱还不够路由本身花的，简单请求反而更慢了。

原因： 路由是一个标签数很少的分类任务，难度远低于 Agent 决策，却被分配了同一档算力。

正确做法： 按成本从低到高选：明确关键词或正则 → embedding 相似度分类 → 小模型分类。并且路由失败时默认走通用 Agent，不要因为分类不确定就拒绝服务。

#Planner：只在路径确实不确定时才用

Planner（规划器） 先把复杂任务拆成有序步骤，再逐步执行。

它的适用条件很窄：任务需要多个工具配合，且步骤之间有依赖。例如"分析写入超时"要依次看日志、看客户端配置、看批量大小、看重试策略，前一步的结果决定后一步看哪里。

⚠️ 给所有请求都加规划 错误操作： 在图里把 Planner 放成 router → planner → agent 的固定一环。

实际结果： 用户问"1+1 等于几"，系统先输出一份三步计划再回答。延迟翻倍，token 翻倍，回答质量没有变化。

原因： 规划的收益来自减少试错；当任务只有一步时没有试错可减，规划就只剩开销。而且过早生成的长计划会在环境变化后迅速失效，参见概念笔记的误区 2。

正确做法： 把 Planner 挂在 Router 的一个分支上，只有被判定为复杂任务时才进入；其余请求由 agent 节点每轮决定下一步。

#Validator：把"可验证目标"变成一个节点

概念笔记的第一步要求目标必须可验证。Validator（校验器）就是这个要求的执行体——它解决 Agent 最典型的失败模式：模型宣称完成，实际没有。

修改代码 → 跑 pytest → 失败 → 带着报错回到 agent → 再改 → 通过 → 结束

关键在于用什么来判定。优先级从高到低：

1. 程序化判定：测试通过、JSON 通过 schema 校验、必填字段齐全、金额对得上；
2. 规则判定：引用的文档 ID 必须出现在 retrieved_docs 里；
3. 模型判定（LLM-as-judge）：只在前两种做不到时用，且要意识到判定本身也会错。

能用第 1 种就不要用第 3 种。一个用模型判定模型的闭环，出错时你连哪一端错了都分不清。

这三个节点要不要现在就加

• 请求类型明显分档、且有大量轻量请求 → 加 Router
• 任务需要多工具配合且步骤有依赖 → 加 Planner，否则跳过
• 结果有客观判定标准 → 一定加 Validator，这是三者里性价比最高的

#五、Context Engine：模型真正读到的是什么？

一句话结论：**Agent 的输入不是"历史消息列表"，而是一份每轮重新组装出来的上下文。**负责这件事的模块叫 Context Engine（上下文引擎）。

左边是彼此独立的来源，中间是组装与压缩，右边才是模型实际收到的 messages。读图的关键是中间那个方块：它做的是取舍，不是拼接——当来源总量超过预算时，必须有明确的规则决定丢掉谁。

def build_context(state: AgentState, budget: int) -> list[Message]:
    """按固定优先级组装上下文，超出预算时从低优先级开始压缩。"""
    blocks = [
        ("system",   render_system_prompt(state)),      # 必留
        ("task",     render_task_state(state)),         # 必留
        ("tools",    render_tool_results(state)),       # 本轮结果，必留
        ("rag",      render_docs(state["retrieved_docs"])),
        ("recent",   state["messages"][-10:]),
        ("summary",  load_summary(state["session_id"])),
        ("memory",   load_user_memory(state["user_id"])),
    ]
    return fit_to_budget(blocks, budget)   # 从后往前裁剪或摘要

顺序本身就是策略：越靠前越不能丢。系统指令和本轮工具结果丢了会直接导致错误行为，早期对话丢了通常只损失一点连贯性。

⚠️ 把全部历史消息直接喂给模型 错误操作： messages = 该会话的所有历史，然后整个传进去。

实际结果： 长会话里 token 成本线性增长；超出窗口后请求直接报错；即使没超，模型也更容易忽略中间部分的信息，出现"前面说过的约束被忘掉"。

原因： 上下文窗口是有限且非均匀有效的资源——位置靠中间的内容更容易被稀释。全量塞入等于让最新的关键信息和几十轮前的寒暄争夺同一份注意力。

正确做法： 固定预算 + 分级压缩：最近若干轮保留原文，更早的滚动摘要，长期记忆按当前问题检索后再注入。并且记录本轮丢弃了什么，否则出问题时无法复现模型当时看到的输入。

#RAG 只是其中一个来源

这条对项目结构的影响很直接：Agent 节点只应该看见 retrieve(query)。

Agent 看到的：   retrieve(query) -> list[Doc]
Agent 不该看到： Milvus 连接、BM25 权重、RRF 融合、chunk_size、reranker 型号

把混合检索、重排、查询改写全部关进 RAG Service 内部，日后从 Milvus 换到别的向量库，Agent 一行都不用改。检索侧本身的分层见RAG 架构设计，调优手段见RAG 优化方案。

本节总结

• Context Engine 的职责是取舍，输出长度必须可预测。
• 组装顺序 = 丢弃优先级，写死在代码里，不靠模型自觉。
• RAG 是来源之一，不是与 Agent 并列的系统。

#六、Model Gateway：让模型成为可替换的配置

一句话结论：ChatOpenAI(...) 只应该在一个文件里出现。

class ModelGateway(Protocol):
    async def chat(self, messages: list, *,
                   tools: list | None = None,
                   schema: type | None = None,
                   tier: str = "smart") -> Response: ...

节点里调用 gateway.chat(...)，而不是直接构造某个厂商的客户端。这样带来两件事：

第一，换模型是改配置而不是改代码。 供应商故障时的降级、区域切换、私有化部署替换，都收敛在网关内部。

第二，可以按任务难度分档路由。 这是 Agent 项目最容易被忽略的成本杠杆：

什么时候引入网关 项目只有一个模型时也值得写这一层——它此刻只有二十行，而等到 ChatOpenAI 已经散落在十几个文件里再收拢，成本会高一个量级。但不要在第一版就实现自动降级、自动重试、自动路由的全套策略，先留出接口。

#七、Middleware：横切能力放在哪一层？

概念笔记说清了护栏"要有什么"，工程上的问题是"写在哪"。答案是：能力如果对每个请求都一样，就不该写进 Agent，而应做成中间件。

请求 → 鉴权 → 限流 → 输入护栏 → 上下文注入
                                    ↓
                                  Agent 图
                                    ↓
     ← 日志 / 追踪 ← token 统计 ← 输出护栏 ← 结果

适合放进这一层的：鉴权与权限、限流、超时、重试与降级、输入输出护栏、提示注入检测、token 与费用统计、结构化日志与链路追踪。

判断标准很简单：如果一个能力需要在每个节点里重复写一遍，它就属于中间件。 LangChain 目前也提供了 middleware 机制，覆盖模型调用前后处理、工具执行包装、重试与 fallback、动态模型/工具选择等场景，可以直接用，不必自己造。

中间件不能替代工具执行层的强制检查 中间件在调用边界上生效，而删除、付款、发信这类动作的真正边界在工具函数内部。权限校验、确认令牌、幂等键必须落在工具实现里——否则任何绕过中间件的调用路径（内部重试、后台任务、直接调函数的测试代码）都是缺口。

#八、持久化与人工审批为什么必须成对出现？

Checkpointer（检查点） 在每个 super-step（超步）边界上把当前 State 快照写入存储。

一个超步是「图节点的一次迭代」：并行执行的节点属于同一个超步，顺序执行的节点属于不同超步。所以「每个节点后存一次」只在纯串行图里成立——有并行分支时，是那一批节点全部跑完才落一次盘。更要紧的是：检查点不会存在节点函数中间，节点跑到一半崩溃，这一步的中间结果不会被保存。

它带来的能力容易被低估：

崩溃恢复：进程挂了，用同一个 thread_id 重新调用，从最后一个检查点继续
时间旅行：回到任意一步，改掉某个字段再重跑，用于调试
人工审批：节点里调 interrupt() 抛出 → 回退到上一个超步边界的快照 → 请求返回
          → 几分钟后用户点确认 → 从该快照恢复，**整个节点从头重跑**

第三条是重点：人工审批之所以能实现，是因为有检查点。 没有持久化，"暂停"就只能靠把请求挂在内存里等待——一旦进程重启或负载均衡切换实例，等待中的任务全部丢失。

注意「停住」是从用户视角说的：interrupt() 从节点内部抛出，图回到上一个超步边界的快照上等待；恢复时不是从 interrupt() 那一行继续，而是把整个节点函数重新执行一遍，走到 interrupt() 时用这次带回来的值返回。这正是下面那句注释的由来——interrupt() 之前不能放任何有副作用的代码。

from langgraph.types import interrupt, Command
 
def dangerous_tool_node(state: AgentState) -> dict:
    # ⚠️ interrupt 之前的代码，在每次恢复时都会重新执行一遍。
    #    这里只做「读」和「组装」，不做任何有副作用的事。
    results = []
    for call in state["messages"][-1].tool_calls:      # 遍历全部，不能只取 [0]
        if call["name"] in HIGH_RISK_TOOLS:
            # 挂起点：State 落盘，本次调用返回；决定回来后整个节点重跑到这里
            decision = interrupt({"tool": call["name"], "args": call["args"]})
            if decision != "approve":
                results.append(tool_message(call, "用户拒绝了该操作"))
                continue
        # 副作用发生在 interrupt 之后，并且带幂等键
        results.append(execute(call, idempotency_key=call["id"]))
    return {"messages": results}

外部恢复时传入用户的决定：graph.invoke(Command(resume="approve"), config)。

需要走这条路径的动作，和概念笔记第五步列的一致：删除、付款、发信、改库、推送代码、部署、执行 shell。注意判据是高风险（不可逆、外部可见、影响范围大），不是「凡写操作都要确认」——给低风险写入也套确认，只会让用户对确认弹窗脱敏。

⚠️ 以为 interrupt 是「从这一行继续」 错误操作： 按 Python 的 yield / await 直觉来理解 interrupt()，于是把创建记录、扣款、发消息这类副作用写在 interrupt() 之前，以为恢复后会跳过它们、直接从 interrupt 那一行往下走。

实际结果： 恢复时整个节点从第一行重新执行。写在 interrupt 之前的副作用会再执行一次——重复建单、重复扣款、重复发信。而且 interrupt 之前每多一次审批往返就多重放一次，审批越久、拒绝重来次数越多，重复得越多。这类 bug 在测试里几乎发现不了，因为测试很少真的走一遍「挂起 → 恢复」。

原因： LangGraph 的恢复机制不是协程续跑，而是重放节点。官方文档写得很直白：「运行时从头重启整个节点——它不会从 interrupt 被调用的那一行恢复，这意味着 interrupt 之前运行过的任何代码都会再执行一次。」框架靠的是：重放到第 N 个 interrupt() 时，直接用已经收到的第 N 个 resume 值返回，而不是再次挂起。

正确做法： 把节点按「interrupt 之前只读、副作用一律放在 interrupt 之后」来切分。做不到时，interrupt 之前的副作用必须幂等（带幂等键、upsert 而非 insert）。另外由于是按顺序索引匹配 resume 值，同一个节点里的多个 interrupt() 调用顺序必须稳定——不能根据随机数或外部状态跳过某个 interrupt，否则重放时索引对不上，会把 A 的决定用到 B 上。

⚠️ 只处理 tool_calls[0] 错误操作： call = state["messages"][-1].tool_calls[0]，取第一个工具调用就完事。

实际结果： 模型在同一条响应里返回多个工具调用时（并行工具调用是现在主流模型的默认能力），第 2 个之后的全部被静默丢弃——不报错、不告警。表现是「模型说要查天气和查酒店，结果只查了天气」，且看日志完全正常。更糟的是，被丢掉的 tool_call 没有对应的 ToolMessage 回填，有些模型 API 会因为「工具调用与结果数量不匹配」在下一轮直接报错。

正确做法： 永远遍历 tool_calls，并保证每个 tool_call_id 都有一条对应的结果消息回填（哪怕内容是「被拒绝」或「执行失败」）。

⚠️ 用内存 checkpointer 上生产 错误操作： 开发时用的 MemorySaver 一路带到线上。

实际结果： 进程重启后所有会话状态清零；多 worker 部署时，用户第二次请求打到另一个实例就"失忆"；等待审批的任务永久卡死。

原因： 内存实现的检查点和进程同生命周期，且不跨进程共享。它是为本地调试设计的。

正确做法： 线上使用数据库支持的 checkpointer（如 Postgres 版本），并把 thread_id 与会话 ID 绑定；同时给检查点设置保留期与清理任务，否则表会无限增长。

#九、项目结构与落地顺序

#目录

app/
├── api/                  # FastAPI 路由、SSE 流式输出
├── graphs/
│   ├── state.py          # AgentState 定义（第二节）
│   ├── nodes/            # router / planner / agent / tools / validator
│   └── main_graph.py     # 建图与 compile（第三节）
├── tools/                # 一个工具一个文件，按域分子目录
├── rag/
│   ├── ingestion/        # 解析 → 清洗 → 切分 → 向量化 → 入库
│   ├── retrieval/        # 向量 / BM25 / 混合 / 重排 / 查询改写
│   └── service.py        # 对外只暴露 retrieve()
├── context/              # builder.py（组装）· compressor.py（压缩）
├── memory/               # 短期 / 长期 / 摘要
├── models/               # gateway.py（统一接口）· router.py（分档）
├── middleware/           # 鉴权 · 限流 · 护栏 · 日志
├── storage/              # postgres / redis / vector 的连接与仓储
├── observability/        # tracing · metrics · logging
├── eval/                 # datasets · evaluators · runner
└── config/

这是一份架构提案，不是官方规范 LangChain / LangGraph 官方只提供 create_agent、middleware、图原语、checkpointer 这几层抽象，以及 Deep Agents 这类更高层的封装（见Deepagents）。上面的目录划分是工程约定，不是框架要求——小项目可以先把 context/、memory/、middleware/ 合并，等真正变厚了再拆。

#技术栈

各框架之间的定位差异见Agent 常见框架与区别，LangChain 自身的抽象层次见LangChain 组件层。

#落地顺序

判断该不该进入下一阶段 不是按时间推进，而是看当前阶段的痛点是否已经出现。没有长会话就不必先做上下文压缩，没有高风险工具就不必先做审批流。唯一建议提前做的是第 2 阶段的显式 State——它是后面所有能力的地基，补起来最贵。

#速查表

#🆚 与概念笔记的分工

#复习重点

可直接复述的结论

1. 分层的目的是让"恢复、审批、路由、限额、排障、替换"这六件事各有归属，而不是让目录变好看。
2. 围绕 State 而不是围绕 Prompt 设计系统；节点只返回差量、不原地修改。节点可以有副作用，但因为会被重放，副作用必须幂等。
3. 把循环改成图，收益是每个分支都有名字，因而可被观测、中断和单独测试；但图和 while 一样需要显式退出条件——而且预算判断要写在每一条回环边上，只写一条等于没写。
4. Router 省成本、Planner 省试错、Validator 防"假装完成"——三者中 Validator 性价比最高，且优先用程序化判定。
5. 模型读到的是每轮重新组装的上下文，不是全部历史；组装顺序就是丢弃优先级。
6. RAG 对 Agent 只暴露 retrieve(query)，换向量库不该影响 Agent 代码。
7. 人工审批依赖检查点：没有可序列化的 State 和持久化存储，就没有真正的暂停。检查点落在超步边界，不是节点函数中间。
8. interrupt 恢复 = 整个节点重跑，不是从那一行继续。副作用要么放在 interrupt 之后，要么幂等。
9. 落地按痛点推进，但显式 State 要在第一天就做，它是补起来最贵的一层。

#自测问题

1. 节点里 state["messages"].append(x) 会引发什么问题？为什么框架要求返回差量？
2. 为什么说"能不能中途暂停等人工确认"取决于 State 的设计？
3. 只在 agent 的出边上判断 iteration >= max_iterations，而 validate → agent 那条边不判断，会发生什么？最终报什么错？
4. 一个节点里，interrupt() 之前写了一行 create_order(...)。用户审批通过后恢复，数据库里会有几张单？为什么？
5. 路由节点用主力大模型做分类，错在哪？
6. Validator 用 LLM 判定和用 pytest 判定，差别在哪？什么时候只能用前者？
7. 上下文组装时，为什么系统指令和本轮工具结果必须排在最前面？
8. MemorySaver 换成 Postgres 版本，除了不丢数据，还解锁了什么能力？