流式输出

总结

有几种不同的方法可以从图运行中流回输出：

value：返回每个节点执行后的 完整状态
updates：返回每个节点执行后，节点名称和更新内容
debug：返回尽可能多信息的调试事件。包括计划执行节点的任务，和任务的执行结果
messages：会 逐个 token 的流式返回 LLM 消息，以及节点或任务内部任何 LLM 调用相关的元数据
custom：使用 StreamWriter 从节点内部发出自定义数据

举例

from typing import TypedDict
from langgraph.graph import StateGraph, START

class State(TypedDict):
  topic: str
  joke: str

def refine_topic(state: State):
  return {"topic": state["topic"] + "和猫"}


def generate_joke(state: State):
  return {"joke": f"这是一个关于{state['topic']}的笑话"}

graph = (
  StateGraph(State)
    .add_sequence([refine_topic, generate_joke])
    .add_edge(START, "refine_topic")
    .compile()
)

`stream_mode="values"`

返回每个节点执行后的 完整状态

for chunk in graph.stream({ "topic": "狗" }, stream_mode="values"):
  print(chunk)

{'topic': '狗'}
{'topic': '狗和猫'}
{'topic': '狗和猫', 'joke': '这是一个关于狗和猫的笑话'

`stream_mode="updates"`

返回每个节点执行后，节点名称和更新内容

for chunk in graph.stream({ "topic": "狗" }, stream_mode="updates"):
  print(chunk)

{'refine_topic': {'topic': '狗和猫'}}
{'generate_joke': {'joke': '这是一个关于狗和猫的笑话'}}

`stream_mode="debug"`

返回尽可能多信息的调试事件。包括计划执行节点的任务，和任务的执行结果

for chunk in graph.stream({ "topic": "狗" }, stream_mode="debug"):
  print(chunk)

{'type': 'task', 'timestamp': '2025-05-16T00:44:27.232856+00:00', 'step': 1, 'payload': {'id': 'eb939d5b-06f2-6fec-da89-4f2808c2a433', 'name': 'refine_topic', 'input': {'topic': '狗'}, 'triggers': ('branch:to:refine_topic',)}}
{'type': 'task_result', 'timestamp': '2025-05-16T00:44:27.233889+00:00', 'step': 1, 'payload': {'id': 'eb939d5b-06f2-6fec-da89-4f2808c2a433', 'name': 'refine_topic', 'error': None, 'result': [('topic', '狗和猫')], 'interrupts': []}}
{'type': 'task', 'timestamp': '2025-05-16T00:44:27.233889+00:00', 'step': 2, 'payload': {'id': '386a0482-ec26-7c4a-b53e-d279a9d1e4e2', 'name': 'generate_joke', 'input': {'topic': '狗和猫'}, 'triggers': ('branch:to:generate_joke',)}}
{'type': 'task_result', 'timestamp': '2025-05-16T00:44:27.233889+00:00', 'step': 2, 'payload': {'id': '386a0482-ec26-7c4a-b53e-d279a9d1e4e2', 'name': 'generate_joke', 'error': None, 'result': [('joke', '这是一个 关于狗和猫的笑话')], 'interrupts': []}}

`stream_mode="messages"`

会 逐个 token 的流式返回 LLM 消息，以及节点或任务内部任何 LLM 调用相关的元数据

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

def generate_joke(state: State):
  res = llm.invoke(
    [HumanMessage(content=f"生成一个关于{state['topic']}的笑话")]
  )
  return {"joke": res.content}

for message_chunk, metadata in graph.stream({ "topic": "狗" }, stream_mode="messages"):
  if message_chunk.content:
    print(message_chunk.content, end="|", flush=True)

好的|，|这里|有一个|关于|狗|和|猫|的|经典|笑话|：

|---

|**|狗|和|猫|的|面试|**|  

|一家|公司|同时|面试|了一只|狗|和|一只|猫|。|  
|HR|问|狗|：“|你有什么|技能|？”|  
|狗|骄傲|地说|：“|我会|忠诚|、|看|家|、|捡|飞|盘|，|还能|哄|主人|开心|！”|  
|HR|点点头|，|又问|猫|：“|你呢|？”|  
|猫|淡定|地|舔|了|舔|爪子|说|：“|我|……|会|开|视频|会议|。”|  
|HR|疑惑|：“|啊|？”|  
|猫|指了指|电脑|：“|你|键盘|上|现在|全是|我的|毛|。”|  

|---|  

|（|笑|点|：|猫|用|“|掉|毛|”|暗示|自己|早已|“|占领|”|办公|设备|，|而|狗|还在|努力|展示|传统|技能|😂|）|  

|需要|其他|风格|的笑|话|可以|告诉我|哦|！|

print(metadata)

{
'langgraph_step': 2,
'langgraph_node': 'generate_joke',
'langgraph_triggers': ('branch:to:generate_joke',),
'langgraph_path': ('__pregel_pull', 'generate_joke'),
'langgraph_checkpoint_ns': 'generate_joke:3c066034-a385-6d5f-18bb-e7ca9fe140f9',
'checkpoint_ns': 'generate_joke:3c066034-a385-6d5f-18bb-e7ca9fe140f9',
'ls_provider': 'openai',
'ls_model_name': 'deepseek-chat',
'ls_model_type': 'chat',
'ls_temperature': 0.0,
'ls_max_tokens': 1024
}