图执行 | LangChain 中文文档

LangGraph 智能体不是黑盒。每个图都由按顺序或并行执行的命名节点组成：classify（分类）、research（研究）、analyze（分析）、synthesize（综合）。图执行卡片会为每个节点渲染一张卡片，使这条流水线一目了然：显示节点状态、实时流式输出其内容，并追踪整个工作流的完成情况。用户可以准确看到智能体正在做什么、进行到哪一步，以及每一步产出了什么。

这种模式对生产环境的智能体尤其有用，因为它将图结构转化为产品界面体验。与其把一次运行当作单一的助手响应，不如将 LangGraph 内部使用的检查点、节点名称、状态键和流式输出元数据同样暴露出来。

import { PatternEmbed } from "/snippets/pattern-embed.jsx"

图节点如何映射到界面卡片

LangGraph 图定义了一系列节点，每个节点负责一项特定任务。例如，研究流水线可能包含：

Classify：对用户的查询进行分类
Research：收集相关信息
Analyze：从研究中得出结论
Synthesize：生成最终的精炼响应

每个节点都会将其输出写入图状态中的特定键。在前端，您无需硬编码这种映射，因为 useStream 会在运行时通过 stream.subgraphs 发现每个节点，并为每个被观察到的步骤暴露一个 SubgraphDiscoverySnapshot：

// 节点会被自动发现——无需硬编码列表
const graphNodes = [...stream.subgraphs.values()];

// 每个快照都带有节点名称和当前状态
graphNodes.forEach((node) => {
  console.log(node.nodeName, node.status); // "classify", "running"
});

使用 node.nodeName 作为进度条和卡片标题中的标签。将每个快照传给 useMessages(stream, node)，以渲染限定到节点的流式输出内容，而无需将界面与图状态键名耦合。

这种映射成为您的图与界面之间的契约。后端作者可以有目的地添加、重命名或重排节点，而前端作者决定每个状态键应如何可视化：状态徽章、markdown 面板、表格、图表、追踪视图或审批卡片。

设置 `useStream`

像往常一样接好 useStream。您会用到的主要属性是 messages（用于对话）和 subgraphs（用于当前运行中发现的图节点）。将每个已发现的子图快照传给选择器，即可读取限定到该节点的消息。

INFO

The code examples use useStream<typeof myAgent> for type-safe stream state. See Type inference for Python or JavaScript backends.

ReactVueSvelteAngular

tsx

import { useStream } from "@langchain/react";

const AGENT_URL = "http://localhost:2024";

export function PipelineChat() {
  const stream = useStream<typeof myAgent>({
    apiUrl: AGENT_URL,
    assistantId: "graph_execution_cards",
  });
  const graphNodes = [...stream.subgraphs.values()];

  return (
      <PipelineProgress nodes={graphNodes} isLoading={stream.isLoading} />
      <NodeCardList nodes={graphNodes} stream={stream} isLoading={stream.isLoading} />
  );
}

vue

<script setup lang="ts">
import { useStream } from "@langchain/vue";

const AGENT_URL = "http://localhost:2024";

const stream = useStream<typeof myAgent>({
  apiUrl: AGENT_URL,
  assistantId: "graph_execution_cards",
});
</script>

<template>
    <PipelineProgress
      :nodes="[...stream.subgraphs.value.values()]"
      :is-loading="stream.isLoading.value"
    />
    <NodeCardList
      :nodes="[...stream.subgraphs.value.values()]"
      :stream="stream"
      :is-loading="stream.isLoading.value"
    />
</template>

svelte

<script lang="ts">
  import { useStream } from "@langchain/svelte";

  const AGENT_URL = "http://localhost:2024";

  const stream = useStream<typeof myAgent>({
    apiUrl: AGENT_URL,
    assistantId: "graph_execution_cards",
  });
</script>

  <PipelineProgress nodes={[...stream.subgraphs.values()]} isLoading={stream.isLoading} />
  <NodeCardList
    nodes={[...stream.subgraphs.values()]}
    {stream}
    isLoading={stream.isLoading}
  />

import { Component, computed } from "@angular/core";
import { injectStream } from "@langchain/angular";

const AGENT_URL = "http://localhost:2024";

