写在前面：网上 90% 的 Agent 教程还停留在 langgraph 0.2.x 的写法，AgentExecutor、initialize_agent 这些 API 早就废弃了，复制过去只会满屏 ImportError。

这篇文章基于 2026 年 8 月的最新稳定版：langgraph==1.2.11 + langchain==1.3.15 + langchain-core==1.5.5。文中每一段代码我都在本地实际执行过并贴出了真实输出，复制即可运行。

全文约 8000 字，建议先收藏再看。看完你会得到一个能扛住生产流量的 Agent 骨架，而不是一个只能演示的玩具。

目录

一、先泼盆冷水：你以为的 Agent 和真实的 Agent

很多人对 Agent 的理解停留在"给大模型接几个 API"。真跑到线上你会发现，能不能调工具根本不是难点——难点全在工具调完之后：

你以为的问题线上真正的问题
模型不会调工具模型陷入死循环，一小时烧掉 300 块 token 费
提示词不够好聊到第 30 轮上下文超限，整个会话崩溃
工具接口没写对工具报错后模型不知所措，反复重试同一个错误参数
输出格式不稳定Agent 擅自执行了退款/删库这类不可逆操作
响应有点慢服务重启后所有会话记忆全部丢失

Agent 的本质是一个带状态的循环，不是一次性的函数调用：

用户输入
   ↓
[ 模型思考 ] ←─────────────┐
   ↓                       │
需要调工具？               │
   ├── 是 → [ 执行工具 ] ──┘   (把结果塞回上下文，再想一轮)
   └── 否 → 输出最终答案

这个循环叫 ReAct（Reasoning + Acting）。而 LangGraph 干的事，就是把这个循环建模成一张有向图：节点是计算单元，边是流转逻辑，状态在图里流动，每一步都可以被持久化、被中断、被回放。

理解这一点，后面所有 API 都是自然的。

1.1 三种主流范式，别只会 ReAct

范式工作方式适合场景代价
ReAct想一步做一步，边走边看步骤数不确定的探索型任务（查资料、排障）容易在中途跑偏、绕远路
Plan-and-Execute先出完整计划，再逐条执行步骤明确的流程型任务（数据处理流水线）计划错了整条链都错，纠偏成本高
Reflection做完自我批判一轮再改质量优先的生成任务（写代码、写文案）token 成本直接翻倍甚至三倍

工程实践里最稳的是混合：主干用 ReAct，用 TodoListMiddleware 给它挂一个待办清单（引入 Plan 的骨架感），关键节点插一个 review 节点（引入 Reflection）。第七节会讲怎么一行配置实现。

选型建议：先无脑上 ReAct。只有当你观察到"Agent 明明知道要做 5 件事却总忘掉第 3 件"时，才需要引入 Plan；只有当输出质量是核心 KPI 时，才引入 Reflection。过早上复杂范式是新手最常见的浪费。

二、环境搭建：版本锁定是第一生产力

2.1 依赖安装

# 建议 Python 3.11 / 3.12（3.13 部分依赖仍有兼容问题）
python -m venv .venv && source .venv/bin/activate   # Windows: .venv\Scripts\activate

pip install -U langgraph langchain langchain-openai
pip install langgraph-checkpoint-sqlite              # 持久化用

请务必把版本写死进 requirements.txt，LangChain 生态迭代极快，不锁版本三个月后必炸：

langgraph==1.2.11
langchain==1.3.15
langchain-core==1.5.5
langchain-openai==1.5.1
langgraph-checkpoint==4.2.0
langgraph-checkpoint-sqlite

验证一下：

import importlib.metadata as md
for p in ["langgraph", "langchain", "langchain-core"]:
    print(p, md.version(p))

langgraph 1.2.11
langchain 1.3.15
langchain-core 1.5.5

2.2 接入国产模型（DeepSeek / 通义 / Kimi / 智谱）

国内绝大多数模型都提供了 OpenAI 兼容接口，不需要装额外的 SDK，改 base_url 就行：

import os
from langchain_openai import ChatOpenAI

# DeepSeek
llm = ChatOpenAI(
    model="deepseek-chat",
    base_url="https://api.deepseek.com/v1",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    temperature=0,          # Agent 场景务必调低，创造力是敌人
    timeout=60,
    max_retries=2,
)

# 通义千问：base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", model="qwen-max"
# 月之暗面：base_url="https://api.moonshot.cn/v1",  model="kimi-k2"
# 智谱  ：base_url="https://open.bigmodel.cn/api/paas/v4/", model="glm-4.6"

