目录

1. 为什么需要流式输出？

2. 核心改动点：模型与 Agent 初始化

2.1. 启用 Chat Model 的 Streaming

2.2. Agent 的其他配置

3. 实现流式调用与解析

3.1. 理解流式输出的结构

4. 总结与展望

---

1. 为什么需要流式输出？

传统的 Agent invoke 调用需要等待整个推理和工具执行过程完成后，才能返回最终结果。这在高延迟或包含复杂工具调用的场景中，会导致用户长时间等待。

通过 流式输出 (stream)，我们可以实时接收模型生成的 Token，让用户感受到 Agent 正在“思考”和“回复”，极大地改善等待体验。

文本输出：https://blog.csdn.net/W_Meng_H/article/details/155595693

2. 核心改动点：模型与 Agent 初始化

要启用流式输出，我们需要在两个地方进行关键配置：

2.1. 启用 Chat Model 的 Streaming

在初始化 ChatTongyi 时，必须将 streaming 参数设置为 True。这是模型级别支持流式传输的前提。

# chat/qwen.py

# 初始化模型
tongyi_chat = ChatTongyi(
    model=MODEL_NAME,
    top_p=MODEL_TOP_P,
    dashscope_api_key=DASHSCOPE_API_KEY,
    # === 关键改动点 1：启用流式输出 ===
    streaming=True, 
)

2.2. Agent 的其他配置

Agent 的核心逻辑、工具定义 (tools/weather_tool.py) 和动态系统提示 (user_role_prompt) 保持不变，这些机制自然地集成到流式执行流程中。

3. 实现流式调用与解析

与同步调用 agent.invoke() 不同，流式调用使用 agent.stream()。这个方法返回一个迭代器，我们可以通过 for 循环实时获取数据块。

官方文档：流式传输 | LangChain 中文文档

# chat/qwen.py

# --- 第一次调用：景点推荐（流式输出） ---
print("\n--- 第一次调用: 北京景点推荐 (流式输出) ---")
print("最终AI回复内容:")

# 关键改动点 2: 使用 agent.stream()
for token, metadata in agent.stream(
    {"messages": [{"role": "user", "content": "北京景点推荐"}]},
    context={"user_role": "user"},
    stream_mode="messages", # 推荐使用 "messages" 模式获取结构化输出
):
    # token: 包含内容块（content_blocks）
    # metadata: 包含 Agent 状态（如 langgraph_node）

    print(f"node: {metadata['langgraph_node']}")
    print(f"content: {token.content_blocks}")
    print("\n")

3.1. 理解流式输出的结构

当我们使用 stream_mode="messages" 时，每次迭代会返回两个核心部分：

部分	类型	描述	关键信息
token	MessageChunk	消息内容的分块	token.content_blocks 包含实际生成的文本
metadata	dict	Agent 运行的元数据	metadata['langgraph_node'] 可以显示 Agent 当前所处的阶段（如思考、调用工具、生成答案）

通过打印 metadata['langgraph_node']，我们可以清晰地看到 Agent 的决策路径。例如：

• 如果 Agent 决定调用工具，您会先看到 tool-calling 相关的节点信息。

• 当 Agent 最终生成回复时，您会看到 final-answer 或类似的节点信息，紧随其后的是回复文本。

4. 总结与展望

通过简单的两步配置（模型 streaming=True 和调用 agent.stream())，我们成功地将基于 LangChain 和阿里通义千问构建的 Agent 升级到了流式输出版本。

在实际的 Web 或客户端应用中，我们只需要将 token.content_blocks 中的内容实时追加到屏幕上，即可实现一个高性能、用户友好的智能应用。