Files
deerflow-code/frontend-web/src/core/threads/use-multi-round-rag.ts
T
董宏凯 438772f0b7 feat(frontend): 上线报告协作与多轮 RAG 消息聚合体验
功能说明:\n- 新增报告协作页(Report Collaboration)核心状态层、消息适配与会话面板,支持按节点/阶段选择和协作上下文展示\n- 补齐 report-collaboration API 与类型定义,并接入页面路由/构建路径,形成可运行的协作链路\n- 重构多轮 RAG 与 write/artefact 消息分组逻辑,优化 AgentChat 与 DeepResearch 场景下的消息展示顺序与可读性\n- 新增/更新 DeepResearch 前端页面(Workbench、报告配置、沙箱联动)以承接报告生成与复用能力\n- 增加报表协作相关单测,覆盖事件归并、消息适配与关键 reducer 行为
2026-09-04 23:18:27 +08:00

968 lines
30 KiB
TypeScript

import type { Message } from "@langchain/langgraph-sdk";
import { useCallback, useEffect, useRef, useState } from "react";
import { readRuntimeEnv } from "@/env";
import { createClientId } from "@/lib/client-id";
const STORAGE_PREFIX = "workspace.multi-round-rag.v2.";
export type MultiRoundRagRetrieval = {
id?: string;
title: string;
content: string;
source?: string;
time?: string;
recUuid?: string;
url?: string;
type?: string;
keywords?: string;
taskDirection?: string;
platform?: string;
fans?: number;
comment_count?: number;
message_id?: string;
hot_num?: number;
};
export type MultiRoundRagRound = {
id: string;
title: string;
thoughtMarkdown: string;
queryTitle?: string;
query: string;
queryLoading?: boolean;
queryShow: boolean;
textidContent?: string;
extractInfo?: string;
extractContent?: string;
extractLoading?: boolean;
retrievals: MultiRoundRagRetrieval[];
};
export type MultiRoundRagTrace = {
title?: string;
rounds: MultiRoundRagRound[];
finalAnswerMarkdown: string;
thinkingComplete?: boolean;
};
export type MultiRoundRagSource = {
title: string;
type: string;
snippet: string;
source?: string;
time?: string;
url?: string;
content?: string;
recUuid?: string;
};
export type MultiRoundRagChatMessage = {
id: string;
role: "user" | "assistant";
content: string;
timestamp: string;
thinkingTrace?: MultiRoundRagTrace;
sources?: MultiRoundRagSource[];
};
type MultiRoundEvent = {
event?: unknown;
data?: unknown;
};
function storageKey(threadId: string) {
return `${STORAGE_PREFIX}${threadId}`;
}
function readStoredMessages(threadId: string): MultiRoundRagChatMessage[] {
if (typeof window === "undefined" || !threadId) return [];
try {
const raw = window.sessionStorage.getItem(storageKey(threadId));
const parsed: unknown = raw ? JSON.parse(raw) : [];
if (!Array.isArray(parsed)) return [];
const messages = (parsed as Array<MultiRoundRagChatMessage & { isDemo?: boolean }>)
.filter((message) => !message.isDemo)
.map(({ isDemo: _isDemo, ...message }) => message);
if (messages.length !== parsed.length) {
writeStoredMessages(threadId, messages);
}
return messages;
} catch {
return [];
}
}
function writeStoredMessages(threadId: string, messages: MultiRoundRagChatMessage[]) {
if (typeof window === "undefined" || !threadId) return;
try {
window.sessionStorage.setItem(storageKey(threadId), JSON.stringify(messages));
} catch {
// Storage quota failures must not interrupt a running answer.
}
}
function createId(prefix: string) {
return createClientId(prefix);
}
function isAbortError(error: unknown) {
return error instanceof DOMException && error.name === "AbortError";
}
function asRecord(value: unknown): Record<string, unknown> | null {
return value && typeof value === "object" && !Array.isArray(value)
? (value as Record<string, unknown>)
: null;
}
function asText(value: unknown): string {
if (typeof value === "string") return value;
if (typeof value === "number" && Number.isFinite(value)) return String(value);
return "";
}
function parseJson(value: unknown): unknown {
if (typeof value !== "string") return value;
try {
return JSON.parse(value) as unknown;
} catch {
return value;
}
}
function normalizeToolName(name: string) {
return name.trim().replace(/-/g, "_");
}
function platformTypeChange(platform: string) {
if (platform.includes("fb")) return "facebook";
if (platform.includes("x")) return "x";
if (platform.includes("in")) return "instagram";
if (platform.includes("re")) return "reddit";
if (platform.includes("th")) return "threads";
return platform || "facebook";
}
function createEmptyRound(title: string, thoughtMarkdown = ""): MultiRoundRagRound {
return {
id: createId("multi-rag-round"),
title,
thoughtMarkdown,
query: "",
queryLoading: false,
queryShow: false,
retrievals: [],
};
}
function createEmptyTrace(): MultiRoundRagTrace {
return {
title: "多轮检索思考过程",
rounds: [createEmptyRound("正在准备检索")],
thinkingComplete: false,
finalAnswerMarkdown: "",
};
}
function cloneTrace(trace: MultiRoundRagTrace): MultiRoundRagTrace {
return {
...trace,
rounds: trace.rounds.map((round) => ({
...round,
retrievals: [...round.retrievals],
})),
};
}
function lastRound(trace: MultiRoundRagTrace): MultiRoundRagRound {
const current = trace.rounds.at(-1);
if (current) return current;
const next = createEmptyTrace().rounds[0]!;
trace.rounds.push(next);
return next;
}
function eventsFromLine(line: string): MultiRoundEvent[] {
const trimmed = line.trim();
if (!trimmed || trimmed.startsWith(":")) return [];
let payload = trimmed;
if (/^data:\s?/i.test(trimmed)) {
payload = trimmed.replace(/^data:\s?/i, "").trim();
} else if (/^(id|event|retry):/i.test(trimmed)) {
return [];
}
if (!payload) return [];
if (payload === "[DONE]") return [{ event: "done" }];
if (payload.startsWith("{") || payload.startsWith("[")) {
try {
return expandStreamEvents(JSON.parse(payload));
} catch {
return [];
}
}
return expandStreamEvents(payload);
}
function asStreamEvent(value: unknown): MultiRoundEvent | null {
const parsed = parseJson(value);
if (Array.isArray(parsed)) {
return null;
}
const record = asRecord(parsed);
if (!record) {
if (typeof parsed === "string" && parsed.trim() && !/^[{[]/.test(parsed.trim())) {
return { event: "message", data: parsed };
}
return null;
}
const eventName = asText(record.event ?? record.type ?? record.event_type);
if (eventName) {
return {
event: eventName,
data: record.data !== undefined ? record.data : (record.payload ?? record),
};
}
if (record.tool_name || record.name) {
return { event: "tool_call", data: record };
}
return record;
}
function expandStreamEvents(value: unknown): MultiRoundEvent[] {
const parsed = parseJson(value);
if (Array.isArray(parsed)) {
return parsed.flatMap((item) => expandStreamEvents(item));
}
const event = asStreamEvent(parsed);
return event ? [event] : [];
}
function extractJsonObjects(raw: string): { objects: string[]; rest: string } {
const objects: string[] = [];
let depth = 0;
let stringQuote: '"' | "'" | null = null;
let escaped = false;
let start = -1;
for (let index = 0; index < raw.length; index += 1) {
const char = raw[index]!;
if (stringQuote) {
if (escaped) escaped = false;
else if (char === "\\") escaped = true;
else if (char === stringQuote) stringQuote = null;
continue;
}
if (char === '"' || char === "'") {
stringQuote = char;
continue;
}
if (char === "{") {
if (depth === 0) start = index;
depth += 1;
} else if (char === "}") {
depth -= 1;
if (depth === 0 && start !== -1) {
objects.push(raw.slice(start, index + 1));
start = -1;
}
}
}
return depth > 0 && start !== -1
? { objects, rest: raw.slice(start) }
: { objects, rest: "" };
}
function pullStreamEvents(buffer: string): { events: MultiRoundEvent[]; rest: string } {
const events: MultiRoundEvent[] = [];
const lines = buffer.split(/\r?\n/);
const restLine = lines.pop() ?? "";
for (const line of lines) {
events.push(...eventsFromLine(line));
}
if (events.length > 0) {
return { events, rest: restLine };
}
const extracted = extractJsonObjects(buffer);
for (const jsonText of extracted.objects) {
events.push(...expandStreamEvents(parseJson(jsonText)));
}
return { events, rest: extracted.rest || restLine };
}
function mapReferenceToSource(item: unknown, index: number): MultiRoundRagSource {
const source = asRecord(item) ?? {};
const content = asText(source.content ?? source.m_content ?? source.snippet ?? source.text);
const rawUrl = asText(source.url ?? source.link ?? source.href);
const recUuid = asText(source.recUuid ?? source.rec_uuid ?? source.uuid);
return {
title: asText(source.title ?? source.m_title) || `参考资料 ${index + 1}`,
type: asText(source.type) || "参考资料",
snippet: asText(source.snippet) || content,
content,
source: asText(source.source ?? source.source1 ?? source.m_source ?? source.platform) || undefined,
time: asText(source.time ?? source.m_publish ?? source.publish_time ?? source.date) || undefined,
url: /^https?:\/\//i.test(rawUrl) ? rawUrl : undefined,
recUuid: recUuid || undefined,
};
}
function parseResult<T>(raw: unknown): T | null {
const parsed = parseJson(raw);
return parsed == null ? null : (parsed as T);
}
function extractInfoText(result: unknown): string {
const parsed = parseJson(result);
if (typeof parsed === "string") return parsed;
const record = asRecord(parsed);
return asText(
record?.content ??
record?.text ??
record?.extracted ??
record?.info ??
record?.summary ??
record?.result,
);
}
function mergeThink(existing: string, incoming: string) {
if (!incoming) return existing;
if (!existing) return incoming;
if (incoming.startsWith(existing)) return incoming;
if (existing.endsWith(incoming)) return existing;
return `${existing}${incoming}`;
}
function extractThinkContent(chatText: string) {
const closed = chatText.match(/<think>([\s\S]*?)<\/think>/i);
if (closed?.[1] != null) return closed[1].trim();
const open = chatText.match(/<think>([\s\S]*)$/i);
if (open?.[1] != null) return open[1].trim();
return chatText.replace(/<\/?think>/gi, "").trim();
}
/**
* GeneralQA `handleToolCall` 的独立实现:按 tool_name 把 SSE 工具事件填进最近一轮 trace。
* `show_text` / `show-text` 负责思考与最终回答;检索类工具先写检索词,再把 result 渲染成卡片。
*/
export function applyMultiRoundToolCall(trace: MultiRoundRagTrace, toolValue: unknown) {
const tool = asRecord(toolValue);
if (!tool) return;
const toolName = normalizeToolName(asText(tool.tool_name ?? tool.name));
const input =
asRecord(parseJson(tool.tool_input ?? tool.input ?? tool.arguments)) ?? {};
const round = lastRound(trace);
switch (toolName) {
case "show_text": {
const chatText = asText(input.text ?? input.content);
if (!chatText) return;
if (chatText.includes("<think>")) {
const thinkContent = extractThinkContent(chatText);
const firstRound = trace.rounds[0];
const stillOnFirstThink =
firstRound?.title === "正在准备检索" ||
((!round.query && round.retrievals.length === 0) &&
(round.title === "思考过程" || round.title === "思考中" || round.title === "正在准备检索"));
if (stillOnFirstThink && firstRound) {
firstRound.title = "思考过程";
firstRound.thoughtMarkdown = mergeThink(firstRound.thoughtMarkdown, thinkContent);
} else if (!round.query && round.retrievals.length === 0) {
round.title = round.title === "正在准备检索" ? "思考过程" : round.title;
round.thoughtMarkdown = mergeThink(round.thoughtMarkdown, thinkContent);
} else {
round.queryLoading = false;
round.queryShow = Boolean(round.query) || round.retrievals.length > 0;
trace.rounds.push(createEmptyRound("思考中", thinkContent));
}
} else {
round.queryLoading = false;
trace.finalAnswerMarkdown += chatText;
trace.thinkingComplete = true;
}
return;
}
case "tavily_search":
case "local_search": {
if (input.result) {
const result = parseResult<{
query?: string;
results?: Array<{ title?: string; content?: string; url?: string }>;
}>(input.result);
round.query = result?.query || round.query;
round.queryLoading = false;
round.title = toolName === "local_search" ? "本地知识检索" : "联网检索";
round.retrievals = (result?.results || []).map((item, idx) => ({
id: `${toolName}-${idx}`,
title: item.title || "未命名结果",
content: item.content || "",
source: item.url || "",
url: item.url,
}));
}
return;
}
case "google_search": {
if (input.result) {
const result = parseResult<{
organic?: Array<{
title?: string;
content?: string;
url?: string;
recUuid?: string;
publish_time?: string;
}>;
}>(input.result);
round.queryLoading = false;
round.title = "思考过程";
round.retrievals = (result?.organic || []).map((item) => ({
id: item.recUuid,
title: item.title || "未命名结果",
content: item.content || "",
source: item.recUuid || "",
time: item.publish_time,
recUuid: item.recUuid,
url: item.url,
type: "more",
}));
} else if (asText(input.q)) {
round.query = asText(input.q);
round.queryTitle = "检索词";
round.queryShow = true;
round.queryLoading = true;
}
return;
}
case "getTaskLikeTop": {
if (input.result) {
const result = parseResult<{
data?: Array<{
taskName?: string;
overview?: string;
id?: number;
keywords?: string;
taskDirection?: string;
taskStartDate?: string;
taskEndDate?: string;
}>;
}>(input.result);
round.queryLoading = false;
round.title = "思考过程";
round.retrievals = (result?.data || []).map((item) => ({
id: String(item.id ?? ""),
title: item.taskName || "未命名结果",
content: item.overview || "",
keywords: item.keywords,
taskDirection: item.taskDirection,
time: `${item.taskStartDate ?? ""}-${item.taskEndDate ?? ""}`,
type: "task",
}));
} else if (asText(input.keyword)) {
round.query = asText(input.keyword);
round.queryShow = true;
round.queryLoading = true;
}
return;
}
case "search_user_profile": {
round.title = "靶标平台画像库";
if (input.result) {
const result = parseResult<
Array<{
user_name?: string;
user_id?: string;
page_content?: string;
平台?: string;
粉丝数?: number;
}>
>(input.result);
round.queryLoading = false;
round.retrievals = (result || []).map((item) => ({
id: item.user_id,
title: item.user_name || item.user_id || "-",
content: item.page_content || "",
platform: platformTypeChange(asText(item.平台)),
fans: item.粉丝数 || 0,
type: "hx",
}));
} else if (asText(input.keyword) || asText(input.query) || asText(input.user_fig)) {
round.queryTitle = "检索词";
round.query = asText(input.keyword) || asText(input.query) || asText(input.user_fig);
round.queryShow = true;
round.queryLoading = true;
}
return;
}
case "get_taiwan_news":
case "search_keyword_taiwan_news":
case "get_world_news":
case "get_america_news":
case "get_root_messaage_comment": {
const titles: Record<string, string> = {
get_taiwan_news: "台湾热点新闻",
search_keyword_taiwan_news: "台湾热点新闻",
get_world_news: "世界热点新闻",
get_america_news: "美国热点新闻",
get_root_messaage_comment: "获取该消息的全部评论",
};
const types: Record<string, string> = {
get_taiwan_news: "taiwan_news",
search_keyword_taiwan_news: "taiwan_news",
get_world_news: "world_news",
get_america_news: "america_news",
get_root_messaage_comment: "messaage_comment",
};
round.title = titles[toolName] ?? toolName;
if (input.result) {
const result = parseResult<
Array<{
user_name?: string;
user_id?: string;
content?: string;
publish_time?: string;
message_id?: string;
platform?: string;
comment_count?: number;
}>
>(input.result);
round.queryLoading = false;
round.retrievals = (result || []).map((item) => ({
id: item.user_id,
title: item.user_name || item.user_id || "-",
content: item.content || "",
platform: platformTypeChange(asText(item.platform)),
comment_count: item.comment_count,
message_id: item.message_id,
time: item.publish_time,
type: types[toolName],
}));
} else if (asText(input.keyword) || asText(input.query) || asText(input.platform)) {
round.queryTitle =
toolName === "get_america_news"
? "查询美国热点新闻"
: toolName === "get_world_news"
? "搜索世界新闻"
: toolName === "get_root_messaage_comment"
? "查找平台"
: "搜索台湾新闻";
round.query =
asText(input.keyword) ||
asText(input.query) ||
platformTypeChange(asText(input.platform));
round.queryShow = true;
round.queryLoading = true;
}
return;
}
case "search_user_behavior": {
round.title = "查询该用户的历史行为";
if (input.result) {
const result = parseResult<unknown[]>(input.result) ?? [];
round.queryLoading = false;
round.retrievals = result.slice(0, 10).map((item) => ({
id: asText(item),
title: asText(item),
content: asText(item),
type: "user_behavior",
}));
} else if (asText(input.keyword) || asText(input.query) || asText(input.platform)) {
round.queryTitle = "查找平台";
round.query =
asText(input.keyword) ||
asText(input.query) ||
platformTypeChange(asText(input.platform));
round.queryShow = true;
round.queryLoading = true;
}
return;
}
case "get_time_data": {
round.title = "态势新闻搜索";
if (input.result) {
const result = parseResult<
Array<{ title?: string; textid?: string; hot_num?: number; date?: string; content?: string }>
>(input.result);
round.queryLoading = false;
round.queryTitle = "查询态势新闻";
round.retrievals = (result || []).map((item) => ({
id: item.textid,
title: item.title || "未命名结果",
content: item.content || "",
time: item.date,
hot_num: item.hot_num,
type: "get_time_data",
}));
} else if (asText(input.keyword) || asText(input.query) || asText(input.s) || asText(input.e)) {
round.query = [asText(input.s), asText(input.e)].filter(Boolean).join(" - ");
round.queryShow = true;
round.queryLoading = true;
}
return;
}
case "get_textid_content": {
round.title = "获取新闻详细内容";
if (input.result) {
const result = parseResult<{ content?: string }>(input.result);
round.queryLoading = false;
round.textidContent = asText(result?.content);
}
return;
}
case "scrape_and_extract_info": {
const info = asText(input.info_to_extract ?? input.infoToExtract);
if (info) {
round.extractInfo = info;
round.extractLoading = !input.result;
}
if (input.result) {
round.extractLoading = false;
round.extractContent = extractInfoText(input.result);
}
return;
}
default:
return;
}
}
function consumeToolCallPayload(trace: MultiRoundRagTrace, payload: unknown) {
const parsed = parseJson(payload);
if (Array.isArray(parsed)) {
for (const tool of parsed) applyMultiRoundToolCall(trace, tool);
return;
}
const record = asRecord(parsed);
if (record?.tool_name || record?.name) {
applyMultiRoundToolCall(trace, record);
return;
}
if (record?.data != null) {
consumeToolCallPayload(trace, record.data);
return;
}
applyMultiRoundToolCall(trace, parsed);
}
function textFromEventData(value: unknown): string {
const parsed = parseJson(value);
if (typeof parsed === "string") return parsed;
const record = asRecord(parsed);
return asText(record?.text ?? record?.content ?? record?.message ?? record?.answer ?? record?.output);
}
function applyStreamEvent(
trace: MultiRoundRagTrace,
event: MultiRoundEvent,
onSources?: (sources: MultiRoundRagSource[]) => void,
onError?: (message: string) => void,
) {
const eventType = normalizeToolName(asText(event.event));
const data = event.data;
if (eventType === "tool_call") {
consumeToolCallPayload(trace, data);
} else if (eventType === "message") {
const text = textFromEventData(data);
if (text) {
lastRound(trace).queryLoading = false;
trace.finalAnswerMarkdown += text;
trace.thinkingComplete = true;
}
} else if (eventType === "references") {
const record = asRecord(parseJson(data));
const refs = Array.isArray(record?.references)
? record.references
: Array.isArray(parseJson(data))
? (parseJson(data) as unknown[])
: [];
onSources?.(refs.map(mapReferenceToSource));
} else if (eventType === "error") {
const message = textFromEventData(data) || "多轮 RAG 服务返回错误。";
onError?.(message);
if (!trace.finalAnswerMarkdown) trace.finalAnswerMarkdown = message;
finishTrace(trace);
} else if (eventType === "done") {
const finalText = textFromEventData(data);
if (finalText && !trace.finalAnswerMarkdown) trace.finalAnswerMarkdown += finalText;
finishTrace(trace);
} else if (asRecord(data)?.tool_name || asRecord(parseJson(data))?.tool_name) {
consumeToolCallPayload(trace, data);
}
}
function finishTrace(trace: MultiRoundRagTrace) {
trace.thinkingComplete = true;
for (const round of trace.rounds) {
round.queryLoading = false;
round.extractLoading = false;
if (round.retrievals.length > 0) round.queryShow = true;
}
}
async function* readMultiRoundSse(
baseUrl: string,
question: string,
signal: AbortSignal,
): AsyncGenerator<MultiRoundEvent> {
const response = await fetch(`${baseUrl.replace(/\/$/, "")}/generate`, {
method: "POST",
headers: {
"Content-Type": "application/json",
Accept: "text/event-stream, application/json, text/plain",
},
signal,
body: JSON.stringify({ prompt: question }),
});
if (!response.ok) {
throw new Error(`多轮 RAG 服务请求失败(${response.status})。`);
}
if (!response.body) {
throw new Error("多轮 RAG 服务未返回流式响应。");
}
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const pulled = pullStreamEvents(buffer);
buffer = pulled.rest;
for (const event of pulled.events) {
yield event;
if (normalizeToolName(asText(event.event)) === "done") return;
}
}
buffer += decoder.decode();
const pulled = pullStreamEvents(`${buffer}\n`);
for (const event of pulled.events) {
yield event;
}
if (pulled.rest.trim()) {
for (const event of [
...eventsFromLine(pulled.rest),
...expandStreamEvents(parseJson(pulled.rest)),
]) {
yield event;
}
}
}
/**
* Frontend adapter for GeneralQA's multi-round endpoint. It keeps the external
* SSE protocol outside DeerFlow and stores only this mode's own chat messages.
*/
export function useMultiRoundRagMessages(threadId: string) {
const [messages, setMessages] = useState<MultiRoundRagChatMessage[]>(() =>
readStoredMessages(threadId),
);
const messagesRef = useRef(messages);
const [isLoading, setIsLoading] = useState(false);
const [error, setError] = useState<string | null>(null);
const controllerRef = useRef<AbortController | null>(null);
const inFlightThreadIdRef = useRef<string | null>(null);
useEffect(() => {
if (inFlightThreadIdRef.current === threadId) return;
const stored = readStoredMessages(threadId);
messagesRef.current = stored;
setMessages(stored);
setError(null);
}, [threadId]);
useEffect(() => {
messagesRef.current = messages;
}, [messages]);
useEffect(() => () => controllerRef.current?.abort(), []);
const updateMessages = useCallback(
(targetThreadId: string, updater: (current: MultiRoundRagChatMessage[]) => MultiRoundRagChatMessage[]) => {
setMessages((current) => {
const next = updater(current);
messagesRef.current = next;
writeStoredMessages(targetThreadId, next);
return next;
});
},
[],
);
const stop = useCallback(() => controllerRef.current?.abort(), []);
const runWithEvents = useCallback(
async ({
targetThreadId,
query,
createEvents,
}: {
targetThreadId: string;
query: string;
createEvents: (signal: AbortSignal) => AsyncIterable<MultiRoundEvent>;
}) => {
const question = query.trim();
if (!question) return;
const humanId = createId("multi-rag-human");
const assistantId = createId("multi-rag-ai");
const trace = createEmptyTrace();
let sources: MultiRoundRagSource[] = [];
const timestamp = new Date().toISOString();
const userMessage: MultiRoundRagChatMessage = {
id: humanId,
role: "user",
content: question,
timestamp,
};
const assistantMessage: MultiRoundRagChatMessage = {
id: assistantId,
role: "assistant",
content: "",
timestamp,
thinkingTrace: cloneTrace(trace),
};
updateMessages(targetThreadId, (current) => [...current, userMessage, assistantMessage]);
inFlightThreadIdRef.current = targetThreadId;
setError(null);
setIsLoading(true);
const controller = new AbortController();
controllerRef.current = controller;
const refreshAssistant = () => {
const traceSnapshot = cloneTrace(trace);
updateMessages(targetThreadId, (current) =>
current.map((message) =>
message.id === assistantId
? {
...message,
content: traceSnapshot.finalAnswerMarkdown,
thinkingTrace: traceSnapshot,
sources: sources.length > 0 ? sources : message.sources,
}
: message,
),
);
};
try {
for await (const event of createEvents(controller.signal)) {
if (controller.signal.aborted) throw new DOMException("Aborted", "AbortError");
applyStreamEvent(
trace,
event,
(nextSources) => {
sources = nextSources;
},
setError,
);
refreshAssistant();
if (normalizeToolName(asText(event.event)) === "done") break;
}
finishTrace(trace);
refreshAssistant();
} catch (caught) {
finishTrace(trace);
if (!isAbortError(caught)) {
const message = caught instanceof Error ? caught.message : "多轮 RAG 问答失败。";
setError(message);
if (!trace.finalAnswerMarkdown) trace.finalAnswerMarkdown = message;
}
refreshAssistant();
} finally {
if (inFlightThreadIdRef.current === targetThreadId) {
inFlightThreadIdRef.current = null;
}
if (controllerRef.current === controller) controllerRef.current = null;
setIsLoading(false);
}
},
[updateMessages],
);
const submit = useCallback(
async ({ targetThreadId, query }: { targetThreadId: string; query: string }) => {
if (isLoading) return;
const question = query.trim();
if (!question) return;
const baseUrl = readRuntimeEnv("VITE_AIPI_GENERATE_CHAT_URL") || "";
if (!baseUrl) {
throw new Error("未配置多轮 RAG 服务地址(VITE_AIPI_GENERATE_CHAT_URL)。");
}
await runWithEvents({
targetThreadId,
query: question,
createEvents: (signal) => readMultiRoundSse(baseUrl, question, signal),
});
},
[isLoading, runWithEvents],
);
return { messages, isLoading, error, submit, stop };
}
export function toLangGraphMessages(messages: MultiRoundRagChatMessage[]): Message[] {
return messages.map((message) => {
if (message.role === "user") {
return {
id: message.id,
type: "human",
content: [{ type: "text", text: message.content }],
additional_kwargs: {
client_ts: message.timestamp,
multi_round_rag: true,
},
} as Message;
}
return {
id: message.id,
type: "ai",
content: message.content,
additional_kwargs: {
client_ts: message.timestamp,
multi_round_rag: true,
multi_round_rag_trace: message.thinkingTrace,
},
} as Message;
});
}
function messageTimestamp(message: Message, fallback: number) {
const raw = message.additional_kwargs?.client_ts;
if (typeof raw === "number" && Number.isFinite(raw)) return raw;
if (typeof raw === "string") {
const parsed = Date.parse(raw);
if (!Number.isNaN(parsed)) return parsed;
}
return fallback;
}
export function mergeMessagesByTime(base: Message[], extra: Message[]): Message[] {
if (extra.length === 0) return base;
if (base.length === 0) return extra;
const stamp = (messages: Message[], src: number) => {
let last = 0;
return messages.map((msg, idx) => {
const t = messageTimestamp(msg, last);
last = t;
return { msg, t, src, idx };
});
};
return [...stamp(base, 0), ...stamp(extra, 1)]
.sort((a, b) => a.t - b.t || a.src - b.src || a.idx - b.idx)
.map((item) => item.msg);
}
export function isMultiRoundRagMessage(message: { additional_kwargs?: Record<string, unknown> }): boolean {
return message.additional_kwargs?.multi_round_rag === true;
}
export function getMultiRoundRagTrace(
message: { additional_kwargs?: Record<string, unknown> },
): MultiRoundRagTrace | null {
const trace = message.additional_kwargs?.multi_round_rag_trace;
return trace && typeof trace === "object" ? (trace as MultiRoundRagTrace) : null;
}