116.Agent-LangChain核心组件-Event事件流(Beta测试版)

摘要本文介绍 LangChain 1.3 版本对 Streaming 流式输出的重构——Event 事件流。文章首先说明事件流将原本通过 stream_mode 区分的内容整合为独立的类型投影Projection流并逐一讲解 stream.messages、stream.values、stream.tool_calls、stream.output、stream.subgraphs、stream.extensions 等常用投影的用途。随后作者因千问模型对事件流支持不佳、无法生成 tool_id改用 DeepSeek 模型并给出安装 langchain-deepseek 库及完整代码示例演示如何通过事件流获取模型消息、工具调用、最终输出与思考过程实现打字机效果与 Agent 全流程观测。内容参考于图灵AI大模型全栈Event事件流是LangChain的1.3版本对Streaming也就是流式输出的重构它把通过stream_mode区分的内容改成了一个独立的类型投影Projection流投影流如上图它把原始流Streaming投影成了不同的数据类型通过去不同的变量就可以得到对应的数据stream.messages通过stream点messages就可以看到模型的讲话信息也只能看到模型讲话的信息stream.values通过stream点values就可以看到状态State的变化也就可以看到Agent中的过程也只能看到状态的变化stream.tool_calls通过stream点tool_calls就可以看到工具的执行过程也只能看到工具的执行过程stream.output通过stream点output就可以看到Agent的最终结果也只能看到最终结果stream.subgraphs通过stream点subgraphs就可以看到子图或子Agent的运行状态也只能看到子图或子Agent的运行状态stream.extensions通过stream点extensions就可以看到我们自定义的投影也只能看到我们自定义的投影上图中有官方提供的常用投影览表可以看出它把原始流进行的整合规划我们想要什么直接点出来就行不需要再传递模式类型就是如果想看模型的讲话就stream.messages如果想看Agent的过程就stream.values就跟电视机切换频道一样之前一直使用的千问的模型但是千问模型对于事件流支持的不好没办法生成tool_id这里换成了deepseek再说一次事件流是测试版所以存在这种问题安装deepseek库uv add langchain-deepseek如下图红框LangChain的官网都有示例地址https://docs.langchain.com/oss/python/integrations/chat/deepseek地址https://docs.langchain.com/oss/python/integrations/chat效果图原始流可以看到全部的数据下图是messages的消息可以看到大模型说话的内容但是存在空内容下图是values下图是tool_calls的信息下方是messages中的内容通过messages中的text可以实现打字机效果下图就是大模型调用工具的内容注意下图蓝框里的id如果使用千问模型会是空的模型本次输出的内容模型本次的思考过程代码from langchain.agents import create_agent from langchain_deepseek import ChatDeepSeek from langchain_core.utils.uuid import uuid7 from langgraph.checkpoint.memory import InMemorySaver from dotenv import load_dotenv import os # 加载环境变量 load_dotenv() # 初始化模型 qwen模型对事件流支持不行,不能生成tool_id,换成deepseek llm ChatDeepSeek( modeldeepseek-v4-flash, api_keyos.getenv(DEEPSEEK_API_KEY), ) # 创建一个工具 def get_weather(city: str) - str: 获取指定城市的天气信息 return f{city}的天气是晴朗的! # 创建代理 agent create_agent( modelllm, tools[get_weather], system_prompt你是一个乐于助人的助手能够回答用户关于天气的问题。请使用提供的工具来获取天气信息。, checkpointerInMemorySaver(), # 使用InMemorySaver来保存对话状态这样在同一线程中进行的对话可以共享状态 ) # 创建线程id config {configurable: {thread_id: str(uuid7())}} # 使用事件流获取代理全部执行过程 stream agent.stream_events( {messages: [{role: user, content: 济南天气怎么样}]}, configconfig, versionv3, ) # 获取原始的流事件 # print(获取原始的流事件) # for event in stream: # print(event) # 获取llm返回的每条消息 for message in stream.messages: # print(获取消息的文本增量和最终文本:) # for delta in message.text: # print(delta, end, flushTrue) # print(获取模型生成工具调用时的内容) # print(message.tool_calls.get()) # print(获取模型本次输出消息:) # print(message.output) print(获取模型本次推理消息:) for delta in message.reasoning: print(f[thinking] {delta}, end, flushTrue) print() # print(------获取最终状态-------) # print(stream.output)