时间旅行

LangGraph 代理中的每次状态变化都会创建一个 checkpoint，即该时刻代理状态的完整快照。时间旅行让你可以检查任意 checkpoint，查看代理当时持有的确切状态，并从该点恢复执行以探索替代路径。它同时是调试器、撤销按钮和审计日志。

此功能需要 LangGraph Agent Server。使用 langgraph dev 本地运行您的智能体，或将其部署到 LangSmith，以使用此模式。

Checkpoints 的工作原理

LangGraph 会在每个节点执行后持久化代理状态。每个持久化状态都是一个 ThreadState 对象，捕获以下内容：

checkpoint：标识该特定快照的元数据（ID、时间戳）
values：该点的完整代理状态（messages、自定义键）
tasks：计划接下来运行的图节点
next：执行计划中即将运行的节点名称

这会创建一条线性时间线，记录代理做出的每个决策、调用的每个工具和生成的每个响应。你的 UI 可以渲染这条时间线，并允许用户跳转到任意点。

设置 `useStream`

为你的代理创建流，然后从 LangGraph client 显式获取活动线程的 checkpoint 历史。从 checkpoint 恢复时使用 forkFrom: { checkpointId }。

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

import { useStream } from "@langchain/react";
import { useEffect, useState } from "react";

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

export function TimeTravelChat() {
  const [threadId, setThreadId] = useState<string | null>(null);
  const [history, setHistory] = useState<ThreadState[]>([]);
  const stream = useStream<typeof myAgent>({
    apiUrl: AGENT_URL,
    assistantId: "time_travel",
    threadId,
    onThreadId: setThreadId,
  });

  useEffect(() => {
    if (!threadId || stream.isLoading) return;
    stream.client.threads.getHistory(threadId).then(setHistory);
  }, [stream.client, threadId, stream.isLoading]);

  function resumeFrom(cp: ThreadState) {
    stream.submit({}, {
      forkFrom: { checkpointId: cp.checkpoint.checkpoint_id },
    });
  }

  return (
    <div className="flex h-screen">
      <ChatPanel messages={stream.messages} />
      <TimelineSidebar history={history} onSelect={resumeFrom} />
    </div>
  );
}

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

const AGENT_URL = "http://localhost:2024";
const threadId = ref<string | null>(null);
const history = ref<ThreadState[]>([]);

const stream = useStream<typeof myAgent>({
  apiUrl: AGENT_URL,
  assistantId: "time_travel",
  threadId,
  onThreadId: (id) => (threadId.value = id),
});

watch(
  [threadId, stream.isLoading],
  async ([id, isLoading]) => {
    if (isLoading) return;
    history.value = id
      ? ((await stream.client.threads.getHistory(id)) as ThreadState[])
      : [];
  },
  { immediate: true },
);

function resumeFrom(cp: ThreadState) {
  stream.submit({}, {
    forkFrom: { checkpointId: cp.checkpoint.checkpoint_id },
  });
}
</script>

<template>
  <div class="flex h-screen">
    <ChatPanel :messages="stream.messages.value" />
    <TimelineSidebar :history="history" @select="resumeFrom" />
  </div>
</template>

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

  const AGENT_URL = "http://localhost:2024";
  let threadId = $state<string | null>(null);
  let history = $state<ThreadState[]>([]);

  const stream = useStream<typeof myAgent>({
    apiUrl: AGENT_URL,
    assistantId: "time_travel",
    threadId: () => threadId,
    onThreadId: (id) => (threadId = id),
  });

  $effect(() => {
    if (!threadId) {
      history = [];
      return;
    }
    if (stream.isLoading) return;
    stream.client.threads.getHistory(threadId).then((states) => {
      history = states as ThreadState[];
    });
  });

  function resumeFrom(cp: ThreadState) {
    stream.submit({}, {
      forkFrom: { checkpointId: cp.checkpoint.checkpoint_id },
    });
  }
</script>

<div class="flex h-screen">
  <ChatPanel messages={stream.messages} />
  <TimelineSidebar {history} onSelect={resumeFrom} />
</div>

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

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

@Component({
  selector: "app-time-travel-chat",
  template: `
    <div class="flex h-screen">
      <app-chat-panel [messages]="stream.messages()" />
      <app-timeline-sidebar
        [history]="history()"
        (select)="resumeFrom($event)"
      />
    </div>
  `,
})
export class TimeTravelChatComponent {
  threadId = signal<string | null>(null);
  history = signal<ThreadState[]>([]);

  stream = injectStream<typeof myAgent>({
    apiUrl: AGENT_URL,
    assistantId: "time_travel",
    threadId: this.threadId,
    onThreadId: (id) => this.threadId.set(id),
  });

  constructor() {
    effect(() => {
      if (this.stream.isLoading()) return;
      void this.refreshHistory(this.threadId());
    });
  }

  async refreshHistory(id: string | null) {
    this.history.set(id
      ? ((await this.stream.client.threads.getHistory(id)) as ThreadState[])
      : []);
  }

  resumeFrom(cp: ThreadState) {
    this.stream.submit({}, {
      forkFrom: { checkpointId: cp.checkpoint.checkpoint_id },
    });
  }
}

