流式传输 (Streaming)

import { StateGraph, StateSchema, START, END } from "@langchain/langgraph";
import { z } from "zod/v4";

const State = new StateSchema({
  topic: z.string(),
  joke: z.string(),
});

const graph = new StateGraph(State)
  .addNode("refineTopic", (state) => {
    return { topic: state.topic + " and cats" };
  })
  .addNode("generateJoke", (state) => {
    return { joke: `This is a joke about ${state.topic}` };
  })
  .addEdge(START, "refineTopic")
  .addEdge("refineTopic", "generateJoke")
  .addEdge("generateJoke", END)
  .compile();

for await (const chunk of await graph.stream(
  { topic: "ice cream" },
  // 设置 streamMode: "updates" 以仅流式传输每个节点后对图状态的更新
  // 其他流式传输模式也可用。有关详细信息，请参阅支持的流式传输模式
  { streamMode: "updates" }
)) {
  console.log(chunk);
}

{'refineTopic': {'topic': 'ice cream and cats'}}
{'generateJoke': {'joke': 'This is a joke about ice cream and cats'}}

import { StateGraph, StateSchema, START } from "@langchain/langgraph";
import { z } from "zod/v4";

// 定义子图
const SubgraphState = new StateSchema({
  foo: z.string(), // 请注意，此键与父图状态共享
  bar: z.string(),
});

const subgraphBuilder = new StateGraph(SubgraphState)
  .addNode("subgraphNode1", (state) => {
    return { bar: "bar" };
  })
  .addNode("subgraphNode2", (state) => {
    return { foo: state.foo + state.bar };
  })
  .addEdge(START, "subgraphNode1")
  .addEdge("subgraphNode1", "subgraphNode2");
const subgraph = subgraphBuilder.compile();

// 定义父图
const ParentState = new StateSchema({
  foo: z.string(),
});

const builder = new StateGraph(ParentState)
  .addNode("node1", (state) => {
    return { foo: "hi! " + state.foo };
  })
  .addNode("node2", subgraph)
  .addEdge(START, "node1")
  .addEdge("node1", "node2");
const graph = builder.compile();

for await (const chunk of await graph.stream(
  { foo: "foo" },
  {
    streamMode: "updates",
    // 设置 subgraphs: true 以流式传输来自子图的输出
    subgraphs: true,
  }
)) {
  console.log(chunk);
}

[[], {'node1': {'foo': 'hi! foo'}}]
[['node2:dfddc4ba-c3c5-6887-5012-a243b5b377c2'], {'subgraphNode1': {'bar': 'bar'}}]
[['node2:dfddc4ba-c3c5-6887-5012-a243b5b377c2'], {'subgraphNode2': {'foo': 'hi! foobar'}}]
[[], {'node2': {'foo': 'hi! foobar'}}]

import { ChatOpenAI } from "@langchain/openai";
import { StateGraph, StateSchema, GraphNode, START } from "@langchain/langgraph";
import * as z from "zod";

// jokeModel 标记为 "joke"
const jokeModel = new ChatOpenAI({
  model: "gpt-4.1-mini",
  tags: ["joke"]
});
// poemModel 标记为 "poem"
const poemModel = new ChatOpenAI({
  model: "gpt-4.1-mini",
  tags: ["poem"]
});

const State = new StateSchema({
  topic: z.string(),
  joke: z.string(),
  poem: z.string(),
});

const callModel: GraphNode<typeof State> = async (state) => {
  const topic = state.topic;
  console.log("Writing joke...");

  const jokeResponse = await jokeModel.invoke([
    { role: "user", content: `Write a joke about ${topic}` }
  ]);

  console.log("\n\nWriting poem...");
  const poemResponse = await poemModel.invoke([
    { role: "user", content: `Write a short poem about ${topic}` }
  ]);

  return {
    joke: jokeResponse.content,
    poem: poemResponse.content
  };
};

const graph = new StateGraph(State)
  .addNode("callModel", callModel)
  .addEdge(START, "callModel")
  .compile();

