图执行

可视化多步骤图管道，并展示每个节点的状态和流式内容

LangGraph 智能体不是黑箱。每个图都由具名节点组成，它们按顺序或并行执行：分类、研究、分析、综合。图执行卡片通过为每个节点渲染一张卡片，使这条管道可见，显示其状态、实时流式输出其内容，并跟踪整个工作流的完成情况。用户可以确切地看到智能体在做什么、处于哪个步骤，以及每个步骤产出了什么。

这种模式对生产环境的智能体特别有用，因为它将图结构转化为产品 UX。与其将运行视为单个助手响应，你可以暴露 LangGraph 内部使用的相同的检查点、节点名称、状态键和流元数据。

在线演示

此处为官方文档中的交互式演示（PatternEmbed），静态站点无法呈现。请前往官方原文查看。

图节点如何映射到 UI 卡片

一个 LangGraph 图定义了一系列节点，每个节点负责一项特定的任务。例如，一条研究管道可能有：

分类：对用户的查询进行分类
研究：收集相关信息
分析：从研究中得出结论
综合：生成最终打磨好的回复

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

// Nodes are discovered automatically — no hardcoded list needed
const graphNodes = [...stream.subgraphs.values()];

// Each snapshot carries the node name and current status
graphNodes.forEach((node) => {
  console.log(node.nodeName, node.status); // "classify", "running"
});

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

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

设置 `useStream`

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

INFO

代码示例使用 useStream<typeof myAgent> 来实现类型安全的流状态。类型推断请参阅 Python 或 JavaScript 后端。

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 (
    <div>
      <PipelineProgress nodes={graphNodes} isLoading={stream.isLoading} />
      <NodeCardList nodes={graphNodes} stream={stream} isLoading={stream.isLoading} />
    </div>
  );
}

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>
  <div>
    <PipelineProgress
      :nodes="[...stream.subgraphs.value.values()]"
      :is-loading="stream.isLoading.value"
    />
    <NodeCardList
      :nodes="[...stream.subgraphs.value.values()]"
      :stream="stream"
      :is-loading="stream.isLoading.value"
    />
  </div>
</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>

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

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

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

@Component({
  selector: "app-pipeline-chat",
  template: `
    <div>
      <app-pipeline-progress
        [nodes]="graphNodes()"
        [isLoading]="stream.isLoading()"
      />
      <app-node-card-list
        [nodes]="graphNodes()"
        [stream]="stream"
        [isLoading]="stream.isLoading()"
      />
    </div>
  `,
})
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 (
    <div className="flex items-center gap-1">
      {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 key={node.id} className="flex items-center">
            <div
              className={`rounded-full px-3 py-1 text-xs font-medium ${colors[status]}`}
            >
              {node.nodeName}
            </div>
            {i < nodes.length - 1 && (
              <div
                className={`mx-1 h-0.5 w-6 ${
                  status === "complete" ? "bg-green-500" : "bg-gray-200"
                }`}
              />
            )}
          </div>
        );
      })}
    </div>
  );
}

构建可折叠的 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 (
    <div className="rounded-lg border bg-white shadow-sm">
      <button
        onClick={() => setOpen(!open)}
        className="flex w-full items-center justify-between p-4"
      >
        <div className="flex items-center gap-3">
          <h3 className="font-semibold">{node.nodeName}</h3>
          <StatusBadge status={node.status} />
        </div>
        <span className={open ? "rotate-90" : ""}>▶</span>
      </button>

      {open && (
        <div className="border-t px-4 py-3">
          <div className="prose prose-sm max-w-none">
            {lastAIMessage?.text?.trim()
              ? <Markdown>{lastAIMessage.text}</Markdown>
              : <p className="italic text-gray-500">Processing...</p>}
          </div>
        </div>
      )}
    </div>
  );
}

流式内容与已完成内容

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

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

该模式是：在节点卡片中显示最近的限定作用域 AI 消息，并且仅当你确实需要一个图状态字段时才使用 stream.values。

由于限定作用域的消息与产生它们的节点绑定，UI 可以支持并行的图路径，而无需根据消息顺序进行猜测。每张卡片都会根据属于其节点的事件进行更新，而已完成的值仍然可以通过 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 (
    <div className="space-y-3">
      {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} />;
      })}
    </div>
  );
}

使用场景

图执行卡片非常适合任何可视性至关重要的多步骤管道：

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

处理动态管道

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

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

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

INFO

如果你的图有条件分支（例如，对于简单的事实性查询跳过 "Research"），跳过的节点不会出现在 stream.subgraphs 中。你的管道进度条可以只渲染已发现的节点，或将没有匹配快照的预期节点淡化显示。

最佳实践

从流中发现节点。根据 stream.subgraphs 渲染卡片，而不是硬编码预期的节点；条件性或被跳过的步骤在运行之前不会出现。
将状态键视为 UI 契约。确定哪些图输出应足够稳定以便前端渲染，并将这些键记录在图定义旁边。
为节点卡片使用限定作用域的消息。它们在节点流式输出期间以及完成之后都能工作，而无需将 UI 卡片与状态键名耦合。
自动折叠已完成的节点。在长管道中，自动折叠已完成的卡片，以便用户可以专注于当前活动的步骤。
显示预估时间。如果你有关于每个节点耗时的历史数据，请显示一个时间预估，以设定用户预期。
添加全局进度指示器。用管道视图顶部的整体进度条（例如 "Step 2 of 4"）补充每个节点的卡片。
按节点处理错误。如果某个节点失败，请在其卡片中显示错误，而不折叠整个管道。其他节点可能仍会成功完成。

数据集

创建数据集

运行评估

评估技术

定义评估目标

打分方法

实验配置

多模态评估

分析实验结果

微调

教程

常见数据类型

评估器类型

UI

SDK

框架与集成

改进评估器

反馈

追踪设置

集成

手动埋点

配置与故障排除

项目与环境设置

高级追踪技术

数据与隐私

故障排除指南

查看与管理追踪

批量导出追踪数据

轨迹视图

数据类型参考

监控与告警

在线评估器

自动化

开发你的应用

设置依赖

持久化

能力

助手

运行

并发消息处理

如何构建

鉴权与访问控制

自定义鉴权教程

服务器定制

静态加密

替换内置后端

扩展 HTTP 服务器

请求头与日志

部署指南

参考

配置

参考

提示词

创建与管理提示词

连接模型

上下文中枢

教程

全栈 Web 应用

Agent Server API

控制平面 API

计费与用量

参考

按云服务商起步

用 Terraform 部署

AWS

GCP

Azure

安装实例

配置实例

连接外部服务

平台鉴权与访问控制

管理实例

自托管可观测性

混合部署

脚本

参考

组织