fix(agent): smooth Chinese stream chunks
This commit is contained in:
@@ -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 的基础地址。
|
||||
|
||||
+209
-14
@@ -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<string, unknown>) => void;
|
||||
};
|
||||
|
||||
type SmoothTextStreamPart = TextStreamPart<Record<string, never>>;
|
||||
type SmoothTextDeltaPart = Extract<SmoothTextStreamPart, { type: "text-delta" }>;
|
||||
type SmoothTextEndPart = Extract<SmoothTextStreamPart, { type: "text-end" }>;
|
||||
|
||||
const segmenters = new Map<string, Intl.Segmenter | null>();
|
||||
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<Record<string, never>>({
|
||||
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<string, unknown>) => 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);
|
||||
|
||||
@@ -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<string, unknown> }> = [];
|
||||
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<string, unknown> }> = [];
|
||||
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<string, unknown> }> = [];
|
||||
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<string, unknown> }> = [];
|
||||
|
||||
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 () =>
|
||||
|
||||
Reference in New Issue
Block a user