图执行 - AI 开发中文文档

LangGraph 智能体并非黑盒。每个图由命名的节点组成，这些节点按顺序或并行执行：分类、研究、分析、综合。图执行卡片通过为每个节点渲染卡片来使此流程可视化，显示其状态、实时流式传输其内容，并跨整个工作流程跟踪完成情况。用户可以确切地看到智能体正在做什么、它处于哪个步骤以及每个步骤产生了什么。这种模式对生产环境智能体特别有用，因为它将图结构转化为产品用户体验。您不必将运行视为单个助手响应，而是可以暴露 LangGraph 内部使用的相同检查点、节点名称、状态键和流元数据。

图节点如何映射到 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（用于当前运行中发现的图节点）。将每个发现的子图快照传递给选择器以读取该节点范围内的消息。

代码示例使用 useStream<typeof myAgent> 进行类型安全的流状态管理。请参考 Python 或 JavaScript 后端的类型推断说明。

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>
  );
}

<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>

<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 路由到节点

当图流式传输时，每个发现的子图快照都会标识它所属的节点。将此快照传递给选择器钩子或可组合组件以读取该节点范围内的消息：

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;

构建流程进度条

顶部有一个水平进度条，为用户提供整个流程的鸟瞰视图。每个步骤都是一个带标签的片段，随着节点的完成而填充：

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 组件

每个节点都有自己的卡片，显示状态徽章、内容（流式或最终）以及用于长输出的可折叠正文：

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>
  );
}

流式传输与已完成内容

节点卡片读取流式传输和最终内容的范围消息。这避免了假设图节点名称与其写入的状态键匹配（例如，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>;
}

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

将所有内容整合在一起

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

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 仅显示与当前执行相关的卡片，避免出现空的占位符卡片。

如果您的图具有条件分支（例如，对于简单的事实性查询跳过“研究”），则跳过的节点将不会出现在 stream.subgraphs 中。您的流程进度条可以仅渲染发现的节点，或淡化没有匹配快照的预期节点。

最佳实践

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

官方原文：src/oss/langgraph/frontend/graph-execution.mdx · 非官方翻译，译文由 AI 生成，以英文原文为准。

​图节点如何映射到 UI 卡片

​设置 useStream

​将流式 Token 路由到节点

​确定节点状态

​构建流程进度条

​构建可折叠的 NodeCard 组件

​流式传输与已完成内容

​将所有内容整合在一起

​用例

​处理动态流程

​最佳实践

图节点如何映射到 UI 卡片

设置 `useStream`

将流式 Token 路由到节点

确定节点状态

构建流程进度条

构建可折叠的 NodeCard 组件

流式传输与已完成内容

将所有内容整合在一起

用例

处理动态流程

最佳实践