LangChain 实现了流式传输系统，用于实时更新。 流式传输对于增强基于 LLM 构建的应用程序的响应能力至关重要。通过渐进式显示输出，甚至在完整响应准备好之前，流式传输显著改善了用户体验 (UX)，尤其是在处理 LLM 的延迟时。
LangChain 的流式传输系统允许您将代理运行的实时反馈呈现在您的应用程序中。 LangChain 流式传输可能实现的功能：
代理进度
要流式传输代理进度，请使用 streamMode: "updates" 调用 stream 方法。这将在每个代理步骤后发出一个事件。 例如，如果您有一个调用工具一次的代理，您应该会看到以下更新：
- LLM 节点：带有工具调用请求的
AIMessage
- 工具节点：@[
ToolMessage] 带有执行结果
- LLM 节点：最终的 AI 响应
import z from "zod";
import { createAgent, tool } from "langchain";
const getWeather = tool(
async ({ city }) => {
return `The weather in ${city} is always sunny!`;
},
{
name: "get_weather",
description: "Get weather for a given city.",
schema: z.object({
city: z.string(),
}),
}
);
const agent = createAgent({
model: "gpt-5-nano",
tools: [getWeather],
});
for await (const chunk of await agent.stream(
{ messages: [{ role: "user", content: "what is the weather in sf" }] },
{ streamMode: "updates" }
)) {
const [step, content] = Object.entries(chunk)[0];
console.log(`step: ${step}`);
console.log(`content: ${JSON.stringify(content, null, 2)}`);
}
/**
* step: model
* content: {
* "messages": [
* {
* "kwargs": {
* // ...
* "tool_calls": [
* {
* "name": "get_weather",
* "args": {
* "city": "San Francisco"
* },
* "type": "tool_call",
* "id": "call_0qLS2Jp3MCmaKJ5MAYtr4jJd"
* }
* ],
* // ...
* }
* }
* ]
* }
* step: tools
* content: {
* "messages": [
* {
* "kwargs": {
* "content": "The weather in San Francisco is always sunny!",
* "name": "get_weather",
* // ...
* }
* }
* ]
* }
* step: model
* content: {
* "messages": [
* {
* "kwargs": {
* "content": "The latest update says: The weather in San Francisco is always sunny!\n\nIf you'd like real-time details (current temperature, humidity, wind, and today's forecast), I can pull the latest data for you. Want me to fetch that?",
* // ...
* }
* }
* ]
* }
*/
LLM Tokens
要流式传输 LLM 生成的令牌，请使用 streamMode: "messages"
import z from "zod";
import { createAgent, tool } from "langchain";
const getWeather = tool(
async ({ city }) => {
return `The weather in ${city} is always sunny!`;
},
{
name: "get_weather",
description: "Get weather for a given city.",
schema: z.object({
city: z.string(),
}),
}
);
const agent = createAgent({
model: "gpt-4o-mini",
tools: [getWeather],
});
for await (const [token, metadata] of await agent.stream(
{ messages: [{ role: "user", content: "what is the weather in sf" }] },
{ streamMode: "messages" }
)) {
console.log(`node: ${metadata.langgraph_node}`);
console.log(`content: ${JSON.stringify(token.contentBlocks, null, 2)}`);
}
自定义更新
要流式传输工具执行时的更新，您可以使用配置中的 writer 参数。
import z from "zod";
import { tool, createAgent } from "langchain";
import { LangGraphRunnableConfig } from "@langchain/langgraph";
const getWeather = tool(
async (input, config: LangGraphRunnableConfig) => {
// Stream any arbitrary data
config.writer?.(`Looking up data for city: ${input.city}`);
// ... fetch city data
config.writer?.(`Acquired data for city: ${input.city}`);
return `It's always sunny in ${input.city}!`;
},
{
name: "get_weather",
description: "Get weather for a given city.",
schema: z.object({
city: z.string().describe("The city to get weather for."),
}),
}
);
const agent = createAgent({
model: "gpt-4o-mini",
tools: [getWeather],
});
for await (const chunk of await agent.stream(
{ messages: [{ role: "user", content: "what is the weather in sf" }] },
{ streamMode: "custom" }
)) {
console.log(chunk);
}
Looking up data for city: San Francisco
Acquired data for city: San Francisco
如果您将 writer 参数添加到您的工具中，您将无法在不提供 writer 函数的情况下在 LangGraph 执行上下文之外调用该工具。
流式传输多种模式
您可以通过将 streamMode 作为数组传递来指定多种流式传输模式：streamMode: ["updates", "messages", "custom"]
import z from "zod";
import { tool, createAgent } from "langchain";
import { LangGraphRunnableConfig } from "@langchain/langgraph";
const getWeather = tool(
async (input, config: LangGraphRunnableConfig) => {
// Stream any arbitrary data
config.writer?.(`Looking up data for city: ${input.city}`);
// ... fetch city data
config.writer?.(`Acquired data for city: ${input.city}`);
return `It's always sunny in ${input.city}!`;
},
{
name: "get_weather",
description: "Get weather for a given city.",
schema: z.object({
city: z.string().describe("The city to get weather for."),
}),
}
);
const agent = createAgent({
model: "gpt-4o-mini",
tools: [getWeather],
});
for await (const [streamMode, chunk] of await agent.stream(
{ messages: [{ role: "user", content: "what is the weather in sf" }] },
{ streamMode: ["updates", "messages", "custom"] }
)) {
console.log(`${streamMode}: ${JSON.stringify(chunk, null, 2)}`);
}
禁用流式传输
在某些应用程序中，您可能需要禁用给定模型的单个令牌流式传输。 这在多代理系统中很有用，可以控制哪些代理流式传输其输出。 请参阅模型指南以了解如何禁用流式传输。
