大模型应用开发学习 26：怎样看见任务的执行进度和流式结果

用户为 A3003 提供“尺寸不合适”的退货原因，助手开始查询、核对、准备草稿。界面却只有转圈图标：用户看不出它在查单、等待确认，还是已经失败。另一种界面不断追加文字，却把“正在核对”与最终答复拼成一段，用户取消后仍显示执行中。

模型文字逐片返回只解决部分等待体验。多步骤任务还要表达阶段、来源、暂停和失败。任务进度与模型回复可以同时出现，但不能混成一种结果。

等待与恢复已经能保存暂停位置，并从用户回答继续处理。这里沿同一个 build_flow() 观察执行过程，核心问题是：程序怎样把内部发生的事转成可理解的进度，并在等待、失败或结束时给出准确展示。不会重新实现退货规则，也不会把模型生成的文字当作业务事实。

参考代码：先看事件，再核对任务

代码位于 GitHub共享示例目录，完整仓库、固定依赖与配套包入口见 README。本次主要复用 flow.py 的退货图，新增 stream_demo.py 逐条打印事件，并观察暂停、固定取消和受控失败；不连接模型或真实电商。

按 README 准备环境后，在仓库根目录运行：

1.\.venv-series\Scripts\python.exe -X utf8 examples/23-32-assistant/stream_demo.py.\.venv-series\Scripts\python.exe -X utf8 examples/23-32-assistant/stream_demo.py

脚本使用临时业务数据库和内存检查点，结束后清理。它验证一次进程里的事件消费与暂停恢复；第25篇的跨进程恢复仍需文件检查点。脚本默认取消教学申请，不请求真实授权，也不会自动批准。

先按需要选择观察对象。下面四种是 stream_mode 的取值，不是四种业务阶段：

排错时还有 tasks、checkpoints、debug 等模式，分别用于查看任务调度、检查点与更详细的运行信息。它们帮助开发者定位问题，不等于给用户显示的业务步骤，也不宜未经筛选直接公开；当前演示主要保留上面四种视角。

先观察任务更新

查单产生事实，核对产生资格结论，准备产生草稿。updates回答刚发生什么，values回答现在已知什么。

若每次把全量状态追加到屏幕，会反复显示旧事实；只看增量却不维护当前状态，又可能丢掉此前结果。消费方式应服从展示目的，不能看到字典就直接输出。

invoke() 通常在本次调用结束时交付状态；stream() 返回可迭代的输出，消费端可以边推进边处理。流式观察不改变节点的退货规则，也不会把一个耗时函数自动拆成更多步骤。

先暂时关闭审批，只看草稿形成过程，再接回等待路径。以下调用实际配套图，approval=False 让任务在草稿准备后结束；在共享示例目录运行片段或保存为该目录的练习脚本：

1import tempfile2from pathlib import Path3from business import Business4from flow import RequestContext, build_flow5 6with tempfile.TemporaryDirectory() as folder:7    graph = build_flow(Business(Path(folder) / "business.sqlite"), approval=False)8    for mode, data in graph.stream(9        {"task_id": "observe-26", "order_id": "A3003", "reason": "尺寸不合适"},10        context=RequestContext("demo-user"),11        stream_mode=["updates", "custom"],12    ):13        print(mode, data)import tempfile
from pathlib import Path
from business import Business
from flow import RequestContext, build_flow

with tempfile.TemporaryDirectory() as folder:
    graph = build_flow(Business(Path(folder) / "business.sqlite"), approval=False)
    for mode, data in graph.stream(
        {"task_id": "observe-26", "order_id": "A3003", "reason": "尺寸不合适"},
        context=RequestContext("demo-user"),
        stream_mode=["updates", "custom"],
    ):
        print(mode, data)

这里沿用项目依赖的默认 v1 多模式形式 (mode, data)。单独传 stream_mode="updates" 时直接得到更新字典；传列表后才多出外层模式名。当前官方文档还提供显式 version="v2" 的统一字典形式，用 type、ns、data 表示事件；它不能直接套用这里的二元组解包。本文不切换项目接口版本。

先把一次输出按含义读出来，而不是按打印行数判断进度。updates可能包含lookup更新的facts，随后包含check更新的checked，最后包含prepare更新的draft。这几个对象是节点结束时交付的局部结果。

values则会让查询结果在后续状态中继续存在。读者看到同一个订单出现多次，不一定是重复查询，可能只是每次都在展示新的完整快照。区分这两者，能避免错误统计工具调用次数。

