invoke和stream概述invoke和stream都是编译后 graph 的执行入口用来启动整个图运行底层真正干活的是同一套 Pregel BSP 引擎。invoke一次性同步阻塞调用全部 Superstep 跑完返回最终完整 state。stream返回迭代器每完成一个 Superstep立刻产出这一轮的增量片段可以拿到中间过程。二者底层跑的是同一套图逻辑执行结果完全一致只是返回数据时机不一样。graph.invoke(input)函数签名final_state graph.invoke( input{messages: [(user, 帮我执行压测分析jmeter报告)]}, config{configurable: {thread_id:test‑001}} )执行行为同步阻塞主线程卡住整个图全部执行完毕才返回所有 Superstep 完整跑完循环、重试全部跑完。返回完整最终的 AgentState 全部状态字典。包含所有 messages、trajectory、中间变量。thread_id 配合Checkpointer会自动保存每一步检查点快照。优点代码极简非常适合 Eval‑Harness 自动化评测 Eval‑Harness 跑自动化用例不需要中间输出只需要最终结果直接拿final_state做指标判断ToolCorrectness、结果正确性。# Eval‑Harness典型写法 result_state graph.invoke(golden_case_input, configconfig) actual_output result_state[messages][-1].content # deepeval metric直接评测 actual_output不需要处理迭代循环不需要解析中间片段写自动化测试脚本非常清爽。缺点看不到中间执行轨迹运行过程中看不到哪个节点跑了、工具调用、中间报错全部等跑完才拿到。如果 Agent 循环很多、LLM 调用耗时久会长时间阻塞控制台没有任何输出看起来像卡死。如果图内部设置interrupt()人工暂停invoke遇到中断会直接抛出异常invoke 不适合处理人机暂停场景。关键点invoke 遇到interrupt()会报错因为 invoke 设计是 “一口气跑完”遇到暂停无法继续。处理人工介入只能用stream。适用场景Eval‑Harness 自动化回归测试你的场景后台服务只需要最终输出不展示中间过程脚本批量跑用例graph.stream(input)函数签名stream_output graph.stream( input{messages: [(user, 帮我执行压测分析jmeter报告)]}, config{configurable: {thread_id:test‑001}} ) for chunk in stream_output: print(chunk) # 每一个Superstep结束产出一轮chunk增量执行行为返回一个 Python 迭代器不会阻塞等待全部完成。每完成 1 个 Superstep一轮 BSP 超级步、屏障同步完成就产出一个 chunk 片段。chunk 不是完整 statechunk 只是本轮 Superstep 产生的增量更新只包含本轮被修改过的字段。for 循环每一轮拿到的chunk格式示例# chunk示例key节点名称value该节点返回的增量字典 { supervisor: { next_agent: lighthouse_agent, messages: [AIMessage(...)] } }chunk 只有本轮变更字段不会携带全部历史 state。如果你想要完整 state需要借助 checkpointergraph.get_state(config)读取完整快照。优点实时观测中间 Trajectory 轨迹 每跑完一个节点Superstep立刻拿到输出可以打印哪个节点执行、工具调用参数、中间报错调试 Agent 必备。原生支持识别interrupt()人工暂停 当 Superstep 边界触发中断stream 迭代器产出__interrupt__标记迭代停止外部系统可以展示 UI 给操作人员调用graph.resume()继续执行。可以做前端流式 UI一边跑 Agent一边把中间步骤展示给用户。缺点拿到的是增量片段不是完整 state业务代码需要自己拼装完整上下文写 Eval‑Harness 自动化测试会多写一层 for 循环代码更啰嗦chunk 结构会随节点变化解析要做判断。获取完整状态小技巧stream 循环内随时读取完整快照for chunk in graph.stream(input, config): full_snapshot graph.get_state(config) # 读取checkpointer里完整state print(full_snapshot.values[messages])适用场景Agent 调试观察每一步 Superstep 执行轨迹前端流式交互展示 Agent 思考、工具调用过程Human‑in‑the‑loop需要人工暂停、确认再继续需要采集每一步 trajectory 日志。关键对比表格项目graph.invoke()graph.stream()返回值完整最终 State 字典迭代器每 Superstep 返回增量 chunk阻塞行为完全阻塞全部跑完返回迭代产出不会阻塞等待全部完成拿到的数据全部字段仅本轮修改的增量字段处理 interrupt 中断遇到中断直接抛异常识别中断标记迭代停止获取中间 Trajectory拿不到中间过程每一轮 Superstep 拿到中间输出Eval‑Harness 测试首选代码简洁可以用但需要额外 get_state 拿完整状态调试 Agent不推荐首选看每一步节点输出高频踩坑点坑 1stream 拿到 chunk 以为是完整 state# 错误写法chunk只是增量没有全部字段 for chunk in graph.stream(input, config): print(chunk[messages]) # 有可能KeyError本轮没有更新messages就不存在 # 正确想要完整状态调用 graph.get_state(config).values坑 2invoke 为什么不能处理 interruptinterrupt()生效点在Superstep 屏障同步边界。invoke 希望一口气跑完所有 Superstep一旦遇到中断没有对外暴露暂停接口直接抛出错误。想要人机交互必须用 stream。坑 3同一个 thread_idinvoke 和 stream 可以混用checkpointer 保存快照invoke 跑完之后可以用同一个 thread_id 调用 stream 继续执行反之也可以。Eval‑Harness 测试实践建议自动化回归测试用例优先使用 invoke# golden case自动化测试不需要中间步骤 final_state graph.invoke(test_input, configthread_config) # final_state拿到完整state提取输出、工具调用记录交给deepeval指标评测本地调试 Agent看 trajectory 轨迹使用 stream get_state ()for chunk in graph.stream(test_input, configthread_config): snapshot graph.get_state(configthread_config).values print( Superstep完成 ) print(当前轨迹messages, snapshot[messages])时序小例子Supervisor 多 Agent流程START → supervisor → lighthouse_agent → supervisor → END一共 3 个 SuperstepSuperstep1执行 supervisorstream 产出 chunk:{supervisor: {...}}Superstep2执行 lighthouse_agentstream 产出 chunk:{lighthouse_agent: {...}}Superstep3再次执行 supervisorstream 产出 chunk:{supervisor: {...}}stream循环会收到 3 次 chunkinvoke全部 3 轮跑完一次性返回合并后的完整 state。实例Demo下面是完整可运行最小 Demo实现一个简单的反思循环 Agent同时演示graph.invoke()和graph.stream()打印对比输出。依赖pip install langgraph不需要大模型全部模拟逻辑直接跑就能看到效果。# -*- coding:utf-8 -*- from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver # 1. 定义状态 class State(TypedDict): messages: Annotated[list, operator.add] count: int # 2. 定义节点模拟业务逻辑不调用LLM def think_node(state: State): 思考节点计数1模拟agent思考 cnt state[count] new_msg f第{cnt1}次思考完成 return { messages: [new_msg], count: cnt 1 } def judge_node(state: State): 判断节点最多循环3次就结束 cnt state[count] if cnt 3: return {next: end} else: return {next: loop} # 3. 构建图 builder StateGraph(State) builder.add_node(think, think_node) builder.add_node(judge, judge_node) builder.add_edge(START, think) builder.add_edge(think, judge) # 条件边循环回退到 think builder.add_conditional_edges( judge, lambda s: s[next], { loop: think, end: END } ) checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: demo‑001}} init_input {messages: [], count: 0} if __name__ __main__: print( 【1】演示 graph.stream() 每一个Superstep返回增量chunk \n) stream_iter graph.stream(init_input, configconfig) for chunk in stream_iter: print(f stream收到chunk(本轮增量): {chunk}) # 读取checkpoint里面的完整快照状态 full_snap graph.get_state(config).values print(f 当前完整state: count{full_snap[count]}, messages{full_snap[messages]}\n) print(\n 重置thread_id执行 graph.invoke() 阻塞等待全部完成直接拿最终完整state \n) config_invoke {configurable: {thread_id: demo‑002}} final_state graph.invoke(init_input, configconfig_invoke) print(finvoke返回最终完整state) print(fcount {final_state[count]}) print(fmessages {final_state[messages]})输出样例控制台打印 【1】演示 graph.stream() 每一个Superstep返回增量chunk stream收到chunk(本轮增量): {think: {messages: [第1次思考完成], count: 1}} 当前完整state: count1, messages[第1次思考完成] stream收到chunk(本轮增量): {judge: {next: loop}} 当前完整state: count1, messages[第1次思考完成] stream收到chunk(本轮增量): {think: {messages: [第2次思考完成], count: 2}} 当前完整state: count2, messages[第1次思考完成, 第2次思考完成] stream收到chunk(本轮增量): {judge: {next: loop}} 当前完整state: count2, messages[第1次思考完成, 第2次思考完成] stream收到chunk(本轮增量): {think: {messages: [第3次思考完成], count: 3}} 当前完整state: count3, messages[第1次思考完成, 第2次思考完成, 第3次思考完成] stream收到chunk(本轮增量): {judge: {next: end}} 当前完整state: count3, messages[第1次思考完成, 第2次思考完成, 第3次思考完成] 重置thread_id执行 graph.invoke() 阻塞等待全部完成直接拿最终完整state invoke返回最终完整state count 3 messages [第1次思考完成, 第2次思考完成, 第3次思考完成]重点观察现象stream每一轮 Superstep节点执行完毕 屏障同步产出一个chunkchunk只是本轮节点返回的增量字典不是全部 state如果想要完整数据需要调用graph.get_state(config).values读取检查点快照循环过程中间每一步都可以打印适合调试、采集 trajectoryinvoke程序卡住阻塞所有 Superstep 全部执行完毕才返回返回直接就是完整合并后的最终 state看不到中间每一轮增量适合 Eval‑Harness 自动化测试代码干净直接拿结果做指标判断一句话总结invoke 等全部戏演完一次性拿到完整剧本 stream 每演完一幕就把这一幕的剧本片段递给你。