# 使用LangChain+LangGraph开发AI应用（5） - Agent 深入


![](https://static.xiongneng.me/langchain-langgraph-cover-20261017170947.png)

## 学习进度

Agent 是一台完整机器，模型是发动机，`create_agent` 是把油路电路接好的线束，系统提示、工具、中间件都在里面。这一篇把 create_agent 的完整签名摊开看，十五个参数逐个过，六个钩子各守一个位置跑通，流式执行也顺一遍。

示例在 `source/agent_deep.py`，九个 demo。完整案例是一个值守简报员 `source/briefing.py`，定时段落一到，自动收资料出简报。

## 前言用途

Agent 要做的事，是把「模型调工具」这个循环交给框架去转，你只管把工具和规则放进去，等它跑完拿结果。

问题出在「你只管放进去」这六个字上。模型陷入死循环，某类问题根本不该让模型碰，工具调用每一步要留痕，最终产出必须是固定结构，这些事没有一件能靠改提示词解决，需要在循环的各个位置插手的能力，LangChain 把它做成了中间件。

本篇围绕「怎么插手」展开。六个钩子各守一个位置，读状态、改提示、拦调用、掐循环，核心用法里逐个过。

## 原理概念

官方文档给的定义很直接，Agent 就是模型循环调工具直到任务完成。循环之外的一切，提示、工具、中间件，合起来叫 harness。create_agent 是官方提供的线束组装器，你把零件递给它，它还你一个编译好的 LangGraph 图。

这个图长什么样，拆开看一眼。图里有一批节点，model 节点调模型，tools 节点执行工具，每加一个中间件，它的钩子也会变成图上的独立节点。节点的连线上挂着条件，模型返回里有工具调用就走 tools，没有就出循环。插手就是往这张图的节点和边上做文章。

状态是这些节点之间传递的东西。`AgentState` 是个 TypedDict，默认只有三个字段。`messages` 装完整对话历史，只增不删。`structured_response` 装 response_format 解析出来的结构化对象，没配结构化输出时用不到。`jump_to` 是个私有字段，管控制流跳转，后面讲中间件时会用到它。想要自己的字段，写个子类传给 `state_schema`。

create_agent 的完整签名有十五个参数，比官方文档页面上出现的多。我把它们全打出来过。

```text
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, debug = False, name = None, cache = None,
transformers = None
```

常用的就前八个。`middleware` 是本篇的主角。`checkpointer` 让对话历史跨 invoke 保留，多轮对话靠它。`context_schema` 定义随每次调用传入的数据形状，工具和中间件经 runtime.context 读它。`name` 给 agent 起名，多 agent 协作时用于区分。剩下的 `interrupt_before`、`interrupt_after`、`debug`、`cache`、`transformers` 本篇用不上，知道有就行。

中间件有六个钩子，按触发时机分三对。`before_agent` 和 `after_agent` 包住整个 run，一次运行只触发一对。`before_model` 和 `after_model` 包住每轮模型调用，模型调几次就触发几对。`wrap_model_call` 和 `wrap_tool_call` 是包裹式的，前者接住即将发给模型的请求，后者接住即将执行的工具调用，两者都能改内容也能拒绝执行。

![](https://static.xiongneng.me/agent-hooks-20261017170947.png)

钩子函数的签名是固定的。四个 before/after 钩子收 `state` 和 `runtime` 两个参数，返回一个字典或者 None，字典里的键值会被合并回状态。两个 wrap 钩子收 `request` 和 `handler`，`request` 里有本轮的模型请求或工具调用详情，`handler(request)` 让流程继续，不调它就等于拦截。返回值除了字典，还可以是 langgraph 的 `Command`，带上状态更新一起走。

钩子返回的字典里有个特殊的键 `jump_to`，合法值只有 `tools`、`model`、`end` 三个，对应三种跳转。用它的前提是钩子声明了自己能跳去哪，装饰器写法是 `@before_model(can_jump_to=["end"])`。拦截问题这类需求用它正合适，核心用法里有个完整例子。

结构化输出在 agent 里有自己的玩法。直接给 `response_format=你的Pydantic模型`，框架默认走 AutoStrategy，先把 schema 包装成一个工具挂到模型面前，模型调这个工具，参数就按 schema 校验，校验过的对象放进状态的 `structured_response`。模型忘了带必填字段时，校验失败会变成一条 error ToolMessage 喂回去，模型有机会补全重交，和第四篇讲的工具参数自我修正是同一套机制。

## 核心用法

九个用法，示例在 `source/agent_deep.py`，不设 `AGENT_MODEL` 时用本地替身模型跑。每个 demo 独立成函数，`uv run python agent_deep.py` 全跑，或跟名字单跑一个。

### 1. 全参数一览与 name

先看看线束上到底有多少个旋钮。

```python
import inspect
from langchain.agents import create_agent

for name, param in inspect.signature(create_agent).parameters.items():
    default = "" if param.default is inspect.Parameter.empty else f" = {param.default!r}"
    print(f"  {name}{default}")
```

```text
== 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
  debug = False
  name = None
  cache = None
  transformers = None
```

`name` 参数给 agent 标识，不影响行为，单 agent 用不上，嵌进更大的系统做子图时用来区分是谁在说话。

### 2. 动态系统提示

`system_prompt` 收静态字符串，运行时要变的提示交给 `dynamic_prompt` 装饰器。函数收到 `ModelRequest`，里面带完整状态，返回什么字符串，模型这轮就收什么系统提示。

```python
from langchain.agents.middleware import dynamic_prompt

@dynamic_prompt
def round_aware_prompt(request) -> str:
    """按对话轮数生成不同的系统提示。"""
    rounds = len(request.state["messages"])
    if rounds > 2:
        return "对话已经很长，回答务必精简，不超过一句话。"
    return "你是简洁的天气助手，回答不超过两句话。"
```

多轮对话要配 `checkpointer`，不然每次 invoke 都是全新状态，提示永远停在第一档。

```python
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model=model, tools=[get_weather],
    middleware=[round_aware_prompt], checkpointer=InMemorySaver(),
)
config = {"configurable": {"thread_id": "t1"}}
agent.invoke({"messages": [HumanMessage("北京天气")]}, config=config)
result = agent.invoke({"messages": [HumanMessage("那上海呢？该带伞吗？")]}, config=config)
```

```text
== 动态系统提示 ==
两轮对话后最终回答: 上海多云，不用带伞。
```

第二轮进来时历史里已有四条消息，提示切到精简档。每轮模型调用前都重新算一遍提示，dynamic 的意思就在这。

### 3. 自定义 state 字段

写一个 `AgentState` 的子类，加上自己的字段，传给 `state_schema`。字段想跨节点累积就用 `Annotated` 配一个合并函数。

```python
import operator
from typing import Annotated
from langchain.agents import AgentState, AgentMiddleware

class AgentNotes(AgentState):
    """在默认 messages 之外加一个自定义字段。"""
    audit_log: Annotated[list[str], operator.add]

class NoteTaking(AgentMiddleware):
    """开工记一笔，每次工具调用再记一笔。"""
    def before_agent(self, state, runtime):
        return {"audit_log": ["开工"]}

    def wrap_tool_call(self, request, handler):
        from langgraph.types import Command
        result = handler(request)
        name = request.tool_call["name"]
        return Command(update={"messages": [result], "audit_log": [f"调用 {name}"]})
```

`wrap_tool_call` 里没有直接写状态的地方，返回 `Command` 并把更新装进 `update` 是标准做法。跑一遍。

```text
== 自定义 state ==
audit_log: ['开工', '调用 get_weather']
```

两条记录都在。`Annotated[list[str], operator.add]` 的意思是同一步里有多处写入就列表相加，不写这个合并策略，多处写入会直接抛 `Can receive only one value per step`。

### 4. 钩子时序

六个钩子什么顺序触发，一个带打印的中间件跑一遍就清楚。

```python
class Trace(AgentMiddleware):
    def before_agent(self, state, runtime):
        print("  1 before_agent   整个 run 开始前，一次")
    def before_model(self, state, runtime):
        print("  2 before_model   每轮调模型前")
    def wrap_tool_call(self, request, handler):
        print("  3 wrap_tool_call 工具执行外包一层")
        return handler(request)
    def after_model(self, state, runtime):
        print("  4 after_model    每轮调模型后")
    def after_agent(self, state, runtime):
        print("  5 after_agent    整个 run 结束前，一次")
```

一次「查天气再回答」的完整运行。

```text
== 钩子时序 ==
  1 before_agent   整个 run 开始前，一次
  2 before_model   每轮调模型前
  4 after_model    每轮调模型后
  3 wrap_tool_call 工具执行外包一层
  2 before_model   每轮调模型前
  4 after_model    每轮调模型后
  5 after_agent    整个 run 结束前，一次
```

两轮模型调用，before/after 各触发两次，工具执行夹在两轮中间，所以 wrap_tool_call 出现在第一轮 after_model 之后。整个 run 前后的那一对只出现一次。写中间件时对照这张时序表选钩子，位置就不会放错。

### 5. wrap_tool_call 审计

`wrap_tool_call` 拿到的 `request` 里有工具名、入参、调用 id，`handler(request)` 的返回是 `ToolMessage`。改入参、改结果、拒绝执行、记日志，都在这一个函数里。

```python
from langchain.agents.middleware import wrap_tool_call

@wrap_tool_call
def audit(request, handler):
    """打印每次工具调用的入参与出参。"""
    print(f"  [审计] {request.tool_call['name']} 入参 {request.tool_call['args']}")
    result = handler(request)
    print(f"  [审计] 出参 {result.content[:24]}")
    return result
```

```text
== 工具审计 ==
  [审计] get_weather 入参 {'city': '深圳'}
  [审计] 出参 深圳今天晴，气温 26℃。
```

不想让某个调用继续，不调 `handler` 就行，返回一条你自己构造的 `ToolMessage`，模型只会看到你给它的话。

### 6. before_model 拦截

有些事轮不到模型。问题里带着敏感词，最干净的处理是压根不让模型跑。`jump_to` 控制流就是干这个的。

```python
from langchain.agents.middleware import before_model

@before_model(can_jump_to=["end"])
def sensitive_guard(state, runtime):
    """问题里带敏感词就不让模型跑，直接结束。"""
    question = state["messages"][0].content
    if "密码" in str(question):
        return {"jump_to": "end"}
    return None
```

```text
== before_model 拦截 ==
拦截后消息序列: ['HumanMessage']
```

消息序列里只剩用户那句话，模型一轮都没跑，token 一次都没花。`can_jump_to=["end"]` 是声明，告诉框架这个钩子有权跳向 end，不声明，条件边根本不会建。注意拦截要用 `before_model`，写在 `after_model` 里值会进状态但条件边不跳转，实测如此。

### 7. 模型调用轮数限制

模型偶尔会陷进死循环，反复调同一个工具。LangChain 自带了限制器。

```python
from langchain.agents.middleware import ModelCallLimitMiddleware

agent = create_agent(
    model=model, tools=[get_weather],
    middleware=[ModelCallLimitMiddleware(thread_limit=3, exit_behavior="end")],
)
```

```text
== 轮数限制 ==
模型要求调 5 次工具，实际执行 3 次后被掐断
```

替身模型被脚本安排着连调五次，跑到第三次被强制结束。`thread_limit` 管一条对话线程的总量，还有 `run_limit` 管单次运行。`exit_behavior` 选 `end` 是安静收工，选 `error` 会抛异常，看你的场景要哪种。

### 8. response_format 结构化产出

agent 的最终产出想是个校验过的对象，而不是一段自由文本，给 `response_format` 传 Pydantic 模型。

```python
from pydantic import BaseModel

class Brief(BaseModel):
    """简报结构。"""
    summary: str
    need_umbrella: bool

agent = create_agent(model=model, tools=[get_weather], response_format=Brief)
result = agent.invoke({"messages": [HumanMessage("北京需要带伞吗？")]})
```

```text
== 结构化输出 ==
keys: ['messages', 'structured_response']
structured_response: summary='北京晴，不用带伞' need_umbrella=False
```

框架把 Brief 包装成一个工具挂给模型，模型调用它，参数按 schema 校验，校验过的实例放进 `structured_response`。替身模型被脚本安排着直接调用了 Brief 工具，真实模型同样会去调它。必填字段漏了会怎样，框架转成 error ToolMessage 让模型重交，我测过，补全后正常通过。

### 9. 流式执行

`invoke` 只在跑完后一次给全。用户盯着界面时需要过程感，用 `stream`。`updates` 模式流出每个节点完成后的状态更新，`messages` 模式流出模型 token，工具或钩子里手动写的进度走 `custom` 模式。

```python
from langgraph.config import get_stream_writer

@tool
def lookup_city(city: str) -> str:
    """查询城市资料，过程中流出进度。"""
    writer = get_stream_writer()
    writer(f"正在检索 {city} 的资料")
    return f"{city}：人口 2100 万，今天的天气晴。"

agent = create_agent(model=model, tools=[lookup_city])
for chunk in agent.stream(inputs, stream_mode="updates"):
    for node, update in chunk.items():
        kinds = [type(m).__name__ for m in update.get("messages", [])]
        print(f"  {node}: {kinds}")
```

```text
== stream_mode=updates（每步状态更新） ==
  model: ['AIMessage']
  tools: ['ToolMessage']
  model: ['AIMessage']
```

模型发起调用、工具执行、模型给出回答，三个节点依次流出。换 `custom` 模式跑同一个 agent，工具内部的进度也收得到。

```text
== stream_mode=custom（工具内部自定义进度） ==
  [进度] 正在检索 北京 的资料
```

`get_stream_writer()` 只能在图的执行上下文里用，工具单拿出来直接调会拿不到 writer。三种模式可以同时开，`stream_mode=["updates", "custom"]`，这时每个 chunk 是 `(模式, 数据)` 元组，按第一项分发。

## 完整案例

九个用法摊开讲完了。收个尾，把它们装进同一个 agent，写一个值守简报员 `briefing.py`。它接到问题先调工具核实，最后交一份结构固定的简报，审计、动态提示、轮数限制、结构化输出、流式观察都派上了用场。

```bash
uv run python briefing.py --list
```

```text
工具: get_weather, calculate
中间件: NoteKeeper（审计）, duty_prompt（动态提示）, ModelCallLimitMiddleware（限 4 轮）
输出: Brief(title, points, confidence)
```

自定义状态带一个 `audit_log`，中间件 NoteKeeper 用 `wrap_tool_call` 把每次工具调用连同时间戳记进去。动态提示 `duty_prompt` 把审计内容拼进系统提示，模型每一轮都知道自己已经干过什么。

```python
class DutyState(AgentState):
    audit_log: Annotated[list[str], operator.add]

class NoteKeeper(AgentMiddleware):
    def wrap_tool_call(self, request, handler):
        result = handler(request)
        call = request.tool_call
        note = f"{datetime.now().strftime('%H:%M:%S')} 调用 {call['name']}({call['args']})"
        return Command(update={"messages": [result], "audit_log": [note]})

@dynamic_prompt
def duty_prompt(request: ModelRequest) -> str:
    logs = request.state.get("audit_log") or []
    audit = "；".join(logs) if logs else "还没有调用过任何工具"
    return (
        "你是值守简报员，用工具核实信息后输出简报，回答不超过三句话。"
        f"已完成的操作：{audit}。"
    )
```

组装就一行 create_agent，注意 `state_schema` 换成了自定义的 DutyState。

```python
agent = create_agent(
    model=model,
    tools=[get_weather, calculate],
    middleware=[
        NoteKeeper(),
        duty_prompt,
        ModelCallLimitMiddleware(thread_limit=4, exit_behavior="end"),
    ],
    response_format=Brief,
    state_schema=DutyState,
    name="duty_briefing",
)
```

先看本地替身的完整输出。

```bash
uv run python briefing.py "北京适合跑步吗？顺便算下 6 乘 8"
```

```text
模型: ScriptedChatModel（本地替身）
问题: 北京适合跑步吗？顺便算下 6 乘 8

审计日志:
  16:55:18 调用 get_weather({'city': '北京'})

简报: 北京天气简报
  · 北京今天晴，气温 26℃
  · 湿度 40%，微风，适合户外活动
可信度: high
```

结构化简报从 `structured_response` 里取出来，审计日志同步落账。加 `--stream` 能看着它一步步走。

```text
  [model] AIMessage -> get_weather({'city': '北京'})
  [tools] ToolMessage
  [model] AIMessage -> Brief({'title': '北京天气简报', ...})
  [model] ToolMessage
```

四行输出就是四次状态更新，最后那次 `Brief` 调用是结构化输出工具被执行，产出进 `structured_response`。接上真实模型再跑同一条命令，看它怎么处理两个诉求。

```bash
uv run python briefing.py "北京适合跑步吗？顺便算下 128 乘 4" --plain --stream
```

```text
模型: openai:deepseek-v4.1-flash
问题: 北京适合跑步吗？顺便算下 128 乘 4

  [model] AIMessage -> get_weather({'city': '北京'}); calculate({'expression': '128 * 4'})
  [tools] ToolMessage
  [tools] ToolMessage
  [model] AIMessage

审计日志:
  17:00:23 调用 get_weather({'city': '北京'})
  17:00:23 调用 calculate({'expression': '128 * 4'})

回答: 北京今天晴、26℃、湿度40%、微风，适合跑步。128×4=512。建议避开正午强日照时段。
```

真实模型一轮就把两个工具都挑了出来，两个 tools 更新并行流出，审计日志两条都落了账。并行写入能共存，靠的是 `audit_log` 上声明的 `operator.add`。本地替身只调一个工具，这个雷不响，真实模型并行调用的那一刻才见分晓。`--plain` 是降级开关，网关不支持结构化输出时报 400，改走提示词约束加 `json.loads` 解析的纯文本路线。

测试在 `tests/test_agent_deep.py` 十四条，`tests/test_briefing.py` 七条，钩子时序、拦截、限流、自我修正、并行合并、流式收账都钉住了。加上前四篇的，全仓七十一条，全绿。

## 源码

[https://github.com/yidao620c/xneng-langchain-langgraph](https://github.com/yidao620c/xneng-langchain-langgraph)

## 参考

- [LangChain 官方文档 - Agents](https://docs.langchain.com/oss/python/langchain/agents)
- [LangChain 官方文档 - Middleware](https://docs.langchain.com/oss/python/langchain/middleware/overview)
- [LangChain 官方文档 - Streaming](https://docs.langchain.com/oss/python/langchain/streaming)


---

> 作者: XiongNeng  
> URL: https://xiongneng.me/posts/ai/langchain/langchain-langgraph-05-agents/