在本例中，可把输出对照为下表。表格只列关键字段，不是完整序列化结果：

updates 的外层键是节点名，内层才是该节点返回的更新。values 则已经由图按字段规则合并，通常还会先发出入口状态。第23篇为 trace 设置了追加规则，所以每个节点的单项轨迹会变成四步轨迹；消费端若用普通字典覆盖来拼接更新，会错误地只留下最后一项。需要完整状态时，直接订阅 values 或在运行后读检查点，比自行猜测合并规则更清楚。

如果某次查单只用了100毫秒，而准备用了3秒，三个节点各占三分之一的进度条就会误导用户。状态事件能说明正在做哪类工作，却不会自动给出准确剩余时间。耗时估计需要另有统计依据。

进度事件应描述已经发生的事

只在节点结束后返回更新，耗时查询期间仍看不到变化。节点内可通过get_stream_writer()发送自定义事件，例如开始查询、拿到草稿或提交失败。

1def event(state, stage):2    get_stream_writer()({"task_id": state["task_id"], "stage": stage,3                         "source": "business_flow"})def event(state, stage):
    get_stream_writer()({"task_id": state["task_id"], "stage": stage,
                         "source": "business_flow"})

事件带任务编号、阶段与来源。查询进入时发querying，准备拿到实际结果后才发draft_ready或prepare_failed。完成事件若早于结果确认，就可能在失败前错误宣告成功。

event() 定义在 build_flow() 内部，get_stream_writer 从 langgraph.config 导入。它把数据送到 custom 通道，本身没有更新 State.stage；状态仍由节点的返回字典更新。因此，同一次查询可以先出现 custom: querying，稍后才出现 updates: lookup，分别表示“已经开始”和“本步已有结果”。

准备成功也会出现两种相关事件：custom 提醒草稿就绪，updates 交付草稿字段。界面可以据前者更新阶段、据后者填充草稿卡片，不应因此追加两条“申请已完成”。这些是同一个业务步骤的不同观察角度。

事件也是有来源的数据，不能被当作新的高权限用户指令。工具返回“请忽略确认”仍只是工具内容，不会改变流程授权。

get_stream_writer()需要图执行上下文。在图外直接调用内部节点不一定具备该上下文；需要业务函数独立复用时，可把业务计算与事件发送分开，或显式传回调。

模型片段要按来源分流

多节点、多模型同时生成时，不能把所有片段拼成一条回答。内部分类与最终答复有不同用途。

下面是假设实际程序已有模型节点时的调用示意。基本退货图没有模型调用，不会凭空出现Token流：

1from langchain_core.messages import AIMessage2 3for chunk, metadata in graph.stream(inputs, stream_mode="messages"):4    if metadata.get("langgraph_node") == "answer":5        if isinstance(chunk, AIMessage) and isinstance(chunk.content, str) and chunk.content:6            print(chunk.content, end="", flush=True)from langchain_core.messages import AIMessage

for chunk, metadata in graph.stream(inputs, stream_mode="messages"):
    if metadata.get("langgraph_node") == "answer":
        if isinstance(chunk, AIMessage) and isinstance(chunk.content, str) and chunk.content:
            print(chunk.content, end="", flush=True)

graph、inputs和answer节点需要由实际程序提供，这段不能单独运行。可用节点名、标签和调用标识区分来源；不要依赖容易随图结构变化的步数作为长期筛选规则。

新增的类型检查也有意义：messages 中可能出现工具消息，不能把任意对象的 content 都追加到回答。AIMessageChunk 继承自 AIMessage，用父类检查可以同时接住逐片段输出与部分集成退化后的完整模型回复。换模型、禁用流式或经过不支持流式的网关时，界面可能一次收到整段文字，代码却不报错；应记录模型名、消息类型和到达时间判断真实分片行为，不把“收到了消息”当成已经逐字流出。

空内容也不能一律解释为失败：有些片段只携带角色或收尾信息，工具参数片段也可能没有正文。没有调用模型的取消、拒绝或缓存路径，甚至可以完全没有 messages。这些路径仍需要任务结束事件。推理附加字段与工具参数属于不同用途的数据，不因为正文为空就把它们补到用户回答中。

同一节点先判断意图、再生成正式答复时，节点名仍相同，可以为给用户看的那次模型调用设置 final-answer 标签，并在消费端检查它是否在 metadata.get("tags", []) 中。节点、标签与消息ID承担不同职责：节点说明步骤，标签说明用途，消息ID帮助把同一次回复的片段拼到一起；不能用一个节点名区分它的全部重复调用。

