（十六）观察节点、消息与异步执行

1. 先看过程，再看最终结果

第十节的 stream_mode="updates" 每次给出一个节点返回的更新，适合看“刚发生了什么”；values 给出合并后的状态，适合看“现在保存了什么”；messages 关注模型逐段输出。它们观察的是不同层面。

同样是 retrieve 的一次输出，updates 中只含它返回的 hits 和 status；合并后的 values 还保留原问题、时长与 operation_id。可以分别观察这两种流，核对“节点返回了哪些字段”和“图目前保留了哪些字段”。若把 updates 的一个局部字典当成完整状态，就会误以为旧问题被删除了。

LangGraph 1.2 还提供按投影区分的事件流接口。本章沿用已经实测的 stream-mode API，以便把 updates 与节点返回值直接对应；继续扩展时可查流式输出文档比较接口。

日志至少保留任务编号、节点名、开始结束时间、成功或错误种类；模型调用可再记录用量，检索可记录来源编号。沿同一个任务号串起事件，比把所有完整问题、私人笔记和令牌都写进日志更便于排查。

2. 异步接口接到已有协程知识

LangChain 和 LangGraph 的异步入口常为 ainvoke、astream。它们要与实际异步客户端和异步持久化组件配合；同步 sqlite3、同步模型调用仍会占用当前线程。

# 这个例子使用内存中的纯 Python 节点，观察异步入口和异步迭代。
import asyncio
from typing import TypedDict
from langgraph.graph import END, START, StateGraph

class State(TypedDict):
    text: str

async def tidy(state: State):
    # 实际网络节点可以在这里 await 异步 HTTP 或模型客户端。
    return {"text": state["text"].strip()}

async def main():
    builder = StateGraph(State)
    builder.add_node("tidy", tidy)
    builder.add_edge(START, "tidy")
    builder.add_edge("tidy", END)
    graph = builder.compile()
    async for update in graph.astream({"text": "  NumPy  "}, stream_mode="updates"):
        print(update)  # 本次更新为 {"tidy": {"text": "NumPy"}}。

if __name__ == "__main__":
    asyncio.run(main())

这一段展示调用形式；它没有并发网络任务，所以 async 不会让去空格的计算变快。完整项目使用同步图与同步 SQLite，因此 FastAPI 路由也用普通 def，让框架在线程池中处理；需要扩展为异步服务时，再一起更换模型、工具和检查点的异步实现。