AI创想

标题: LangGraph 流式输出全攻略：从基础用法到高级特性的深度解析 [打印本页]
作者: 创想小编    时间: 5 小时前
标题: LangGraph 流式输出全攻略：从基础用法到高级特性的深度解析
作者：佑瞻
在开发智能应用时，我们常常面临这样的场景：需要实时获取流程的中间结果，或者逐步展示大语言模型的生成过程。这时候，LangGraph 的流式输出功能就成了关键利器。今天，我们就来深入聊聊 LangGraph 的流式 API，从基础用法到高级特性，帮助你在项目中实现更流畅的实时交互体验。
一、流式 API 的核心概念与基本用法

1.1 流式处理的优势

相比传统的一次性返回结果，流式输出有两大核心优势：
1.2 基本用法示例

LangGraph 提供了.stream()（同步）和.astream()（异步）两种流式方法，以迭代器形式 yield 输出结果。来看一个基础示例：
python
  1. # 同步流式处理
  2. for chunk in graph.stream(inputs, stream_mode="updates"):
  3.     print(chunk)
  5. # 异步流式处理（需配合async/await使用）
  6. async for chunk in graph.astream(inputs, stream_mode="updates"):
  7.     print(chunk)
复制代码
1.3 支持的流模式

LangGraph 支持五种核心流模式，每种模式解决不同的场景需求：
模式描述
values流式传输每一步后的完整状态值
updates流式传输每一步的状态更新内容，同一步的多次更新会分开流式传输
custom从节点内部流式传输自定义数据
messages流式传输 LLM 调用的令牌及元数据，适用于实时展示模型生成过程
debug流式传输执行过程中的详细调试信息，包含节点名称和完整状态
二、多模式流式处理与状态流

2.1 同时流式传输多种模式

通过传递模式列表，我们可以同时获取多种类型的流数据：
python
  1. for mode, chunk in graph.stream(inputs, stream_mode=["updates", "custom"]):
  2.     print(f"模式: {mode}, 内容: {chunk}")
复制代码
输出会是(mode, chunk)元组，让我们能清晰区分不同模式的数据。
2.2 状态流的两种形态

python
  1. from langgraph.graph import StateGraph, START, END
  3. class State(TypedDict):
  4.     topic: str
  5.     joke: str
  7. def refine_topic(state: State):
  8.     return {"topic": state["topic"] + " and cats"}
  10. def generate_joke(state: State):
  11.     return {"joke": f"This is a joke about {state['topic']}"}
  13. graph = (
  14.     StateGraph(State)
  15.     .add_node(refine_topic)
  16.     .add_node(generate_joke)
  17.     .add_edge(START, "refine_topic")
  18.     .add_edge("refine_topic", "generate_joke")
  19.     .add_edge("generate_joke", END)
  20.     .compile()
  21. )
  23. # 仅流式传输更新内容
  24. print("--- updates模式 ---")
  25. for chunk in graph.stream({"topic": "ice cream"}, stream_mode="updates"):
  26.     print(chunk)
  28. # 流式传输完整状态
  29. print("\n--- values模式 ---")
  30. for chunk in graph.stream({"topic": "ice cream"}, stream_mode="values"):
  31.     print(chunk)
复制代码
三、子图流式处理与调试模式

3.1 包含子图输出的流式处理

通过设置subgraphs=True，我们可以在父图的流中包含子图的输出：
python
  1. from langgraph.graph import START, StateGraph
  3. # 定义子图
  4. class SubgraphState(TypedDict):
  5.     foo: str
  6.     bar: str
  8. def subgraph_node_1(state: SubgraphState):
  9.     return {"bar": "bar"}
  11. def subgraph_node_2(state: SubgraphState):
  12.     return {"foo": state["foo"] + state["bar"]}
  14. subgraph_builder = StateGraph(SubgraphState)
  15. subgraph_builder.add_node(subgraph_node_1)
  16. subgraph_builder.add_node(subgraph_node_2)
  17. subgraph_builder.add_edge(START, "subgraph_node_1")
  18. subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
  19. subgraph = subgraph_builder.compile()
  21. # 定义父图
  22. class ParentState(TypedDict):
  23.     foo: str
  25. def node_1(state: ParentState):
  26.     return {"foo": "hi! " + state["foo"]}
  28. builder = StateGraph(ParentState)
  29. builder.add_node("node_1", node_1)
  30. builder.add_node("node_2", subgraph)
  31. builder.add_edge(START, "node_1")
  32. builder.add_edge("node_1", "node_2")
  33. graph = builder.compile()
  35. # 流式处理父图和子图的更新
  36. for chunk in graph.stream({"foo": "foo"}, stream_mode="updates", subgraphs=True):
  37.     print(chunk)
复制代码
输出会包含命名空间元组，明确标识数据来自哪个图或子图，例如：
plaintext
  1. ((), {'node_1': {'foo': 'hi! foo'}})
  2. (('node_2:dfddc4ba-c3c5-6887-5012-a243b5b377c2',), {'subgraph_node_1': {'bar': 'bar'}})
复制代码
3.2 调试模式：获取最详细的执行信息

debug模式会流式传输执行过程中的所有可用信息，是排查问题的利器：
python
  1. for chunk in graph.stream({"topic": "ice cream"}, stream_mode="debug"):
  2.     print(chunk)
复制代码
四、LLM 令牌流与精确过滤

4.1 实时流式传输 LLM 生成内容