这里单独订阅 messages，每项是 (chunk, metadata)；若把它加入前面的多模式列表，应先拆出 (mode, data)，在 mode == "messages" 时再拆 data。messages 里的一个片段未必对应一个汉字或一个完整词，不能按界面刷新次数统计 Token。LangChain 模型集成可以将调用中的模型输出接入这条通道，实际分片还取决于模型集成与服务支持；直接使用其他 SDK 时，可能需要自行转成 custom 事件。

片段也可能是内容块、工具参数增量或其他结构。参数尚未完整时不能执行工具，一段文字显示完也不能证明申请已提交。

模型片段与业务事件还存在“可修订”和“可承诺”的差别。模型刚生成“这笔订单……”时，后续文字仍可能转成“不符合条件”；工具参数也可能分几段才接收完整。在响应完成并通过必要校验前，不能仅根据前几个字提前触发业务操作。

界面可以为模型回复维护一个临时缓冲区，按消息或调用标识拼接；业务进度维护另一份任务状态。等待确认时停止执行中的动画，并显示具体待处理项。这样文本继续生成、任务暂时挂起和业务已经完成不会互相覆盖。

对用户不可见的中间分类也可能产生Token事件。过滤节点只是最小方法，实际图中同一节点多次调用模型时还需要标签或调用身份。把所有模型事件直接转发到用户，可能意外展示内部草稿或与当前任务无关的内容。

并行与子流程需要关联信息

同一节点可能处理多笔订单，或在重试中多次执行。只有节点名仍不足以确定当前片段属于哪一项。应保留任务编号、子任务位置与本次执行身份；断线重连还需要稳定事件ID或序号。

包含子图时，可用subgraphs=True观察命名空间，分清父流程和子流程。但命名空间不替代业务关联，父子状态映射也要核对，避免把子助手全部工具轨迹直接交给用户。

把一个节点拆成子图以后，业务最终结果可能仍然正确，但子图内部的进度不一定仍出现在原来的订阅范围。要观察内部事件，应显式启用 subgraphs=True 并按当前版本核对输出形状。默认 v1 形式有四种组合：

主图的命名空间通常为 ()，子图事件带调用实例路径。给原来的二元组循环加 subgraphs=True，不能仍照旧解包；单模式子图的二元组和多模式主图的二元组看起来一样，字段含义却不同。需要统一形状时，可选择已核查文档提供的 version="v2"，按 type、ns、data 读取；v2的中断另在 values 事件的 interrupts 字段中读取，不能继续套用v1特殊键判断。

若需要观察模型、工具与 Runnable 的调用起止，可使用更细的 astream_events()。其事件用 event、name 分派，data 随开始、片段、结束而变化，run_id 与 parent_ids 关联调用，tags 与 metadata 帮助筛选。include_types、include_names、include_tags 分别按组件类型、名称、用途缩小范围。它的事件协议 version="v2" 与 stream(version="v2") 的返回包装不是同一件事。

例如需要的是“工具实际开始执行时，已解析的参数是什么”，应关注工具开始事件；工具参数的生成片段尚未完整时，只能做过程展示，不能据它提前执行。这里只说明选择依据，未把异步调用事件集成进当前命令行脚本。也不要为同时看任务进度和工具调用，再独立执行一遍图：两次流调用通常就是两次执行。能由工具通过 custom 汇报的信息，优先复用同一次运行。

本例最小事件只有三个字段，没有实现完整持久事件重放。若用于网页重连，需要补充事件账本、最后确认位置与去重规则。存在循环时，仅以task_id + stage去重，会错误丢掉第二轮的真实进度，应加入业务版本或尝试标识。

等待、失败与完成分别表达

流结束不一定代表业务完成。人工确认暂停后，本次调用可返回，业务仍等待用户；客户端断开也不代表服务端已取消。

最小消费端需要维护任务状态。Assistant.show从检查点提取中断，命令行据此展示追问或草稿。意外异常还要由适配层转成失败展示；画一条归档边不会让异常自动进入归档节点。

先走完等待，再观察取消。stream_demo.py 的 approval_demo() 启用审批并使用 InMemorySaver，为本次图调用传入固定 thread_id。事件消费者的实际实现如下；其中 context 是 RequestContext("demo-user")，inputs 可以是初始字段，也可以是恢复用的 Command：