关键提醒：选模型时唯一的硬指标是是否支持 Function Calling / Tool Use。不支持的模型（比如一些纯 base 模型、部分推理模型的早期版本）在 Agent 场景下基本没法用，靠提示词硬解析 JSON 的年代已经过去了。

2.3 一个技巧：离线验证图结构

调试 Agent 时每次都调真模型，既慢又烧钱。我在写这篇文章时用的是一个脚本化假模型，按顺序吐出预设的 AIMessage，用来验证图的连线、中断、记忆是否正确——这个类建议你直接抄进项目当测试工具用：

# fake.py
from typing import List
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage
from langchain_core.outputs import ChatGeneration, ChatResult


class ScriptedChatModel(BaseChatModel):
    """按脚本依次返回 AIMessage，用于离线验证图的连线是否正确。"""

    responses: List[AIMessage] = []
    i: int = 0

    def _generate(self, messages, stop=None, run_manager=None, **kwargs) -> ChatResult:
        msg = self.responses[min(self.i, len(self.responses) - 1)]
        object.__setattr__(self, "i", self.i + 1)
        return ChatResult(generations=[ChatGeneration(message=msg)])

    def bind_tools(self, tools, **kwargs):
        return self

    @property
    def _llm_type(self) -> str:
        return "scripted"

单元测试里用它，CI 跑 Agent 测试可以零成本、零网络。下文所有示例都能用它复现。

三、Level 1：15 行代码跑通第一个 Agent

LangChain 1.x 里，创建 Agent 的唯一推荐入口是 langchain.agents.create_agent（不是 AgentExecutor，也不再优先推荐 langgraph.prebuilt.create_react_agent）：

from langchain_core.tools import tool
from langchain.agents import create_agent

@tool
def get_weather(city: str) -> str:
    """查询指定城市的实时天气。"""
    return f"{city}：晴，28℃"

agent = create_agent(
    model=llm,                      # 也可以直接传字符串 "openai:gpt-4.1"
    tools=[get_weather],
    system_prompt="你是天气助手，回答简洁。",
)

result = agent.invoke({"messages": [{"role": "user", "content": "上海天气"}]})
print(result["messages"][-1].content)

用假模型跑出来的完整消息链（真实输出）：

HumanMessage | 上海天气      | None
AIMessage    |               | [{'name': 'get_weather', 'args': {'city': '上海'}, 'id': 'c1', 'type': 'tool_call'}]
ToolMessage  | 上海：晴，28℃ | None
AIMessage    | 上海今天晴，28℃。 | []

这四条消息就是 Agent 的全部秘密：用户提问 → 模型决定调工具（content 为空，tool_calls 有值）→ 框架执行工具产出 ToolMessage → 模型看到结果给出最终答案。

create_agent 的完整签名值得贴出来，后面每一个参数我们都会用到：

create_agent(
    model, tools=None, *,
    system_prompt=None,
    middleware=(),          # 核心！第七节详解
    response_format=None,   # 结构化输出
    state_schema=None,      # 自定义状态字段
    context_schema=None,    # 运行时上下文（依赖注入）
    checkpointer=None,      # 短期记忆
    store=None,             # 长期记忆
    interrupt_before=None, interrupt_after=None,
    name=None, cache=None, debug=False,
)

四、Level 2：工具（Tool）的正确写法

工具写得好不好，直接决定 Agent 智商上限。模型看不到你的实现，它只能看到：函数名 + docstring + 参数 schema。这三样就是你给模型的全部提示词。

4.1 反面教材 vs 正面教材

# 反面教材：模型完全不知道这是干嘛的、参数啥格式
@tool
def query(p1: str, p2: str) -> str:
    """查询"""
    ...

# 正面教材：用 pydantic 把约束写死
from pydantic import BaseModel, Field
from langchain_core.tools import tool

class QueryInput(BaseModel):
    city: str = Field(description="城市名，如 上海")
    days: int = Field(default=1, ge=1, le=7, description="预报天数，1-7")

@tool(args_schema=QueryInput)
def weather(city: str, days: int = 1) -> str:
    """查询城市未来 N 天的天气预报。"""
    if city == "火星":
        raise ValueError("暂不支持该地区")
    return f"{city} 未来{days}天：晴"

weather.args 打印出来（真实输出）就是最终喂给模型的 schema：

{'city': {'description': '城市名，如 上海', 'title': 'City', 'type': 'string'},
 'days': {'default': 1, 'description': '预报天数，1-7',
          'maximum': 7, 'minimum': 1, 'title': 'Days', 'type': 'integer'}}

