上一章——> 【LangGraph】 流式处理入门

前言

当你的 Agent 需要执行多个内部步骤（比如搜索、计算、调用多个 API）时，用户希望看到“它在做什么”，而不是看到一个空白屏幕等待最终结果
LangGraph 的 custom 流模式允许你在工具或节点中发送任意结构化数据，从而实现思考过程的实时展示
本文将以一个带有多步骤搜索的天气工具为例，教你如何设计“进度事件”、模拟耗时操作，并最终在前端/终端实现类似 ChatGPT 的“思考中…”动态效果

---

1、为什么需要“思考过程”可视化？

在很多复杂任务中，Agent 需要调用多个工具或者一个工具内部分为多个子任务
例如：

• 搜索房产时，先查询区域房价，再筛选户型，最后计算总价

• 分析股票时，先获取历史数据，再计算技术指标，最后生成报告

• 如果不对这些内部步骤进行展示，用户会疑惑“是不是卡住了”

通过 custom 流，我们可以将每个子步骤的开始、进度、结果实时发送给客户端，大幅提升用户体验和信任度

---

2、多步骤工具的设计模式

2.1 模拟一个带步骤的搜索工具

假设我们的天气查询工具实际上需要两个子步骤：“获取天气基础信息”和“获取温度范围”
我们可以在工具内部定义一个步骤列表，循环执行，每完成一步就通过 get_stream_writer() 发送进度事件

import time
from langgraph.config import get_stream_writer

def search(runtime: ToolRuntime[Context]) -> str:
    writer = get_stream_writer()
    user_id = runtime.context.user_id
    user_name = runtime.state["user_name"]
    
    # 步骤定义
    search_steps = [
        {"name": "搜索1", "time": 1, "result": "晴天"},
        {"name": "搜索2", "time": 1.5, "result": "15-20°"}
    ]
    all_result = "查询天气: "
    
    # 发送开始事件
    writer({
        "type": "search_node",
        "status": "start",
        "user_id": user_id,
        "user_name": user_name,
    })
    
    for i, step in enumerate(search_steps, 1):
        writer({
            "type": "search_node",
            "status": "searching",
            "cur_step": i,
            "step": step["name"],
            "all_steps": len(search_steps),
            "user_id": user_id,
            "user_name": user_name,
        })
        time.sleep(step["time"])   # 模拟耗时
        all_result += step["result"]
    
    writer({
        "type": "search_node",
        "status": "end",
        "user_id": user_id,
        "user_name": user_name,
        "result": all_result,
    })
    return all_result

关键设计点：

1. status 字段区分阶段：start、searching、end

2. 在 searching 事件中包含 cur_step、all_steps 以便前端计算进度百分比

3. 每个事件携带 user_id、user_name 等上下文信息，便于日志追踪

---

2.2 在 LLM 节点中也添加开始/结束事件

def llm_call(state: State):
    writer = get_stream_writer()
    writer({
        "type": "llm_call",
        "status": "start",
        "messages": "开始调用大模型",
        "content": state["messages"][-1].content,
    })
    # ... 调用模型
    writer({
        "type": "llm_call",
        "status": "end",
        "messages": "结束调用大模型",
    })
    return {"messages": [result]}

这样客户端可以知道“模型正在思考”和“模型思考完成”

---

三、接收并处理自定义流：构建终端进度指示器

在客户端（或简单地用 Python 脚本）接收流时，我们区分 custom 和 values 两种模式，并针对不同类型的自定义事件做出不同的 UI 反应

