LangChain 中文文档

外观

LangChain 智能体构建在 LangGraph 之上，因此它们支持相同的流式传输技术栈，并提供面向智能体的消息、工具调用、状态和自定义更新的投影。

对于大多数应用和前端用例，请通过 stream_events(..., version="v3") 使用事件流。事件流返回一个带有类型化投影的 run 对象，因此每个投影都可以独立消费，而不必解析流模式元组。

py
from langchain.agents import create_agent

def get_weather(city: str) -> str:
    """Get weather for a city."""
    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": "What is the weather in SF?"}],
}, version="v3")

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

final_state = stream.output
ts
import { createAgent, tool } from "langchain";
import * as z from "zod";

const getWeather = tool(
  async ({ city }) => `It's always sunny in ${city}!`,
  {
    name: "get_weather",
    description: "Get weather for a city.",
    schema: z.object({ city: z.string() }),
  }
);

const agent = createAgent({
  model: "gpt-5-nano",
  tools: [getWeather],
});

const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "What is the weather in SF?" }] },
  { version: "v3" }
);

for await (const message of stream.messages) {
  for await (const delta of message.text) {
    process.stdout.write(delta);
  }
}

const finalState = await 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自定义 transformer 投影。
stream.tool_calls工具执行生命周期、输入、输出增量、最终输出和错误。
投影用途
for event in stream带有完整信封并可访问每个通道的原始协议事件。
stream.messages模型消息流，每次 LLM 调用一个。
message.text一条消息的文本增量与最终文本。
message.reasoning针对暴露推理内容的模型的推理增量。
message.toolCalls工具调用参数分块与最终确定的工具调用。
message.output模型调用完成后的最终消息对象。
message.usage当提供商返回时的 token 用量元数据。
stream.values智能体状态快照。
stream.output最终的智能体状态。
stream.subgraphs嵌套图运行（子智能体和普通子图）。
stream.extensions自定义 transformer 投影。
stream.toolCalls工具执行生命周期、输入、输出增量、最终输出和错误。

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

stream.messages 会产生消息流。每个消息流都暴露 .text、.reasoning、.toolCalls、.output 和 .usage。异步投影可以迭代以获取实时增量，也可以 await 以获取最终值。

智能体消息 ​

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

py
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)
ts
const stream = await agent.streamEvents(input, { version: "v3" });

for await (const message of stream.messages) {
  process.stdout.write(`[${message.node}] `);
  for await (const delta of message.text) {
    process.stdout.write(delta);
  }

  const fullMessage = await message.output;
  console.log(fullMessage.content);

  const usage = await message.usage;
  if (usage) {
    console.log(usage);
  }
}

message.output 为你提供最终确定的 AI 消息，包括提供商特定的内容块。在 TypeScript 中，如果你只需要 token 数量或其他用量元数据，请使用 message.usage；在 Python 中，请从 message.output.usage_metadata 读取用量。

推理内容 ​

推理内容与文本内容使用相同的结构，但只有在所选模型发出推理块时才可用。

py
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)
ts
const stream = await agent.streamEvents(input, { version: "v3" });

for await (const message of stream.messages) {
  for await (const delta of message.reasoning) {
    process.stdout.write(`[thinking] ${delta}`);
  }

  for await (const delta of message.text) {
    process.stdout.write(delta);
  }
}

关于模型配置的详细信息，请参阅推理指南和你所用提供商的集成页面。

工具调用 ​

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

  • message.tool_calls 在模型生成工具调用时流式传输工具调用参数分块。
  • stream.tool_calls 在工具调用开始后流式传输工具执行的生命周期。
py
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)
ts
const stream = await agent.streamEvents(input, { version: "v3" });

await Promise.all([
  (async () => {
    for await (const message of stream.messages) {
      for await (const chunk of message.toolCalls) {
        console.log("tool call chunk", chunk);
      }
    }
  })(),
  (async () => {
    for await (const call of stream.toolCalls) {
      console.log(call.name, call.input);
      console.log(await call.output, await call.error);
    }
  })(),
]);

流式子智能体 ​

当一次 create_agent 调用调用另一个具名的 create_agent（通常通过包装工具）时，内部智能体的事件会在嵌套的命名空间流动。你传给 create_agent 的 name= 会在流中标识该内部智能体，因此你可以按智能体进行过滤和标记。

具名的子智能体会出现在专门的 stream.subagents 投影上。每个句柄都暴露内部智能体自己的 .messages、.values、.tool_calls 和 .output，以及 .name（你传入的 name=）和 .cause（派发该子智能体的工具调用）。因为这里只出现具名的 create_agent 运行，所以你不必过滤掉普通子图。

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

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    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:
    """Query the weather agent."""
    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": "What's the weather in Boston?"}]},
    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()

当一次 createAgent 调用调用另一个具名的 createAgent（通常通过包装工具）时，内部智能体的事件会在嵌套的命名空间流动。你传给 createAgent 的 name 会在流中标识该内部智能体，因此你可以按智能体进行过滤和标记。

具名的子智能体会出现在专门的 stream.subagents 投影上。每个句柄都暴露内部智能体自己的 .messages、.toolCalls 和 .output，以及 .name（你传入的 name=）、.cause（派发该子智能体的工具调用）和嵌套的 .subagents。因为这里只出现具名的 createAgent 运行，所以你不必过滤掉普通子图。

ts
import { createAgent, tool } from "langchain";
import { z } from "zod";