// streamMode 设置为 "messages" 以流式传输 LLM 令牌
// metadata 包含有关 LLM 调用的信息，包括 tags
for await (const [msg, metadata] of await graph.stream(
  { topic: "cats" },
  { streamMode: "messages" }
)) {
  // 按 metadata 中的 tags 字段过滤流式传输的令牌，仅包括
  // 带有 "joke" 标签的 LLM 调用的令牌
  if (metadata.tags?.includes("joke")) {
    console.log(msg.content + "|");
  }
}

import { ChatOpenAI } from "@langchain/openai";
import { StateGraph, StateSchema, GraphNode, START } from "@langchain/langgraph";
import * as z from "zod";

const model = new ChatOpenAI({ model: "gpt-4.1-mini" });

const State = new StateSchema({
  topic: z.string(),
  joke: z.string(),
  poem: z.string(),
});

const writeJoke: GraphNode<typeof State> = async (state) => {
  const topic = state.topic;
  const jokeResponse = await model.invoke([
    { role: "user", content: `Write a joke about ${topic}` }
  ]);
  return { joke: jokeResponse.content };
};

const writePoem: GraphNode<typeof State> = async (state) => {
  const topic = state.topic;
  const poemResponse = await model.invoke([
    { role: "user", content: `Write a short poem about ${topic}` }
  ]);
  return { poem: poemResponse.content };
};

const graph = new StateGraph(State)
  .addNode("writeJoke", writeJoke)
  .addNode("writePoem", writePoem)
  // 同时写入笑话和诗歌
  .addEdge(START, "writeJoke")
  .addEdge(START, "writePoem")
  .compile();

// "messages" 流式传输模式返回 [messageChunk, metadata] 的元组
// 其中 messageChunk 是 LLM 流式传输的令牌，metadata 是一个字典
// 包含有关调用 LLM 的图节点的信息和其他信息
for await (const [msg, metadata] of await graph.stream(
  { topic: "cats" },
  { streamMode: "messages" }
)) {
  // 按 metadata 中的 langgraph_node 字段过滤流式传输的令牌
  // 仅包括来自 writePoem 节点的令牌
  if (msg.content && metadata.langgraph_node === "writePoem") {
    console.log(msg.content + "|");
  }
}

import { tool } from "@langchain/core/tools";
import { ChatOpenAI } from "@langchain/openai";
import { createAgent } from "@langchain/langgraph";
import { MemorySaver } from "@langchain/langgraph-checkpoint-memory";
import { z } from "zod/v4";

const searchFlights = tool(
  async function* (input) {
    const airlines = ["United", "Delta", "American", "JetBlue"];
    const completed: string[] = [];

    for (let i = 0; i < airlines.length; i++) {
      await new Promise((r) => setTimeout(r, 600));
      completed.push(`${airlines[i]}: checked`);
      yield {
        message: `Searching ${airlines[i]}...`,
        progress: (i + 1) / airlines.length,
        completed,
      };
    }

    return JSON.stringify({
      flights: [
        { airline: "United", price: 450, duration: "5h 30m" },
        { airline: "Delta", price: 520, duration: "5h 15m" },
      ],
    });
  },
  {
    name: "search_flights",
    description: "Search for available flights.",
    schema: z.object({
      destination: z.string(),
      departure_date: z.string(),
    }),
  }
);

const checkHotels = tool(
  async function* (input) {
    const hotels = ["Grand Hyatt", "Marriott", "Hilton"];
    const completed: string[] = [];

    for (let i = 0; i < hotels.length; i++) {
      await new Promise((r) => setTimeout(r, 400));
      completed.push(`${hotels[i]}: available`);
      yield {
        message: `Checking ${hotels[i]}...`,
        progress: (i + 1) / hotels.length,
        completed,
      };
    }

    return JSON.stringify({
      hotels: [
        { name: "Grand Hyatt", price: 250, rating: 4.5 },
        { name: "Marriott", price: 180, rating: 4.2 },
      ],
    });
  },
  {
    name: "check_hotels",
    description: "Check hotel availability.",
    schema: z.object({
      city: z.string(),
      check_in: z.string(),
      nights: z.number(),
    }),
  }
);

