主题
模型调用与 Agent 调用的入参、出参对比
本文横向对比第 02 章「模型的创建与调用」和第 07 章「智能体」的调用方式,重点比较 Chat 模型的 invoke() / stream() 与 create_agent() 创建出的 Agent 在同一调用接口下的入参、出参差异,并单独说明流式模式下的结构。
本文基于课件已核验的内容整理。涉及 LangGraph 流式二级参数(如
stream_mode的扩展选项、消息过滤)的部分,课件未展开,本文只写到课件明确给出的范围,未逐项在线核验最新版本。
1. 先建立两个调用接口在结构上的区别
| 维度 | 模型调用 model.invoke() | Agent 调用 agent.invoke() |
|---|---|---|
| 调用对象 | BaseChatModel 实例 | create_agent() 返回的编译后图(基于 LangGraph) |
| 入参类型 | str / list[dict] / list[Message] | 必须是 dict:{"messages": [...]} |
| 入参语义 | 一次对话的完整快照,直接发给模型 | 图的初始状态,messages 只是状态中的一个键 |
| 出参类型 | 单个 AIMessage | dict,即图的最终状态 |
| 出参中的 message 数 | 1 条 | 多条,含工具调用中间步骤 |
| 底层模型请求次数 | 1 次 | 1~N 次,由工具调用轮次决定 |
| 系统提示词位置 | 放在入参消息列表里 | create_agent(system_prompt=...) 或放在入参 messages 里 |
config 常见用途 | callbacks、tags、metadata | 额外支持 recursion_limit、thread_id 等 |
一句话概括:模型 invoke 是“一次请求 → 一条消息”;Agent invoke 是“一次请求 → 一个状态字典”,内部装着 N 次模型请求与工具执行拼接出的完整消息链。
2. 入参对比
2.1 模型 invoke 的入参
python
response = model.invoke(input, config=None)input 支持三种形态:
| 形态 | 示例 | role 来源 | 适用场景 |
|---|---|---|---|
| 字符串 | model.invoke("你好") | 自动包成 HumanMessage | 单轮测试,无法带 system 和历史 |
| 字典列表 | [{"role": "user", "content": "你好"}] | 显式指定 | 生产环境,可序列化、可带历史 |
| 消息对象列表 | [SystemMessage("..."), HumanMessage("...")] | 由类决定 | 需要类型检查与 IDE 补全 |
role 支持 system / user / assistant / tool,分别对应 SystemMessage / HumanMessage / AIMessage / ToolMessage。
config 用于回调、标签、元数据等运行配置,不改变对话内容。
2.2 Agent invoke 的入参
python
response = agent.invoke({"messages": [...]}, config=None)相比模型调用,有三处差异:
① 外层多一个 dict
messages 是 LangGraph 状态里的键。状态中还可能有其他字段,例如启用 response_format 后的 structured_response。这就是外形必须是 dict 的原因。
② messages 的值形态与模型调用一致
python
agent.invoke({"messages": ["你好"]}) # 自动转 HumanMessage
agent.invoke({"messages": [{"role": "user", "content": "你好"}]})
agent.invoke({"messages": [HumanMessage("你好")]})差异只在外层包装,内层的消息列表写法可以沿用模型调用的习惯。
③ 大量配置在“创建时”而非“调用时”
python
agent = create_agent(
model=model,
tools=[tool1, tool2],
system_prompt="Agent 的行为指令",
middleware=(),
name="chat_assistant",
)同一个 Agent 可以反复 invoke,每次只换 messages。模型调用则每次都要把 system 提示带进消息列表。
课件中同时出现过
create_agent(system_prompt=...)和向 messages 里塞{"role": "system", ...}两种写法。建议统一使用system_prompt参数,避免两个来源同时存在时行为不一致。
④ config 多出 Agent 专属项
python
response = agent.invoke(input, config={"recursion_limit": 5})recursion_limit 限制图的最大执行步数,用于防止工具循环不停。模型调用不存在这一项,因为单次模型调用没有循环。
2.3 入参对比小结
| 对比点 | 模型 invoke | Agent invoke |
|---|---|---|
| 是否可直接传字符串 | 可以 | 不可以,必须 {"messages": ...} |
| 系统提示词 | 在入参里 | 在创建时(推荐) |
| 工具 | 通过 bind_tools() 绑定,只产生调用意图 | 在创建时传入,框架负责执行 |
| 历史管理 | 调用方自己拼 | 调用方自己拼(Agent 不会自动记忆跨轮历史) |
| 循环控制 | 无 | recursion_limit |
3. 出参对比
3.1 模型 invoke 的出参:单个 AIMessage
AIMessage
├── content 正文,通常是字符串;多模态或内容块时可能是列表
├── additional_kwargs 集成层附加字段(各厂商不同)
├── response_metadata 服务端原始元数据(token_usage / finish_reason / model_name 等)
├── usage_metadata LangChain 标准化后的 token 用量
├── tool_calls 模型“想调”的工具(意图,没有结果)
├── invalid_tool_calls 解析失败的工具调用
├── id 本条消息 ID
└── name 可选关键点:tool_calls 在模型层只是意图。 模型说“要调 get_weather(city=北京)”,但没有任何代码去执行它,也拿不到执行结果。
3.2 Agent invoke 的出参:状态字典
python
{
"messages": [
SystemMessage(...), # 入参里的 system(若通过 messages 传入)
HumanMessage(...), # 入参里的 user
AIMessage( # 第 1 次模型响应:决定调用工具
content="",
tool_calls=[{"name": "get_weather", "args": {"city": "北京"}, "id": "call_xxx"}],
response_metadata={"finish_reason": "tool_calls"},
),
ToolMessage( # 框架执行工具后生成的结果
content="北京的天气为晴朗,25°C。",
name="get_weather",
tool_call_id="call_xxx",
),
AIMessage( # 第 2 次模型响应:最终回答
content="北京天气晴朗,25°C。",
response_metadata={"finish_reason": "stop"},
),
]
}- 返回值类型是
dict,取正文要用response["messages"][-1].content。 messages是本轮完整消息序列,包含入参消息和全部中间步骤。- 工具重试时序列会更长,例如重试两次后成功,序列为
Human → AI(tool_calls) → Tool → AI(tool_calls) → Tool → AI。 - 启用
response_format时,字典中会多一个structured_response键,这是模型层出参中不存在的内容。
3.3 各类消息由谁产生
| 消息类型 | 模型 invoke 中 | Agent invoke 中 |
|---|---|---|
SystemMessage / HumanMessage | 由调用方产生 | 由调用方产生,原样出现在出参中 |
AIMessage(带 tool_calls) | 模型产生,只停留在意图 | 模型产生意图,框架负责执行 |
ToolMessage | 不会出现 | 框架自动生成 |
AIMessage(最终回答) | 就是返回值本身 | 模型产生,位于序列末尾 |
ToolMessage 是唯一一类模型层永远产不出、只有 Agent 才会产生的消息。判断工具往返是否配对,看 ToolMessage.tool_call_id 是否等于上一条 AIMessage.tool_calls[].id。
3.4 模型出参与 Agent 出参的映射关系
Agent 一轮执行可以理解为“模型调用 + 工具执行的串行拼装”:
text
Agent.invoke({"messages": [...]})
│
├─ 第 1 次 model.invoke(入参 messages) → AIMessage#1 (finish_reason=tool_calls)
├─ 执行工具 → ToolMessage#1
├─ 第 2 次 model.invoke(入参 + AI#1 + Tool#1) → AIMessage#2 (finish_reason=tool_calls)
├─ 执行工具 → ToolMessage#2
└─ 第 N 次 model.invoke(...) → AIMessage#N (finish_reason=stop)
↓
{"messages": 入参 + 全部中间消息}(另有 structured_response 可选)推论:
- 序列中每条
AIMessage的内部结构,与直接调模型得到的AIMessage完全相同。 finish_reason可以区分步骤:tool_calls是中间步,stop是最终回答。- token 用量按轮生成、分布在各条
AIMessage上,Agent 出参没有直接给出总用量,需要自行汇总。 - Agent 的“循环”本质就是消息列表不断增长,每次模型请求都携带此前全部消息。
4. 取值姿势对照
| 需求 | 模型 invoke | Agent invoke |
|---|---|---|
| 正文 | response.content | response["messages"][-1].content |
| token 用量 | response.usage_metadata | 遍历 messages,汇总各 AIMessage.usage_metadata |
| 结束原因 | response.response_metadata["finish_reason"] | 同一个键,但需逐条 AIMessage 查看 |
| 是否调用工具 | response.tool_calls 非空 | 统计 ToolMessage 条数,或看中间 AIMessage.tool_calls |
| 结构化结果 | 需 with_structured_output() 包装 | response["structured_response"] |
| 完整执行过程 | 拿不到,只有意图没有结果 | response["messages"] 全序列 |
5. 流式调用的入参、出参对比
5.1 总体差异
| 维度 | 模型 model.stream() | Agent agent.stream() |
|---|---|---|
| 入参 | 与 invoke() 完全相同 | 与 invoke() 完全相同,外加 stream_mode |
| 出参类型 | 迭代器(iterator) | 迭代器(iterator) |
| 单个产出物 | AIMessageChunk(消息片段) | 取决于 stream_mode |
| 是否需要拼装 | 需要,累积所有 chunk 得到完整 AIMessage | messages 模式需要拼装;values / updates 模式不需要 |
| 流式开启条件 | 依赖服务端支持流式输出 | 同样依赖模型/服务端支持 |
核心结论一:两个 stream() 的入参写法与各自的 invoke() 保持一致。
核心结论二:出参的“粒度”完全不同。 模型 stream() 流的是字(token);Agent stream() 流的是执行步骤,想要“字”的效果必须显式指定 stream_mode="messages"。
5.2 模型 stream() 的入参
python
for chunk in model.stream("写一首七言律诗,总结大模型的发展"):
print(chunk.text, end="", flush=True)input、config与invoke()完全一致,支持字符串、字典列表、消息对象列表。- 调用后立即返回迭代器,不等待完整结果。
- 流式输出依赖模型供应商支持,不支持时可能退化为一次性返回。
5.3 模型 stream() 的出参
每次循环得到的是一个 AIMessageChunk,而不是完整的 AIMessage:
AIMessageChunk
├── content 本片段新增的文字(不是全文)
├── text 文本快捷属性(课件示例使用 chunk.text)
├── tool_call_chunks 工具调用的增量片段
├── usage_metadata 用量信息(通常在最后一个 chunk 才完整)
└── id / response_metadata 等拼装完整响应需要累加:
python
full = None
for chunk in model.stream("你好"):
full = chunk if full is None else full + chunk # AIMessageChunk 支持相加- 不能把单个 chunk 当成答案:
chunk.content只是这一小段。 - 工具调用在流式下表现为
tool_call_chunks增量,参数是分多次拼出来的,需要合并。 finish_reason、完整usage_metadata往往只在末尾出现。- 模型
stream()不会执行工具,与invoke()一样只产出意图。
5.4 Agent stream() 的入参
python
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "..."}]},
stream_mode="values", # 关键差异点
config=None,
):
...入参相比 agent.invoke() 只多了一个关键参数:
| 参数 | 说明 |
|---|---|
input | 同 invoke(),必须是 {"messages": [...]} |
stream_mode | 决定“每一步流什么”,可取 values / updates / messages / tasks / debug / checkpoints / custom;默认为 updates |
config | 同 invoke(),含 recursion_limit、thread_id 等 |
stream_mode 还可以传列表组合使用,此时循环体拿到的是 (模式名, 数据) 元组:
python
for stream_mode, chunk in agent.stream(input, stream_mode=["tasks", "updates"]):
print(f"当前流模式: {stream_mode}, 当前数据: {chunk}")5.5 Agent stream() 各模式的出参
| 模式 | 每次产出 | 典型结构 | 适用场景 |
|---|---|---|---|
values | 每一步后的完整状态 | 与 invoke() 返回的字典相同,messages 持续增长 | 每步都要完整状态、状态持久化 |
updates(默认) | 每一步的增量更新 | {节点名: 该节点更新的状态片段} | 观察工具调用、执行进度 |
messages | 流式 token 与元数据 | (消息片段, 元数据字典) 元组,元数据含来源节点(model/tool) | 类似 ChatGPT 的打字机效果 |
tasks | 任务开始/结束信息 | 含任务结果与错误信息 | 监控任务生命周期 |
debug | 比 tasks 更详细的步骤信息 | 含步骤、时间戳、任务类型(task / task_result) | 调试、监控 |
checkpoints | 检查点创建时的状态 | 含检查点中的完整状态 | 状态持久化、工作流恢复、分布式跟踪 |
custom | 工具/节点内自定义数据 | 由 get_stream_writer() 写入的任意内容 | 业务进度日志、自定义指标 |
5.5.1 values 模式
python
for chunk in agent.stream(input, stream_mode="values"):
print(chunk["messages"][-1])- 每个 chunk 都是一份完整状态字典,结构与
invoke()的返回值相同。 - 相邻 chunk 之间,
messages只增不减,可以直接取[-1]看最新消息。 - 这是最接近“
invoke()的逐步骤版本”的模式,适合把invoke代码平滑迁移成流式。
5.5.2 updates 模式(默认)
python
for chunk in agent.stream(input, stream_mode="updates"):
print(chunk)- 每个 chunk 只包含发生变化的部分,例如
{"model": {"messages": [AIMessage(...)]}}、{"tools": {"messages": [ToolMessage(...)]}}。 - 键是节点名;值是该节点新增或更新的状态字段。
- 适合做“执行步骤日志”或进度展示,但不能直接当成完整状态使用。
5.5.3 messages 模式
python
for chunk, metadata in agent.stream(input, stream_mode="messages"):
print(chunk.content, end="", flush=True)
print(metadata) # 包含来自哪个节点(model / tool)- 每个 chunk 是
(消息片段, 元数据)元组,消息片段是AIMessageChunk,元数据标明来源节点。 - 这才是“打字机效果”对应的模式;默认的
updates不会逐 token 输出。 - 与工具一起使用时,
tool节点的输出也会进入该流,前端通常需要按metadata过滤出model节点的片段。
5.5.4 tasks / debug / checkpoints
python
for chunk in agent.stream(input, stream_mode="tasks"):
print(chunk) # 任务开始/结束、结果、错误python
for chunk in agent.stream(input, stream_mode="debug"):
print(chunk) # 更细的步骤、时间戳、task / task_resultpython
for chunk in agent.stream(input, config={"configurable": {"thread_id": "session01"}},
stream_mode="checkpoints"):
print(chunk) # 检查点及其中的状态checkpoints需要 Agent 创建时传入checkpointer(如InMemorySaver()),并用thread_id标识会话。- 这三种模式偏运维与调试,产出不是直接展示给用户的文本。
5.5.5 custom 模式
python
from langgraph.config import get_stream_writer
@tool
def report_generator(...):
"""生成报告"""
writer = get_stream_writer()
writer({"progress": "已处理 10/100 条记录"})
...python
for chunk in agent.stream(input, stream_mode="custom"):
print(chunk)- 由工具或节点内部通过
get_stream_writer()主动发送,内容是自定义结构。 - 适合业务进度、进度条、阶段性提示,不适合替代正文。
5.6 流式对比小结
| 对比点 | 模型 stream() | Agent stream() |
|---|---|---|
| 入参是否变化 | 不变 | 增加 stream_mode(默认 updates),可传组合 |
| 产出粒度 | token 片段 | 步骤级(默认),messages 模式下为 token 片段 |
| 产出类型 | AIMessageChunk | 字典 / 元组,视模式而定 |
| 是否包含工具执行过程 | 否 | 是(values / updates / messages 均可观察) |
| 是否需要手动拼装 | 必须 | 仅 messages 模式需要 |
| 结构化输出 | 不涉及 | values 模式最终状态中可见 structured_response |
| 入参消息列表是否被修改 | 否 | 否,历史仍需自行维护 |
5.7 选模式的口诀
- 做聊天界面,要“打字机效果” → Agent
stream_mode="messages" - 观察 Agent 调了哪些工具、执行到哪一步 →
updates(默认) - 每一步都需要完整状态 →
values - 需要展示业务进度 →
custom - 调试或监控任务生命周期 →
tasks/debug - 需要断点续跑、状态持久化 →
checkpoints+checkpointer
6. 常见易错点
agent.invoke("你好")会报错。 Agent 要求 dict 入参,模型调用可以直接传字符串。- 对 Agent 返回值直接
.content会失败。 必须response["messages"][-1].content。 [-1]不一定是最终答案。 触发recursion_limit或异常时,最后一条可能是带tool_calls的AIMessage。稳妥做法是倒序找第一条tool_calls为空的AIMessage。- 只保存
[-1].content会丢上下文。 续聊应保存整个response["messages"]的副本。 response["messages"]包含入参消息。 循环里对同一列表反复 append 会越积越长,需要配合裁剪或记忆策略。- Agent 默认
stream_mode="updates"不会逐字输出。 没指定messages却期待打字机效果,是最常见的误解。 - 模型
stream()的 chunk 不是完整消息。 单个chunk.content只是片段,工具调用参数也分散在tool_call_chunks中。 - 两个
stream()的消费代码不能互换。 模型侧循环变量是chunk,Agent 侧在messages模式下是(chunk, metadata)元组,组合模式下是(mode, chunk)。 - 流式同样受服务端能力限制。 供应商不支持流式时,行为可能退化为一次性返回。
7. 速查表
| 场景 | 推荐写法 | 取结果 |
|---|---|---|
| 单次问答 | model.invoke(prompt) | response.content |
| 多轮问答 | model.invoke([...messages]) | response.content,并 append 回历史 |
| 单次流式生成 | for chunk in model.stream(prompt) | 累加 chunk,或逐段 chunk.text |
| Agent 同步执行 | agent.invoke({"messages": [...]}) | response["messages"][-1].content |
| Agent 结构化输出 | agent.invoke({...}) + response_format | response["structured_response"] |
| Agent 打字机效果 | agent.stream(input, stream_mode="messages") | for chunk, meta in ... |
| Agent 过程追踪 | agent.stream(input)(默认 updates) | 每个 chunk 即一个步骤增量 |
| Agent 每步完整状态 | agent.stream(input, stream_mode="values") | chunk["messages"] |
| Agent 业务进度 | agent.stream(input, stream_mode="custom") | 工具内 get_stream_writer() 写入 |