Skip to main content
LangGraph 代理中的每个状态变更都会创建一个检查点,即该时刻代理状态的完整快照。时间旅行功能允许您检查任意检查点,查看代理当时持有的确切状态,并从该点恢复执行以探索替代路径。它集调试器、撤销按钮和审计日志于一体。
此功能需要 LangGraph Agent Server。使用 langgraph dev 在本地运行您的代理,或将其部署到 LangSmith 以使用此模式。

检查点工作原理

LangGraph 在每次节点执行后持久化代理状态。每个持久化状态都是一个 ThreadState 对象,包含:
  • checkpoint:标识此特定快照的元数据(ID、时间戳)
  • values:此时的完整代理状态(消息、自定义键)
  • tasks:计划接下来运行的图节点
  • next:执行计划中即将运行的节点名称
这创建了一个线性时间线,记录代理做出的每个决策、调用的每个工具以及生成的每个响应。您的 UI 可以渲染此时间线,并允许用户跳转到任意时间点。

设置 useStream

通过向 useStream 传递 fetchStateHistory: true 来启用检查点历史记录。这会告诉钩子加载当前线程的完整检查点时间线。 导入您的代理并将 typeof myAgent 作为类型参数传递给 useStream,以实现对状态值的类型安全访问:
import type { myAgent } from "./agent";
import { useStream } from "@langchain/react";

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

export function TimeTravelChat() {
  const stream = useStream<typeof myAgent>({
    apiUrl: AGENT_URL,
    assistantId: "time_travel",
    fetchStateHistory: true,
  });

  const history = stream.history ?? [];

  return (
    <div className="flex h-screen">
      <ChatPanel messages={stream.messages} />
      <TimelineSidebar
        history={history}
        onSelect={(cp) => stream.submit(null, { checkpoint: cp.checkpoint })}
      />
    </div>
  );
}

ThreadState 对象

history 数组中的每个条目都是一个 ThreadState,表示时间线中的一个检查点:
interface ThreadState {
  checkpoint: {
    checkpoint_id: string;
    checkpoint_ns: string;
  };
  values: Record<string, unknown>;
  tasks: Array<{
    id: string;
    name: string;
    interrupts?: unknown[];
  }>;
  next: string[];
}
属性描述
checkpoint标识此快照。将其传递给 submit 以从此处恢复
values此时的完整代理状态,包括 messages 和任何自定义状态键
tasks在此检查点运行的图节点,包括其名称和任何中断
next计划在此检查点之后执行的节点名称

构建检查点时间线

时间线侧边栏将每个检查点显示为可点击的条目。每个条目显示运行的节点以及该时刻存在的消息数量:
function TimelineSidebar({
  history,
  onSelect,
}: {
  history: ThreadState[];
  onSelect: (cp: ThreadState) => void;
}) {
  return (
    <aside className="w-80 overflow-y-auto border-l bg-gray-50 p-4">
      <h2 className="mb-4 text-sm font-semibold uppercase text-gray-500">
        Checkpoint Timeline
      </h2>
      <div className="space-y-2">
        {history.map((cp, i) => {
          const taskName = cp.tasks?.[0]?.name ?? "unknown";
          const msgCount = (cp.values?.messages as unknown[])?.length ?? 0;

          return (
            <button
              key={cp.checkpoint.checkpoint_id}
              onClick={() => onSelect(cp)}
              className="w-full rounded-lg border bg-white p-3 text-left
                         hover:border-blue-400 hover:shadow-sm transition-all"
            >
              <div className="flex items-center justify-between">
                <span className="text-xs text-gray-400">#{i + 1}</span>
                <NodeBadge name={taskName} />
              </div>
              <p className="mt-1 text-sm font-medium">{taskName}</p>
              <p className="text-xs text-gray-500">
                {msgCount} message{msgCount !== 1 ? "s" : ""}
              </p>
            </button>
          );
        })}
      </div>
    </aside>
  );
}

检查检查点状态

点击检查点应显示该时刻的完整状态。JSON 查看器让开发人员完全了解代理知道和决定的内容:
function CheckpointInspector({ checkpoint }: { checkpoint: ThreadState }) {
  const [expanded, setExpanded] = useState(false);

  return (
    <div className="rounded-lg border bg-white p-4">
      <div className="flex items-center justify-between">
        <h3 className="font-semibold">
          Checkpoint {checkpoint.checkpoint.checkpoint_id.slice(0, 8)}...
        </h3>
        <button
          onClick={() => setExpanded(!expanded)}
          className="text-sm text-blue-600 hover:underline"
        >
          {expanded ? "Collapse" : "Expand"} state
        </button>
      </div>

      <div className="mt-2 space-y-1 text-sm">
        <p>
          <strong>Node:</strong>{" "}
          {checkpoint.tasks?.[0]?.name ?? "—"}
        </p>
        <p>
          <strong>Next:</strong>{" "}
          {checkpoint.next?.join(", ") || "—"}
        </p>
        <p>
          <strong>Messages:</strong>{" "}
          {(checkpoint.values?.messages as unknown[])?.length ?? 0}
        </p>
      </div>

      {expanded && (
        <div className="mt-3 max-h-96 overflow-auto rounded bg-gray-900 p-3">
          <pre className="text-xs text-gray-200">
            {JSON.stringify(checkpoint.values, null, 2)}
          </pre>
        </div>
      )}
    </div>
  );
}
对于生产环境的 UI,建议使用具有可折叠节点的专用 JSON 查看器组件,而不是原始的 JSON.stringify。像 react-json-viewreact-json-tree 这样的库能为用户提供更好的探索体验。

