Browse Source

feat: 优化 chat-engine.ts 中的消息流处理逻辑

feat/character-card
npmrun 2 months ago
parent
commit
9a0b48ca61
  1. BIN
      packages/drizzle-pkg/db.sqlite
  2. 93
      server/service/agent/chat-engine.ts

BIN
packages/drizzle-pkg/db.sqlite

Binary file not shown.

93
server/service/agent/chat-engine.ts

@ -506,6 +506,9 @@ export async function executeChat(params: ChatEngineParams): Promise<ChatEngineR
let streamedReasoning = ""; let streamedReasoning = "";
let streamedToolCalls: Array<Record<string, unknown>> = []; let streamedToolCalls: Array<Record<string, unknown>> = [];
let chunkPartId = 0; let chunkPartId = 0;
let streamedParts: Array<Record<string, unknown>> = [];
let activeTextPartIdx = -1;
let activeReasoningPartIdx = -1;
params.onClientClose?.(() => { params.onClientClose?.(() => {
if (!serverAbortController.signal.aborted) { if (!serverAbortController.signal.aborted) {
@ -561,8 +564,39 @@ export async function executeChat(params: ChatEngineParams): Promise<ChatEngineR
onChunk: ({ chunk }) => { onChunk: ({ chunk }) => {
if (chunk.type === "text-delta") { if (chunk.type === "text-delta") {
streamedText += chunk.text; streamedText += chunk.text;
if (activeTextPartIdx < 0) {
activeTextPartIdx = streamedParts.length;
streamedParts.push({ id: `p_${chunkPartId++}`, type: "text", text: "" });
}
const tp = streamedParts[activeTextPartIdx] as Record<string, unknown>;
tp.text = ((tp.text as string) || "") + chunk.text;
} else if (chunk.type === "text-start") {
activeTextPartIdx = streamedParts.length;
streamedParts.push({ id: `p_${chunkPartId++}`, type: "text", text: "" });
} else if (chunk.type === "text-end") {
activeTextPartIdx = -1;
} else if (chunk.type === "reasoning-delta") { } else if (chunk.type === "reasoning-delta") {
streamedReasoning += chunk.text; streamedReasoning += chunk.text;
if (activeReasoningPartIdx < 0) {
activeReasoningPartIdx = streamedParts.length;
streamedParts.push({ id: `p_${chunkPartId++}`, type: "reasoning", text: "" });
}
const rp = streamedParts[activeReasoningPartIdx] as Record<string, unknown>;
rp.text = ((rp.text as string) || "") + chunk.text;
} else if (chunk.type === "reasoning-start") {
activeReasoningPartIdx = streamedParts.length;
streamedParts.push({ id: `p_${chunkPartId++}`, type: "reasoning", text: "" });
} else if (chunk.type === "reasoning-end") {
activeReasoningPartIdx = -1;
} else if (chunk.type === "tool-input-start") {
streamedParts.push({
id: `p_${chunkPartId++}`,
type: "tool-call",
toolCallId: chunk.id,
toolName: chunk.toolName,
args: undefined,
state: "preparing",
});
} else if (chunk.type === "tool-call") { } else if (chunk.type === "tool-call") {
streamedToolCalls.push({ streamedToolCalls.push({
id: `p_${chunkPartId++}`, id: `p_${chunkPartId++}`,
@ -572,6 +606,22 @@ export async function executeChat(params: ChatEngineParams): Promise<ChatEngineR
args: chunk.input, args: chunk.input,
state: "call", state: "call",
}); });
const existing = streamedParts.find(
(p) => p.type === "tool-call" && p.toolCallId === chunk.toolCallId,
);
if (existing) {
existing.args = chunk.input;
existing.state = "call";
} else {
streamedParts.push({
id: `p_${chunkPartId++}`,
type: "tool-call",
toolCallId: chunk.toolCallId,
toolName: chunk.toolName,
args: chunk.input,
state: "call",
});
}
} else if (chunk.type === "tool-result") { } else if (chunk.type === "tool-result") {
const tcIdx = streamedToolCalls.findIndex((tc) => tc.toolCallId === chunk.toolCallId); const tcIdx = streamedToolCalls.findIndex((tc) => tc.toolCallId === chunk.toolCallId);
const tcEntry = tcIdx >= 0 ? streamedToolCalls[tcIdx] : undefined; const tcEntry = tcIdx >= 0 ? streamedToolCalls[tcIdx] : undefined;
@ -579,6 +629,13 @@ export async function executeChat(params: ChatEngineParams): Promise<ChatEngineR
tcEntry.result = chunk.output; tcEntry.result = chunk.output;
tcEntry.state = "result"; tcEntry.state = "result";
} }
const part = streamedParts.find(
(p) => p.type === "tool-call" && p.toolCallId === chunk.toolCallId,
);
if (part) {
part.result = chunk.output;
part.state = "result";
}
} }
}, },
onFinish: async ({ finishReason, usage, steps, text: assistantText }) => { onFinish: async ({ finishReason, usage, steps, text: assistantText }) => {
@ -602,6 +659,12 @@ export async function executeChat(params: ChatEngineParams): Promise<ChatEngineR
for (const step of steps) { for (const step of steps) {
const toolResultMap = new Map(step.toolResults.map((tr) => [tr.toolCallId, tr.output])); const toolResultMap = new Map(step.toolResults.map((tr) => [tr.toolCallId, tr.output]));
const contentArr = step.content as Array<Record<string, unknown>>; const contentArr = step.content as Array<Record<string, unknown>>;
logger.info(
"[%s] [CHAT-ENGINE] step %d content order: %s",
requestId,
step.stepNumber,
contentArr.map((cp) => cp.type + (cp.type === "tool-call" ? `(${(cp as any).toolName})` : "")).join(" → "),
);
const approvalToolCallIds = new Set<string>(); const approvalToolCallIds = new Set<string>();
for (const cp of contentArr) { for (const cp of contentArr) {
if (cp.type === "tool-approval-request" && (cp as any).approvalId && (cp as any).toolCall) { if (cp.type === "tool-approval-request" && (cp as any).approvalId && (cp as any).toolCall) {
@ -746,33 +809,21 @@ export async function executeChat(params: ChatEngineParams): Promise<ChatEngineR
const partialText = streamedText; const partialText = streamedText;
logger.info("[%s] [CHAT-ENGINE] aborted: streamedTextLen=%d streamedToolCalls=%d", requestId, partialText.length, streamedToolCalls.length); logger.info("[%s] [CHAT-ENGINE] aborted: streamedTextLen=%d streamedToolCalls=%d", requestId, partialText.length, streamedToolCalls.length);
const parts: Array<Record<string, unknown>> = []; const parts: Array<Record<string, unknown>> = streamedParts
let hasTextPart = false; .filter((p) => {
if (streamedReasoning.trim()) { if (p.type === "text") return ((p.text as string) || "").trim().length > 0;
parts.push({ id: `p_${parts.length}`, type: "reasoning", text: streamedReasoning }); if (p.type === "reasoning") return ((p.text as string) || "").trim().length > 0;
} return true;
if (partialText.trim()) { })
parts.push({ id: `p_${parts.length}`, type: "text", text: partialText }); .map((p, i) => ({ ...p, id: `p_${i}` }));
hasTextPart = true; let hasTextPart = parts.some((p) => p.type === "text");
}
for (const tc of streamedToolCalls) {
parts.push({
id: `p_${parts.length}`,
type: "tool-call",
toolCallId: tc.toolCallId,
toolName: tc.toolName,
args: tc.args,
result: tc.result,
state: tc.result !== undefined ? "result" : "call",
});
}
const hasToolCallPart = parts.some((p) => p.type === "tool-call"); const hasToolCallPart = parts.some((p) => p.type === "tool-call");
let finalText = partialText; let finalText = partialText;
if (!hasTextPart && hasToolCallPart) { if (!hasTextPart && hasToolCallPart) {
const noticeText = "(对话已停止,以上是已完成的工具调用记录。)"; const noticeText = "(对话已停止,以上是已完成的工具调用记录。)";
parts.push({ id: `p_${parts.length}`, type: "text", text: noticeText }); parts.push({ id: `p_${parts.length}`, type: "text", text: noticeText });
finalText = (finalText || "") + noticeText; finalText = (finalText || "") + noticeText;
hasTextPart = true;
logger.info("[%s] [CHAT-ENGINE] aborted with tool calls but no text, appended notice", requestId); logger.info("[%s] [CHAT-ENGINE] aborted with tool calls but no text, appended notice", requestId);
} }

Loading…
Cancel
Save