export const agent = createAgent({
  model: new ChatOpenAI({ model: "gpt-4o-mini" }),
  tools: [searchFlights, checkHotels],
  checkpointer: new MemorySaver(),
});

import { useStream } from "@langchain/langgraph-sdk/react";

function TravelPlanner() {
  const stream = useStream<typeof agent>({
    assistantId: "travel-agent",
    streamMode: ["values", "tools"],
  });

  const activeTools = stream.toolProgress.filter(
    (t) => t.state === "starting" || t.state === "running"
  );

  return (
    <div>
      {stream.messages.map((msg) => (
        <div key={msg.id}>{msg.content}</div>
      ))}

      {activeTools.map((tool) => {
        const data = tool.data as {
          message?: string;
          progress?: number;
          completed?: string[];
        } | undefined;

        return (
          <div key={tool.toolCallId ?? tool.name}>
            <strong>{tool.name}</strong>
            {data?.message && <p>{data.message}</p>}
            {data?.progress != null && (
              <div style={{ width: "100%", background: "#eee" }}>
                <div
                  style={{
                    width: `${data.progress * 100}%`,
                    background: "#4CAF50",
                    height: 8,
                    transition: "width 0.3s ease",
                  }}
                />
              </div>
            )}
            {data?.completed?.map((step, i) => (
              <div key={i}>&#10003; {step}</div>
            ))}
          </div>
        );
      })}
    </div>
  );
}

import { StateGraph, StateSchema, MessagesValue, GraphNode, START, LangGraphRunnableConfig } from "@langchain/langgraph";
import { tool } from "@langchain/core/tools";
import * as z from "zod";
import OpenAI from "openai";

const openaiClient = new OpenAI();
const modelName = "gpt-4.1-mini";

async function* streamTokens(modelName: string, messages: any[]) {
  const response = await openaiClient.chat.completions.create({
    messages,
    model: modelName,
    stream: true,
  });

  let role: string | null = null;
  for await (const chunk of response) {
    const delta = chunk.choices[0]?.delta;

    if (delta?.role) {
      role = delta.role;
    }

    if (delta?.content) {
      yield { role, content: delta.content };
    }
  }
}

// 这是我们的工具
const getItems = tool(
  async (input, config: LangGraphRunnableConfig) => {
    let response = "";
    for await (const msgChunk of streamTokens(
      modelName,
      [
        {
          role: "user",
          content: `Can you tell me what kind of items i might find in the following place: '${input.place}'. List at least 3 such items separating them by a comma. And include a brief description of each item.`,
        },
      ]
    )) {
      response += msgChunk.content;
      config.writer?.(msgChunk);
    }
    return response;
  },
  {
    name: "get_items",
    description: "Use this tool to list items one might find in a place you're asked about.",
    schema: z.object({
      place: z.string().describe("The place to look up items for."),
    }),
  }
);

const State = new StateSchema({
  messages: MessagesValue,
});

const callTool: GraphNode<typeof State> = async (state) => {
  const aiMessage = state.messages.at(-1);
  const toolCall = aiMessage.tool_calls?.at(-1);

  const functionName = toolCall?.function?.name;
  if (functionName !== "get_items") {
    throw new Error(`Tool ${functionName} not supported`);
  }

  const functionArguments = toolCall?.function?.arguments;
  const args = JSON.parse(functionArguments);

  const functionResponse = await getItems.invoke(args);
  const toolMessage = {
    tool_call_id: toolCall.id,
    role: "tool",
    name: functionName,
    content: functionResponse,
  };
  return { messages: [toolMessage] };
};

const graph = new StateGraph(State)
  // 这是工具调用图节点
  .addNode("callTool", callTool)
  .addEdge(START, "callTool")
  .compile();

const inputs = {
  messages: [
    {
      content: null,
      role: "assistant",
      tool_calls: [
        {
          id: "1",
          function: {
            arguments: '{"place":"bedroom"}',
            name: "get_items",
          },
          type: "function",
        }
      ],
    }
  ]
};

for await (const chunk of await graph.stream(
  inputs,
  { streamMode: "custom" }
)) {
  console.log(chunk.content + "|");
}