AI Agent自学(3):Persistence and Streaming

我爱免费

作者：微信文章
祝大家马年快乐！我衷心的祝愿真诚热爱生活的大家马上暴富！又是姗姗来迟的更新，这短短的十天内，新的前沿技术又纷至沓来。不知道大家有没有尝试Seeddance 2.0，生成视频的效果越来越真假难辨了,如果早几天有这个技术，部门年会中场过渡环节就可以轻而易举的增加视频环节的新花样了，大家可以感受一下Seeddance 官网上的披萨店案例：这个画面细节、镜头调度以及长度十五秒的视频一致性确实颠覆了文生视频的现有格局，去拜年时一个长辈甚至问我知不知道这个Seeddance怎么用，该怎么充钱，和deepseek有什么区别？我虽然知道一些基础概念，交流结束后还是觉得非常惭愧，身为年轻人的我求知欲应该更强一点，应该更主动的学习前沿事物并且总结和分享！我后续要增加一个栏目，分享更多AI新产品的使用指南和体验！马年来了！除了文生视频，春晚机器人的轮番轰炸也让人害怕，我有好多需要学习渴望分享的内容！这个公众号一定要频繁更新高质量内容！回到上面的视频一致性，我们的AI Agent概念AI Agent自学（2）：Agentic Search下一步也正好到Persistence(持久化）与Streaming(流式传播）的内容。我们来看看AI Agent是怎么解决金鱼般七秒钟的记忆，并且能把自己的思考过程流畅完整的表达出来的。

---

01. 问题定位假如你打开一个网页端的Deepseek或者豆包，没有登录的情况下，你的问题在刷新后就会重置（缺乏持久性Persistence），同时它在回答问题的过程中，你完全处于盲盒的等待状态（缺乏流式显示Streaming）。这个过程人们对于中间的状态处于无法掌控的状态，倘若你想回到过去的某一步，也是完全做不到的。那这会带来什么问题呢？假如你是一个AI工程师,你希望通过Agent反复调试来实现真正满足你需求的代码，那这可能需要Agent实现多步节点。举个例子，我们需要做数据分析->特征提取->模型构建->可视化等共十个节点任务，倘若前7个节点都能正常解决问题，8-10节点出现问题，如果我们不能回到第7个节点后的那个状态，我们将不得不重启整个任务（Token在燃烧，我们会浪费大量的时间以及金钱）。

所以我们需要为工作流加入Checkpoints，只要有问题我们可以读档重来，就像你用豆包一样你可以回到上一步的聊天编辑其中的内容，然后等待新的内容生成。同样我们可以用Streaming的方式来看看大模型到底是怎么调用工具，怎么思考并整理成最终呈现给我们的内容的。

---

02. Persistence细节
在LangGraph里实现Persistence并不困难，也即是开启记忆功能，只需要引入一个Checkpointer。在DeepLearning开设的AI Agents in LangGraph（https://learn.deeplearning.ai/courses/ai-agents-in-langgraph/lesson/nj9ok/persistence-and-streaming）中,用一个简单的天气查询功能，以gpt-4o作为模型的案例，简单的实现了整个过程。
简单来说，这个例子引入了一个叫 SqliteSaver的库。

底层逻辑：每当 Agent 运行完流水线上的一个节点（Node），LangGraph 都会自动触发一个“写操作”，把当前的 State（包括对话记录、变量等）序列化成数据，存入数据库。

结果体现：这就好比你在打单机游戏时，每过一个关卡系统自动帮你 Auto-save。

这样做，你每个节点的信息就都可以轻松地读取，并且及时“回档”。不过这又诞生了第二个问题，如果我同时有好几个复杂问题在进行，我不希望他们互相干扰，单纯的Checkpointer似乎不能适用于这种完全独立的任务环境，应该怎么解决呢？
这其实是很经典的多进程问题！就跟打游戏一样，你短时间玩累了，希望开一个新角色档，那新的档就是一个新的进程（Thread），这是完全独立的，里面就有崭新的任务过程等等。在LangGraph中，我们是通过传入参数来控制相应进程的写入和读取的。

原理：如果你想使用进程管理，你调用 graph.invoke 时，必须传一个 config={"configurable": {"thread_id": "1"}}。

深入理解：数据库里有一张表，主键就是 thread_id + checkpoint_id。


如果你换个 thread_id，Agent 就会开启一段全新的、失忆的对话。

如果你沿用旧的 thread_id，LangGraph 会先去数据库里执行“读操作”，把最后一次保存的状态反序列化，装载进内存，Agent 瞬间“回魂”。

其实很简单对吧，一个thread_id和checkpoint_id就可以解决进程和状态的同时管理，你可以轻松定位关键点。不过，我还想知道每一步做了什么或者每一步新做了什么，LangGraph里该怎么做？

---

03. Streaming细节

在普通的 ChatGPT 网页里，流式输出只是为了视觉上的快——字一个一个蹦出来。 但在 LangGraph 的 Agent 架构中，流是为了逻辑上的透明。因为 Agent 内部会有很多“黑盒”动作：查数据库、调 Python 脚本、甚至在多个 AI 之间传球。



在 LangGraph 中，Streaming 不仅仅是让文字一个一个跳出来，它是为了让工程师能实时监控状态机的内部状态，这其实已经封装到了graph.stream函数中，根据 DeepLearning.ai 的课程，我们主要使用以下三种模式：

1. stream_mode="values"（数值流）：全量快照

原理：每当一个 Node 运行结束，它会推送当前 State 的完整副本。

代码示例：

# 每次循环，v 都会包含整个对话历史for v in graph.stream(inputs, config, stream_mode="values"):    print(v["messages"][-1]) # 我们通常只取最后一条来看

结果体现：你会看到 messages 列表在不断变长。就像拍电影，每一帧你都能看到整张画面的所有角色。

2. stream_mode="updates"（更新流）：增量更新

原理：它不给你看全貌，只推送刚刚那个节点产生了什么变化。

代码示例：

# v 的结构是 {节点名: {该节点输出的增量数据}}for v in graph.stream(inputs, config, stream_mode="updates"):    print(v)

结果体现：输出类似于 {'agent': {'messages': [...]}}。这对于精准调试极度重要：如果第 8 步错了，你一眼就能从这行 update 里抓出是哪个 Node 吐出了脏数据，而不需要去几千字的完整 State 里翻找。

3. astream_events（异步事件流）：X 光机级监控

这是最底层的监控，它甚至不以 Node 为单位，而是以原子动作为单位。

代码示例：

# 我们可以订阅极其细微的事件async for event in graph.astream_events(inputs, config, version="v1"):    kind = event["event"]    if kind == "on_chat_model_stream":        # 这里实现像 ChatGPT 那样的打字机吐字效果        print(event["data"]["chunk"].content, end="|")    elif kind == "on_tool_start":        # 实时监控：Agent 这一秒正要去调用哪个工具        print(f"正在调用工具: {event['name']}")

原理深度：这是基于异步钩子（Hooks）实现的。它让你能在大模型“思考中”就切入进去。

结果体现：


on_tool_start：看到 Agent 构造的工具参数，参数错了你可以直接预警。

on_chat_model_stream：实现极致的 UI 交互，让用户看到 Agent “正在打草稿”。

通过这样的方式，我们就可以做到实时查看对应LLM的所作所为，对输出内容进行一些过滤（纯输入输出是json格式，人会觉得信息太过繁杂）。最终就可以看到一个很不错的结果。

下面是通过tavily查询San Francisco的温度等信息过滤后的结果。messages = [HumanMessage(content="What is the weather in SF?")]thread = {"configurable": {"thread_id": "4"}}async for event in abot.graph.astream_events({"messages": messages}, thread, version="v1"):    kind = event["event"]    if kind == "on_chat_model_stream":        content = event["data"]["chunk"].content        if content:            # Empty content in the context of OpenAI means            # that the model is asking for a tool to be invoked.            # So we only print non-empty content            print(content, end="|")



我们基本上了解LangGraph的相关内容，之后的推文我们要开始进行一些真实情况的应用了！我觉得分享有用的内容很有意思！除此之外我会很快分享Kaggle、Deepdance等各种各样的内容！

---

坚持新内容学习与总结复盘！专题内容：AI Agent自学（1）：LangGraph 与循环AI Agent自学（2）：Agentic Search