1. 从“黑盒”到“白盒”为什么我们需要Token级流式输出如果你用过ChatGPT的网页版一定对那种逐字蹦出来的回答方式不陌生。这种体验在技术层面我们称之为“流式输出”Streaming。但很多时候我们调用的API接口返回的只是一个完整的字符串就像你发一条短信对方要等全部编辑完才发给你中间你只能干等着。这种“非流式”的体验在构建需要实时交互的应用时简直是灾难。而“Token级流式输出”则是将这种体验做到了极致。它不仅仅是“流式”更是“细粒度”的流式。这里的“Token”可以简单理解为语言模型处理文本的最小单位可能是一个词、一个字甚至是一个标点符号。astream和astream_events这两个方法就是实现这种极致体验的关键工具。它们让AI的“思考”过程对你变得透明你不再需要等待一个漫长的生成过程结束而是可以像看一个打字高手现场创作一样实时看到每一个“字”是如何被敲出来的。这不仅仅是用户体验的提升。在AI面试、实时对话助手、代码补全、内容创作等场景中这种能力至关重要。想象一下在一个模拟面试中面试官AI的问题是一个词一个词显示出来的这能极大地增加临场感和互动性或者当你让AI写一篇长文时你能实时看到行文逻辑一旦发现跑偏可以立即中断或引导这大大提升了可控性。因此掌握astream和astream_events意味着你掌握了构建下一代高交互性AI应用的核心钥匙。2. 核心概念拆解astream 与 astream_events 到底有何不同很多初学者甚至一些有经验的开发者在面对astream和astream_events时都会感到困惑它们不都是流式输出吗有什么区别选哪个这里我们必须彻底讲清楚因为选择错误可能会让你的应用逻辑变得复杂甚至无法实现预期功能。简单来说astream提供的是“内容流”而astream_events提供的是“事件流”。这是两种不同维度的“透明化”。2.1 astream专注于最终输出的“词流”astream方法的设计哲学非常直接我给你模型生成的最终文本内容并且是一个Token一个Token地给。你订阅这个流就像打开了一个水龙头文本内容像水一样涓涓流出。它的核心特点是输出物是纯文本你收到的是str类型的Token。粒度是Token级你可以在每个Token生成后立即处理它。上下文简单你只知道“来了一个新词”但不知道这个词是来自模型的思考、工具调用还是其他环节。典型使用场景当你只关心模型最终生成的回答内容并且需要实时显示时。例如构建一个仿ChatGPT的聊天界面或者一个实时翻译工具。你只需要把收到的Token不断追加到前端页面上即可。一个极简的伪代码示例async for token in chain.astream({input: 请介绍一下你自己}): print(token, end, flushTrue) # 模拟实时打印你会看到“你”、“好”、“”、“我”、“是”……这样一个字一个字地出现。2.2 astream_events洞察LLM应用内部运行的“事件总线”astream_events则强大得多也复杂得多。它不仅仅流式输出最终文本而是将整个LangChain或其他框架执行过程“解剖”开来把其中发生的每一个重要事件都推送给你。你可以把它想象成给你的AI应用装了一个“飞行数据记录仪”黑匣子或者“调试器”。它告诉你现在执行到哪一步了例如开始运行链chain开始调用大模型llm开始调用工具tool这一步的输入和输出是什么例如发给模型的提示词prompt是什么模型返回的原始响应response是什么最终内容是如何被组合出来的它的输出不是一个简单的字符串而是一个个结构化的事件对象。每个事件都包含event事件类型、name组件名、data数据等字段。核心事件类型通常包括on_chain_start/end: 链开始/结束。on_llm_start/end: 语言模型调用开始/结束。在end事件中你能拿到模型的原始响应可能包含推理过程。on_tool_start/end: 工具调用开始/结束。on_parser_start/end: 输出解析开始/结束。on_chat_model_stream: 聊天模型流式输出Token这是获取Token级内容的关键事件。典型使用场景高级调试与监控你需要知道一个复杂的AI链到底在哪一步出错了耗时在哪。构建复杂交互界面例如你想在UI上区分显示“模型在思考”、“模型在调用搜索引擎”、“模型在输出答案”等不同状态并展示相应内容。实现特定中间过程拦截比如在模型调用工具前你需要人工确认或者在模型生成过程中你想根据已生成的内容动态修改后续提示词。关键区别对比表特性astreamastream_events输出内容最终文本的Token字符串结构化事件对象字典信息粒度仅最终输出内容全链路生命周期事件复杂度低易于使用高需要理解事件体系控制力弱只能接收内容极强可感知和控制几乎所有环节主要用途简单的内容流式展示调试、监控、构建复杂交互应用选择建议如果你的需求只是“把AI的回答一个字一个字打出来”用astream就够了简单高效。如果你需要“看清AI大脑里的每一个念头和动作”或者要构建有状态、多步骤的复杂应用那么astream_events是你的不二之选。3. 实战演练从零构建一个带流式输出的AI面试模拟器理论说再多不如一行代码。让我们以一个“AI面试模拟器”的场景来实战演练如何运用这两种流式模式。假设我们的模拟器会流式地提出问题并在面试者回答后流式地给出评价。3.1 基础环境搭建与链的构建首先我们需要一个能进行多轮对话、且有明确流程的链。这里我们使用LangGraph来构建一个简单的有状态工作流它比简单的LCEL链更适合多轮对话场景。# 环境准备安装必要库 # pip install langchain langchain-openai langgraph import os from typing import Dict, TypedDict, Annotated, List from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, AIMessage, SystemMessage # 定义状态。我们使用LangGraph推荐的“消息列表”方式来维护对话历史。 class State(TypedDict): messages: Annotated[List, add_messages] # 自动追加消息的列表 interview_phase: str # 记录面试阶段如“greeting”, “q1”, “feedback1” # 初始化大模型。使用GPT-4o-mini兼顾效果与成本并确保其支持流式输出。 llm ChatOpenAI(modelgpt-4o-mini, streamingTrue, temperature0.7) # 1. 定义节点面试官提问节点 def interviewer_node(state: State): system_prompt 你是一位专业的资深技术面试官正在对候选人进行Python开发工程师的面试。 当前面试阶段是{phase}。 请根据当前的对话历史提出一个专业、有深度的技术问题。问题应该聚焦于Python核心概念、数据结构、算法或系统设计。 你只输出问题本身不要有任何前缀或后缀。 # 根据阶段生成不同的提示词 phase state.get(interview_phase, start) if phase start: question 你好欢迎参加本次Python开发工程师的面试。我们开始第一个问题请谈谈Python中的装饰器Decorator是如何工作的并举一个你项目中实际使用的例子。 elif phase q1: question 很好。那么第二个问题在处理大规模数据时Python的生成器Generator和列表List有何本质区别为什么生成器能节省内存 else: # 默认使用LLM生成问题 messages [ SystemMessage(contentsystem_prompt.format(phasephase)), *state[“messages”][-4:] # 取最近几条历史 ] response llm.invoke(messages) question response.content # 将面试官的问题添加到消息历史 state[“messages”].append(AIMessage(contentquestion)) # 更新阶段 state[“interview_phase”] “waiting_for_answer” return state # 2. 定义节点分析答案并给出反馈的节点 def feedback_node(state: State): # 从历史中提取候选人的最后一次回答最后一条HumanMessage last_human_msg None for msg in reversed(state[“messages”]): if isinstance(msg, HumanMessage): last_human_msg msg break if not last_human_msg: return state feedback_prompt 你是一位面试官。针对候选人刚才的回答 “{answer}” 请给出简要、专业的反馈。反馈应包含1) 回答中的亮点2) 可能存在的不足或可以深入的点3) 一个改进建议。 请用流式的方式自然地输出这段反馈。 messages [ SystemMessage(contentfeedback_prompt.format(answerlast_human_msg.content)), *state[“messages”] ] # 注意这里我们直接返回一个调用LangGraph会处理流式 response llm.invoke(messages) feedback response.content state[“messages”].append(AIMessage(contentfeedback)) state[“interview_phase”] “feedback_given” return state # 3. 构建图 workflow StateGraph(State) # 添加节点 workflow.add_node(“interviewer”, interviewer_node) workflow.add_node(“feedback”, feedback_node) # 设置边和入口 workflow.set_entry_point(“interviewer”) workflow.add_edge(“interviewer”, “feedback”) workflow.add_edge(“feedback”, END) # 一轮结束实际中可以加条件边进行多轮 # 编译图 app workflow.compile()这个图定义了一个简单的两节点流程interviewer提问 -feedback给出反馈。interviewer_node在初始阶段会问一个预设问题后续可以根据状态扩展。feedback_node会分析用户的上一条消息并生成流式反馈。3.2 使用 astream 实现基础流式问答现在我们使用astream来让面试官的提问和反馈都以流式形式输出。import asyncio async def simulate_interview_with_astream(): print(“ AI面试模拟开始 (使用astream) \n”) # 初始化状态 initial_state {“messages”: [SystemMessage(content“你是技术面试官”)], “interview_phase”: “start”} # 1. 流式获取面试官问题 print(“面试官: ”, end“”, flushTrue) async for event in app.astream(initial_state, config{“configurable”: {“thread_id”: “test-1”}}): # app.astream 会返回图的每个节点的输出状态 # 我们需要从状态中提取最新的AI消息并流式输出其内容 # 注意这里为了演示astream我们简化处理直接取最终state。 # 实际上对于流式输出每个token需要更精细的控制这引出了astream_events的必要性。 pass # 上面的写法无法用astream直接实现节点内LLM调用的token级流式。 # 这是因为astream流的是图的“状态快照”而不是内部LLM的token。 # 要实现真正的、细粒度的流式必须让节点内的LLM调用本身支持流式并暴露出来。 # 这恰恰是astream力所不及而astream_events擅长的领域。 print(“\n--- 说明 ---”) print(“app.astream() 流式返回的是整个图节点的状态变更对于节点内部LLM的逐词输出它不够直接。”) print(“要实现‘面试官问题逐字出现’的效果需要将LLM的流式调用提升到节点函数层面或者使用 astream_events。”)上面的代码揭示了一个关键点对于LangGraph或复杂链astream方法通常流式输出的是整个链或图的中间状态而不是内部LLM调用的Token。要捕获内部LLM的Token流我们需要深入到事件层面。3.3 使用 astream_events 实现全链路Token级流式与状态追踪这才是重头戏。我们将改造节点函数使其内部的LLM调用支持流式并通过astream_events来捕获这些流式事件实现完美的交互体验。async def simulate_interview_with_astream_events(): print(“\n AI面试模拟开始 (使用astream_events) \n”) initial_state {“messages”: [], “interview_phase”: “start”} # 关键使用 astream_events并指定我们关心的事件 # include_names 可以过滤只关心特定节点的事件这里我们先看全部。 async for event in app.astream_events( initial_state, config{“configurable”: {“thread_id”: “test-2”}}, version“v1” # 使用稳定的事件API版本 ): event_type event[“event”] name event.get(“name”, “N/A”) data event.get(“data”, {}) # 1. 监听面试官节点结束事件获取其流式输出 if event_type “on_chain_end” and name “interviewer”: output data.get(“output”, {}) # 假设节点输出中包含了流式生成的完整问题 if “question” in output: print(f“\n[面试官-节点输出] 问题: {output[‘question’]}”) # 2. 监听LLM流式Token事件这是核心 # 当feedback节点内的LLM开始流式输出时我们会收到这个事件。 if event_type “on_chat_model_stream”: # data[‘chunk’] 是一个 AIMessageChunk 或类似对象 chunk data.get(“chunk”) if chunk and hasattr(chunk, ‘content’): token chunk.content if token: # 过滤空token print(token, end“”, flushTrue) # 这才是真正的逐词输出 # 3. 监听工具调用或其他事件本例未使用工具 elif event_type “on_tool_start”: print(f“\n[系统] AI正在调用工具: {name}”) elif event_type “on_tool_end”: print(f“\n[系统] 工具调用完成。”) # 4. 通过事件区分不同节点的LLM调用 # 事件会包含所属父节点的信息我们可以通过 event[‘tags’] 或 event[‘parent_ids’] 来区分 # 例如判断当前流式Token是来自‘interviewer’节点还是‘feedback’节点 tags event.get(“tags”, []) if “feedback_node” in tags and event_type “on_chat_model_stream”: # 可以在这里为来自feedback节点的流式内容添加前缀 pass这段代码虽然看起来复杂但它给了我们无与伦比的掌控力。on_chat_model_stream事件让我们能直接抓到LLM吐出的每一个Token。通过分析事件的tags或name我们可以精确知道这个Token是属于面试官的提问还是对回答的反馈从而在前端UI上做不同的样式渲染比如提问用蓝色反馈用绿色。3.4 整合与优化一个更完善的流式面试官节点为了让interviewer_node也支持流式提问我们需要重构它使其内部使用LLM的流式调用并将流式结果通过状态或特殊字段传递出来。这里展示一种设计思路from langchain_core.runnables import RunnableConfig from langchain_core.messages import AIMessageChunk import inspect async def streaming_interviewer_node(state: State, config: RunnableConfig): 一个支持流式提问的面试官节点 phase state.get(“interview_phase”, “start”) if phase “start”: # 预设问题也可以流式“打出来” question_text “你好欢迎参加本次Python开发工程师的面试。我们开始第一个问题请谈谈Python中的装饰器Decorator是如何工作的并举一个你项目中实际使用的例子。” # 模拟流式效果 for char in question_text: yield {“token”: char, “type”: “interview_question”} # 通过yield流式输出 await asyncio.sleep(0.05) # 控制速度 full_question question_text else: # 使用LLM生成流式问题 prompt f”基于对话历史提出下一个技术问题。历史{state[‘messages’][-3:]}” messages [HumanMessage(contentprompt)] full_question_chunks [] # 关键调用 astream 而不是 invoke async for chunk in llm.astream(messages, configconfig): if hasattr(chunk, ‘content’): token chunk.content full_question_chunks.append(token) # 同样可以yield出去 yield {“token”: token, “type”: “interview_question”} await asyncio.sleep(0.03) full_question “”.join(full_question_chunks) # 更新状态在流式结束后 new_messages state[“messages”] [AIMessage(contentfull_question)] state.update({“messages”: new_messages, “interview_phase”: “waiting_for_answer”}) # 注意在LangGraph中异步生成器节点的写法需要适配这里仅为逻辑示意。实操心得在实际开发中将astream_events与前端如WebSocket结合是常见模式。后端异步迭代astream_events一旦收到on_chat_model_stream事件就将chunk.content通过WebSocket推送到前端。前端根据事件携带的tags判断内容类型更新不同的UI区域。这样你就能构建出一个和ChatGPT官网体验相媲美甚至更强大的交互应用因为你能区分“思考中”、“调用工具中”、“输出中”等不同状态。4. 避坑指南与高级技巧流式实践中的那些“坑”流式输出听起来很美但在实际生产和复杂应用中你会遇到一系列预料之外的问题。下面是我在多个项目中趟过的坑以及对应的解决方案。4.1 坑一流式中断与连接稳定性问题描述在网络微抖动或服务器处理时间较长时客户端到服务器的流式连接如SSE或WebSocket可能超时中断导致回答显示到一半就停了。根因分析HTTP流Server-Sent Events或WebSocket连接有超时机制。如果LLM生成一个Token的时间过长例如在处理复杂推理时服务器在这段时间内没有发送任何数据客户端或代理服务器如Nginx可能会主动断开连接。解决方案发送心跳包即使在LLM思考间隙也定期从服务器向客户端发送注释行如: ping\n\n或空数据帧保持连接活跃。async def stream_with_heartbeat(chain, input_data): import asyncio async for chunk in chain.astream(input_data): if chunk: yield chunk else: # 发送心跳 yield “data: :ping\n\n” # SSE格式的心跳 # 或者设置一个后台心跳任务调整超时配置在Nginx或你的ASGI服务器如Uvicorn中显著增加proxy_read_timeout,keepalive_timeout等参数。客户端自动重连在前端实现重连逻辑当连接异常断开时尝试带着上下文重新连接并请求继续生成。这需要服务端支持“续写”功能。4.2 坑二astream_events 的事件风暴与性能问题描述一个复杂的链可能包含数十个步骤astream_events会产生大量事件。如果不加过滤处理每个事件都会消耗CPU并在网络间传输大量冗余数据可能拖慢整体响应速度。根因分析astream_events的设计目标是提供最大透明度因此默认会发出所有事件。对于生产环境很多如on_chain_start的中间事件可能并非前端所需。解决方案使用include_names或include_types进行过滤只订阅你关心的事件。# 只关心名为 ‘feedback’ 的节点事件和所有LLM流事件 async for event in app.astream_events(…, include_names[“feedback”], include_types[“on_chat_model_stream”]): # 处理事件在服务端进行聚合不要每收到一个事件就立刻向前端推送。可以稍作缓冲例如将连续的多个on_chat_model_stream事件合并为一个包含一段文本的数据包再发送减少网络请求次数。区分开发与生产模式在开发调试时启用完整事件流在生产环境则只开启必要的事件如Token流和关键错误事件。4.3 坑三上下文管理与异步迭代器生命周期问题描述在使用async for消费流时如果循环体内发生未处理的异常或者你希望提前中断流比如用户点击了“停止生成”按钮如何确保资源如数据库连接、LLM会话被正确清理根因分析流式响应是一个长时间的异步操作。如果迭代器异常退出其内部的__aexit__方法可能没有被正确调用导致资源泄露。解决方案使用try…finally或异步上下文管理器async def handle_stream(request): stream app.astream_events(…) try: async for event in stream: # 处理事件 if user_cancelled: # 用户取消 await stream.aclose() # 主动关闭流 break finally: # 确保清理工作 await cleanup_resources()利用框架的生命周期钩子如果你在使用FastAPI可以利用其BackgroundTasks或依赖项的退出逻辑来确保流关闭后的清理工作被执行。4.4 坑四Token拼接与格式处理问题描述LLM返回的Token流可能包含一些特殊格式如Markdown的代码块标记“。如果前端简单拼接可能会出现“被拆散在两段数据里导致高亮渲染失败。或者流式输出中文时一个UTF-8字符可能被拆成多个字节传输。根因分析流式传输是基于字节或Token的不保证语义完整性。网络传输和缓冲机制可能导致数据包边界出现在任何位置。解决方案前端智能拼接前端不要直接innerText token。对于可能被拆分的标记符如“、**可以设置一个小的缓冲区延迟渲染或者使用专门的Markdown流式渲染库。服务端最小化传输单元虽然以Token为单位最实时但对于中文可以考虑在服务端稍微聚合比如凑够一个完整的短句或至少一个完整字符再发送。但这会牺牲一定的实时性。使用专门协议可以考虑使用更复杂的协议如WebSocket并在消息中携带类型标记如{“type”: “text”, “data”: “…”}或{“type”: “delta”, “data”: “…”}帮助前端更精确地处理。流式输出尤其是Token级的细粒度流式是构建现代AI应用体验的基石。astream提供了简单直接的入门路径而astream_events则打开了深度控制和可观察性的大门。从简单的聊天界面到复杂的多智能体工作流调试台都离不开这两件利器。理解它们之间的差异并根据场景正确选择是每个AI应用开发者必须掌握的技能。在实际操作中耐心处理好连接、事件风暴和资源管理这些“魔鬼细节”你的应用才能真正流畅稳定。
1. 从“黑盒”到“白盒”为什么我们需要Token级流式输出如果你用过ChatGPT的网页版一定对那种逐字蹦出来的回答方式不陌生。这种体验在技术层面我们称之为“流式输出”Streaming。但很多时候我们调用的API接口返回的只是一个完整的字符串就像你发一条短信对方要等全部编辑完才发给你中间你只能干等着。这种“非流式”的体验在构建需要实时交互的应用时简直是灾难。而“Token级流式输出”则是将这种体验做到了极致。它不仅仅是“流式”更是“细粒度”的流式。这里的“Token”可以简单理解为语言模型处理文本的最小单位可能是一个词、一个字甚至是一个标点符号。astream和astream_events这两个方法就是实现这种极致体验的关键工具。它们让AI的“思考”过程对你变得透明你不再需要等待一个漫长的生成过程结束而是可以像看一个打字高手现场创作一样实时看到每一个“字”是如何被敲出来的。这不仅仅是用户体验的提升。在AI面试、实时对话助手、代码补全、内容创作等场景中这种能力至关重要。想象一下在一个模拟面试中面试官AI的问题是一个词一个词显示出来的这能极大地增加临场感和互动性或者当你让AI写一篇长文时你能实时看到行文逻辑一旦发现跑偏可以立即中断或引导这大大提升了可控性。因此掌握astream和astream_events意味着你掌握了构建下一代高交互性AI应用的核心钥匙。2. 核心概念拆解astream 与 astream_events 到底有何不同很多初学者甚至一些有经验的开发者在面对astream和astream_events时都会感到困惑它们不都是流式输出吗有什么区别选哪个这里我们必须彻底讲清楚因为选择错误可能会让你的应用逻辑变得复杂甚至无法实现预期功能。简单来说astream提供的是“内容流”而astream_events提供的是“事件流”。这是两种不同维度的“透明化”。2.1 astream专注于最终输出的“词流”astream方法的设计哲学非常直接我给你模型生成的最终文本内容并且是一个Token一个Token地给。你订阅这个流就像打开了一个水龙头文本内容像水一样涓涓流出。它的核心特点是输出物是纯文本你收到的是str类型的Token。粒度是Token级你可以在每个Token生成后立即处理它。上下文简单你只知道“来了一个新词”但不知道这个词是来自模型的思考、工具调用还是其他环节。典型使用场景当你只关心模型最终生成的回答内容并且需要实时显示时。例如构建一个仿ChatGPT的聊天界面或者一个实时翻译工具。你只需要把收到的Token不断追加到前端页面上即可。一个极简的伪代码示例async for token in chain.astream({input: 请介绍一下你自己}): print(token, end, flushTrue) # 模拟实时打印你会看到“你”、“好”、“”、“我”、“是”……这样一个字一个字地出现。2.2 astream_events洞察LLM应用内部运行的“事件总线”astream_events则强大得多也复杂得多。它不仅仅流式输出最终文本而是将整个LangChain或其他框架执行过程“解剖”开来把其中发生的每一个重要事件都推送给你。你可以把它想象成给你的AI应用装了一个“飞行数据记录仪”黑匣子或者“调试器”。它告诉你现在执行到哪一步了例如开始运行链chain开始调用大模型llm开始调用工具tool这一步的输入和输出是什么例如发给模型的提示词prompt是什么模型返回的原始响应response是什么最终内容是如何被组合出来的它的输出不是一个简单的字符串而是一个个结构化的事件对象。每个事件都包含event事件类型、name组件名、data数据等字段。核心事件类型通常包括on_chain_start/end: 链开始/结束。on_llm_start/end: 语言模型调用开始/结束。在end事件中你能拿到模型的原始响应可能包含推理过程。on_tool_start/end: 工具调用开始/结束。on_parser_start/end: 输出解析开始/结束。on_chat_model_stream: 聊天模型流式输出Token这是获取Token级内容的关键事件。典型使用场景高级调试与监控你需要知道一个复杂的AI链到底在哪一步出错了耗时在哪。构建复杂交互界面例如你想在UI上区分显示“模型在思考”、“模型在调用搜索引擎”、“模型在输出答案”等不同状态并展示相应内容。实现特定中间过程拦截比如在模型调用工具前你需要人工确认或者在模型生成过程中你想根据已生成的内容动态修改后续提示词。关键区别对比表特性astreamastream_events输出内容最终文本的Token字符串结构化事件对象字典信息粒度仅最终输出内容全链路生命周期事件复杂度低易于使用高需要理解事件体系控制力弱只能接收内容极强可感知和控制几乎所有环节主要用途简单的内容流式展示调试、监控、构建复杂交互应用选择建议如果你的需求只是“把AI的回答一个字一个字打出来”用astream就够了简单高效。如果你需要“看清AI大脑里的每一个念头和动作”或者要构建有状态、多步骤的复杂应用那么astream_events是你的不二之选。3. 实战演练从零构建一个带流式输出的AI面试模拟器理论说再多不如一行代码。让我们以一个“AI面试模拟器”的场景来实战演练如何运用这两种流式模式。假设我们的模拟器会流式地提出问题并在面试者回答后流式地给出评价。3.1 基础环境搭建与链的构建首先我们需要一个能进行多轮对话、且有明确流程的链。这里我们使用LangGraph来构建一个简单的有状态工作流它比简单的LCEL链更适合多轮对话场景。# 环境准备安装必要库 # pip install langchain langchain-openai langgraph import os from typing import Dict, TypedDict, Annotated, List from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, AIMessage, SystemMessage # 定义状态。我们使用LangGraph推荐的“消息列表”方式来维护对话历史。 class State(TypedDict): messages: Annotated[List, add_messages] # 自动追加消息的列表 interview_phase: str # 记录面试阶段如“greeting”, “q1”, “feedback1” # 初始化大模型。使用GPT-4o-mini兼顾效果与成本并确保其支持流式输出。 llm ChatOpenAI(modelgpt-4o-mini, streamingTrue, temperature0.7) # 1. 定义节点面试官提问节点 def interviewer_node(state: State): system_prompt 你是一位专业的资深技术面试官正在对候选人进行Python开发工程师的面试。 当前面试阶段是{phase}。 请根据当前的对话历史提出一个专业、有深度的技术问题。问题应该聚焦于Python核心概念、数据结构、算法或系统设计。 你只输出问题本身不要有任何前缀或后缀。 # 根据阶段生成不同的提示词 phase state.get(interview_phase, start) if phase start: question 你好欢迎参加本次Python开发工程师的面试。我们开始第一个问题请谈谈Python中的装饰器Decorator是如何工作的并举一个你项目中实际使用的例子。 elif phase q1: question 很好。那么第二个问题在处理大规模数据时Python的生成器Generator和列表List有何本质区别为什么生成器能节省内存 else: # 默认使用LLM生成问题 messages [ SystemMessage(contentsystem_prompt.format(phasephase)), *state[“messages”][-4:] # 取最近几条历史 ] response llm.invoke(messages) question response.content # 将面试官的问题添加到消息历史 state[“messages”].append(AIMessage(contentquestion)) # 更新阶段 state[“interview_phase”] “waiting_for_answer” return state # 2. 定义节点分析答案并给出反馈的节点 def feedback_node(state: State): # 从历史中提取候选人的最后一次回答最后一条HumanMessage last_human_msg None for msg in reversed(state[“messages”]): if isinstance(msg, HumanMessage): last_human_msg msg break if not last_human_msg: return state feedback_prompt 你是一位面试官。针对候选人刚才的回答 “{answer}” 请给出简要、专业的反馈。反馈应包含1) 回答中的亮点2) 可能存在的不足或可以深入的点3) 一个改进建议。 请用流式的方式自然地输出这段反馈。 messages [ SystemMessage(contentfeedback_prompt.format(answerlast_human_msg.content)), *state[“messages”] ] # 注意这里我们直接返回一个调用LangGraph会处理流式 response llm.invoke(messages) feedback response.content state[“messages”].append(AIMessage(contentfeedback)) state[“interview_phase”] “feedback_given” return state # 3. 构建图 workflow StateGraph(State) # 添加节点 workflow.add_node(“interviewer”, interviewer_node) workflow.add_node(“feedback”, feedback_node) # 设置边和入口 workflow.set_entry_point(“interviewer”) workflow.add_edge(“interviewer”, “feedback”) workflow.add_edge(“feedback”, END) # 一轮结束实际中可以加条件边进行多轮 # 编译图 app workflow.compile()这个图定义了一个简单的两节点流程interviewer提问 -feedback给出反馈。interviewer_node在初始阶段会问一个预设问题后续可以根据状态扩展。feedback_node会分析用户的上一条消息并生成流式反馈。3.2 使用 astream 实现基础流式问答现在我们使用astream来让面试官的提问和反馈都以流式形式输出。import asyncio async def simulate_interview_with_astream(): print(“ AI面试模拟开始 (使用astream) \n”) # 初始化状态 initial_state {“messages”: [SystemMessage(content“你是技术面试官”)], “interview_phase”: “start”} # 1. 流式获取面试官问题 print(“面试官: ”, end“”, flushTrue) async for event in app.astream(initial_state, config{“configurable”: {“thread_id”: “test-1”}}): # app.astream 会返回图的每个节点的输出状态 # 我们需要从状态中提取最新的AI消息并流式输出其内容 # 注意这里为了演示astream我们简化处理直接取最终state。 # 实际上对于流式输出每个token需要更精细的控制这引出了astream_events的必要性。 pass # 上面的写法无法用astream直接实现节点内LLM调用的token级流式。 # 这是因为astream流的是图的“状态快照”而不是内部LLM的token。 # 要实现真正的、细粒度的流式必须让节点内的LLM调用本身支持流式并暴露出来。 # 这恰恰是astream力所不及而astream_events擅长的领域。 print(“\n--- 说明 ---”) print(“app.astream() 流式返回的是整个图节点的状态变更对于节点内部LLM的逐词输出它不够直接。”) print(“要实现‘面试官问题逐字出现’的效果需要将LLM的流式调用提升到节点函数层面或者使用 astream_events。”)上面的代码揭示了一个关键点对于LangGraph或复杂链astream方法通常流式输出的是整个链或图的中间状态而不是内部LLM调用的Token。要捕获内部LLM的Token流我们需要深入到事件层面。3.3 使用 astream_events 实现全链路Token级流式与状态追踪这才是重头戏。我们将改造节点函数使其内部的LLM调用支持流式并通过astream_events来捕获这些流式事件实现完美的交互体验。async def simulate_interview_with_astream_events(): print(“\n AI面试模拟开始 (使用astream_events) \n”) initial_state {“messages”: [], “interview_phase”: “start”} # 关键使用 astream_events并指定我们关心的事件 # include_names 可以过滤只关心特定节点的事件这里我们先看全部。 async for event in app.astream_events( initial_state, config{“configurable”: {“thread_id”: “test-2”}}, version“v1” # 使用稳定的事件API版本 ): event_type event[“event”] name event.get(“name”, “N/A”) data event.get(“data”, {}) # 1. 监听面试官节点结束事件获取其流式输出 if event_type “on_chain_end” and name “interviewer”: output data.get(“output”, {}) # 假设节点输出中包含了流式生成的完整问题 if “question” in output: print(f“\n[面试官-节点输出] 问题: {output[‘question’]}”) # 2. 监听LLM流式Token事件这是核心 # 当feedback节点内的LLM开始流式输出时我们会收到这个事件。 if event_type “on_chat_model_stream”: # data[‘chunk’] 是一个 AIMessageChunk 或类似对象 chunk data.get(“chunk”) if chunk and hasattr(chunk, ‘content’): token chunk.content if token: # 过滤空token print(token, end“”, flushTrue) # 这才是真正的逐词输出 # 3. 监听工具调用或其他事件本例未使用工具 elif event_type “on_tool_start”: print(f“\n[系统] AI正在调用工具: {name}”) elif event_type “on_tool_end”: print(f“\n[系统] 工具调用完成。”) # 4. 通过事件区分不同节点的LLM调用 # 事件会包含所属父节点的信息我们可以通过 event[‘tags’] 或 event[‘parent_ids’] 来区分 # 例如判断当前流式Token是来自‘interviewer’节点还是‘feedback’节点 tags event.get(“tags”, []) if “feedback_node” in tags and event_type “on_chat_model_stream”: # 可以在这里为来自feedback节点的流式内容添加前缀 pass这段代码虽然看起来复杂但它给了我们无与伦比的掌控力。on_chat_model_stream事件让我们能直接抓到LLM吐出的每一个Token。通过分析事件的tags或name我们可以精确知道这个Token是属于面试官的提问还是对回答的反馈从而在前端UI上做不同的样式渲染比如提问用蓝色反馈用绿色。3.4 整合与优化一个更完善的流式面试官节点为了让interviewer_node也支持流式提问我们需要重构它使其内部使用LLM的流式调用并将流式结果通过状态或特殊字段传递出来。这里展示一种设计思路from langchain_core.runnables import RunnableConfig from langchain_core.messages import AIMessageChunk import inspect async def streaming_interviewer_node(state: State, config: RunnableConfig): 一个支持流式提问的面试官节点 phase state.get(“interview_phase”, “start”) if phase “start”: # 预设问题也可以流式“打出来” question_text “你好欢迎参加本次Python开发工程师的面试。我们开始第一个问题请谈谈Python中的装饰器Decorator是如何工作的并举一个你项目中实际使用的例子。” # 模拟流式效果 for char in question_text: yield {“token”: char, “type”: “interview_question”} # 通过yield流式输出 await asyncio.sleep(0.05) # 控制速度 full_question question_text else: # 使用LLM生成流式问题 prompt f”基于对话历史提出下一个技术问题。历史{state[‘messages’][-3:]}” messages [HumanMessage(contentprompt)] full_question_chunks [] # 关键调用 astream 而不是 invoke async for chunk in llm.astream(messages, configconfig): if hasattr(chunk, ‘content’): token chunk.content full_question_chunks.append(token) # 同样可以yield出去 yield {“token”: token, “type”: “interview_question”} await asyncio.sleep(0.03) full_question “”.join(full_question_chunks) # 更新状态在流式结束后 new_messages state[“messages”] [AIMessage(contentfull_question)] state.update({“messages”: new_messages, “interview_phase”: “waiting_for_answer”}) # 注意在LangGraph中异步生成器节点的写法需要适配这里仅为逻辑示意。实操心得在实际开发中将astream_events与前端如WebSocket结合是常见模式。后端异步迭代astream_events一旦收到on_chat_model_stream事件就将chunk.content通过WebSocket推送到前端。前端根据事件携带的tags判断内容类型更新不同的UI区域。这样你就能构建出一个和ChatGPT官网体验相媲美甚至更强大的交互应用因为你能区分“思考中”、“调用工具中”、“输出中”等不同状态。4. 避坑指南与高级技巧流式实践中的那些“坑”流式输出听起来很美但在实际生产和复杂应用中你会遇到一系列预料之外的问题。下面是我在多个项目中趟过的坑以及对应的解决方案。4.1 坑一流式中断与连接稳定性问题描述在网络微抖动或服务器处理时间较长时客户端到服务器的流式连接如SSE或WebSocket可能超时中断导致回答显示到一半就停了。根因分析HTTP流Server-Sent Events或WebSocket连接有超时机制。如果LLM生成一个Token的时间过长例如在处理复杂推理时服务器在这段时间内没有发送任何数据客户端或代理服务器如Nginx可能会主动断开连接。解决方案发送心跳包即使在LLM思考间隙也定期从服务器向客户端发送注释行如: ping\n\n或空数据帧保持连接活跃。async def stream_with_heartbeat(chain, input_data): import asyncio async for chunk in chain.astream(input_data): if chunk: yield chunk else: # 发送心跳 yield “data: :ping\n\n” # SSE格式的心跳 # 或者设置一个后台心跳任务调整超时配置在Nginx或你的ASGI服务器如Uvicorn中显著增加proxy_read_timeout,keepalive_timeout等参数。客户端自动重连在前端实现重连逻辑当连接异常断开时尝试带着上下文重新连接并请求继续生成。这需要服务端支持“续写”功能。4.2 坑二astream_events 的事件风暴与性能问题描述一个复杂的链可能包含数十个步骤astream_events会产生大量事件。如果不加过滤处理每个事件都会消耗CPU并在网络间传输大量冗余数据可能拖慢整体响应速度。根因分析astream_events的设计目标是提供最大透明度因此默认会发出所有事件。对于生产环境很多如on_chain_start的中间事件可能并非前端所需。解决方案使用include_names或include_types进行过滤只订阅你关心的事件。# 只关心名为 ‘feedback’ 的节点事件和所有LLM流事件 async for event in app.astream_events(…, include_names[“feedback”], include_types[“on_chat_model_stream”]): # 处理事件在服务端进行聚合不要每收到一个事件就立刻向前端推送。可以稍作缓冲例如将连续的多个on_chat_model_stream事件合并为一个包含一段文本的数据包再发送减少网络请求次数。区分开发与生产模式在开发调试时启用完整事件流在生产环境则只开启必要的事件如Token流和关键错误事件。4.3 坑三上下文管理与异步迭代器生命周期问题描述在使用async for消费流时如果循环体内发生未处理的异常或者你希望提前中断流比如用户点击了“停止生成”按钮如何确保资源如数据库连接、LLM会话被正确清理根因分析流式响应是一个长时间的异步操作。如果迭代器异常退出其内部的__aexit__方法可能没有被正确调用导致资源泄露。解决方案使用try…finally或异步上下文管理器async def handle_stream(request): stream app.astream_events(…) try: async for event in stream: # 处理事件 if user_cancelled: # 用户取消 await stream.aclose() # 主动关闭流 break finally: # 确保清理工作 await cleanup_resources()利用框架的生命周期钩子如果你在使用FastAPI可以利用其BackgroundTasks或依赖项的退出逻辑来确保流关闭后的清理工作被执行。4.4 坑四Token拼接与格式处理问题描述LLM返回的Token流可能包含一些特殊格式如Markdown的代码块标记“。如果前端简单拼接可能会出现“被拆散在两段数据里导致高亮渲染失败。或者流式输出中文时一个UTF-8字符可能被拆成多个字节传输。根因分析流式传输是基于字节或Token的不保证语义完整性。网络传输和缓冲机制可能导致数据包边界出现在任何位置。解决方案前端智能拼接前端不要直接innerText token。对于可能被拆分的标记符如“、**可以设置一个小的缓冲区延迟渲染或者使用专门的Markdown流式渲染库。服务端最小化传输单元虽然以Token为单位最实时但对于中文可以考虑在服务端稍微聚合比如凑够一个完整的短句或至少一个完整字符再发送。但这会牺牲一定的实时性。使用专门协议可以考虑使用更复杂的协议如WebSocket并在消息中携带类型标记如{“type”: “text”, “data”: “…”}或{“type”: “delta”, “data”: “…”}帮助前端更精确地处理。流式输出尤其是Token级的细粒度流式是构建现代AI应用体验的基石。astream提供了简单直接的入门路径而astream_events则打开了深度控制和可观察性的大门。从简单的聊天界面到复杂的多智能体工作流调试台都离不开这两件利器。理解它们之间的差异并根据场景正确选择是每个AI应用开发者必须掌握的技能。在实际操作中耐心处理好连接、事件风暴和资源管理这些“魔鬼细节”你的应用才能真正流畅稳定。