构建 checkpoint 时间线

时间线侧边栏会将每个 checkpoint 显示为可点击条目。每个条目显示运行过的节点，以及该点存在的消息数量：

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

检查 checkpoint 状态

点击 checkpoint 应显示该点的完整状态。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-view 或 react-json-tree 等库可以为用户提供更好的探索体验。

从 checkpoint 恢复

时间旅行的核心能力是从任意先前 checkpoint 恢复执行。用户选择 checkpoint 时，使用 null input 调用 submit 并传入 checkpoint ID：

stream.submit({}, {
  forkFrom: { checkpointId: selectedCheckpoint.checkpoint.checkpoint_id },
});

这会告诉 LangGraph：

回滚到所选 checkpoint 的状态
从该点开始重新执行图
将新结果流式传输到客户端

所选 checkpoint 之后的现有消息会被新的执行路径替换。这实际上会在对话时间线中创建一个分支。

从 checkpoint 恢复不会删除原始时间线。先前 checkpoints 仍会保留在历史中。这意味着用户始终可以回去尝试不同路径，而不会丢失任何先前工作。

SplitView 布局

时间旅行最适合使用分栏布局，主聊天位于左侧，时间线位于右侧：

function TimeTravelLayout() {
  const [threadId, setThreadId] = useState<string | null>(null);
  const [history, setHistory] = useState<ThreadState[]>([]);
  const stream = useStream<typeof myAgent>({
    apiUrl: AGENT_URL,
    assistantId: "time_travel",
    threadId,
    onThreadId: setThreadId,
  });

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

  useEffect(() => {
    if (!threadId || stream.isLoading) return;
    stream.client.threads.getHistory(threadId).then(setHistory);
  }, [stream.client, threadId, stream.isLoading]);

  return (
    <div className="flex h-screen">
      {/* Main chat area */}
      <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>

      {/* Timeline sidebar */}
      <aside className="w-96 overflow-y-auto border-l bg-gray-50">
        <TimelineSidebar
          history={history}
          selected={selectedCheckpoint}
          onSelect={setSelectedCheckpoint}
          onResume={(cp) =>
            stream.submit({}, {
              forkFrom: { checkpointId: cp.checkpoint.checkpoint_id },
            })
          }
        />
        {selectedCheckpoint && (
          <CheckpointInspector checkpoint={selectedCheckpoint} />
        )}
      </aside>
    </div>
  );
}

提取 checkpoint 元数据

将原始 checkpoint 数据转换为适合在时间线中显示的条目：

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。

用例

时间旅行在许多场景中都很有价值：

调试代理行为：逐步查看代理决策，理解它为什么选择特定路径
撤销动作：如果代理走错方向，从较早的 checkpoint 恢复并重试
探索替代路径：从对话中途的 checkpoint fork，查看不同输入如何改变结果
审计：查看代理动作的完整历史，用于合规、质量保证或事故后分析
教学：逐步走查代理执行过程，解释多步推理如何工作

时间旅行与 human-in-the-loop 模式结合时尤其强大。如果人工审核者在 interrupt 处拒绝了代理动作，可以从该动作执行前的 checkpoint 恢复，并提供纠正输入。

在时间线中处理 interrupts

包含 interrupts（human-in-the-loop 暂停）的 checkpoints 应该有特殊视觉处理。它们代表代理停止并等待人工输入的时刻：

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

最佳实践

惰性加载历史：对于包含数百个 checkpoints 的线程，使用分页或只加载最近 N 个条目，以保持 UI 响应。
显示有意义的标签：显示节点名称和消息数，而不是原始 checkpoint ID。用户需要上下文，而不是 UUID。
恢复前确认：从旧 checkpoint 恢复会替换当前执行路径。显示确认对话框，避免用户意外丢失当前对话状态。
高亮当前 checkpoint：让用户清楚看到哪个 checkpoint 对应当前对话状态。
支持键盘导航：高级用户会想用方向键逐步浏览 checkpoints。为时间线添加键盘处理器，提供流畅的调试体验。
对比 checkpoints 之间的状态差异：对于高级用户，展示两个连续 checkpoints 之间发生了什么变化，可以揭示代理状态在每一步的具体演化。

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

Get started

Core components

Middleware

Frontend

Advanced usage

Agent development

Production

Checkpoints 的工作原理

设置 `useStream`

构建 checkpoint 时间线

检查 checkpoint 状态

从 checkpoint 恢复

SplitView 布局

提取 checkpoint 元数据

用例

在时间线中处理 interrupts

最佳实践

​Checkpoints 的工作原理

​设置 useStream

​构建 checkpoint 时间线

​检查 checkpoint 状态

​从 checkpoint 恢复

​SplitView 布局

​提取 checkpoint 元数据

​用例

​在时间线中处理 interrupts

​最佳实践

Checkpoints 的工作原理

设置 `useStream`

构建 checkpoint 时间线

检查 checkpoint 状态

从 checkpoint 恢复

SplitView 布局

提取 checkpoint 元数据

用例

在时间线中处理 interrupts

最佳实践