langgraph agent流式加streamlit实现流式输出教程

在传统的AI应用开发中，用户点击“发送”后面对的是一个漫长的空白期。这种“黑盒等待”不仅消磨耐心，更掩盖了模型思考的过程。当大语言模型需要调用工具、检索知识库或进行多步推理时，几秒钟甚至十几秒的静默是常态。对于开发者而言，这不仅是体验问题，更是架构设计的缺陷。

真正的智能交互应当像人与人对话一样自然：有停顿、有思考、有即时反馈。流式输出（Streaming）并非简单的技术优化，而是重塑用户信任的关键交互范式。 本文将跳出参数配置的琐碎细节，从实际交互场景出发，探讨如何利用 LangGraph 的状态编排能力与 Streamlit 的前端渲染优势，构建一个能够实时“吐露心声”的 AI Agent。

同步阻塞的困境：为什么传统回调不够用

许多初学者的做法是在后端完整生成回答后，一次性返回给前端。在简单问答中这尚可接受，但在涉及 LangGraph 的多节点工作流中，问题会被放大。假设一个 Agent 需要先搜索新闻，再总结观点，最后生成回复。如果采用同步方式，用户必须等待所有步骤完成才能看到第一个字。

更糟糕的是，如果中间某个环节出错或耗时过长，用户无法感知进度，只能怀疑系统是否卡死。传统的 HTTP 请求-响应模式在这种长链路任务中显得笨重且低效。我们需要一种机制，让数据像水流一样，一旦产生就立即推送到界面，而不是蓄满水库再开闸。

同步模式下用户需等待全部生成完毕，流式模式下首字即刻可见

LangGraph 的状态流：将思考过程拆解为可观测节点

LangGraph 的核心价值在于它将 LLM 的调用封装为有向图中的节点。每个节点代表一个具体的动作，如“检索”、“推理”或“工具调用”。要实现流式输出，首先需要在图执行层面支持增量数据的获取。

在 LangGraph 中，astream_events 方法是实现这一目标的关键。它允许开发者订阅图执行过程中的各类事件，包括节点的启动、结束以及 LLM Token 的生成。不同于普通的 invoke 方法返回最终状态，astream_events 能够捕捉到模型内部生成的每一个 token。

在这种架构下，Agent 不再是单一的函数调用，而是一个可观测的状态机。我们可以区分哪些事件是“中间思考”，哪些是“最终回答”。例如，当 Agent 调用搜索工具时，前端可以显示“正在搜索...”，而当 LLM 开始生成文本时，前端则开始逐字渲染内容。这种细粒度的控制，使得复杂工作流的透明度大幅提升。

Streamlit 的实时渲染：打通前后端的数据管道

有了后端的事件流，前端的接收与渲染同样重要。Streamlit 虽然以简洁著称，但其默认的组件更新机制是基于脚本重新运行的。要实现流畅的流式效果，必须巧妙利用 st.empty() 和会话状态（Session State）。

常见的错误做法是在循环中直接调用 st.write()，这会导致页面不断刷新或追加混乱的布局。正确的策略是创建一个占位符容器，随着数据的流入，动态更新该容器内的内容。对于 Markdown 格式的流式输出，还需要处理不完整标签带来的渲染闪烁问题。

LangGraph后端事件流通过异步生成器传输至Streamlit前端占位符

在代码实现上，我们通常使用异步生成器来桥接 LangGraph 的事件流和 Streamlit 的 UI 更新。每当接收到一个新的 token 事件，就将其拼接到当前缓冲区，并刷新占位符的显示。这种“增量更新”而非“全量替换”的策略，是保证界面流畅不卡顿的核心。

import asyncio
import streamlit as st
from langgraph_client import AsyncClient

async def run_agent_stream(prompt):
    # 模拟 LangGraph 客户端调用
    client = AsyncClient()
    async for event in client.astream_events({"messages": [{"role": "user", "content": prompt}]}):
        if event['event'] == 'on_chat_model_stream':
            yield event['data']['chunk'].content

st.title("流式 AI 助手")

if prompt := st.chat_input("请输入问题"):
    with st.chat_message("user"):
        st.markdown(prompt)
    
    with st.chat_message("assistant"):
        message_placeholder = st.empty()
        full_response = ""
        
        async for chunk in run_agent_stream(prompt):
            full_response += chunk
            # 实时更新占位符，模拟打字机效果
            message_placeholder.markdown(full_response + "▌")
        
        message_placeholder.markdown(full_response)

上述代码展示了基本的骨架。关键在于 message_placeholder.markdown() 的频繁调用并未造成页面重载，因为 Streamlit 在现代浏览器中通过 WebSocket 或轮询机制高效地更新了 DOM 节点。注意，实际生产中需要处理网络异常和中断情况，确保用户体验的鲁棒性。

边界与权衡：流式并非万能解药

尽管流式输出提升了感知速度，但它也带来了新的复杂性。首先是调试难度的增加。当数据以碎片化形式流动时，追踪某个特定错误变得困难，尤其是当错误隐藏在某个中间节点的日志中而非最终输出里时。

其次，对于需要严格格式输出的场景（如 JSON 结构化数据），流式展示可能并不合适。用户更需要的是完整的、校验通过的数据块，而不是半成品的 JSON 片段。在这种情况下，混合模式可能是更好的选择：在思考阶段显示流式日志，在最终结果阶段一次性展示结构化内容。

思考过程以折叠日志形式呈现，最终结构化结果清晰展示

此外，LangGraph 的事件类型繁多，并非所有事件都适合推送给前端。过滤无关的系统事件、仅展示对用户有价值的信息，是设计良好 UX 的必要步骤。过度透明的“裸奔”式日志输出，反而会增加用户的认知负担。

回到最初的取舍：我们牺牲了一定的架构简单性和调试便利性，换来了用户交互的即时感和掌控感。对于面向最终用户的 AI 产品，这种交换通常是值得的；而对于内部数据处理管道，同步阻塞或许依然更高效。

构建流式 Agent 不仅仅是技术的堆叠，更是对“人机协作节奏”的重新定义。当机器学会像人一样“边想边说”，冰冷的算法便多了几分温度。在未来的 AI 应用中，等待将被消除，取而代之的是持续的、流动的对话流。