LangGraph 代理中的每次状态变化都会创建一个 checkpoint,即该时刻代理状态的完整快照。时间旅行让你可以检查任意 checkpoint,查看代理当时持有的确切状态,并从该点恢复执行以探索替代路径。它同时是调试器、撤销按钮和审计日志。
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>
);
}
构建 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 之间发生了什么变化,可以揭示代理状态在每一步的具体演化。