1def consume(graph, inputs, config, *, context=None):2    """逐条打印，不先用list收完整条流；最后由调用方核对快照。"""3    for mode, data in graph.stream(4        inputs, config, context=context, stream_mode=["updates", "custom"]5    ):6        print(json.dumps({"mode": mode, "data": data}, ensure_ascii=False, default=str), flush=True)def consume(graph, inputs, config, *, context=None):
    """逐条打印，不先用list收完整条流；最后由调用方核对快照。"""
    for mode, data in graph.stream(
        inputs, config, context=context, stream_mode=["updates", "custom"]
    ):
        print(json.dumps({"mode": mode, "data": data}, ensure_ascii=False, default=str), flush=True)

json 在脚本顶部导入，flush=True 让控制台及时显示。这里输出完整数据只为教学观察，不能原样作为网页公开接口。default=str 会把中断对象打印成字符串供人阅读，不能据此解析协议或恢复任务；恢复所需的结构化内容仍来自检查点快照。

首次流中先后能看到查询开始、查询与核对更新、草稿就绪，然后出现带 __interrupt__ 的 updates。它是框架中断信息，不是名为 __interrupt__ 的业务节点。此时 review 尚未正常返回，不能期待已有 review.decision。

流耗尽后，脚本通过 graph.get_state(config) 取回中断，核对 snapshot.next 为 ("review",)、中断类型为 approval、申请数为0，再显示“等待确认”。随后固定传入 Command(resume={"action": "cancel", "draft_id": 当前草稿编号})，只为演示取消路径。恢复值通过审批校验后，才出现 review 的 decision="cancel" 更新，以及 cancelled 事件与最终状态。

若换成批准，显示内容与业务结果仍遵守第25篇约束：有效草稿才能提交，成功只是创建待审核申请，过期则失败。当前独立脚本固定走取消，不把两种路径都说成本次已演示。

再看一个独立故障实验：节点先发自定义“正在核对”，再抛受控错误。消费者必须结束执行中状态，不能留下成功标记。stream_demo.py 的 fault_demo() 直接运行这一部分，不依赖 Assistant 的 SQLite 检查点集成；既有 experiments.py stream 也保留同类演示。

1def fail(state):2    get_stream_writer()({"task_id": state["task_id"], "stage": "checking"})3    raise RuntimeError("受控教学故障")def fail(state):
    get_stream_writer()({"task_id": state["task_id"], "stage": "checking"})
    raise RuntimeError("受控教学故障")

FaultState 只保存 task_id，独立图从 START 走到这个 fail 节点；虽然结构上连向 END，异常仍会中断本次调用。外层只捕获演示预定的 RuntimeError，再打印 application_error。后者是程序自己生成的失败展示记录，不是 LangGraph 的 stream_mode，也不表示 State.stage 已被写成 failed。故障图没有成功完成的节点更新或业务申请，不能从最后一条 checking 推断成功。

把内部事件翻译成产品含义

内部状态包含归属、参数和调试轨迹，不宜全部公开。用户需要的可能只有“已查到订单”“等待确认”“申请已创建”。敏感细节留在有权限的记录里。

百分比也要有依据。动态任务可能循环或追问，走过两个节点不等于完成50%。总量已知时显示“已完成2/5项”；总量未知时展示阶段更准确。

控制台、网页和其他客户端可以使用不同传输，但来源、关联与状态语义应一致。慢客户端、背压、连接保活和重放属于传输适配问题，不需要混入图状态的基本解释。

走一遍“先有进度，后失败”的消费过程。演示首先发出checking，消费端应显示核对中；节点随后抛出受控异常，外层捕获并产生application_error，界面转为失败。此时没有成功的业务结果，不能把最后收到的进度当成最终状态。

如果只写一个for循环追加消息，异常会打断循环，而原来“核对中”的提示可能永远留着。消费端需要在成功、暂停、异常和连接中断时分别收束状态，而不是只在正常结尾添加“完成”。

连接中断尤其不同于节点失败。服务器可能继续执行，也可能已被取消。客户端重新连接时，应通过任务查询取得当前事实，再决定是否继续读事件；不要简单从头发一个新申请来恢复页面。

当系统需要可靠重放时，可保存事件编号与最后消费位置，采用“可能重复投递、消费端去重”的约定。事件持久化又会带来保留期限与存储成本。最小本地程序不必立即实现整个平台，但正文应清楚说明哪一步由谁负责，而不是让读者误以为stream()已经包含所有能力。

