时间旅行 (Time travel)

在对话历史中审查、漫游并从任意检查点恢复

LangGraph 智能体中的每一次状态变化都会创建一个检查点（checkpoint），即智能体在这一时刻状态的完整快照。时间旅行（Time travel）允许你审查任何检查点，查看智能体当时所处的精确状态，并从该点恢复执行以探索替代路径。它集调试器、撤销按钮和审计日志于一身。

NOTE

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

检查点的工作原理

LangGraph 在每次节点执行后都会持久化智能体状态。每个持久化的状态都是一个 ThreadState 对象，它包含：

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

这将创建智能体所做出的每一个决策、调用的每一个工具以及生成的每一个响应的线性时间线。你的 UI 可以渲染此时间线，并允许用户跳转到任意点。

配置 `useStream`

为你的智能体创建流，然后显式地从 LangGraph 客户端获取活动线程的检查点历史记录。从检查点恢复使用 forkFrom: { checkpointId }。

NOTE

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

ReactVueSvelteAngular

tsx

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

vue

<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 },
  });
}
`&lt;/script&gt;`

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

svelte

<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 },
    });
  }
`&lt;/script&gt;`

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

构建检查点时间线

时间线侧边栏将每个检查点展示为一个可点击的条目。每个条目显示所运行的节点以及该时刻存在的消息数量：

tsx

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 查看器让开发者能够完全看清智能体当时所了解的信息和所做的决定：

tsx

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

TIP

对于生产环境的 UI，建议考虑使用支持折叠节点的专用 JSON 查看器组件，而不是原生的 JSON.stringify。类似 react-json-view 或 react-json-tree 这样的库能为用户带来更好的浏览体验。

从检查点恢复

时间旅行的核心在于能够从任意先前的检查点恢复执行。当用户选择一个检查点时，传入 null 输入调用 submit 并传递检查点 ID：

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

这指示 LangGraph：

回滚到所选检查点的状态
从该点向前重新执行图
将新结果流式传输给客户端

所选检查点之后的已有消息将被新的执行路径替换。这实际上在对话时间线中创建了一个分支（branch）。

NOTE

从检查点恢复不会删除原始时间线。先前的检查点在历史记录中仍然可用。这意味着用户随时可以回退并尝试不同的路径，而不会丢失任何先前的工作。

SplitView 布局

时间旅行最适合采用左右分割布局，左侧为主聊天区，右侧为时间线：

tsx

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 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({}, {
              forkFrom: { checkpointId: cp.checkpoint.checkpoint_id },
            })
          }
        />
        {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。

使用场景

时间旅行在许多场景中都极具价值：

调试智能体行为：单步跟踪智能体的决策过程，了解它为什么选择某条特定路径
撤销操作：如果智能体误入歧途，可以从更早的检查点恢复并重试
探索替代方案：从对话途中的检查点派生分支，观察不同输入如何改变输出结果
安全审计：审查智能体操作的完整历史记录，用于合规性、质量保证或事后事件分析
教学演示：逐步演示智能体的执行过程，解释多步推理的工作原理

NOTE

当与人在回路（human-in-the-loop）模式结合时，时间旅行尤为强大。如果人工审核员在中断处拒绝了智能体的操作，他们可以从执行该操作之前的检查点恢复，并提供修正输入。

在时间线中处理中断

包含中断（人在回路暂停）的检查点值得进行特殊的视觉处理。它们代表了智能体停止并等待人工输入的时刻：

tsx

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。
恢复前进行确认：从旧检查点恢复会替换当前的执行路径。显示确认对话框，防止用户意外丢失当前的对话状态。
高亮当前检查点：在视觉上清晰标明哪个检查点对应当前对话状态。
支持键盘导航：高级用户希望能够使用方向键单步浏览检查点。为时间线添加键盘事件处理，以提供流畅的调试体验。
比较检查点之间的状态差异（Diff）：对于高级用户，展示两个连续检查点之间的变化可以确切揭示智能体状态在每一步是如何演变的。

时间旅行 (Time travel) ​

检查点的工作原理 ​

配置 useStream ​

构建检查点时间线 ​

检查检查点状态 ​

从检查点恢复 ​

SplitView 布局 ​

提取检查点元数据 ​

使用场景 ​

在时间线中处理中断 ​

最佳实践 ​

时间旅行 (Time travel)

检查点的工作原理

配置 `useStream`

构建检查点时间线

检查检查点状态

从检查点恢复

SplitView 布局

提取检查点元数据

使用场景

在时间线中处理中断

最佳实践