const getWeather = tool(
  async ({ city }) => `It's always sunny in ${city}!`,
  { name: "get_weather", schema: z.object({ city: z.string() }) }
);

const weatherAgent = createAgent({
  model: "openai:gpt-5.5",
  tools: [getWeather],
  name: "weather_agent",
});

const callWeather = tool(
  async ({ query }) => {
    const result = await weatherAgent.invoke({
      messages: [{ role: "user", content: query }],
    });
    return result.messages.at(-1)?.text ?? "";
  },
  { name: "call_weather", schema: z.object({ query: z.string() }) }
);

const supervisor = createAgent({
  model: "openai:gpt-5.5",
  tools: [callWeather],
  name: "supervisor",
});

const stream = await supervisor.streamEvents(
  { messages: [{ role: "user", content: "What's the weather in Boston?" }] },
  { version: "v3" }
);

for await (const subagent of stream.subagents) {
  process.stdout.write(`${subagent.name}: `);
  for await (const message of subagent.messages) {
    for await (const token of message.text) {
      process.stdout.write(token);
    }
  }
  process.stdout.write("\n");
}
// 输出："weather_agent: The weather in Boston is sunny!"

从工具调用的普通 StateGraph 子图也会出现在 stream.subgraphs 上——在 .compile(name=...) 上设置 name=，即可在 subagent.graph_name 中获得标签。

stream.subagents 是具名 create_agent 子智能体的聚焦视图，而 stream.subgraphs 涵盖所有嵌套图。使用与你界面（UI）匹配的任意一个。

stream.subagents 是具名 createAgent 子智能体的聚焦视图，而 stream.subgraphs 涵盖所有嵌套图。使用与你界面（UI）匹配的任意一个。

状态与最终输出 ​

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

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

for snapshot in stream.values:
    print(snapshot)

final_state = stream.output
ts
const stream = await agent.streamEvents(input, { version: "v3" });

for await (const snapshot of stream.values) {
  console.log(snapshot);
}

const finalState = await stream.output;

多个投影 ​

在异步代码中进行并发消费时，请将 astream_events 与 asyncio.gather 一起使用：

py
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(...)：

py
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)

在 JavaScript 中需要多个投影时，请使用并发消费者：

ts
const stream = await agent.streamEvents(input, { version: "v3" });

await Promise.all([
  (async () => {
    for await (const message of stream.messages) {
      console.log(await message.text);
    }
  })(),
  (async () => {
    for await (const call of stream.toolCalls) {
      console.log(call.name, call.input);
    }
  })(),
]);

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

py
for event in stream:
    print(event["method"], event["params"]["namespace"], event["params"]["data"])
ts
for await (const event of stream) {
  console.log(event.method, event.params.namespace, event.params.data);
}

自定义更新 ​

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

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

for activity in stream.extensions["tool_activity"]:
    print(activity)
ts
const stream = await agent.streamEvents(input, {
  version: "v3",
  transformers: [toolActivityTransformer],
});

for await (const activity of stream.extensions.toolActivity) {
  console.log(activity);
}

在中间件上注册 transformer ​

INFO

Middleware-registered transformers require langchain>=1.3.2.

INFO

Middleware-registered transformers require langchain@1.4.3 or later.

中间件可以与其钩子和工具一起声明流式 transformer 工厂。工厂的形状因语言而异：

在 AgentMiddleware 子类上将 transformers 属性设置为一组工厂。每个工厂的形状为 Callable[[tuple[str, ...]], StreamTransformer]，并以 factory(scope) 的形式调用，其中 scope 是 mini-mux 作用域元组（() 表示根 mux，非空表示子图）。每次调用返回一个新的 transformer 可以保持每个子图相互隔离。

py
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()],
)

将 streamTransformers 以工厂元组的形式传给 createMiddleware。每个工厂的形状为 () => StreamTransformer<any>（零参数），每个作用域会调用一次。每次调用返回一个新的 transformer 可以保持每个子图相互隔离。

ts
import { createAgent, createMiddleware } from "langchain";

const toolActivityMiddleware = createMiddleware({
  name: "ToolActivityMiddleware",
  streamTransformers: [toolActivityTransformer],
});

const agent = createAgent({
  model: "gpt-5-nano",
  tools: [getWeather],
  middleware: [toolActivityMiddleware],
});

在编译时，create_agent 会将中间件注册的工厂与传给其自身 transformers= 参数的工厂合并。编译后的图上的最终顺序为：

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

这使内置的工具调用投影排在消费者 transformer 之前，并让调用者提供的条目拥有最终决定权。

内置的 PIIMiddleware 使用这个钩子来对流式传输的网络输出中的 PII 进行脱敏。启用 apply_to_output=True 时，其注册的 transformer 会在文本增量、工具调用参数、工具输出和状态快照离开运行之前，从这些内容中清除检测到的 PII，从而堵住原本会通过 after_model 状态级脱敏让原始 PII 流向 stream_events(version="v3") 实时读者的窗口。

py
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 检测。

在编译时，createAgent 会将中间件注册的工厂与传给其自身 streamTransformers 选项的工厂合并。编译后的图上的最终顺序为：

  1. 内置的 ToolCallTransformer。
  2. 中间件注册的工厂，按中间件顺序排列。
  3. 调用者从 createAgent 提供的 streamTransformers。

这使内置的工具调用投影排在消费者 transformer 之前，并让调用者提供的条目拥有最终决定权。

关于 transformer 约定，请参见构建你自己的投影。

相关资源 ​