如何在没有 LangChain 模型的情况下流式传输 LLM 令牌¶

在本指南中，我们将从为代理提供支持的语言模型流式传输令牌，而无需使用 LangChain 聊天模型。我们将以 ReAct 代理为例，直接使用 OpenAI 客户端库。

设置¶

要开始，请分别安装 openai 和 langgraph 包

$ npm install openai @langchain/langgraph @langchain/core

兼容性

本指南需要 @langchain/core>=0.2.19，如果您使用的是 LangSmith，则需要 langsmith>=0.1.39。有关升级帮助，请参阅 本指南.

您还需要确保已将您的 OpenAI 密钥设置为 process.env.OPENAI_API_KEY。

定义模型和工具模式¶

首先，初始化 OpenAI SDK 并使用 OpenAI 的格式 定义模型要填充的工具模式

在 [1]

import OpenAI from "openai";

const openaiClient = new OpenAI({});

const toolSchema: OpenAI.ChatCompletionTool = {
  type: "function",
  function: {
    name: "get_items",
    description: "Use this tool to look up which items are in the given place.",
    parameters: {
      type: "object",
      properties: {
        place: {
          type: "string",
        },
      },
      required: ["place"],
    }
  }
};

import OpenAI from "openai"; const openaiClient = new OpenAI({}); const toolSchema: OpenAI.ChatCompletionTool = { type: "function", function: { name: "get_items", description: "使用此工具查找给定位置有哪些物品。", parameters: { type: "object", properties: { place: { type: "string", }, }, required: ["place"], } } };

调用模型¶

现在，为 LangGraph 节点定义一个方法，该方法将调用模型。它将处理对模型的工具调用的格式化以及通过 自定义回调事件 进行流式传输。

如果您使用的是 LangSmith，您还可以包装 OpenAI 客户端以获得与 LangChain 聊天模型相同的出色跟踪效果。

以下是代码示例

在 [2]

import { dispatchCustomEvent } from "@langchain/core/callbacks/dispatch";
import { wrapOpenAI } from "langsmith/wrappers/openai";
import { Annotation } from "@langchain/langgraph";

const StateAnnotation = Annotation.Root({
  messages: Annotation<OpenAI.ChatCompletionMessageParam[]>({
    reducer: (x, y) => x.concat(y),
  }),
});

// If using LangSmith, use "wrapOpenAI" on the whole client or
// "traceable" to wrap a single method for nicer tracing:
// https://docs.smith.langchain.com/how_to_guides/tracing/annotate_code
const wrappedClient = wrapOpenAI(openaiClient);

const callModel = async (state: typeof StateAnnotation.State) => {
  const { messages } = state;
  const stream = await wrappedClient.chat.completions.create({
    messages,
    model: "gpt-4o-mini",
    tools: [toolSchema],
    stream: true,
  });
  let responseContent = "";
  let role: string = "assistant";
  let toolCallId: string | undefined;
  let toolCallName: string | undefined;
  let toolCallArgs = "";
  for await (const chunk of stream) {
    const delta = chunk.choices[0].delta;
    if (delta.role !== undefined) {
      role = delta.role;
    }
    if (delta.content) {
      responseContent += delta.content;
      await dispatchCustomEvent("streamed_token", {
        content: delta.content,
      });
    }
    if (delta.tool_calls !== undefined && delta.tool_calls.length > 0) {
      // note: for simplicity we're only handling a single tool call here
      const toolCall = delta.tool_calls[0];
      if (toolCall.function?.name !== undefined) {
        toolCallName = toolCall.function.name;
      }
      if (toolCall.id !== undefined) {
        toolCallId = toolCall.id;
      }
      await dispatchCustomEvent("streamed_tool_call_chunk", toolCall);
      toolCallArgs += toolCall.function?.arguments ?? "";
    }
  }
  let finalToolCalls;
  if (toolCallName !== undefined && toolCallId !== undefined) {
    finalToolCalls = [{
      id: toolCallId,
      function: {
        name: toolCallName,
        arguments: toolCallArgs
      },
      type: "function" as const,
    }];
  }

  const responseMessage = {
    role: role as any,
    content: responseContent,
    tool_calls: finalToolCalls,
  };
  return { messages: [responseMessage] };
}