把原始流收成一份稳定的展示协议

打印 (mode, data) 可以帮助开发者调试，但网页不应该认识所有节点和内部字段。配套 stream_demo.py 新增 to_events()，把当前无模型、无子图的退货图翻译成六类事件：

对于被完整消费的本地调用，paused、done、error 恰好产生一个。done 只表示本次执行正常收束，取消也会产生 done，不能翻译成“申请成功”。断线时客户端可能收不到终止事件，仍要按任务查询恢复界面，不把这一约定当作网络送达保证。

to_events() 把 try 放在整个 for 循环外，因为图执行与异常是在迭代时发生。失败分支产生 error 后立即 return，避免后面再发 done。它只捕获当前实验使用的 RuntimeError；接真实服务时，其他已定义运行错误也要由适配层统一处理，不能把这段教学实现称作通用错误处理器。

在共享示例目录执行：

1python -X utf8 stream_demo.py --events2python -X utf8 stream_demo.py --ssepython -X utf8 stream_demo.py --events
python -X utf8 stream_demo.py --sse

第一条先输出草稿流程，以 interrupt → paused 收尾；随后固定取消，以 done 收尾，状态为 cancelled。独立故障图则是 stage: checking → error，没有 done。第二条执行同样的机制，只把每个事件序列化为 SSE（Server-Sent Events，服务端向客户端持续发送事件的一种格式）报文：

1event: paused2data: {"type": "paused", "task_id": "events-26"}3 event: paused
data: {"type": "paused", "task_id": "events-26"}

这里的空行是报文分隔，data 使用不带缩进的单行 JSON。脚本只打印报文，没有启动 HTTP 服务、连接浏览器或实现断线重放。真正接网页时还要配置对应响应类型并验证传输缓冲，不能把控制台逐条输出等同于浏览器已经实时收到。

若后续接入模型与检索，可在翻译层增加带来源的 token 和引用事件，不能把完整内部状态直接推给用户。当前程序未接真实模型，因而没有用固定字符串伪造 Token 流；新增接口也不声称已经实现模型消息过滤、子图适配或持久事件账本。

恢复时要另外处理进度重发：中断前的 writer() 会随节点重新进入而再次调用。去重信息若需要跨调用生效，就必须比单次消费循环活得更久。可按任务、阶段、草稿版本和尝试编号关联同一事件；改稿的第二轮进度必须保留，而恢复时同一版本的重复提示可合并。不要对文字本身去重，重复的字可能本来就是答案的一部分。现有脚本不发送审批节点内部的重复进度，也未实现跨进程去重存储。

用过程定位问题，用记录确认结果

既有 experiments.py stream 先用 list() 收集输出，再统一打印，便于保留实验报告；它不演示用户边执行边看到变化。新增 stream_demo.py 在迭代过程中逐条打印，能直接观察消费顺序。若网页服务先把整条流收完再发送，底层即使使用 stream()，用户仍要等到最后才看到内容。

本轮实际使用 Python 3.14.6、LangGraph 1.2.5、langgraph-checkpoint 4.1.1 运行独立脚本，核对等待时申请数0、取消后仍为0，以及 checking 后转为 application_error。另用同一退货图订阅 updates、values、custom、messages，得到四个节点更新与五个状态快照，最后为 draft_ready，模型事件为0，申请数为0。这里核对的是本地控制流，没有重新验证 README 锁定的完整依赖环境。

独立脚本不依赖 SQLite 检查点集成，因此本次内存暂停恢复可以运行；这不消除第25篇所记录的持久恢复环境限制。未调用真实模型、重跑完整共享测试或实现网页重连、事件账本及最终发布渲染。

补充运行 --events 与 --sse，核对等待以 paused 结束、取消后以 done 结束、故障以 error 结束且没有额外 done；SSE报文可按空行分隔并解析单行JSON。这里验证报文与本地终止事件，不代表HTTP传输、真实模型退化、标签过滤或子图流已完成集成测试。

从 A3003 和退货原因开始，查询事件说明程序已开始工作，节点更新交付事实与草稿，中断记录让界面转入等待；取消后有明确结束结果，受控失败也不会被半截进度掩盖。模型文字另按来源进入回复区域，不参与推断申请是否提交。

进度让等待可解释，也让排错有线索。把过程消息、当前快照和最终业务记录分别核对，用户才能知道助手正在做什么、是否需要自己回答，以及实际完成了什么。