事件流

LangChain 智能体基于 LangGraph 构建，因此支持相同的流式处理堆栈，并提供了针对智能体的、面向消息、工具调用、状态和自定义更新的投影。对于大多数应用和前端用例，请使用通过 stream_events(..., version="v3") 实现的事件流。事件流返回一个带有类型化投影的运行对象，因此每个投影可以独立消费，而无需解析流模式的元组。

from langchain.agents import create_agent


def get_weather(city: str) -> str:
    """获取指定城市的天气。"""
    return f"It's always sunny in {city}!"


agent = create_agent(
    model="gpt-5-nano",
    tools=[get_weather],
)

stream = agent.stream_events({
    "messages": [{"role": "user", "content": "旧金山的天气怎么样？"}],
}, version="v3")

for message in stream.messages:
    for delta in message.text:
        print(delta, end="", flush=True)

final_state = stream.output

可流式传输的内容

投影	用途
`for event in stream`	带有完整信封的原始协议事件，可访问每个通道。
`stream.messages`	模型消息流，每次 LLM 调用一个流。
`message.text`	消息的文本增量和最终文本。
`message.reasoning`	对于暴露推理内容的模型，其推理过程的增量。
`message.tool_calls`	工具调用参数块和已确定的工具调用。
`message.output`	模型调用完成后的最终消息对象。
`stream.values`	智能体状态快照。
`stream.output`	最终智能体状态。
`stream.subgraphs`	嵌套图运行（子智能体和普通子图）。
`stream.extensions`	自定义转换器投影。
`stream.tool_calls`	工具执行生命周期、输入、输出增量、最终输出和错误。

stream.messages 生成 ChatModelStream 对象。每个消息流都暴露了 .text、.reasoning、.tool_calls 和 .output。同步投影是可迭代的，用于获取实时增量，也可排空以获取最终值：使用 str(message.text) 获取最终文本，使用 message.tool_calls.get() 获取已确定的工具调用。

智能体消息

当你想要获取每次 LLM 调用的模型输出时，请使用 stream.messages。

stream = agent.stream_events(input, version="v3")

for message in stream.messages:
    print(f"[{message.node}] ", end="")
    for delta in message.text:
        print(delta, end="", flush=True)

    full_message = message.output
    usage = full_message.usage_metadata
    if usage:
        print(usage)

message.output 为你提供最终确定的 AI 消息，包括特定于提供者的内容块。在 TypeScript 中，当你只需要令牌计数或其他使用元数据时使用 message.usage；在 Python 中，从 message.output.usage_metadata 读取使用情况。

推理内容

推理内容使用与文本内容相同的形状，但仅当所选模型发出推理块时才可用。

stream = agent.stream_events(input, version="v3")

for message in stream.messages:
    for delta in message.reasoning:
        print(f"[thinking] {delta}", end="", flush=True)

    for delta in message.text:
        print(delta, end="", flush=True)

有关模型配置详情，请参阅推理指南和你的提供者的集成页面。

工具调用

有两个有用的工具调用投影：

message.tool_calls 在模型生成工具调用时，流式传输工具调用参数块。
stream.tool_calls 在工具调用开始后，流式传输工具执行的生命周期。

stream = agent.stream_events(input, version="v3")

for message in stream.messages:
    for chunk in message.tool_calls:
        print(f"tool call chunk: {chunk}")

    finalized = message.tool_calls.get()
    if finalized:
        print(f"finalized tool calls: {finalized}")

for call in stream.tool_calls:
    print(f"{call.tool_name}({call.input})")
    for delta in call.output_deltas:
        print(delta, end="", flush=True)
    print(call.output, call.error)

流式传输子智能体

当一个 create_agent 调用（通常通过包装工具）调用另一个具名的 create_agent 时，内部智能体的事件会在嵌套的命名空间中流动。你传递给 create_agent 的 name= 会在流中标识该内部智能体，因此你可以按智能体进行过滤和标记。具名子智能体出现在专用的 stream.subagents 投影上。每个句柄都暴露了内部智能体自己的 .messages、.values、.tool_calls 和 .output，以及 .name（你传递的 name=）和 .cause（调度子智能体的工具调用）。因为只有具名的 create_agent 运行才会出现在这里，你不需要过滤掉普通子图。

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model


def get_weather(city: str) -> str:
    """获取给定城市的天气。"""
    return f"It's always sunny in {city}!"


weather_agent = create_agent(
    model=init_chat_model("openai:gpt-5.5"),
    tools=[get_weather],
    name="weather_agent",
)


def call_weather(query: str) -> str:
    """查询天气智能体。"""
    result = weather_agent.invoke({"messages": [{"role": "user", "content": query}]})
    return result["messages"][-1].text