import { dispatchCustomEvent } from "@langchain/core/callbacks/dispatch"; import { wrapOpenAI } from "langsmith/wrappers/openai"; import { Annotation } from "@langchain/langgraph"; const StateAnnotation = Annotation.Root({ messages: Annotation({ reducer: (x, y) => x.concat(y), }), }); // 如果使用 LangSmith，请在整个客户端上使用“wrapOpenAI”或 // “traceable” 包装单个方法以获得更出色的跟踪： // https://docs.smith.langchain.com/how_to_guides/tracing/annotate_code const wrappedClient = wrapOpenAI(openaiClient); const callModel = async (state: typeof StateAnnotation.State) => { const { messages } = state; const stream = await wrappedClient.chat.completions.create({ messages, model: "gpt-4o-mini", tools: [toolSchema], stream: true, }); let responseContent = ""; let role: string = "assistant"; let toolCallId: string | undefined; let toolCallName: string | undefined; let toolCallArgs = ""; for await (const chunk of stream) { const delta = chunk.choices[0].delta; if (delta.role !== undefined) { role = delta.role; } if (delta.content) { responseContent += delta.content; await dispatchCustomEvent("streamed_token", { content: delta.content, }); } if (delta.tool_calls !== undefined && delta.tool_calls.length > 0) { // 注意：为简便起见，我们在此处只处理单个工具调用 const toolCall = delta.tool_calls[0]; if (toolCall.function?.name !== undefined) { toolCallName = toolCall.function.name; } if (toolCall.id !== undefined) { toolCallId = toolCall.id; } await dispatchCustomEvent("streamed_tool_call_chunk", toolCall); toolCallArgs += toolCall.function?.arguments ?? ""; } } let finalToolCalls; if (toolCallName !== undefined && toolCallId !== undefined) { finalToolCalls = [{ id: toolCallId, function: { name: toolCallName, arguments: toolCallArgs }, type: "function" as const, }]; } const responseMessage = { role: role as any, content: responseContent, tool_calls: finalToolCalls, }; return { messages: [responseMessage] }; }

请注意，您不能在 LangGraph 节点之外调用此方法，因为 dispatchCustomEvent 如果在不适当的上下文中调用，将失败。

定义工具和工具调用节点¶

接下来，设置实际的工具函数和节点，该节点将在模型填充工具调用时调用它

在 [3]

const getItems = async ({ place }: { place: string }) => {
  if (place.toLowerCase().includes("bed")) {  // For under the bed
    return "socks, shoes and dust bunnies";
  } else if (place.toLowerCase().includes("shelf")) {  // For 'shelf'
    return "books, pencils and pictures";
  } else {  // if the agent decides to ask about a different place
    return "cat snacks";
  }
};

const callTools = async (state: typeof StateAnnotation.State) => {
  const { messages } = state;
  const mostRecentMessage = messages[messages.length - 1];
  const toolCalls = (mostRecentMessage as OpenAI.ChatCompletionAssistantMessageParam).tool_calls;
  if (toolCalls === undefined || toolCalls.length === 0) {
    throw new Error("No tool calls passed to node.");
  }
  const toolNameMap = {
    get_items: getItems,
  };
  const functionName = toolCalls[0].function.name;
  const functionArguments = JSON.parse(toolCalls[0].function.arguments);
  const response = await toolNameMap[functionName](functionArguments);
  const toolMessage = {
    tool_call_id: toolCalls[0].id,
    role: "tool" as const,
    name: functionName,
    content: response,
  }
  return { messages: [toolMessage] };
}