print("已思考.....")
for chunk in graph.stream({
    "messages":[HumanMessage(content="今天烟台莱山区天气怎么样")],
    "user_name":"小猫"
},context={
    "user_id":"123"
# values → 输出完整状态
},stream_mode=["custom","values"]):
    # print(chunk)
    # ('custom', {'type': 'llm_call', 'status': 'start', 'messages': '开始调用大模型', 'content': '今天烟台莱山区天气怎么样'})
    if chunk[0]=="custom":
        info=chunk[-1]
        if info.get("type")=="search_node":
            status=info.get("status")
            if status=="start":
                pass
                print(f"用户id :{info["user_id"]},用户名称:{info["user_name"]}  开始调用工具.......")
            elif status=="searing":
                print(f"进度{info["cur_step"]}/{info["all_steps"]}正在处理{info["step"]}....")
            elif status=="end":
                print(f"搜索完成!{info["result"]}")
        elif info.get("type")=="llm_call":
            pass
    elif chunk[0] == "values":
        info = chunk[-1]
        # 是AIMessages且不包含tool_calls就是最后一条消息就打印
        if isinstance(info["messages"][-1],AIMessage) and not info["messages"][-1].tool_calls:
            print("最终结果：")
            print(info["messages"][-1].content)

运行这段代码，你会看到类似下面的输出：

我们可以打开deepseek的深度思考模式做个对比：

我们可以看到这种设计，让体验已经非常接近真实的“Agent 思考过程”了

---

四、与其他模式组合使用

如果你还希望看到每个节点的状态更新，可以同时监听 updates
例如：

stream_mode=["custom", "updates", "values"]
updates 可以用于在 UI 上高亮当前正在执行的节点。

values 可以在最后保存完整的对话历史

注意：
同时开启多个模式会增加事件数量，但通常不会对性能产生显著影响

---

五、最佳实践与注意事项

5.1 自定义事件的数据结构

这里我们建议统一使用一个 type 字段区分事件种类
（如 “search_node”、“llm_call”、“data_fetch” 等），并用 status 字段标记阶段
这有助于客户端编写清晰的 switch-case

---

5.2 避免发送过多微小事件

如果你的工具内部有循环 1000 次，每次发送一个事件可能会导致流消息风暴
可以考虑：

• 每 N 步发送一次汇总进度

• 仅发送开始、关键节点、结束事件

• 使用节流（throttle）技术

---

5.3 确保事件数据可序列化

注意：
writer() 接受的对象必须是 JSON 可序列化的（dict、list、str、int、float、bool、None）
不要传入自定义对象或函数

---

5.4 幂等性考虑

由于 LangGraph 在中断恢复时可能会重新执行节点（例如 interrupt 后从检查点恢复），工具内部的 writer 也会被重复调用
如果进度事件包含不可逆的副作用（比如扣费记录），要谨慎处理
一般情况下，只发送日志性质的事件没有大问题

---

六、扩展：构建一个真实的 PDF 解析工具

假设wm有一个解析大型 PDF 文档的工具，它需要逐页提取文本并分析
我们可以这样发送自定义事件：

def parse_pdf(runtime: ToolRuntime, file_path: str) -> str:
    writer = get_stream_writer()
    total_pages = get_total_pages(file_path)
    
    writer({
    "type": "pdf_parse",
     "status": "start", 
     "total_pages": total_pages
     })
      
    for page in range(1, total_pages + 1):
        text = extract_page(file_path, page)
        writer({"type": "pdf_parse", 
        "status": "processing", 
        "page": page, 
        "total": total_pages})
        # 可选：发送部分提取结果
        if page % 10 == 0:
            writer({
            "type": "pdf_parse",
            "status": "milestone", 
            "message": f"已处理 {page} 页"})
            
    writer({
    		"type": "pdf_parse", 
    		"status": "end", 
    		"summary": "解析完成"
    	  })
    				
    return aggregated_text

前端可以实时显示进度条：“正在解析第 5/100 页…”

---

七、总结

本文深入讲解了 custom 流模式的高级用法，重点包括：

1. 在工具内部模拟多步骤任务，发送进度事件

2. 使用 get_stream_writer() 在不同节点中发送自定义数据结构

3. 客户端接收混合流，并分别处理 custom 和 values 数据

custom 模式让我们的 Agent 不再是黑盒，而是透明、可理解、可干预的智能系统
然而，对于 LLM 生成的文本，逐 token 输出仍然是聊天应用的核心需求

---

本期分享先到这里，在下一篇文章中，我们将详细讨论 stream_mode=“messages” 模式，学习如何按标签或节点过滤 token 流，以及如何管理多模型的并发输出，下期再见~~~~~