本文是「LangGraph 教程系列」第 9 篇。写作时基于 langgraph 1.2.10、langchain 1.3.14、langchain-openai 1.4.1、Python 3.12。配套代码仓库 https://github.com/wxj006007/deep-research-assistant 本篇对应 tagv2.3。上一版的研究助手已经能停在审批点也能从长期记忆里拿到用户偏好。它看起来终于像个能交付的程序了。可真把它放到终端或网页里跑一次问题就冒出来了。用户等了半天只在最后看到一整段答案。检索开始了吗资料够不够程序是在等待人工确认还是已经卡死了完全不知道。这一篇不再给助手加新能力而是把它正在做什么交出来。不是所有东西都要逐字显示真正有用的是把不同层次的信息送到合适的地方。一、v2.2 撞的墙用户只看见终点同步调用很适合把流程讲清楚。finalgraph.invoke({question:LangGraph 的 stream_mode 有什么区别},configconfig,contextcontext,)print(final[answer])但 invoke 有一个天然取舍它只在结束时给结果。研究助手内部已经走过读取记忆、规划、审批、检索、评估、成稿这些步骤调用方却像在门外等消息。图里的状态也不能直接拿去当界面协议。state 要保存问题、资料、轮次和审批结果它是工作台。用户需要的是回答正文、进度提示和必要的错误信息。两者重合一部分但不是同一个东西。二、四条流装不同的东西v2.3 仍是同一张研究图只是在节点外包了一层进度事件。渲染错误:Mermaid 渲染失败: Parse error on line 2: flowchart LR graph[研究图] -- msg[m ----------------^ Expecting SEMI, NEWLINE, SPACE, EOF, subgraph, end, acc_title, acc_descr, acc_descr_multiline_value, AMP, COLON, STYLE, LINKSTYLE, CLASSDEF, CLASS, CLICK, DOWN, DEFAULT, NUM, COMMA, NODE_STRING, BRKT, MINUS, MULT, UNICODE_TEXT, direction_tb, direction_bt, direction_rl, direction_lr, direction_td, got GRAPHLangGraph 的流模式可以这样分工。模式传出的内容本例放到哪里values每个超级步后的完整 state调试、检查快照updates某个节点刚写入的 state 增量执行轨迹、开发调试messages模型消息块和元数据最终回答的逐 token 输出custom节点主动写出的任意业务事件前端进度提示values 很直观刚开始学图时尤其有帮助。但资料一多完整 state 会一遍遍重复。把它原样推给浏览器既浪费带宽也会把不该展示的内部字段泄露出去。所以 v2.3 的终端演示订阅 messages、updates、custom。values 保留给排障和教学不把它包装成生产前端的默认数据格式。三、进度不进 state写进 custom读取记忆和检索结果本来就会更新 state。正在读取、正在检索这样的提示只是这一轮运行的即时消息没必要跟着 checkpoint 长期保存。节点通过 get_stream_writer 写出结构化事件。fromlanggraph.configimportget_stream_writerdefemit_progress(stage:str,message:str,**details:object)-None:get_stream_writer()({type:research_progress,stage:stage,message:message,**details,})检索包装节点只增加事件不改变 v2.2 的搜索函数。defsearch_with_progress(state:MemoryResearchState)-dict[str,object]:emit_progress(search.start,正在检索资料,querystate[current_query])updatessearch_node(state)emit_progress(search.done,本轮检索完成,new_documentslen(updates[docs]),)returnupdates同样的模式用在读写长期记忆、规划、评估和成稿上。业务状态仍由原来的节点返回界面进度则独立走 custom 通道。这样做还有一个实际好处换终端、网页或日志系统时不必改图的 state 设计。四、同一次运行订阅三种模式本篇的 build_graph 入口仍然保留 v2.2 的依赖边界。defbuild_graph(checkpointer:MemorySaver|NoneNone,store:BaseStore|NoneNone,):builderStateGraph(MemoryResearchState,context_schemaResearchContext)# 注册带进度事件的包装节点returnbuilder.compile(checkpointercheckpointerorMemorySaver(),storestoreorInMemoryStore(),)消费端一次传入三个模式。LangGraph 会把每条数据组织为 mode 和 payload。formode,payloadingraph.stream(graph_input,configconfig,contextcontext,stream_mode[messages,updates,custom],):ifmodemessages:chunk,metadatapayloadelifmodeupdates:print(payload)elifmodecustom:print(payload[message])这里有个很容易被忽略的细节。planner、evaluator、writer 都会调用模型。若把所有 messages 都显示在回答区域用户会先看到查询词和评估 JSON界面就乱了。因此示例只渲染 synthesize 节点的消息。ifmetadata.get(langgraph_node)!synthesize:continuecontentgetattr(chunk,content,)ifcontent:print(str(content),end,flushTrue)这样模型仍可在每个节点按自己的方式工作用户却只会逐字看到最后那段回答。progress 区则接收检索和审批这些更适合人读的事件。五、流结束不一定代表任务完成第 7 篇讲过interrupt 会暂停图并交还控制权。放进流式调用后规则没有变化。第一次 stream 消费到带有interrupt的 updates 后迭代器就结束了。interrupted,statsconsume_stream(graph,initial_input,config,context)ifinterrupted:interrupted,statsconsume_stream(graph,Command(resume{action:approve}),config,context,)关键还是同一个 thread_id 和同一个 ResearchContext。thread 决定从哪个 checkpoint 恢复context 决定当前用户和 research_id。还有一件事要提前处理。interrupt 之后恢复时所在节点会重放。review_with_progress 在暂停前发出的 waiting 事件可能再出现一次。前端不要把它当成一条永不重复的数据库记录用 stage 和本轮执行标识做幂等展示即可。这也解释了为什么 custom 事件里不应该偷偷做外部写入。它适合报进度不适合承担记账、发邮件一类副作用。六、跑起来代码在src/v2_3_streaming.py。python-msrc.v2_3_streaming脚本会验证这几件事。custom 通道收到读取记忆、审批、检索、成稿和保存摘要的进度事件。updates 通道收到实际节点写入的字段增量。messages 只把 writer 的 token 输出为最终回答。图暂停后用同一个 thread 恢复成功成稿才会写入长期记忆。拒绝研究时不会新增摘要。回到一开始那个等待页面。流式输出不是把所有内部细节一股脑倒给用户而是给回答、进度和调试各开一条合适的路。用户知道助手正在做什么开发者也还能看到状态究竟在哪一步变化。可节点继续增加后流再清楚也救不了一张过大的图。下一篇要处理的是另一个更难的问题怎么把不断膨胀的研究流程拆成几个能单独理解、又能一起运行的子图。赞或收藏 关注 我们下次再见
本文是「LangGraph 教程系列」第 9 篇。写作时基于 langgraph 1.2.10、langchain 1.3.14、langchain-openai 1.4.1、Python 3.12。配套代码仓库 https://github.com/wxj006007/deep-research-assistant 本篇对应 tagv2.3。上一版的研究助手已经能停在审批点也能从长期记忆里拿到用户偏好。它看起来终于像个能交付的程序了。可真把它放到终端或网页里跑一次问题就冒出来了。用户等了半天只在最后看到一整段答案。检索开始了吗资料够不够程序是在等待人工确认还是已经卡死了完全不知道。这一篇不再给助手加新能力而是把它正在做什么交出来。不是所有东西都要逐字显示真正有用的是把不同层次的信息送到合适的地方。一、v2.2 撞的墙用户只看见终点同步调用很适合把流程讲清楚。finalgraph.invoke({question:LangGraph 的 stream_mode 有什么区别},configconfig,contextcontext,)print(final[answer])但 invoke 有一个天然取舍它只在结束时给结果。研究助手内部已经走过读取记忆、规划、审批、检索、评估、成稿这些步骤调用方却像在门外等消息。图里的状态也不能直接拿去当界面协议。state 要保存问题、资料、轮次和审批结果它是工作台。用户需要的是回答正文、进度提示和必要的错误信息。两者重合一部分但不是同一个东西。二、四条流装不同的东西v2.3 仍是同一张研究图只是在节点外包了一层进度事件。渲染错误:Mermaid 渲染失败: Parse error on line 2: flowchart LR graph[研究图] -- msg[m ----------------^ Expecting SEMI, NEWLINE, SPACE, EOF, subgraph, end, acc_title, acc_descr, acc_descr_multiline_value, AMP, COLON, STYLE, LINKSTYLE, CLASSDEF, CLASS, CLICK, DOWN, DEFAULT, NUM, COMMA, NODE_STRING, BRKT, MINUS, MULT, UNICODE_TEXT, direction_tb, direction_bt, direction_rl, direction_lr, direction_td, got GRAPHLangGraph 的流模式可以这样分工。模式传出的内容本例放到哪里values每个超级步后的完整 state调试、检查快照updates某个节点刚写入的 state 增量执行轨迹、开发调试messages模型消息块和元数据最终回答的逐 token 输出custom节点主动写出的任意业务事件前端进度提示values 很直观刚开始学图时尤其有帮助。但资料一多完整 state 会一遍遍重复。把它原样推给浏览器既浪费带宽也会把不该展示的内部字段泄露出去。所以 v2.3 的终端演示订阅 messages、updates、custom。values 保留给排障和教学不把它包装成生产前端的默认数据格式。三、进度不进 state写进 custom读取记忆和检索结果本来就会更新 state。正在读取、正在检索这样的提示只是这一轮运行的即时消息没必要跟着 checkpoint 长期保存。节点通过 get_stream_writer 写出结构化事件。fromlanggraph.configimportget_stream_writerdefemit_progress(stage:str,message:str,**details:object)-None:get_stream_writer()({type:research_progress,stage:stage,message:message,**details,})检索包装节点只增加事件不改变 v2.2 的搜索函数。defsearch_with_progress(state:MemoryResearchState)-dict[str,object]:emit_progress(search.start,正在检索资料,querystate[current_query])updatessearch_node(state)emit_progress(search.done,本轮检索完成,new_documentslen(updates[docs]),)returnupdates同样的模式用在读写长期记忆、规划、评估和成稿上。业务状态仍由原来的节点返回界面进度则独立走 custom 通道。这样做还有一个实际好处换终端、网页或日志系统时不必改图的 state 设计。四、同一次运行订阅三种模式本篇的 build_graph 入口仍然保留 v2.2 的依赖边界。defbuild_graph(checkpointer:MemorySaver|NoneNone,store:BaseStore|NoneNone,):builderStateGraph(MemoryResearchState,context_schemaResearchContext)# 注册带进度事件的包装节点returnbuilder.compile(checkpointercheckpointerorMemorySaver(),storestoreorInMemoryStore(),)消费端一次传入三个模式。LangGraph 会把每条数据组织为 mode 和 payload。formode,payloadingraph.stream(graph_input,configconfig,contextcontext,stream_mode[messages,updates,custom],):ifmodemessages:chunk,metadatapayloadelifmodeupdates:print(payload)elifmodecustom:print(payload[message])这里有个很容易被忽略的细节。planner、evaluator、writer 都会调用模型。若把所有 messages 都显示在回答区域用户会先看到查询词和评估 JSON界面就乱了。因此示例只渲染 synthesize 节点的消息。ifmetadata.get(langgraph_node)!synthesize:continuecontentgetattr(chunk,content,)ifcontent:print(str(content),end,flushTrue)这样模型仍可在每个节点按自己的方式工作用户却只会逐字看到最后那段回答。progress 区则接收检索和审批这些更适合人读的事件。五、流结束不一定代表任务完成第 7 篇讲过interrupt 会暂停图并交还控制权。放进流式调用后规则没有变化。第一次 stream 消费到带有interrupt的 updates 后迭代器就结束了。interrupted,statsconsume_stream(graph,initial_input,config,context)ifinterrupted:interrupted,statsconsume_stream(graph,Command(resume{action:approve}),config,context,)关键还是同一个 thread_id 和同一个 ResearchContext。thread 决定从哪个 checkpoint 恢复context 决定当前用户和 research_id。还有一件事要提前处理。interrupt 之后恢复时所在节点会重放。review_with_progress 在暂停前发出的 waiting 事件可能再出现一次。前端不要把它当成一条永不重复的数据库记录用 stage 和本轮执行标识做幂等展示即可。这也解释了为什么 custom 事件里不应该偷偷做外部写入。它适合报进度不适合承担记账、发邮件一类副作用。六、跑起来代码在src/v2_3_streaming.py。python-msrc.v2_3_streaming脚本会验证这几件事。custom 通道收到读取记忆、审批、检索、成稿和保存摘要的进度事件。updates 通道收到实际节点写入的字段增量。messages 只把 writer 的 token 输出为最终回答。图暂停后用同一个 thread 恢复成功成稿才会写入长期记忆。拒绝研究时不会新增摘要。回到一开始那个等待页面。流式输出不是把所有内部细节一股脑倒给用户而是给回答、进度和调试各开一条合适的路。用户知道助手正在做什么开发者也还能看到状态究竟在哪一步变化。可节点继续增加后流再清楚也救不了一张过大的图。下一篇要处理的是另一个更难的问题怎么把不断膨胀的研究流程拆成几个能单独理解、又能一起运行的子图。赞或收藏 关注 我们下次再见
)