supervisor = create_agent(
    model=init_chat_model("openai:gpt-5.5"),
    tools=[call_weather],
    name="supervisor",
)

stream = supervisor.stream_events(
    {"messages": [{"role": "user", "content": "波士顿的天气怎么样？"}]},
    version="v3",
)

for subagent in stream.subagents:
    print(f"{subagent.name}: ", end="")
    for message in subagent.messages:
        for token in message.text:
            print(token, end="", flush=True)
    print()

从工具调用的普通 StateGraph 子图也会出现在 stream.subgraphs 上——在 .compile(name=...) 上设置 name= 可以在 subagent.graph_name 中获得一个标签。 stream.subagents 是具名 create_agent 子智能体的聚焦视图，而 stream.subgraphs 覆盖了每个嵌套图。使用与你的 UI 匹配的那个。

状态与最终输出

使用 stream.values 获取状态快照，使用 stream.output 获取最终智能体状态。

stream = agent.stream_events(input, version="v3")

for snapshot in stream.values:
    print(snapshot)

final_state = stream.output

多重投影

对于异步代码中的并发消费，使用带有 asyncio.gather 的 astream_events：

import asyncio

stream = await agent.astream_events(input, version="v3")

async def consume_messages():
    async for message in stream.messages:
        print(await message.text)

async def consume_tool_calls():
    async for call in stream.tool_calls:
        print(call.tool_name, call.input)

await asyncio.gather(consume_messages(), consume_tool_calls())

对于同步代码，请改用 stream.interleave(...)：

stream = agent.stream_events(input, version="v3")

for name, item in stream.interleave("messages", "tool_calls", "values"):
    if name == "messages":
        print(item.text)
    elif name == "tool_calls":
        print(item.tool_name, item.input)
    elif name == "values":
        print(item)

要访问未作为类型化投影暴露的通道，或检查完整的事件信封，请迭代原始协议事件：

for event in stream:
    print(event["method"], event["params"]["namespace"], event["params"]["data"])

自定义更新

当你的应用需要内置投影之外的投影（如检索进度、制品或特定领域的事件）时，请使用自定义流转换器。

stream = agent.stream_events(
    input,
    version="v3",
    transformers=[ToolActivityTransformer],
)

for activity in stream.extensions["tool_activity"]:
    print(activity)

在中间件上注册转换器

中间件注册的转换器需要 langchain>=1.3.2。

中间件可以在其钩子和工具旁边声明流转换器工厂。工厂的形状因语言而异：将 AgentMiddleware 子类的 transformers 属性设置为一个工厂序列。每个工厂的形状是 Callable[[tuple[str, ...]], StreamTransformer]，并作为 factory(scope) 被调用，其中 scope 是迷你多路复用器作用域元组（() 用于根复用器，非空用于子图）。每次调用都返回一个新的转换器，可以保持每个子图隔离。

from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware


class ToolActivityMiddleware(AgentMiddleware):
    transformers = (ToolActivityTransformer,)


agent = create_agent(
    model="gpt-5-nano",
    tools=[get_weather],
    middleware=[ToolActivityMiddleware()],
)

在编译时，create_agent 将中间件注册的工厂与传递给其自身 transformers= 参数的任何内容合并。编译图上的最终顺序是：

内置的 ToolCallTransformer。
中间件注册的工厂，按中间件顺序。
调用者从 create_agent 提供的 transformers=。

这将内置的工具调用投影保持在消费者转换器的前面，并给予调用者提供的条目最终决定权。内置的 PIIMiddleware 使用此钩子从流式传输的有线输出中编辑 PII。通过设置 apply_to_output=True，其注册的转换器会在文本增量、工具调用参数、工具输出和状态快照离开运行之前，清除检测到的 PII，关闭了 after_model 状态级编辑可能会让原始 PII 传递到 stream_events(version="v3") 的实时读取者的窗口。

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware

agent = create_agent(
    model="gpt-5-nano",
    tools=[],
    middleware=[
        PIIMiddleware("email", strategy="redact", apply_to_output=True),
    ],
)

有关完整的配置界面，请参阅 PII 检测。有关转换器契约，请参阅构建你自己的投影。

Get started

Core components

Middleware

Frontend

Advanced usage

Agent development

Production

可流式传输的内容

智能体消息

推理内容

工具调用

流式传输子智能体

状态与最终输出

多重投影

自定义更新

在中间件上注册转换器

相关内容

​可流式传输的内容

​智能体消息

​推理内容

​工具调用

​流式传输子智能体

​状态与最终输出

​多重投影

​自定义更新

​在中间件上注册转换器

​相关内容

可流式传输的内容

智能体消息

推理内容

工具调用

流式传输子智能体

状态与最终输出

多重投影

自定义更新

在中间件上注册转换器

相关内容