通过messages模式，我们可以逐令牌地流式传输 LLM 输出，实现类似 ChatGPT 的实时打字效果：
python
  1. from langchain.chat_models import init_chat_model
  2. from langgraph.graph import StateGraph, START
  4. @dataclass
  5. class MyState:
  6.     topic: str
  7.     joke: str = ""
  9. llm = init_chat_model(model="openai:gpt-4o-mini")
  11. def call_model(state: MyState):
  12.     llm_response = llm.invoke(
  13.         [{"role": "user", "content": f"Generate a joke about {state.topic}"}]
  14.     )
  15.     return {"joke": llm_response.content}
  17. graph = (
  18.     StateGraph(MyState)
  19.     .add_node(call_model)
  20.     .add_edge(START, "call_model")
  21.     .compile()
  22. )
  24. # 流式传输LLM令牌
  25. for message_chunk, metadata in graph.stream({"topic": "ice cream"}, stream_mode="messages"):
  26.     if message_chunk.content:
  27.         print(message_chunk.content, end="|", flush=True)
复制代码
4.2 按标签过滤 LLM 调用

通过为 LLM 添加标签，我们可以精确过滤需要流式传输的令牌：
python
  1. from langchain.chat_models import init_chat_model
  3. # 初始化带标签的LLM
  4. llm_1 = init_chat_model(model="openai:gpt-4o-mini", tags=['joke'])
  5. llm_2 = init_chat_model(model="openai:gpt-4o-mini", tags=['poem'])
  7. # 定义使用这些LLM的图
  8. graph = ...
  10. # 仅流式传输标签为'joke'的LLM输出
  11. async for msg, metadata in graph.astream({"topic": "cats"}, stream_mode="messages"):
  12.     if metadata["tags"] == ["joke"]:
  13.         print(msg.content, end="|", flush=True)
复制代码
4.3 按节点过滤 LLM 输出

通过元数据中的langgraph_node字段，我们可以仅流式传输特定节点的 LLM 输出：
python
  1. for msg, metadata in graph.stream(inputs, stream_mode="messages"):
  2.     if msg.content and metadata["langgraph_node"] == "some_node_name":
  3.         # 处理来自指定节点的LLM输出
  4.         ...
复制代码
五、自定义数据流与任意 LLM 集成

5.1 从节点内部发送自定义数据

通过get_stream_writer()，我们可以在节点内部主动发送自定义数据：
python
  1. from langgraph.config import get_stream_writer
  2. from langgraph.graph import StateGraph, START
  4. class State(TypedDict):
  5.     query: str
  6.     answer: str
  8. def node(state: State):
  9.     writer = get_stream_writer()
  10.     writer({"custom_key": "Generating custom data inside node"})
  11.     return {"answer": "some data"}
  13. graph = (
  14.     StateGraph(State)
  15.     .add_node(node)
  16.     .add_edge(START, "node")
  17.     .compile()
  18. )
  20. inputs = {"query": "example"}
  21. # 流式传输自定义数据
  22. for chunk in graph.stream(inputs, stream_mode="custom"):
  23.     print(chunk)
复制代码
5.2 集成任意 LLM 的流式输出

即使 LLM 没有集成 LangChain，我们也可以通过custom模式实现流式传输：
python
  1. from langgraph.config import get_stream_writer
  3. def call_arbitrary_model(state):
  4.     writer = get_stream_writer()
  5.     # 假设存在自定义流式客户端
  6.     for chunk in your_custom_streaming_client(state["topic"]):
  7.         writer({"custom_llm_chunk": chunk})
  8.     return {"result": "completed"}
  10. graph = (
  11.     StateGraph(State)
  12.     .add_node(call_arbitrary_model)
  13.     .compile()
  14. )
  16. # 流式传输自定义LLM的输出
  17. for chunk in graph.stream({"topic": "cats"}, stream_mode="custom"):
  18.     print(chunk)
复制代码
5.3 禁用特定模型的流式传输

当混合使用支持和不支持流式的模型时，需要显式禁用流式：
python
  1. from langchain.chat_models import init_chat_model
  3. # 禁用流式传输
  4. model = init_chat_model(
  5.     "anthropic:claude-3-7-sonnet-latest",
  6.     disable_streaming=True
  7. )
复制代码
六、Python <3.11 下的异步流式处理

6.1 手动传递 RunnableConfig

在 Python <3.11 中，异步 LLM 调用需要显式传递 RunnableConfig：
python
  1. from langchain.chat_models import init_chat_model
  2. from langgraph.graph import StateGraph, START
  4. llm = init_chat_model("openai:gpt-4o-mini")
  6. async def async_node(state):
  7.     # 显式传递RunnableConfig
  8.     llm_response = await llm.ainvoke(
  9.         [{"role": "user", "content": "Hello"}],
  10.         config=RunnableConfig()
  11.     )
  12.     return {"response": llm_response.content}
  14. # 定义图并执行
  15. ...
复制代码
6.2 手动传递流写入器

在异步节点中无法使用get_stream_writer()，需要手动传递 writer 参数：
python
  1. async def async_node_with_writer(state, writer):
  2.     await writer({"custom_data": "async custom data"})
  3.     return {"result": "done"}
  5. # 定义图时传递writer
  6. graph = StateGraph(...)
  7. graph.invoke(..., config={"writer": your_writer})
复制代码
七、实践建议与总结

通过 LangGraph 的流式输出功能，我们能够：
如果本文对你有帮助，别忘了点赞收藏，关注我，一起探索更高效的开发方式～

原文地址：https://blog.csdn.net/The_Thieves/article/details/148809652



欢迎光临 AI创想 (https://www.llms-ai.com/) Powered by Discuz! X3.4