时间旅行 | LangChain 中文文档

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

import { PatternEmbed } from "/snippets/pattern-embed.jsx"

INFO

This feature requires the LangGraph Agent Server. Run your agent locally with langgraph dev or deploy it to LangSmith to use this pattern.

检查点如何工作

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

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

这会形成一条线性时间线，记录智能体做出的每个决策、调用的每个工具以及产生的每个响应。你的界面可以渲染这条时间线，让用户跳转到任意时间点。

设置 `useStream`

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

INFO

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

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 (
      <ChatPanel messages={stream.messages} />
      <TimelineSidebar history={history} onSelect={resumeFrom} />
  );
}

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

<template>
    <ChatPanel :messages="stream.messages.value" />
    <TimelineSidebar :history="history" @select="resumeFrom" />
</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 },
    });
  }
</script>

  <ChatPanel messages={stream.messages} />
  <TimelineSidebar {history} onSelect={resumeFrom} />

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: `
      <app-chat-panel [messages]="stream.messages()" />
      <app-timeline-sidebar
        [history]="history()"
        (select)="resumeFrom($event)"
      />
  `,
})
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">
        Checkpoint Timeline
        {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"
            >
                #{i + 1}
                <NodeBadge name={taskName} />
              {taskName}
                {msgCount} message{msgCount !== 1 ? "s" : ""}
            </button>
          );
        })}
    </aside>
  );
}

检查检查点状态

点击某个检查点应显示该时刻的完整状态。JSON 查看器能让开发者完整地看到智能体当时知道和决定了什么：

tsx

function CheckpointInspector({ checkpoint }: { checkpoint: ThreadState }) {
  const [expanded, setExpanded] = useState(false);

  return (
          Checkpoint {checkpoint.checkpoint.checkpoint_id.slice(0, 8)}...
        <button
          onClick={() => setExpanded(!expanded)}
          className="text-sm text-blue-600 hover:underline"
        >
          {expanded ? "Collapse" : "Expand"} state
        </button>

          Node:{" "}
          {checkpoint.tasks?.[0]?.name ?? "—"}
          Next:{" "}
          {checkpoint.next?.join(", ") || "—"}
          Messages:{" "}
          {(checkpoint.values?.messages as unknown[])?.length ?? 0}

      {expanded && (
          <pre className="text-xs text-gray-200">
            {JSON.stringify(checkpoint.values, null, 2)}
          </pre>
      )}
  );
}

TIP

对于生产环境界面，考虑使用带有可折叠节点的正规 JSON 查看器组件，而不是原始的 JSON.stringify。像 react-json-view 或 react-json-tree 这样的库能为用户提供更好的探索体验。

从检查点恢复

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

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

这会让 LangGraph：

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

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

INFO

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

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 (
      {/* Main chat area */}
      <main className="flex-1 overflow-y-auto p-6">
          {stream.messages.map((msg) => (
            <Message key={msg.id} message={msg} />
          ))}
        <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>
  );
}

提取检查点元数据

将原始的检查点数据转换为适合展示的时间线条目：

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。

使用场景

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

调试智能体行为：逐步检查智能体的决策，理解它为什么选择某条特定路径
撤销操作：如果智能体走了错误方向，从更早的检查点恢复并重新尝试
探索替代方案：从对话中途的检查点分叉，看看不同输入如何改变结果
审计：为合规、质量保证或事后分析审查智能体行为的完整历史
教学：一步步讲解智能体的执行过程，解释多步骤推理是如何工作的

INFO

时间旅行与 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"
      }`}
    >
        #{index + 1}
        {hasInterrupt && (
            Interrupt
        )}
        {checkpoint.tasks?.[0]?.name ?? "—"}
  );
}

最佳实践

惰性加载历史：对于包含数百个检查点的线程，分页或只加载最近的 N 个条目，以保持界面响应。
显示有意义的标签：展示节点名称和消息数量，而不是原始的检查点 ID。用户需要的是上下文，而不是 UUID。
恢复前先确认：从旧检查点恢复会替换当前的执行路径。显示确认对话框，以免用户意外丢失当前的对话状态。
突出当前检查点：在视觉上清楚标明哪个检查点对应对话的当前状态。
支持键盘导航：高级用户会希望用方向键在检查点之间移动。为时间线添加键盘处理程序，以获得流畅的调试体验。
比较检查点之间的状态差异：对于高级用户，显示两个连续检查点之间变化了什么，可以精确揭示智能体状态在每一步是如何演变的。

Patterns

Integrations

Multi-agent

Test

检查点如何工作

设置 `useStream`

构建检查点时间线

检查检查点状态

从检查点恢复

SplitView 布局

提取检查点元数据

使用场景

在时间线中处理中断

最佳实践

检查点如何工作 ​

设置 useStream ​

构建检查点时间线 ​

检查检查点状态 ​

从检查点恢复 ​

SplitView 布局 ​

提取检查点元数据 ​

使用场景 ​

在时间线中处理中断 ​

最佳实践 ​

检查点如何工作

设置 `useStream`

构建检查点时间线

检查检查点状态

从检查点恢复

SplitView 布局

提取检查点元数据

使用场景

在时间线中处理中断

最佳实践