学习进度
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 的完整签名有十五个参数，比官方文档页面上出现的多。我把它们全打出来过。
1
2
3
4
5
| 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 是包裹式的，前者接住即将发给模型的请求，后者接住即将执行的工具调用，两者都能改内容也能拒绝执行。
钩子函数的签名是固定的。四个 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
先看看线束上到底有多少个旋钮。
1
2
3
4
5
6
| 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}")
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| == 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，里面带完整状态，返回什么字符串，模型这轮就收什么系统提示。
1
2
3
4
5
6
7
8
9
| 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 都是全新状态，提示永远停在第一档。
1
2
3
4
5
6
7
8
9
| 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)
|
1
2
| == 动态系统提示 ==
两轮对话后最终回答: 上海多云，不用带伞。
|
第二轮进来时历史里已有四条消息，提示切到精简档。每轮模型调用前都重新算一遍提示，dynamic 的意思就在这。
3. 自定义 state 字段
写一个 AgentState 的子类，加上自己的字段，传给 state_schema。字段想跨节点累积就用 Annotated 配一个合并函数。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| 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 是标准做法。跑一遍。
1
2
| == 自定义 state ==
audit_log: ['开工', '调用 get_weather']
|
两条记录都在。Annotated[list[str], operator.add] 的意思是同一步里有多处写入就列表相加，不写这个合并策略，多处写入会直接抛 Can receive only one value per step。
4. 钩子时序
六个钩子什么顺序触发，一个带打印的中间件跑一遍就清楚。
1
2
3
4
5
6
7
8
9
10
11
12
| 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 结束前，一次")
|
一次「查天气再回答」的完整运行。
1
2
3
4
5
6
7
8
| == 钩子时序 ==
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 前后的那一对只出现一次。写中间件时对照这张时序表选钩子，位置就不会放错。
wrap_tool_call 拿到的 request 里有工具名、入参、调用 id，handler(request) 的返回是 ToolMessage。改入参、改结果、拒绝执行、记日志，都在这一个函数里。
1
2
3
4
5
6
7
8
9
| 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
|
1
2
3
| == 工具审计 ==
[审计] get_weather 入参 {'city': '深圳'}
[审计] 出参 深圳今天晴，气温 26℃。
|
不想让某个调用继续，不调 handler 就行，返回一条你自己构造的 ToolMessage，模型只会看到你给它的话。
6. before_model 拦截
有些事轮不到模型。问题里带着敏感词，最干净的处理是压根不让模型跑。jump_to 控制流就是干这个的。
1
2
3
4
5
6
7
8
9
| 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
|
1
2
| == before_model 拦截 ==
拦截后消息序列: ['HumanMessage']
|
消息序列里只剩用户那句话，模型一轮都没跑，token 一次都没花。can_jump_to=["end"] 是声明，告诉框架这个钩子有权跳向 end，不声明，条件边根本不会建。注意拦截要用 before_model，写在 after_model 里值会进状态但条件边不跳转，实测如此。
7. 模型调用轮数限制
模型偶尔会陷进死循环，反复调同一个工具。LangChain 自带了限制器。
1
2
3
4
5
6
| from langchain.agents.middleware import ModelCallLimitMiddleware
agent = create_agent(
model=model, tools=[get_weather],
middleware=[ModelCallLimitMiddleware(thread_limit=3, exit_behavior="end")],
)
|
1
2
| == 轮数限制 ==
模型要求调 5 次工具，实际执行 3 次后被掐断
|
替身模型被脚本安排着连调五次，跑到第三次被强制结束。thread_limit 管一条对话线程的总量，还有 run_limit 管单次运行。exit_behavior 选 end 是安静收工，选 error 会抛异常，看你的场景要哪种。
agent 的最终产出想是个校验过的对象，而不是一段自由文本，给 response_format 传 Pydantic 模型。
1
2
3
4
5
6
7
8
9
| 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("北京需要带伞吗？")]})
|
1
2
3
| == 结构化输出 ==
keys: ['messages', 'structured_response']
structured_response: summary='北京晴，不用带伞' need_umbrella=False
|
框架把 Brief 包装成一个工具挂给模型，模型调用它，参数按 schema 校验，校验过的实例放进 structured_response。替身模型被脚本安排着直接调用了 Brief 工具，真实模型同样会去调它。必填字段漏了会怎样，框架转成 error ToolMessage 让模型重交，我测过，补全后正常通过。
9. 流式执行
invoke 只在跑完后一次给全。用户盯着界面时需要过程感，用 stream。updates 模式流出每个节点完成后的状态更新，messages 模式流出模型 token，工具或钩子里手动写的进度走 custom 模式。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| 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}")
|
1
2
3
4
| == stream_mode=updates（每步状态更新） ==
model: ['AIMessage']
tools: ['ToolMessage']
model: ['AIMessage']
|
模型发起调用、工具执行、模型给出回答，三个节点依次流出。换 custom 模式跑同一个 agent，工具内部的进度也收得到。
1
2
| == stream_mode=custom（工具内部自定义进度） ==
[进度] 正在检索 北京 的资料
|
get_stream_writer() 只能在图的执行上下文里用，工具单拿出来直接调会拿不到 writer。三种模式可以同时开，stream_mode=["updates", "custom"]，这时每个 chunk 是 (模式, 数据) 元组，按第一项分发。
完整案例
九个用法摊开讲完了。收个尾，把它们装进同一个 agent，写一个值守简报员 briefing.py。它接到问题先调工具核实，最后交一份结构固定的简报，审计、动态提示、轮数限制、结构化输出、流式观察都派上了用场。
1
| uv run python briefing.py --list
|
1
2
3
| 工具: get_weather, calculate
中间件: NoteKeeper（审计）, duty_prompt（动态提示）, ModelCallLimitMiddleware（限 4 轮）
输出: Brief(title, points, confidence)
|
自定义状态带一个 audit_log，中间件 NoteKeeper 用 wrap_tool_call 把每次工具调用连同时间戳记进去。动态提示 duty_prompt 把审计内容拼进系统提示，模型每一轮都知道自己已经干过什么。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| 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。
1
2
3
4
5
6
7
8
9
10
11
12
| 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",
)
|
先看本地替身的完整输出。
1
| uv run python briefing.py "北京适合跑步吗？顺便算下 6 乘 8"
|
1
2
3
4
5
6
7
8
9
10
| 模型: ScriptedChatModel（本地替身）
问题: 北京适合跑步吗？顺便算下 6 乘 8
审计日志:
16:55:18 调用 get_weather({'city': '北京'})
简报: 北京天气简报
· 北京今天晴，气温 26℃
· 湿度 40%，微风，适合户外活动
可信度: high
|
结构化简报从 structured_response 里取出来，审计日志同步落账。加 --stream 能看着它一步步走。
1
2
3
4
| [model] AIMessage -> get_weather({'city': '北京'})
[tools] ToolMessage
[model] AIMessage -> Brief({'title': '北京天气简报', ...})
[model] ToolMessage
|
四行输出就是四次状态更新，最后那次 Brief 调用是结构化输出工具被执行，产出进 structured_response。接上真实模型再跑同一条命令，看它怎么处理两个诉求。
1
| uv run python briefing.py "北京适合跑步吗？顺便算下 128 乘 4" --plain --stream
|
1
2
3
4
5
6
7
8
9
10
11
12
13
| 模型: 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
参考