注意 ge=1, le=7 变成了 minimum/maximum——约束被带到了模型侧，这比你在函数体里写 if days > 7: raise 有效得多。

工具编写四条铁律：

  1. docstring 是提示词，用一句人话说清"什么时候该用我"，而不是"我是什么"。
  2. 参数越少越好，超过 4 个参数模型的填参准确率断崖式下跌，考虑拆成多个工具。
  3. 返回值必须是模型能读懂的自然语言或紧凑 JSON，别返回一个 30KB 的原始 HTML——那会直接吃掉你的上下文窗口。
  4. 工具名用动词开头（search_order 而不是 order_api），模型对动词更敏感。

4.2 工具报错怎么办？（大多数教程不讲的部分）

工具抛异常时，默认行为是整个图直接崩溃。生产环境这是不可接受的——正确做法是把错误信息转成 ToolMessage 喂回给模型，让它自己纠正：

from langchain.agents.middleware import ToolErrorMiddleware

agent = create_agent(
    model=llm,
    tools=[weather],
    middleware=[
        ToolErrorMiddleware(
            on_error=lambda e, req: f"工具执行失败: {e}。请换个说法或改用其它工具。"
        )
    ],
)
agent.invoke({"messages": [{"role": "user", "content": "火星天气"}]})

真实输出：

HumanMessage | 火星天气
AIMessage    |
ToolMessage  | 工具执行失败: 暂不支持该地区。请换个说法或改用其它工具。
AIMessage    | 抱歉，该地区暂不支持。

模型收到错误后自己降级回答了，整条链路没有崩。

踩坑提醒：ToolErrorMiddleware() 不能空参数构造，会直接抛 ValueError: ToolErrorMiddleware requires on_error and/or aon_error.；且 on_error 的签名是 (exception, request) 两个参数，写三个参数会报 TypeError。这两个坑我都踩过。

如果只是网络抖动这类瞬时错误，用 ToolRetryMiddleware 更合适，它自带指数退避：

from langchain.agents.middleware import ToolRetryMiddleware

ToolRetryMiddleware(
    max_retries=3,
    retry_on=(TimeoutError, ConnectionError),
    backoff_factor=2.0, initial_delay=1.0, max_delay=30.0, jitter=True,
)

4.3 工具超过 20 个怎么办？

一个残酷的事实：工具数量和调用准确率是反比关系。挂 30 个工具时，模型选错工具的概率会显著上升，而且每一轮都要把 30 个工具的 schema 塞进上下文，token 成本恒定高企。

三个解法，按推荐度排序：

  1. 合并同类项：search_order / search_user / search_product 合成一个 search(entity_type, query)。
  2. 两阶段筛选：用 LLMToolSelectorMiddleware，先让一个便宜的小模型从 30 个里挑出最相关的 3-5 个，再交给主模型决策。
from langchain.agents.middleware import LLMToolSelectorMiddleware

middleware=[LLMToolSelectorMiddleware(model=cheap_llm, max_tools=5)]
  1. 拆成多 Agent：每个 Agent 只管自己领域的 5 个工具，由 Supervisor 路由（见第九节）。

4.4 结构化输出：让下游程序能直接消费

Agent 最终答案如果要给前端渲染卡片、或者写进数据库，就不能是一段自由文本。用 response_format：

from pydantic import BaseModel, Field
from langchain.agents import create_agent

class OrderResult(BaseModel):
    order_id: str = Field(description="订单号")
    status: str = Field(description="订单状态：已发货/待付款/已完成")
    amount: float = Field(description="订单金额，单位元")
    need_manual: bool = Field(description="是否需要人工介入")

agent = create_agent(model=llm, tools=[query_order], response_format=OrderResult)

r = agent.invoke({"messages": [{"role": "user", "content": "查一下 B123 的订单"}]})
print(r["structured_response"])     # → OrderResult 实例，可直接 .model_dump()

结果在 result["structured_response"] 里，是一个校验过的 pydantic 对象，不需要你再写正则去抠 JSON。

底层有两种策略，langchain.agents.structured_output 里可以显式指定：

from langchain.agents.structured_output import ToolStrategy, ProviderStrategy

# ToolStrategy：把 schema 伪装成一个工具让模型调用，兼容所有支持 function calling 的模型
create_agent(..., response_format=ToolStrategy(OrderResult, handle_errors=True))

# ProviderStrategy：走厂商原生的 structured output（如 OpenAI 的 json_schema 模式），更可靠但支持的模型少
create_agent(..., response_format=ProviderStrategy(OrderResult))

不显式指定时走 AutoStrategy，框架自己选。国产模型建议显式用 ToolStrategy，原生 structured output 的支持度参差不齐。