const getItems = async ({ place }: { place: string }) => { if (place.toLowerCase().includes("bed")) { // 对于床下 return "socks, shoes and dust bunnies"; } else if (place.toLowerCase().includes("shelf")) { // 对于“shelf” return "books, pencils and pictures"; } else { // 如果代理决定询问其他地方 return "cat snacks"; } }; const callTools = async (state: typeof StateAnnotation.State) => { const { messages } = state; const mostRecentMessage = messages[messages.length - 1]; const toolCalls = (mostRecentMessage as OpenAI.ChatCompletionAssistantMessageParam).tool_calls; if (toolCalls === undefined || toolCalls.length === 0) { throw new Error("未将任何工具调用传递给节点。"); } const toolNameMap = { get_items: getItems, }; const functionName = toolCalls[0].function.name; const functionArguments = JSON.parse(toolCalls[0].function.arguments); const response = await toolNameMap[functionName](functionArguments); const toolMessage = { tool_call_id: toolCalls[0].id, role: "tool" as const, name: functionName, content: response, } return { messages: [toolMessage] }; }

构建图¶

最后，是时候构建您的图了

在 [4]

import { StateGraph } from "@langchain/langgraph";
import OpenAI from "openai";

// We can reuse the same `GraphState` from above as it has not changed.
const shouldContinue = (state: typeof StateAnnotation.State) => {
  const { messages } = state;
  const lastMessage =
    messages[messages.length - 1] as OpenAI.ChatCompletionAssistantMessageParam;
  if (lastMessage?.tool_calls !== undefined && lastMessage?.tool_calls.length > 0) {
    return "tools";
  }
  return "__end__";
}

const graph = new StateGraph(StateAnnotation)
  .addNode("model", callModel)
  .addNode("tools", callTools)
  .addEdge("__start__", "model")
  .addConditionalEdges("model", shouldContinue, {
    tools: "tools",
    __end__: "__end__",
  })
  .addEdge("tools", "model")
  .compile();

import { StateGraph } from "@langchain/langgraph"; import OpenAI from "openai"; // 我们可以重用上述相同的 `GraphState`，因为它没有改变。 const shouldContinue = (state: typeof StateAnnotation.State) => { const { messages } = state; const lastMessage = messages[messages.length - 1] as OpenAI.ChatCompletionAssistantMessageParam; if (lastMessage?.tool_calls !== undefined && lastMessage?.tool_calls.length > 0) { return "tools"; } return "__end__"; } const graph = new StateGraph(StateAnnotation) .addNode("model", callModel) .addNode("tools", callTools) .addEdge("__start__", "model") .addConditionalEdges("model", shouldContinue, { tools: "tools", __end__: "__end__", }) .addEdge("tools", "model") .compile();

在 [5]

import * as tslab from "tslab";

const representation = graph.getGraph();
const image = await representation.drawMermaidPng();
const arrayBuffer = await image.arrayBuffer();

await tslab.display.png(new Uint8Array(arrayBuffer));

import * as tslab from "tslab"; const representation = graph.getGraph(); const image = await representation.drawMermaidPng(); const arrayBuffer = await image.arrayBuffer(); await tslab.display.png(new Uint8Array(arrayBuffer));

流式传输令牌¶

现在，我们可以使用 .streamEvents 方法从 OpenAI 模型获取流式传输的令牌和工具调用

在 [6]

const eventStream = await graph.streamEvents(
  { messages: [{ role: "user", content: "what's in the bedroom?" }] },
  { version: "v2" },
);

for await (const { event, name, data } of eventStream) {
  if (event === "on_custom_event") {
    console.log(name, data);
  }
}

const eventStream = await graph.streamEvents( { messages: [{ role: "user", content: "what's in the bedroom?" }] }, { version: "v2" }, ); for await (const { event, name, data } of eventStream) { if (event === "on_custom_event") { console.log(name, data); } }

streamed_tool_call_chunk {
  index: 0,
  id: 'call_v99reml4gZvvUypPgOpLgxM2',
  type: 'function',
  function: { name: 'get_items', arguments: '' }
}
streamed_tool_call_chunk { index: 0, function: { arguments: '{"' } }
streamed_tool_call_chunk { index: 0, function: { arguments: 'place' } }
streamed_tool_call_chunk { index: 0, function: { arguments: '":"' } }
streamed_tool_call_chunk { index: 0, function: { arguments: 'bed' } }
streamed_tool_call_chunk { index: 0, function: { arguments: 'room' } }
streamed_tool_call_chunk { index: 0, function: { arguments: '"}' } }
streamed_token { content: 'In' }
streamed_token { content: ' the' }
streamed_token { content: ' bedroom' }
streamed_token { content: ',' }
streamed_token { content: ' you' }
streamed_token { content: ' can' }
streamed_token { content: ' find' }
streamed_token { content: ' socks' }
streamed_token { content: ',' }
streamed_token { content: ' shoes' }
streamed_token { content: ',' }
streamed_token { content: ' and' }
streamed_token { content: ' dust' }
streamed_token { content: ' b' }
streamed_token { content: 'unnies' }
streamed_token { content: '.' }

如果您已设置 LangSmith 跟踪，您还将看到 类似这样的跟踪。