@Component({
  selector: "app-pipeline-chat",
  template: `
      <app-pipeline-progress
        [nodes]="graphNodes()"
        [isLoading]="stream.isLoading()"
      />
      <app-node-card-list
        [nodes]="graphNodes()"
        [stream]="stream"
        [isLoading]="stream.isLoading()"
      />
  `,
})
export class PipelineChatComponent {
  stream = injectStream<typeof myAgent>({
    apiUrl: AGENT_URL,
    assistantId: "graph_execution_cards",
  });

  graphNodes = computed(() => [...this.stream.subgraphs().values()]);
}

将流式输出 token 路由到节点

在图进行流式输出时，每个已发现的子图快照都会标识其所属的节点。将该快照传给选择器 hook 或 composable，以读取限定到该节点的消息：

tsx

import { AIMessage } from "langchain";
import { useMessages, type AnyStream, type SubgraphDiscoverySnapshot } from "@langchain/react";

function NodeCard({
  node,
  stream,
}: {
  node: SubgraphDiscoverySnapshot;
  stream: AnyStream;
}) {
  const messages = useMessages(stream, node);
  const lastAIMessage = messages.find(AIMessage.isInstance);
  const streamingContent = lastAIMessage?.text ?? "";

  return <NodeCardBody node={node} content={streamingContent} />;
}

第一个挂载的选择器会为该节点命名空间打开一个限定作用域的订阅。当节点卡片卸载时，该订阅会自动释放。

确定节点状态

每个已发现的节点都带有其当前状态。直接使用 node.status；发现快照会报告 "pending"、"running"、"complete" 或 "error"：

type NodeStatus = SubgraphDiscoverySnapshot["status"];

const status: NodeStatus = node.status;

构建流水线进度条

顶部的水平进度条让用户对整个流水线一目了然。每个步骤都是一个带标签的区段，随着节点完成而填充：

tsx

function PipelineProgress({
  nodes,
  isLoading,
}: {
  nodes: SubgraphDiscoverySnapshot[];
  isLoading: boolean;
}) {
  const firstIncompleteIdx = nodes.findIndex((node) => node.status !== "complete");

  return (
      {nodes.map((node, i) => {
        const isRunning =
          isLoading && node.status !== "complete" && firstIncompleteIdx === i;
        const colors = {
          pending: "bg-gray-200 text-gray-500",
          running: "bg-blue-400 text-white animate-pulse",
          complete: "bg-green-500 text-white",
          error: "bg-red-500 text-white",
        };
        const status = isRunning ? "running" : node.status;

        return (
            <div
              className={`rounded-full px-3 py-1 text-xs font-medium ${colors[status]}`}
            >
              {node.nodeName}
            {i < nodes.length - 1 && (
              <div
                className={`mx-1 h-0.5 w-6 ${
                  status === "complete" ? "bg-green-500" : "bg-gray-200"
                }`}
              />
            )}
        );
      })}
  );
}

构建可折叠的 NodeCard 组件

每个节点都有自己的卡片，显示状态徽章、内容（流式输出中或最终结果），并带有可折叠的主体以容纳较长的输出：

tsx

function NodeCard({
  node,
  stream,
}: {
  node: SubgraphDiscoverySnapshot;
  stream: AnyStream;
}) {
  const [open, setOpen] = useState(node.status === "running");
  const messages = useMessages(stream, node);
  const lastAIMessage = messages.find(AIMessage.isInstance);

  useEffect(() => {
    if (node.status === "running") setOpen(true);
    if (node.status === "complete") setOpen(false);
  }, [node.status]);

  return (
      <button
        onClick={() => setOpen(!open)}
        className="flex w-full items-center justify-between p-4"
      >
          <h3 className="font-semibold">{node.nodeName}</h3>
          <StatusBadge status={node.status} />
        ▶
      </button>

      {open && (
            {lastAIMessage?.text?.trim()
              ? <Markdown>{lastAIMessage.text}</Markdown>
              : Processing...}
      )}
  );
}

流式输出内容与已完成内容

节点卡片同时为流式内容和最终内容读取限定作用域的消息。这避免了假设图节点名称与其写入的状态键匹配（例如，playground 图中的 do_research 写入 research）：

来源	使用时机
`useMessages(stream, node)`	渲染限定到节点的流式消息和最终消息
`stream.values`	使用实际状态键读取整个图的状态，例如最终的 `synthesis` 字段

这种模式是：在节点卡片中显示最近一条限定作用域的 AI 消息，仅在您有意需要图状态字段时才使用 stream.values。