五、Level 3：手写 StateGraph，把黑盒拆开

create_agent 很爽，但只会用它你就永远只能做标准 ReAct。真实业务里你需要插入审核节点、并行检索、条件分支……这时必须自己画图。

LangGraph 的三个核心概念：

概念说明
State一个 TypedDict，在节点间流动的数据。用 Annotated[..., reducer] 指定"新值如何合并进旧值"
Node一个函数 (state) -> dict，返回值是对 state 的局部更新，不是完整 state
Edge普通边（无脑跳转）和条件边（函数决定下一站）

完整的手写 ReAct：

from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import InMemorySaver

@tool
def add(a: int, b: int) -> int:
    """两数相加。"""
    return a + b

class State(TypedDict):
    # add_messages 是官方 reducer：追加而非覆盖，并自动处理消息去重/ID 对齐
    messages: Annotated[list[BaseMessage], add_messages]
    step: int          # 没写 reducer 的字段 = 直接覆盖

tools = [add]
llm_with_tools = llm.bind_tools(tools)

def call_model(state: State):
    return {
        "messages": [llm_with_tools.invoke(state["messages"])],
        "step": state.get("step", 0) + 1,
    }

g = StateGraph(State)
g.add_node("agent", call_model)
g.add_node("tools", ToolNode(tools))

g.add_edge(START, "agent")
g.add_conditional_edges("agent", tools_condition, {"tools": "tools", END: END})
g.add_edge("tools", "agent")          # 这条回边 = ReAct 的"循环"

app = g.compile(checkpointer=InMemorySaver())

跑两轮（同一个 thread_id），真实输出：

turn1: 结果是 3 | step = 2
turn2: 我记得你叫 Leo | history len = 6

注意 step = 2：模型被调用了两次（一次决定调工具，一次总结结果），这就是 Agent 成本翻倍的来源。

5.1 把图画出来（写文档/汇报神器）

print(app.get_graph().draw_mermaid())

真实输出，直接粘进 CSDN 的 mermaid 代码块就能渲染：

__start__

agent

tools

__end__

虚线 = 条件边，实线 = 普通边。图一画出来，逻辑对不对一眼就看出来了。

app.get_graph().draw_mermaid_png() 可以直接出图片，但需要联网调 mermaid.ink，内网环境用 draw_mermaid() 出源码更稳。

5.2 reducer 是最容易翻车的地方

messages: Annotated[list, add_messages]   # 追加
logs:     Annotated[list, operator.add]   # 追加（普通列表）
count:    int                             # 覆盖

没加 reducer 的 list 字段，每次返回都会把之前的数据整个冲掉。我见过太多人在这里 debug 一整天。

六、Level 4：记忆系统——短期 vs 长期

这两个东西完全不是一回事，务必分清：

短期记忆（Checkpointer）长期记忆（Store）
存什么单个会话的完整消息历史、图的执行状态跨会话的用户画像、偏好、知识
隔离维度thread_id（一个会话一条线）namespace（如 ("memories", user_id)）
典型实现InMemorySaver / SqliteSaver / PostgresSaverInMemoryStore / PostgresStore
附带能力时间旅行、断点续跑、人工介入语义检索（可接 embedding）

6.1 短期记忆：thread_id 就是会话 ID

from langgraph.checkpoint.sqlite import SqliteSaver
from langchain.agents import create_agent

with SqliteSaver.from_conn_string("./memo.db") as saver:
    agent = create_agent(model=llm, tools=[], checkpointer=saver)

    cfg = {"configurable": {"thread_id": "user-1024"}}   # 换个 id 就是全新会话
    agent.invoke({"messages": [{"role": "user", "content": "我叫 Leo"}]}, cfg)
    r = agent.invoke({"messages": [{"role": "user", "content": "我叫什么"}]}, cfg)
    print(r["messages"][-1].content, "| 历史条数:", len(r["messages"]))

真实输出：

你叫 Leo | 历史条数: 4

注意第二次调用只传了新消息，历史是框架从 checkpoint 里捞出来自动拼上的——你不需要自己维护 message 数组，这是很多人从裸调 API 迁移过来最不适应的一点。

生产环境请换 PostgresSaver（pip install langgraph-checkpoint-postgres），SQLite 扛不住并发写。

6.2 长期记忆：Store

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
store.put(("memories", "user-1"), "pref", {"text": "喜欢简洁的回答"})
print([i.value for i in store.search(("memories", "user-1"))])
# [{'text': '喜欢简洁的回答'}]

