diff --git a/.opencode/agents/instruction.md b/.opencode/agents/instruction.md index 5eb4daf..5e961ab 100644 --- a/.opencode/agents/instruction.md +++ b/.opencode/agents/instruction.md @@ -2,10 +2,14 @@ description: TJWater Agent，用于供水网络分析和操作员工作流 mode: primary model: deepseek/deepseek-v4-flash -temperature: 0.2 --- 你是 TJWater 供水管网分析 Agent，运用水力专业知识，回复用户时使用简体中文，内容要求简洁准确。 +## 回复要求 + +- 工具执行期间不输出过程说明，全部完成后只回复最终结果 +- 直接给出结论、关键数据和可执行建议，默认仅展示最重要的 Top 5；数据不足或任务失败时简要说明影响和下一步 + ## 工作流生命周期 Skills 树是**动态生长的**——工作流不是预置的，而是从实际任务中沉淀出来的： diff --git a/src/routes/chatStream.ts b/src/routes/chatStream.ts index f06830a..9676382 100644 --- a/src/routes/chatStream.ts +++ b/src/routes/chatStream.ts @@ -111,24 +111,40 @@ const toRuntimeModel = (model?: SupportedModel) => { }; }; -const emitFallbackMessage = async ( +const emitFinalMessage = async ( runtime: OpencodeRuntimeAdapter, sessionId: string, clientSessionId: string, + currentAssistantMessageIds: Set, + assistantTextParts: Map>, write: (event: string, data: Record) => void, ) => { - const messages = await runtime.messages(sessionId); - const assistantMessage = [...messages] + let text = [...currentAssistantMessageIds] .reverse() - .find((message) => message.info.role === "assistant"); - const parts = assistantMessage?.parts ?? []; - const text = collectTextContent(parts); + .map((messageId) => [...(assistantTextParts.get(messageId)?.values() ?? [])].join("")) + .find((content) => content.length > 0) ?? ""; + + if (!text) { + const messages = await runtime.messages(sessionId); + const assistantMessage = [...messages] + .reverse() + .find( + (message) => + message.info.role === "assistant" && + (currentAssistantMessageIds.size === 0 || + currentAssistantMessageIds.has(message.info.id)), + ); + text = collectTextContent(assistantMessage?.parts ?? []); + } + if (text) { write("token", { session_id: clientSessionId, content: text, }); + return true; } + return false; }; export const streamPromptResponse = async ({ @@ -156,15 +172,14 @@ export const streamPromptResponse = async ({ const emittedToolParts = new Set(); const emittedQuestionToolParts = new Set(); const emittedQuestionRequestIds = new Set(); + const currentAssistantMessageIds = new Set(); + const assistantTextParts = new Map>(); const partTypes = new Map(); - const pendingPartTextDeltas = new Map(); - const reasoningDeltas = new Map(); + const pendingTextDeltas = new Map(); const reasoningStatuses = new Map(); const toolStatuses = new Map(); let firstSessionEventLogged = false; let firstNonStatusEventLogged = false; - let firstTokenLogged = false; - let firstReasoningLogged = false; let firstToolEventLogged = false; let lastSessionStatus: string | null = null; let lastSessionStatusMessage: string | null = null; @@ -626,45 +641,26 @@ export const streamPromptResponse = async ({ if (event.type === "message.updated") { if (event.properties.info.role === "assistant") { sawResponseActivity = true; + currentAssistantMessageIds.add(event.properties.info.id); } continue; } if (event.type === "message.part.delta" && event.properties.field === "text") { sawResponseActivity = true; + currentAssistantMessageIds.add(event.properties.messageID); const partType = partTypes.get(event.properties.partID); if (partType === "text") { - if (!firstTokenLogged) { - firstTokenLogged = true; - logDevelopmentDebug("first response token emitted", { - ...debugContext, - partId: event.properties.partID, - elapsedMs: Math.max(0, Date.now() - requestStartedAt), - sincePromptDispatchMs: Math.max(0, Date.now() - promptStartedAt), - }); - } - emittedText = true; - write("token", { - session_id: clientSessionId, - content: event.properties.delta, - }); - } else if (partType === "reasoning") { - if (!firstReasoningLogged) { - firstReasoningLogged = true; - logDevelopmentDebug("first reasoning delta received", { - ...debugContext, - partId: event.properties.partID, - elapsedMs: Math.max(0, Date.now() - requestStartedAt), - sincePromptDispatchMs: Math.max(0, Date.now() - promptStartedAt), - }); - } - const pending = reasoningDeltas.get(event.properties.partID) ?? []; - pending.push(event.properties.delta); - reasoningDeltas.set(event.properties.partID, pending); + const messageParts = assistantTextParts.get(event.properties.messageID) ?? new Map(); + messageParts.set( + event.properties.partID, + `${messageParts.get(event.properties.partID) ?? ""}${event.properties.delta}`, + ); + assistantTextParts.set(event.properties.messageID, messageParts); } else if (!partType) { - const pending = pendingPartTextDeltas.get(event.properties.partID) ?? []; + const pending = pendingTextDeltas.get(event.properties.partID) ?? []; pending.push(event.properties.delta); - pendingPartTextDeltas.set(event.properties.partID, pending); + pendingTextDeltas.set(event.properties.partID, pending); } continue; } @@ -673,23 +669,19 @@ export const streamPromptResponse = async ({ sawResponseActivity = true; const part = event.properties.part; partTypes.set(part.id, part.type); + if (part.type === "text" || part.type === "reasoning" || part.type === "tool") { + currentAssistantMessageIds.add(part.messageID); + } if (part.type === "text") { - const pending = pendingPartTextDeltas.get(part.id) ?? []; - pendingPartTextDeltas.delete(part.id); - for (const content of pending) { - emittedText = true; - write("token", { - session_id: clientSessionId, - content, - }); - } - } else if (part.type === "reasoning") { - const pending = pendingPartTextDeltas.get(part.id) ?? []; - if (pending.length > 0) { - const existing = reasoningDeltas.get(part.id) ?? []; - reasoningDeltas.set(part.id, existing.concat(pending)); - } - pendingPartTextDeltas.delete(part.id); + const pendingText = (pendingTextDeltas.get(part.id) ?? []).join(""); + pendingTextDeltas.delete(part.id); + const messageParts = assistantTextParts.get(part.messageID) ?? new Map(); + messageParts.set(part.id, part.text || pendingText); + assistantTextParts.set(part.messageID, messageParts); + } else { + pendingTextDeltas.delete(part.id); + } + if (part.type === "reasoning") { const reasoningStatus = part.time.end ? "completed" : "running"; if (reasoningStatuses.get(part.id) !== reasoningStatus) { reasoningStatuses.set(part.id, reasoningStatus); @@ -697,14 +689,10 @@ export const streamPromptResponse = async ({ ...debugContext, partId: part.id, status: reasoningStatus, - chunkCount: (reasoningDeltas.get(part.id) ?? []).length, elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); } - const reasoningDetail = buildReasoningProgressDetail( - reasoningDeltas.get(part.id) ?? [], - part.time.end, - ); + const reasoningDetail = buildReasoningProgressDetail(part.time.end); emitProgress({ id: part.id, phase: "planning", @@ -784,9 +772,6 @@ export const streamPromptResponse = async ({ detail: buildToolProgressDetail( part.tool, part.state.status, - toolParams, - reason, - part.state.status === "error" ? part.state.error : undefined, ), }); if ( @@ -932,13 +917,14 @@ export const streamPromptResponse = async ({ } await promptPromise; - if (!emittedText) { - logDevelopmentDebug("no streamed text emitted, falling back to messages()", { - ...debugContext, - elapsedMs: Math.max(0, Date.now() - requestStartedAt), - }); - await emitFallbackMessage(runtime, sessionId, clientSessionId, write); - } + emittedText = await emitFinalMessage( + runtime, + sessionId, + clientSessionId, + currentAssistantMessageIds, + assistantTextParts, + write, + ); emitProgress({ id: "request-received", phase: "start", diff --git a/src/routes/chatStreamEvents.ts b/src/routes/chatStreamEvents.ts index ecb44d1..58b2a03 100644 --- a/src/routes/chatStreamEvents.ts +++ b/src/routes/chatStreamEvents.ts @@ -356,43 +356,6 @@ export const normalizeToolStatus = (status: string) => { return "running"; }; -const formatProgressValue = (value: unknown): string => { - if (typeof value === "string") { - return value.length > 120 ? `${value.slice(0, 117)}...` : value; - } - if ( - typeof value === "number" || - typeof value === "boolean" || - value === null || - value === undefined - ) { - return String(value); - } - try { - const serialized = JSON.stringify(value); - return serialized.length > 120 ? `${serialized.slice(0, 117)}...` : serialized; - } catch { - return "[unserializable]"; - } -}; - -const normalizeProgressText = (chunks: string[]) => - chunks.join("").replace(/\s+/g, " ").trim(); - -const truncateProgressText = (text: string, maxLength: number) => - text.length > maxLength ? `${text.slice(0, maxLength - 3)}...` : text; - -const summarizeToolParams = (params: Record) => { - const ignoredKeys = new Set(["reason", "request_reason", "why", "purpose", "rationale"]); - const summary = Object.entries(params) - .filter(([key]) => !ignoredKeys.has(key)) - .slice(0, 4) - .map(([key, value]) => `${key}=${formatProgressValue(value)}`) - .join(", "); - - return summary || "无附加参数"; -}; - export const buildSessionStatusDetail = (status: { type: string; message?: string }) => { if (status.type === "retry") { return status.message @@ -413,42 +376,25 @@ export const buildSessionStatusDetail = (status: { type: string; message?: strin }; export const buildReasoningProgressDetail = ( - chunks: string[], ended?: string | number | Date | null, -) => { - const reasoningText = truncateProgressText(normalizeProgressText(chunks), 800); - if (ended) { - return reasoningText - ? `推理过程：${reasoningText}` - : "当前推理阶段已完成，Agent 将继续输出答案或进入工具执行。"; - } - return reasoningText - ? `正在推理：${reasoningText}` - : "Agent 正在拆解问题、梳理执行步骤并判断是否需要调用工具。"; -}; +) => ended ? "分析步骤已整理完成。" : "Agent 正在分析问题。"; export const buildToolProgressDetail = ( tool: string, status: string, - params: Record, - reason: string, - error?: string, ) => { const toolName = toolLabels[tool] ?? tool; - const reasonText = reason ? `；调用原因：${reason}` : ""; - const paramsText = `；关键参数：${summarizeToolParams(params)}`; if (status === "error") { - const errorText = error ? `；错误：${error}` : ""; - return `${toolName} 调用失败${reasonText}${paramsText}${errorText}`; + return `${toolName} 调用失败。`; } if (status === "completed") { - return `${toolName} 已执行完成${reasonText}${paramsText}`; + return `${toolName} 已执行完成。`; } if (status === "pending") { - return `${toolName} 已进入待执行状态${reasonText}${paramsText}`; + return `${toolName} 等待执行。`; } - return `${toolName} 正在执行${reasonText}${paramsText}`; + return `${toolName} 正在执行。`; }; export const getToolProgressTitle = (tool: string, status: string) => { diff --git a/tests/routes/chatStream.test.ts b/tests/routes/chatStream.test.ts index f8669c7..e72393e 100644 --- a/tests/routes/chatStream.test.ts +++ b/tests/routes/chatStream.test.ts @@ -15,6 +15,240 @@ const createEventStream = (events: unknown[]) => ({ }); describe("streamPromptResponse", () => { + it("emits only the final assistant text after tool-driven intermediate messages", async () => { + const runtime = { + subscribeEvents: async () => + createEventStream([ + { + type: "message.part.delta", + properties: { + sessionID: "runtime-session-1", + messageID: "assistant-intermediate", + partID: "text-part-intermediate", + field: "text", + delta: "正在加载工作流并尝试分页参数。", + }, + }, + { + type: "message.part.updated", + properties: { + sessionID: "runtime-session-1", + part: { + id: "text-part-intermediate", + sessionID: "runtime-session-1", + messageID: "assistant-intermediate", + type: "text", + text: "正在加载工作流并尝试分页参数。", + time: { start: 1, end: 2 }, + }, + time: 2, + }, + }, + { + type: "message.part.delta", + properties: { + sessionID: "runtime-session-1", + messageID: "assistant-final", + partID: "text-part-final", + field: "text", + delta: "共识别 56 条瓶颈管段，建议优先改造 Top 5。", + }, + }, + { + type: "message.part.updated", + properties: { + sessionID: "runtime-session-1", + part: { + id: "text-part-final", + sessionID: "runtime-session-1", + messageID: "assistant-final", + type: "text", + text: "共识别 56 条瓶颈管段，建议优先改造 Top 5。", + time: { start: 3, end: 4 }, + }, + time: 4, + }, + }, + { + type: "session.idle", + properties: { sessionID: "runtime-session-1" }, + }, + ]), + prompt: async () => undefined, + messages: async () => [ + { + info: { id: "assistant-intermediate", role: "assistant" }, + parts: [ + { + id: "text-part-intermediate", + sessionID: "runtime-session-1", + messageID: "assistant-intermediate", + type: "text", + text: "正在加载工作流并尝试分页参数。", + }, + ], + }, + { + info: { id: "assistant-final", role: "assistant" }, + parts: [ + { + id: "text-part-final", + sessionID: "runtime-session-1", + messageID: "assistant-final", + type: "text", + text: "共识别 56 条瓶颈管段，建议优先改造 Top 5。", + }, + ], + }, + ], + } as unknown as OpencodeRuntimeAdapter; + const events: Array<{ event: string; data: Record }> = []; + + await streamPromptResponse({ + runtime, + sessionId: "runtime-session-1", + clientSessionId: "client-session-1", + message: "分析管网瓶颈", + write: (event, data) => events.push({ event, data }), + }); + + expect( + events + .filter((item) => item.event === "token") + .map((item) => item.data.content) + .join(""), + ).toBe("共识别 56 条瓶颈管段，建议优先改造 Top 5。"); + }); + + it("uses the final text event cache when the messages lookup fails", async () => { + const runtime = { + subscribeEvents: async () => + createEventStream([ + { + type: "message.part.updated", + properties: { + sessionID: "runtime-session-1", + part: { + id: "text-part-final", + sessionID: "runtime-session-1", + messageID: "assistant-final", + type: "text", + text: "最终分析结果。", + time: { start: 1, end: 2 }, + }, + time: 2, + }, + }, + { + type: "session.idle", + properties: { sessionID: "runtime-session-1" }, + }, + ]), + prompt: async () => undefined, + messages: async () => { + throw new Error("transient messages lookup failure"); + }, + } as unknown as OpencodeRuntimeAdapter; + const events: Array<{ event: string; data: Record }> = []; + + const result = await streamPromptResponse({ + runtime, + sessionId: "runtime-session-1", + clientSessionId: "client-session-1", + message: "分析管网瓶颈", + write: (event, data) => events.push({ event, data }), + }); + + expect(result.failed).toBe(false); + expect( + events + .filter((item) => item.event === "token") + .map((item) => item.data.content) + .join(""), + ).toBe("最终分析结果。"); + }); + + it("keeps reasoning, tool parameters, and raw errors out of progress details", async () => { + const runtime = { + subscribeEvents: async () => + createEventStream([ + { + type: "message.part.updated", + properties: { + sessionID: "runtime-session-1", + part: { + id: "reasoning-part-1", + sessionID: "runtime-session-1", + messageID: "assistant-1", + type: "reasoning", + text: "内部推理：尝试 limit=5000 并读取临时路径。", + time: { start: 1, end: 2 }, + }, + time: 2, + }, + }, + { + type: "message.part.delta", + properties: { + sessionID: "runtime-session-1", + messageID: "assistant-1", + partID: "reasoning-part-1", + field: "text", + delta: "内部推理：尝试 limit=5000 并读取临时路径。", + }, + }, + { + type: "message.part.updated", + properties: { + sessionID: "runtime-session-1", + part: { + id: "tool-part-1", + sessionID: "runtime-session-1", + messageID: "assistant-1", + type: "tool", + callID: "call-1", + tool: "tjwater_cli", + state: { + status: "error", + input: { + command: "network get-all-pipes-properties --limit 5000", + reason: "尝试突破分页限制", + }, + error: "HTTP_422 raw backend payload with trace_id=secret-trace", + time: { start: 1, end: 2 }, + }, + }, + time: 2, + }, + }, + { + type: "session.idle", + properties: { sessionID: "runtime-session-1" }, + }, + ]), + prompt: async () => undefined, + messages: async () => [], + } as unknown as OpencodeRuntimeAdapter; + const events: Array<{ event: string; data: Record }> = []; + + await streamPromptResponse({ + runtime, + sessionId: "runtime-session-1", + clientSessionId: "client-session-1", + message: "分析管网瓶颈", + write: (event, data) => events.push({ event, data }), + }); + + const reasoningProgress = events.find( + (item) => item.event === "progress" && item.data.id === "reasoning-part-1", + ); + const toolProgress = events.find( + (item) => item.event === "progress" && item.data.id === "tool-part-1", + ); + expect(reasoningProgress?.data.detail).toBe("分析步骤已整理完成。"); + expect(toolProgress?.data.detail).toBe("tjwater_cli 调用失败。"); + }); + it("forwards opencode permission requests as SSE payloads", async () => { const runtime = { subscribeEvents: async () =>