从检查点恢复

时间旅行的核心功能是能够从任何先前的检查点恢复执行。当用户选择一个检查点时,使用 null 输入调用 submit 并传递检查点引用:
stream.submit(null, { checkpoint: selectedCheckpoint.checkpoint });
这会告诉 LangGraph:
  1. 回滚到所选检查点的状态
  2. 从该点开始重新执行图
  3. 将新结果流式传输到客户端
所选检查点之后的现有消息将被新执行路径替换。这有效地在对话时间线中创建了一个分支
从检查点恢复不会删除原始时间线。先前的检查点仍保留在历史记录中。这意味着用户始终可以返回并尝试不同的路径,而不会丢失任何先前的工作。

SplitView 布局

时间旅行最适合采用分屏布局,左侧是主聊天区域,右侧是时间线:
function TimeTravelLayout() {
  const stream = useStream<typeof myAgent>({
    apiUrl: AGENT_URL,
    assistantId: "time_travel",
    fetchStateHistory: true,
  });

  const [selectedCheckpoint, setSelectedCheckpoint] =
    useState<ThreadState | null>(null);

  const history = stream.history ?? [];

  return (
    <div className="flex h-screen">
      {/* 主聊天区域 */}
      <main className="flex-1 overflow-y-auto p-6">
        <div className="mx-auto max-w-2xl space-y-4">
          {stream.messages.map((msg) => (
            <Message key={msg.id} message={msg} />
          ))}
        </div>
        <ChatInput
          onSubmit={(text) =>
            stream.submit({ messages: [{ type: "human", content: text }] })
          }
          isLoading={stream.isLoading}
        />
      </main>

      {/* 时间线侧边栏 */}
      <aside className="w-96 overflow-y-auto border-l bg-gray-50">
        <TimelineSidebar
          history={history}
          selected={selectedCheckpoint}
          onSelect={setSelectedCheckpoint}
          onResume={(cp) =>
            stream.submit(null, { checkpoint: cp.checkpoint })
          }
        />
        {selectedCheckpoint && (
          <CheckpointInspector checkpoint={selectedCheckpoint} />
        )}
      </aside>
    </div>
  );
}

提取检查点元数据

将原始检查点数据转换为适合时间线显示的条目:
function formatCheckpoints(history: ThreadState[]) {
  return history.map((cp, index) => ({
    index,
    id: cp.checkpoint?.checkpoint_id,
    taskName: cp.tasks?.[0]?.name ?? "unknown",
    messageCount: (cp.values?.messages as unknown[])?.length ?? 0,
    hasInterrupts: cp.tasks?.some((t) => t.interrupts?.length) ?? false,
    nextNodes: cp.next ?? [],
  }));
}
这使得渲染具有有意义标签的时间线条目变得容易,而不是原始 ID。

用例

时间旅行在许多场景中都非常有价值:
  • 调试代理行为:逐步检查代理的决策,以理解其为何选择特定路径
  • 撤销操作:如果代理走错了路,可以从更早的检查点恢复并重试
  • 探索替代方案:从对话中途的检查点分叉,查看不同输入如何改变结果
  • 审计:审查代理操作的完整历史记录,用于合规性、质量保证或事件后分析
  • 教学:逐步讲解代理的执行过程,解释多步推理的工作原理
时间旅行与人机回环模式结合使用时尤其强大。如果人类审查者在中断处拒绝了代理的操作,他们可以从操作执行前的检查点恢复并提供纠正性输入。

处理时间线中的中断

包含中断(人机回环暂停)的检查点需要特殊的视觉处理。它们代表代理停止并等待人类输入的时刻:
function TimelineEntry({
  checkpoint,
  index,
}: {
  checkpoint: ThreadState;
  index: number;
}) {
  const hasInterrupt = checkpoint.tasks?.some(
    (t) => t.interrupts && t.interrupts.length > 0
  );

  return (
    <div
      className={`rounded-lg border p-3 ${
        hasInterrupt
          ? "border-amber-300 bg-amber-50"
          : "border-gray-200 bg-white"
      }`}
    >
      <div className="flex items-center gap-2">
        <span className="text-xs text-gray-400">#{index + 1}</span>
        {hasInterrupt && (
          <span className="rounded bg-amber-200 px-1.5 py-0.5 text-xs font-medium text-amber-800">
            Interrupt
          </span>
        )}
      </div>
      <p className="mt-1 text-sm font-medium">
        {checkpoint.tasks?.[0]?.name ?? "—"}
      </p>
    </div>
  );
}

最佳实践

  • 延迟加载历史记录:对于包含数百个检查点的线程,进行分页或仅加载最近的 N 个条目,以保持 UI 响应性。
  • 显示有意义的标签:显示节点名称和消息数量,而不是原始检查点 ID。用户需要上下文,而不是 UUID。
  • 恢复前确认:从旧检查点恢复会替换当前执行路径。显示确认对话框,以免用户意外丢失当前对话状态。
  • 突出显示当前检查点:在视觉上明确哪个检查点对应于对话的当前状态。
  • 支持键盘导航:高级用户会希望使用箭头键逐步检查点。为时间线添加键盘处理器,以实现流畅的调试体验。
  • 检查点间状态差异:对于高级用户,显示两个连续检查点之间的变化可以揭示代理状态在每个步骤中如何演变。