把 store=store 传进 create_agent，工具里就能通过 InjectedStore 拿到它，实现"Agent 自己记笔记"。

6.3 时间旅行：debug 神技

有了 checkpointer，你可以回到任意一步重跑：

# 列出这条线上所有历史快照
for s in app.get_state_history(cfg):
    print(s.config["configurable"]["checkpoint_id"], "→", s.next)

# 挑一个快照，从那里分叉重跑
past = {"configurable": {"thread_id": "user-1024", "checkpoint_id": "1ef..."}}
app.invoke(None, past)

线上出了 badcase，把 thread_id 捞出来在本地回放，比看日志高效一百倍。

七、Level 5：中间件体系（LangChain 1.x 的杀手锏）

这一节是全文最有价值的部分。 LangChain 1.x 引入的 Middleware 机制，把过去需要手写几百行的生产级能力做成了一行配置。先看完整清单（langchain.agents.middleware 下的实际导出）：

中间件解决什么问题
SummarizationMiddleware上下文超限：自动压缩早期历史
ModelCallLimitMiddleware死循环烧钱：限制单次运行/单线程模型调用次数
ToolCallLimitMiddleware单个工具被反复调用
HumanInTheLoopMiddleware高危操作需人工审批
PIIMiddleware邮箱/身份证/银行卡脱敏
ModelFallbackMiddleware主模型挂了自动降级到备用模型
ModelRetryMiddleware / ToolRetryMiddleware瞬时失败重试
ToolErrorMiddleware工具异常转自然语言
LLMToolSelectorMiddleware工具太多（>20个）时先做一轮筛选
ContextEditingMiddleware精细裁剪历史中的工具输出
TodoListMiddleware给 Agent 装一个任务清单，长任务不跑偏
LLMToolEmulator用 LLM 模拟工具，方便测试

下面挑四个最刚需的实战。

7.1 死循环熔断（上线前必配）

Agent 最恐怖的故障不是答错，是在一个工具上反复横跳几百次。我构造了一个"永远只会调工具"的模型来验证：

from langchain.agents.middleware import ModelCallLimitMiddleware

agent = create_agent(
    model=loop_model,                  # 一个死循环调工具的模型
    tools=[refund],
    middleware=[ModelCallLimitMiddleware(run_limit=3, exit_behavior="end")],
)
r = agent.invoke({"messages": [{"role": "user", "content": "退款"}]})
print("熔断后消息数:", len(r["messages"]))

真实输出：

熔断后消息数: 8

模型调了 3 次就被强制终止（1 条 Human + 3×(AI+Tool) + 1 条终止提示 = 8）。exit_behavior 可选 "end"（优雅结束）或 "error"（抛异常给上层）。

这是我认为最应该默认开启的中间件，没有之一。

7.2 人工审批（HITL）：让 Agent 不敢乱来

退款、发邮件、删数据、下单——这类不可逆操作绝对不能让模型自己拍板。

from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

agent = create_agent(
    model=llm,
    tools=[refund],
    middleware=[HumanInTheLoopMiddleware(interrupt_on={"refund": True})],
    checkpointer=InMemorySaver(),      # HITL 必须配 checkpointer！
)

cfg = {"configurable": {"thread_id": "t-1"}}
out = agent.invoke({"messages": [{"role": "user", "content": "给 B123 退 99 元"}]}, cfg)

if "__interrupt__" in out:
    print("待审批:", out["__interrupt__"][0].value)
    # ……把内容推给运营后台，等人点确认……
    final = agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), cfg)
    print("批准后:", final["messages"][-1].content)

真实输出：