由于限定作用域的消息与产生它的节点绑定，界面可以支持并行的图路径，而无需从消息顺序进行猜测。每张卡片都根据属于其节点的流式事件进行更新，已完成的数值仍可通过 stream.values 获取。

function NodeContent({ stream, node }: { stream: AnyStream; node: SubgraphDiscoverySnapshot }) {
  const messages = useMessages(stream, node);
  const content = messages.find(AIMessage.isInstance)?.text ?? "";

  return <Markdown>{content}</Markdown>;
}

TIP

流式输出内容可能包含尚未完全成型的部分 token 或 markdown。如果您渲染 markdown，请确保您的渲染器能够优雅地处理不完整的语法（例如未闭合的粗体标记 **）。

整合起来

下面是结合了路由、状态检测和卡片渲染的完整卡片列表：

tsx

function NodeCardList({
  nodes,
  stream,
  isLoading,
}: {
  nodes: SubgraphDiscoverySnapshot[];
  stream: AnyStream;
  isLoading: boolean;
}) {
  const firstIncompleteIdx = nodes.findIndex((node) => node.status !== "complete");

  return (
      {nodes.map((node, i) => {
        const isComplete = node.status === "complete";
        const isRunning = isLoading && !isComplete && firstIncompleteIdx === i;
        if (!isComplete && !isRunning) return null;

        return <NodeCard key={node.id} node={node} stream={stream} />;
      })}
  );
}

使用场景

图执行卡片非常适合任何需要可视化的多步骤流水线：

研究流水线：分类 → 收集来源 → 分析 → 综合报告
内容生成：大纲 → 草稿 → 事实核查 → 编辑 → 发布
数据处理：摄取 → 校验 → 转换 → 聚合 → 导出
代码生成：理解需求 → 规划架构 → 编写代码 → 审查 → 测试
决策工作流：收集上下文 → 评估选项 → 对备选方案打分 → 给出建议

处理动态流水线

并非所有图都有固定的节点集合。有些流水线会根据输入添加或跳过节点。发现映射仅包含当前线程中观察到的节点：

const activeNodes = [...stream.subgraphs.values()];

这确保了您的界面只显示与当前执行相关的节点卡片，避免出现空的占位卡片。

INFO

如果您的图有条件分支（例如，对简单的事实性查询跳过 "Research"），被跳过的节点不会出现在 stream.subgraphs 中。您的流水线进度条可以只渲染已发现的节点，或者将没有匹配快照的预期节点置为暗色。

最佳实践

从流中发现问题节点。渲染 stream.subgraphs 中的卡片，而不是硬编码预期的节点；条件性或被跳过的步骤只有在运行后才会出现。
将状态键视为界面契约。决定哪些图输出应足够稳定以供前端渲染，并在图定义旁记录这些键。
为节点卡片使用限定作用域的消息。无论节点正在流式输出还是已经完成，它们都能正常工作，且不会将界面卡片与状态键名耦合。
自动折叠已完成的节点。在较长的流水线中，自动折叠已完成的卡片，以便用户专注于当前活动的步骤。
显示预计耗时。如果您有每个节点耗时的历史数据，请显示时间估算，以设定用户的预期。
添加全局进度指示器。在流水线视图顶部用整体进度条（例如"第 2 步，共 4 步"）来补充每个节点的卡片。
逐节点处理错误。如果某个节点失败，在其卡片中显示错误，而不要折叠整个流水线。其他节点仍可能成功完成。

Graph API

Functional API

图节点如何映射到界面卡片

设置 `useStream`

将流式输出 token 路由到节点

确定节点状态

构建流水线进度条

构建可折叠的 NodeCard 组件

流式输出内容与已完成内容

整合起来

使用场景

处理动态流水线

最佳实践

图节点如何映射到界面卡片 ​

设置 useStream ​

将流式输出 token 路由到节点 ​

确定节点状态 ​

构建流水线进度条 ​

构建可折叠的 NodeCard 组件 ​

流式输出内容与已完成内容 ​

整合起来 ​

使用场景 ​

处理动态流水线 ​

最佳实践 ​

图节点如何映射到界面卡片

设置 `useStream`

将流式输出 token 路由到节点

确定节点状态

构建流水线进度条

构建可折叠的 NodeCard 组件

流式输出内容与已完成内容

整合起来

使用场景

处理动态流水线

最佳实践