与 AI 代理的对话很少是线性的。你可能想重新表述问题、重新生成不满意的响应,或探索不同对话路径,同时不丢失 checkpoint 历史。分支聊天使用 LangGraph checkpoints 作为 fork 点:每次编辑或重新生成都会从所选消息的父 checkpoint 提交一次新的运行。
什么是分支聊天?
分支聊天将对话视为带 checkpoint 的时间线,而不是扁平列表。每条消息都有元数据,指向创建该消息之前的 checkpoint。编辑消息或重新生成响应,会从该 checkpoint 提交一次新的运行。
关键能力:
- 编辑任意用户消息:重写先前提示词,并从该点重新运行代理
- 重新生成任意 AI 响应:让代理基于相同输入生成不同答案
- 检查历史:需要分支时间线时,使用 LangGraph 客户端加载 checkpoints
设置流元数据
对消息使用根流,然后在渲染每条消息的组件中读取逐消息 checkpoint 元数据。该元数据包含用于 fork 的父 checkpoint ID。
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";
const AGENT_URL = "http://localhost:2024";
export function Chat() {
const stream = useStream<typeof myAgent>({
apiUrl: AGENT_URL,
assistantId: "simple_agent",
});
return (
<div>
{stream.messages.map((msg) => (
<MessageWithForkControls key={msg.id} stream={stream} message={msg} />
))}
</div>
);
}
理解消息元数据
useMessageMetadata(stream, messageId) helper 会返回某条消息的 MessageMetadata。在渲染每条消息的组件中使用它,让元数据保持限定在该消息 ID 上:
import type { BaseMessage } from "langchain";
import { useState } from "react";
import { useMessageMetadata, useStream } from "@langchain/react";
function Chat() {
const stream = useStream<typeof myAgent>({
apiUrl: AGENT_URL,
assistantId: "simple_agent",
});
return stream.messages.map((message) => (
<MessageWithForkControls
key={message.id}
stream={stream}
message={message}
/>
));
}
function MessageWithForkControls({
stream,
message,
}: {
stream: ReturnType<typeof useStream>;
message: BaseMessage;
}) {
const metadata = useMessageMetadata(stream, message.id);
const checkpointId = metadata?.parentCheckpointId;
const [editedText, setEditedText] = useState(message.text);
return (
<form
onSubmit={(event) => {
event.preventDefault();
if (!checkpointId) return;
stream.submit(
{ messages: [{ type: "human", content: editedText }] },
{ forkFrom: { checkpointId } }
);
}}
>
<textarea
value={editedText}
onChange={(event) => setEditedText(event.target.value)}
/>
<button disabled={!checkpointId || editedText === message.text}>
Submit edited branch
</button>
</form>
);
}
parentCheckpointId 是消息之前的 checkpoint。将它用作编辑和重新生成的 fork 点。
编辑消息
若要编辑用户消息并 fork 对话:
- 从消息元数据中获取
parentCheckpointId
- 使用
forkFrom: { checkpointId } 提交编辑后的消息
- 代理会从该点重新运行
function handleEdit(
stream: ReturnType<typeof useStream>,
originalMsg: HumanMessage,
metadata: MessageMetadata | undefined,
newText: string
) {
if (!metadata?.parentCheckpointId) return;
stream.submit(
{
messages: [{ type: "human", content: newText }],
},
{ forkFrom: { checkpointId: metadata.parentCheckpointId } }
);
}
编辑后:
- 代理会携带更新后的消息从 fork 点重新运行
- 原始路径仍保留在线程历史中
重新生成响应
若要在不更改输入的情况下重新生成 AI 响应:
- 从 AI 消息元数据中获取
parent_checkpoint
- 使用空输入和
forkFrom: { checkpointId } 提交
- 代理会从该点生成新的响应
function handleRegenerate(
stream: ReturnType<typeof useStream>,
metadata: MessageMetadata | undefined
) {
if (!metadata?.parentCheckpointId) return;
stream.submit(undefined, {
forkFrom: { checkpointId: metadata.parentCheckpointId },
});
}
每次重新生成都会在该位置为 AI 消息创建一条新路径。
重新生成对非确定性代理很有用。由于 LLM 输出会随 temperature 变化,重新生成同一提示词通常会产生有意义的不同响应。
分支底层如何工作
LangGraph 会将每次状态转换持久化为 checkpoint。当你使用 forkFrom 提交时,后端会从该点启动新的执行路径,而不是追加到当前对话。结果是一个树结构:
User: "What is React?"
└─ AI: "React is a JavaScript library..." (branch A)
└─ AI: "React is a UI framework..." (branch B, regenerated)
User: "Tell me about hooks" (branch A)
└─ AI: "Hooks are functions..."
User: "Tell me about JSX" (edited from branch A)
└─ AI: "JSX is a syntax extension..."
每条路径都会持久化在 checkpoint store 中。当你想构建跨 checkpoints 的独立时间线视图时,请使用 stream.client.threads.getHistory(threadId)。
最佳实践
- 在消息附近读取元数据:在渲染消息控件的组件中调用
useMessageMetadata。
- 悬停时显示 fork 控件:编辑和重新生成按钮应在悬停时出现,以保持 UI 简洁。
- 按需刷新历史:仅在渲染时间线或 fork 稳定后调用
client.threads.getHistory()。
- 流式传输时禁用控件:当代理正在主动流式传输响应时,不要允许编辑或重新生成。启用这些操作前检查
stream.isLoading。
- 取消时保留编辑文本:如果用户开始编辑后又取消,请将 textarea 重置为原始消息内容。
- 使用深 checkpoint 树测试:频繁编辑和重新生成的用户可能会创建许多路径。确保时间线渲染仍保持高性能。