是否中断: True
待审批内容: {'action_requests': [{'name': 'refund',
             'args': {'order_id': 'B123', 'amount': 99.0},
             'description': "Tool execution requires approval\nTool: refund\nArgs: {...
批准后: 退款已完成。

整个流程是：图执行到工具前挂起 → 状态落进 checkpoint → 进程可以直接退出 → 人在后台点了确认 → 用同一个 thread_id 恢复执行。这才是工程化的人机协同，不是 input() 卡在那儿。

decisions 支持四种：approve（放行）、edit（改参数后放行）、reject（拒绝并告知模型）、以及直接代替工具给出结果。

重灾区踩坑：resume 的载荷格式在 1.x 变了，必须是 Command(resume={"decisions": [...]})。网上大量老教程写的 Command(resume=[{"type": "accept"}]) 会报 TypeError: list indices must be integers or slices, not str，而且决策类型是 approve 不是 accept。我在这儿卡了半小时。

7.3 PII 脱敏（合规刚需）

from langchain.agents.middleware import PIIMiddleware

agent = create_agent(
    model=llm, tools=[],
    middleware=[
        PIIMiddleware("email", strategy="redact"),
        PIIMiddleware("credit_card", strategy="mask"),
        PIIMiddleware("ip", strategy="hash"),
    ],
)
r = agent.invoke({"messages": [{"role": "user", "content": "我的邮箱是 leo@example.com 请记录"}]})
print(r["messages"][0].content)

真实输出：

我的邮箱是 [REDACTED_EMAIL] 请记录

内置类型：email / credit_card / ip / mac_address / url，四种策略：block（直接拦截）/ redact（替换为标记）/ mask（部分打码）/ hash（哈希化）。要识别身份证、手机号这类中国特色 PII，传自定义 detector 正则即可。参数 apply_to_input / apply_to_output / apply_to_tool_results 控制在哪一侧生效——工具返回值那一侧最容易漏，记得开。

7.4 上下文压缩：解决"聊到 30 轮就崩"

from langchain.agents.middleware import SummarizationMiddleware

SummarizationMiddleware(
    model=llm,                       # 建议用便宜的小模型做摘要
    trigger=("fraction", 0.8),       # 上下文用到 80% 时触发
    keep=("messages", 20),           # 保留最近 20 条原文
)

trigger 支持三种写法：("fraction", 0.8) 按窗口占比、("tokens", 100000) 按绝对 token 数、("messages", 50) 按条数。它会把早期历史压成一段结构化摘要（含"会话意图/关键决策/产物/下一步"四个小节），然后接上最近 N 条原文。

这是长会话 Agent 的生命线。 没有它，你的客服 Agent 聊到第 30 轮必然 context_length_exceeded。

7.5 自定义中间件：埋点、鉴权、灰度都靠它

继承 AgentMiddleware，重写钩子即可：

from langchain.agents.middleware import AgentMiddleware

class TimerMiddleware(AgentMiddleware):
    def before_model(self, state, runtime):
        print("[hook] before_model, 当前消息数 =", len(state["messages"]))
        return None          # 返回 None = 不修改 state

    def after_model(self, state, runtime):
        print("[hook] after_model")
        return None

agent = create_agent(model=llm, tools=[], middleware=[TimerMiddleware()])

真实输出：

[hook] before_model, 当前消息数 = 1
[hook] after_model
[4] 自定义中间件 OK

可用钩子：before_agent / before_model / after_model / after_agent / wrap_tool_call / wrap_model_call。也可以用装饰器形式 @before_model、@dynamic_prompt 快速定义。

中间件的执行顺序是洋葱模型：列表里越靠前的越在外层。把熔断放最外层、把摘要放中间、把埋点放最内层，是我推荐的默认顺序。

八、Level 6：流式输出，别让用户盯着转圈

Agent 动辄十几秒，不做流式体验必崩。LangGraph 的 stream_mode 是个高频面试题：

stream_mode吐什么用在哪
"values"每步之后的完整 state调试
"updates"每个节点的增量更新展示"正在调用 XX 工具"这类进度
"messages"LLM 的 token 级流打字机效果
"custom"你自己 writer() 推的数据工具内部进度条
"debug"全量事件排障
for chunk in agent.stream({"messages": [{"role": "user", "content": "2+3"}]},
                          stream_mode="updates"):
    print(list(chunk.keys()))

真实输出：

['model']
['tools']
['model']

踩坑：create_agent 生成的图，节点名是 model 和 tools，不是 agent。很多教程按 agent 这个 key 取值，结果永远拿不到数据。自己 StateGraph 手写的图才叫你起的名字。

生产环境推荐组合模式，一次订阅拿全：

for mode, chunk in agent.stream(payload, stream_mode=["updates", "messages"]):
    if mode == "messages":
        token, meta = chunk
        print(token.content, end="", flush=True)      # 打字机
    else:
        ...                                           # 侧边栏显示"正在查天气…"

对接 FastAPI SSE：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app_api = FastAPI()

@app_api.post("/chat")
async def chat(q: str, thread_id: str):
    async def gen():
        cfg = {"configurable": {"thread_id": thread_id}}
        async for token, _ in agent.astream(
            {"messages": [{"role": "user", "content": q}]},
            cfg, stream_mode="messages",
        ):
            if token.content:
                yield f"data: {token.content}\n\n"
        yield "data: [DONE]\n\n"
    return StreamingResponse(gen(), media_type="text/event-stream")

九、Level 7：多 Agent 协作

单 Agent 挂 20 个工具，准确率会明显下滑。拆成多个专职 Agent 由 Supervisor 调度是主流解法。核心 API 是 Command——它能同时完成"更新状态"和"决定去哪"两件事：

from typing import Annotated, TypedDict, Literal
from langchain_core.messages import AIMessage, BaseMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.types import Command

class S(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    next: str

def supervisor(state: S) -> Command[Literal["researcher", "writer", "__end__"]]:
    txt = state["messages"][-1].content
    # 生产里这里换成 LLM 结构化输出做路由
    if "写" in txt:
        return Command(goto="writer", update={"next": "writer"})
    if "查" in txt:
        return Command(goto="researcher", update={"next": "researcher"})
    return Command(goto=END)

def researcher(state: S):
    return {"messages": [AIMessage(content="[研究员] 已检索到 3 条资料")]}

def writer(state: S):
    return {"messages": [AIMessage(content="[写手] 稿件已生成")]}

g = StateGraph(S)
g.add_node("supervisor", supervisor)
g.add_node("researcher", researcher)
g.add_node("writer", writer)
g.add_edge(START, "supervisor")
g.add_edge("researcher", END)
g.add_edge("writer", END)
app = g.compile()

真实输出：

"帮我查一下资料" → [研究员] 已检索到 3 条资料
"帮我写一篇稿"   → [写手] 稿件已生成

注意 Command[Literal[...]] 这个返回类型标注不是装饰用的——LangGraph 靠它推断出这个节点可能跳向哪些节点，从而正确绘制图并做校验。漏写会导致 draw_mermaid() 画不出边。

三种主流拓扑，按场景选：

  • Supervisor（星型）：一个总控分发给专家，专家做完回总控。最常用，最好调试。
  • Swarm（网状）：Agent 之间直接 handoff，谁接得住谁接。灵活但容易失控。
  • Hierarchical（树型）：Supervisor 的 Supervisor，适合超大规模，但延迟叠加严重。

忠告：能用单 Agent + 好工具解决的，别上多 Agent。 每多一个 Agent，延迟和 token 成本都是乘法关系，而收益往往是加法。先把单 Agent 的工具描述打磨好，通常就够了。

九点五、成本与延迟优化：让 Agent 便宜一半

Agent 的账单构成很反直觉：大头不是最后那次生成，而是每一轮都要重传的完整历史 + 全部工具 schema。四个见效最快的手段：

1）节点级缓存

相同输入的节点直接吃缓存，对"同一个问题被多人问"的客服场景效果拔群：

from langgraph.cache.memory import InMemoryCache
from langgraph.types import CachePolicy

g.add_node("retrieve", retrieve_node, cache_policy=CachePolicy(ttl=300))
app = g.compile(cache=InMemoryCache())

2）分层模型路由

不是所有环节都需要旗舰模型。典型分配：

环节模型档位理由
意图路由 / 工具筛选小模型（deepseek-chat、qwen-turbo）分类任务，小模型足够
历史摘要小模型纯压缩，不需要推理
主推理 + 最终生成旗舰模型质量瓶颈在这里

SummarizationMiddleware(model=cheap_llm) 和 LLMToolSelectorMiddleware(model=cheap_llm) 就是干这个的。实测下来这一项通常能砍掉 30%~50% 的成本。

3）用好 Prompt Caching

主流厂商都支持前缀缓存（命中的部分按 10%~25% 计费）。要点是把稳定内容放最前面：system prompt → 工具定义 → 少量示例 → 动态历史。千万别在 system prompt 里插当前时间戳，那会让整个前缀每次都失效——我见过团队因为这一行代码多付了一倍的钱。

4）控制工具返回体积

@tool
def fetch_page(url: str) -> str:
    """抓取网页正文。"""
    raw = crawl(url)
    return raw[:2000] + ("\n...(已截断)" if len(raw) > 2000 else "")

一个返回 30KB 的工具，在 10 轮对话里会被重复传输 10 次。截断是最便宜的优化。

十、生产环境踩坑清单（12 条，血泪总结）

  1. 不锁版本——LangChain 生态三个月一次破坏性更新，requirements.txt 必须写死到 patch 位。
  2. temperature 不设 0——Agent 需要的是确定性，不是创造力。路由和填参尤其。
  3. 不配 ModelCallLimitMiddleware——一次死循环够你解释一整天。
  4. HITL 忘配 checkpointer——中断状态无处存放，interrupt 直接失效。
  5. resume 载荷格式写错——必须 Command(resume={"decisions": [{"type": "approve"}]})，类型是 approve 不是 accept。
  6. state 里的 list 忘加 reducer——数据被静默覆盖，最难查的一类 bug。
  7. 按 agent 这个 key 解析流式输出——create_agent 的节点名是 model / tools。
  8. 工具返回超长文本——一个返回 30KB HTML 的爬虫工具能瞬间打爆上下文，务必在工具内部先做摘要/截断。
  9. thread_id 用了全局常量——所有用户的记忆串在一起，事故级问题。用 f"{user_id}:{session_id}"。
  10. ToolErrorMiddleware() 空参构造——直接 ValueError，且 on_error 只接受 (exception, request) 两个参数。
  11. 生产还在用 InMemorySaver——重启即失忆。上 Postgres。
  12. 没有可观测性——Agent 是黑盒中的黑盒，不接 LangSmith 或 OpenTelemetry，线上出问题只能靠猜。

十一、完整项目结构与部署

推荐的工程骨架：

my-agent/
├── requirements.txt        # 版本全部锁死
├── .env                    # API Key，别提交到 git
├── app/
│   ├── llm.py              # 模型工厂（含 fallback 配置）
│   ├── tools/              # 一个工具一个文件，方便单测
│   │   ├── weather.py
│   │   └── order.py
│   ├── middlewares.py      # 中间件组装：熔断 / 摘要 / 脱敏 / HITL
│   ├── graph.py            # create_agent 或自定义 StateGraph
│   └── server.py           # FastAPI + SSE
├── tests/
│   ├── fake.py             # 第 2.3 节的 ScriptedChatModel
│   └── test_graph.py       # 零成本离线测试
└── Dockerfile

middlewares.py 的生产默认配置，直接抄：

from langchain.agents.middleware import (
    ModelCallLimitMiddleware, ToolCallLimitMiddleware,
    SummarizationMiddleware, PIIMiddleware,
    ModelFallbackMiddleware, ToolRetryMiddleware, ToolErrorMiddleware,
)

def build_middlewares(main_llm, cheap_llm, backup_llm):
    return [
        ModelCallLimitMiddleware(run_limit=15, thread_limit=100, exit_behavior="end"),
        ToolCallLimitMiddleware(thread_limit=50, exit_behavior="continue"),
        ModelFallbackMiddleware(backup_llm),
        SummarizationMiddleware(model=cheap_llm, trigger=("fraction", 0.8),
                                keep=("messages", 20)),
        PIIMiddleware("email", strategy="redact", apply_to_tool_results=True),
        ToolRetryMiddleware(max_retries=3, retry_on=(TimeoutError, ConnectionError)),
        ToolErrorMiddleware(on_error=lambda e, req: f"工具执行失败: {e}，请调整参数重试。"),
    ]

可观测性（一行环境变量，强烈建议开）：

export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY=ls__xxx
export LANGSMITH_PROJECT=my-agent-prod

开完之后，每一次模型调用、每一个工具的入参出参、每一步的耗时和 token 消耗，全部可视化。Agent 项目不接可观测性，等于蒙眼开车。

结语

回顾一下这条学习路径：

create_agent 快速起步
    ↓
把工具写好（docstring + pydantic + 错误处理）
    ↓
手写 StateGraph，理解 State/Node/Edge
    ↓
接上 Checkpointer（会话记忆）+ Store（长期记忆）
    ↓
用中间件补齐生产能力：熔断 / 摘要 / 审批 / 脱敏 / 降级
    ↓
流式输出 + 多 Agent + 可观测性

我的核心观点是：Agent 项目的胜负手不在提示词，在工程化。模型能力每半年翻一倍，你今天精心调的提示词半年后可能一文不值；但熔断、审批、记忆、可观测这套骨架，会一直用下去。

代码有任何跑不通的地方，评论区贴报错和版本号，我看到都会回。如果这篇对你有帮助，点个赞收藏一下，后面会写 《MCP 协议实战：把你的 Agent 工具变成全生态通用》 和 《Agent 评测体系：怎么科学地证明你的 Agent 变强了》。

环境版本声明：本文所有代码基于 langgraph==1.2.11 / langchain==1.3.15 / langchain-core==1.5.5 / langchain-openai==1.5.1，于 2026 年 8 月实测通过。若你的版本不同，请优先核对 API 签名。

# 人工智能 # python
Logo
龙虾开发者社区

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用

更多推荐

cover

我用 NVIDIA SkillSpector 纯静态扫了一遍 GitHub 最火的 agent skills：能挖出多少注入？0 个

avatar 龙虾开发者社区
cover

周榜换血九成，Agent Skills 星数翻倍

avatar 龙虾开发者社区
cover

智能体从入门到精通：6个必学GitHub开源项目

avatar 龙虾开发者社区