diff --git a/src/config.ts b/src/config.ts index a528f0c..68a9599 100644 --- a/src/config.ts +++ b/src/config.ts @@ -22,6 +22,21 @@ const optionalString = () => z.string().optional(), ); +const optionalBoolean = (defaultValue: boolean) => + z.preprocess((value) => { + if (typeof value !== "string") { + return value; + } + const normalized = value.trim().toLowerCase(); + if (["1", "true", "yes", "on"].includes(normalized)) { + return true; + } + if (["0", "false", "no", "off"].includes(normalized)) { + return false; + } + return value; + }, z.boolean().default(defaultValue)); + // 统一在启动时解析环境变量,避免业务代码里散落字符串默认值。 const envSchema = z .object({ @@ -51,6 +66,16 @@ const envSchema = z OPENCODE_MODEL: z.string().default("deepseek/deepseek-v4-flash"), // 聊天 UI 和 /stream 允许选择的 opencode 模型完整配置,JSON 数组。 OPENCODE_MODEL_OPTIONS: z.string().default(defaultAgentModelOptionsJson), + // 平滑 token 输出,避免上游模型大块/抖动 token 直接传到前端。 + AGENT_TOKEN_SMOOTHING_ENABLED: optionalBoolean(true), + // 平滑输出每段之间的间隔(毫秒)。 + AGENT_TOKEN_SMOOTHING_DELAY_MS: z.coerce + .number() + .int() + .min(0) + .default(20), + // 平滑输出分词 locale,中文默认使用 zh。 + AGENT_TOKEN_SMOOTHING_LOCALE: z.string().default("zh"), // opencode skills 树目录;会在运行时解析为绝对路径,避免工具 cwd 偏移。 OPENCODE_SKILLS_ROOT_DIR: z.string().default("./.opencode/skills"), // client 模式下,目标 opencode server 的基础地址。 diff --git a/src/routes/chatStream.ts b/src/routes/chatStream.ts index b9adde7..eb5433a 100644 --- a/src/routes/chatStream.ts +++ b/src/routes/chatStream.ts @@ -1,7 +1,9 @@ import type { Event as OpencodeEvent, Part } from "@opencode-ai/sdk/v2"; +import { smoothStream, type TextStreamPart } from "ai"; import { writeLlmRequestAuditLog } from "../audit/llmRequestAudit.js"; import { type SupportedModel } from "../chat/models.js"; +import { config } from "../config.js"; import { logger } from "../logger.js"; import { type PermissionReply, @@ -82,6 +84,196 @@ type ProgressPayload = { detail?: string; }; +type TokenSmootherOptions = { + enabled: boolean; + delayMs: number; + locale: string; + sessionId: string; + write: (event: string, data: Record) => void; +}; + +type SmoothTextStreamPart = TextStreamPart>; +type SmoothTextDeltaPart = Extract; +type SmoothTextEndPart = Extract; + +const segmenters = new Map(); +const CJK_SCRIPT_PATTERN = /[\p{Script=Han}\p{Script=Hiragana}\p{Script=Katakana}\p{Script=Hangul}]/u; +const CJK_PUNCTUATION_PATTERN = /[。!?!?,、;:,;:]/u; +const CJK_SENTENCE_END_PATTERN = /[。!?!?]+(?:["'”’』」))\]\s]+)?/u; +const CJK_PAUSE_PATTERN = /[,、;:,;:]+(?:\s+)?/u; +const CJK_FALLBACK_TRIGGER_CHARS = 24; +const CJK_FALLBACK_MIN_CHARS = 12; +const CJK_FALLBACK_TARGET_CHARS = 18; + +function getTokenSmoothingSegmenter(locale: string) { + if (segmenters.has(locale)) { + return segmenters.get(locale) ?? null; + } + + const segmenter = + typeof Intl !== "undefined" && "Segmenter" in Intl + ? new Intl.Segmenter(locale, { granularity: "word" }) + : null; + segmenters.set(locale, segmenter); + return segmenter; +} + +function sliceCodePoints(value: string, count: number) { + return Array.from(value).slice(0, count).join(""); +} + +function codePointLength(value: string) { + return Array.from(value).length; +} + +function isCjkSmoothingContent(content: string, locale: string) { + return ( + CJK_SCRIPT_PATTERN.test(content) || + (/^(zh|ja|ko)(?:-|$)/i.test(locale) && CJK_PUNCTUATION_PATTERN.test(content)) + ); +} + +function detectCjkPunctuationChunk(content: string) { + const sentenceMatch = CJK_SENTENCE_END_PATTERN.exec(content); + if (sentenceMatch) { + return content.slice(0, sentenceMatch.index + sentenceMatch[0].length); + } + + const pauseMatch = CJK_PAUSE_PATTERN.exec(content); + if (pauseMatch) { + return content.slice(0, pauseMatch.index + pauseMatch[0].length); + } + + return null; +} + +function detectCjkFallbackChunk(content: string, locale: string) { + if (codePointLength(content) < CJK_FALLBACK_TRIGGER_CHARS) { + return null; + } + + const segmenter = getTokenSmoothingSegmenter(locale); + if (segmenter) { + let candidate: string | null = null; + for (const segment of segmenter.segment(content)) { + const nextCandidate = content.slice(0, segment.index + segment.segment.length); + const nextLength = codePointLength(nextCandidate); + if (nextLength > CJK_FALLBACK_TARGET_CHARS) { + break; + } + if (nextLength >= CJK_FALLBACK_MIN_CHARS) { + candidate = nextCandidate; + } + } + if (candidate) { + return candidate; + } + } + + return sliceCodePoints(content, CJK_FALLBACK_TARGET_CHARS); +} + +export function detectTokenSmoothingChunk(content: string, locale = "zh") { + if (!content) { + return null; + } + + if (isCjkSmoothingContent(content, locale)) { + return ( + detectCjkPunctuationChunk(content) ?? + detectCjkFallbackChunk(content, locale) + ); + } + + const wordChunk = content.match(/^\S+\s*/u)?.[0]; + if (wordChunk) { + return wordChunk; + } + + return sliceCodePoints(content, 1); +} + +export function createTokenSmoother({ + enabled, + delayMs, + locale, + sessionId, + write, +}: TokenSmootherOptions) { + const emitChunk = (content: string) => { + write("token", { + session_id: sessionId, + content, + }); + }; + + if (!enabled) { + return { + writeToken(content: string) { + if (content) { + emitChunk(content); + } + }, + async flush() { + return; + }, + }; + } + + const textPartId = `smooth-text-${sessionId}`; + const stream = smoothStream>({ + delayInMs: delayMs, + chunking: (buffer) => detectTokenSmoothingChunk(buffer, locale), + })({ tools: {} }); + const writer = stream.writable.getWriter(); + const reader = stream.readable.getReader(); + let closed = false; + let writeChain = Promise.resolve(); + const readPromise = (async () => { + while (true) { + const result = await reader.read(); + if (result.done) { + return; + } + if (result.value.type === "text-delta") { + emitChunk(result.value.text); + } + } + })(); + + return { + writeToken(content: string) { + if (!content) { + return; + } + if (closed) { + return; + } + writeChain = writeChain.then(() => + writer.write({ + type: "text-delta", + id: textPartId, + text: content, + } satisfies SmoothTextDeltaPart), + ); + }, + async flush() { + if (closed) { + await readPromise; + return; + } + closed = true; + await writeChain; + await writer.write({ + type: "text-end", + id: textPartId, + } satisfies SmoothTextEndPart); + await writer.close(); + await readPromise; + }, + }; +} + const getPermissionTarget = (metadata: unknown) => { if (!isObjectRecord(metadata)) { return undefined; @@ -113,7 +305,7 @@ const emitFallbackMessage = async ( runtime: OpencodeRuntimeAdapter, sessionId: string, clientSessionId: string, - write: (event: string, data: Record) => void, + writeToken: (content: string) => void, ) => { const messages = await runtime.messages(sessionId); const assistantMessage = [...messages] @@ -122,10 +314,7 @@ const emitFallbackMessage = async ( const parts = assistantMessage?.parts ?? []; const text = collectTextContent(parts); if (text) { - write("token", { - session_id: clientSessionId, - content: text, - }); + writeToken(text); } }; @@ -180,6 +369,13 @@ export const streamPromptResponse = async ({ projectId, model: model ?? null, }; + const tokenSmoother = createTokenSmoother({ + enabled: config.AGENT_TOKEN_SMOOTHING_ENABLED, + delayMs: config.AGENT_TOKEN_SMOOTHING_DELAY_MS, + locale: config.AGENT_TOKEN_SMOOTHING_LOCALE, + sessionId: clientSessionId, + write, + }); logDevelopmentDebug("chat stream started", { ...debugContext, @@ -640,10 +836,7 @@ export const streamPromptResponse = async ({ }); } emittedText = true; - write("token", { - session_id: clientSessionId, - content: event.properties.delta, - }); + tokenSmoother.writeToken(event.properties.delta); } else if (partType === "reasoning") { if (!firstReasoningLogged) { firstReasoningLogged = true; @@ -674,10 +867,7 @@ export const streamPromptResponse = async ({ pendingPartTextDeltas.delete(part.id); for (const content of pending) { emittedText = true; - write("token", { - session_id: clientSessionId, - content, - }); + tokenSmoother.writeToken(content); } } else if (part.type === "reasoning") { const pending = pendingPartTextDeltas.get(part.id) ?? []; @@ -881,6 +1071,7 @@ export const streamPromptResponse = async ({ ? getErrorMessage(event.properties.error) : "opencode session error", }); + await tokenSmoother.flush(); write("error", { session_id: clientSessionId, message: event.properties.error @@ -933,10 +1124,12 @@ export const streamPromptResponse = async ({ "failed while waiting for aborted opencode session to become idle", ); }); + await tokenSmoother.flush(); return { aborted: true, failed: false, toolCallCount }; } if (failed) { + await tokenSmoother.flush(); return { aborted: false, failed: true, toolCallCount }; } @@ -946,8 +1139,9 @@ export const streamPromptResponse = async ({ ...debugContext, elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); - await emitFallbackMessage(runtime, sessionId, clientSessionId, write); + await emitFallbackMessage(runtime, sessionId, clientSessionId, tokenSmoother.writeToken); } + await tokenSmoother.flush(); emitProgress({ id: "request-received", phase: "start", @@ -976,6 +1170,7 @@ export const streamPromptResponse = async ({ }); return { aborted: false, failed: false, toolCallCount }; } finally { + await tokenSmoother.flush(); await iterator.return?.(undefined); if (!promptSettled && !aborted) { await promptPromise.catch(() => undefined); diff --git a/tests/routes/chatStream.test.ts b/tests/routes/chatStream.test.ts index a8f5345..3370a39 100644 --- a/tests/routes/chatStream.test.ts +++ b/tests/routes/chatStream.test.ts @@ -1,6 +1,8 @@ import { describe, expect, it } from "bun:test"; import { + createTokenSmoother, + detectTokenSmoothingChunk, streamPromptResponse, type PermissionRequestPayload, } from "../../src/routes/chatStream.js"; @@ -14,7 +16,169 @@ const createEventStream = (events: unknown[]) => ({ }, }); +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + describe("streamPromptResponse", () => { + it("detects Chinese punctuation chunks for smooth token output", () => { + expect(detectTokenSmoothingChunk("管网压力异常,需要继续分析", "zh")).toBe( + "管网压力异常,", + ); + + expect(detectTokenSmoothingChunk("管网压力异常需要继续分析。后续排查阀门", "zh")).toBe( + "管网压力异常需要继续分析。", + ); + }); + + it("does not immediately release a single Chinese character", () => { + expect(detectTokenSmoothingChunk("管", "zh")).toBeNull(); + }); + + it("keeps non-Chinese text on the word path even with zh locale", () => { + expect(detectTokenSmoothingChunk("Agent ", "zh")).toBe("Agent "); + }); + + it("falls back to bounded Chinese chunks when punctuation is absent", () => { + const content = "管网压力异常需要继续分析东部主干供水走廊和相关阀门状态变化"; + const chunk = detectTokenSmoothingChunk(content, "zh"); + + expect(chunk).not.toBeNull(); + if (!chunk) { + return; + } + expect(content.startsWith(chunk)).toBe(true); + expect(Array.from(chunk).length).toBeGreaterThanOrEqual(12); + expect(Array.from(chunk).length).toBeLessThanOrEqual(18); + }); + + it("can bypass token smoothing when disabled", async () => { + const events: Array<{ event: string; data: Record }> = []; + const smoother = createTokenSmoother({ + enabled: false, + delayMs: 20, + locale: "zh", + sessionId: "client-session-1", + write: (event, data) => events.push({ event, data }), + }); + + smoother.writeToken("第一段"); + smoother.writeToken("第二段"); + await smoother.flush(); + + expect(events.map((item) => item.data.content)).toEqual(["第一段", "第二段"]); + }); + + it("buffers Chinese single-character deltas until punctuation", async () => { + const events: Array<{ event: string; data: Record }> = []; + const smoother = createTokenSmoother({ + enabled: true, + delayMs: 1, + locale: "zh", + sessionId: "client-session-1", + write: (event, data) => events.push({ event, data }), + }); + + smoother.writeToken("管"); + await sleep(8); + expect(events).toHaveLength(0); + + for (const char of Array.from("网压力异常,")) { + smoother.writeToken(char); + } + await sleep(20); + + expect(events.map((item) => item.data.content).join("")).toBe("管网压力异常,"); + await smoother.flush(); + }); + + it("smooths long Chinese text without waiting for paragraph-sized buffers", async () => { + const events: Array<{ event: string; data: Record }> = []; + const smoother = createTokenSmoother({ + enabled: true, + delayMs: 1, + locale: "zh", + sessionId: "client-session-1", + write: (event, data) => events.push({ event, data }), + }); + + const content = "管网压力异常需要继续分析东部主干供水走廊和相关阀门状态变化"; + for (const char of Array.from(content)) { + smoother.writeToken(char); + } + await sleep(20); + + expect(events.length).toBeGreaterThan(0); + await smoother.flush(); + expect(events.map((item) => item.data.content).join("")).toBe(content); + }); + + it("flushes smoothed token output before stream completion", async () => { + const runtime = { + subscribeEvents: async () => + createEventStream([ + { + type: "message.part.updated", + properties: { + sessionID: "runtime-session-1", + part: { + id: "text-part-1", + sessionID: "runtime-session-1", + messageID: "message-1", + type: "text", + text: "", + time: { start: Date.now() }, + }, + time: Date.now(), + }, + }, + { + type: "message.part.delta", + properties: { + sessionID: "runtime-session-1", + partID: "text-part-1", + field: "text", + delta: "管网压力异常需要继续分析", + }, + }, + { + type: "message.part.delta", + properties: { + sessionID: "runtime-session-1", + partID: "text-part-1", + field: "text", + delta: "东部主干供水走廊和相关阀门状态。", + }, + }, + { + 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: "analyze", + write: (event, data) => events.push({ event, data }), + }); + + const doneIndex = events.findIndex((item) => item.event === "done"); + const tokenEvents = events.filter((item) => item.event === "token"); + + expect(doneIndex).toBeGreaterThan(-1); + expect(tokenEvents.length).toBeGreaterThan(0); + expect(tokenEvents.every((item) => events.indexOf(item) < doneIndex)).toBe(true); + expect(tokenEvents.map((item) => item.data.content).join("")).toBe( + "管网压力异常需要继续分析东部主干供水走廊和相关阀门状态。", + ); + }); + it("forwards opencode permission requests as SSE payloads", async () => { const runtime = { subscribeEvents: async () =>