From cdb2bfea7cfa63f6fbc0e2feca9eba939e501dbf Mon Sep 17 00:00:00 2001 From: digua Date: Mon, 18 May 2026 22:32:00 +0800 Subject: [PATCH] feat: unify LLM streaming with thinking display and Web/cache API shims - Extract runSimpleLlmStream to @openchatlab/node-runtime for shared Electron/Server streaming with thinking_start/delta/end event handling - Simplify Electron IPC ai.ts by delegating to shared stream function - Add thinking/thinkingDone fields to LlmStreamChunk and preload types - Add collapsible thinking process display in SQL LAB with auto-scroll - Implement Web chatStream via SSE with proper error propagation - Add /_web/cache/save-to-downloads and show-in-folder server routes - Add cacheApi shim in Web registry for server-side file operations - Increase SQL LAB maxTokens to 4096 to accommodate reasoning models - Detect finishReason:length and show token limit error message - Add i18n keys for thinking/thinkingProcess/errorTokenLimit (4 locales) --- electron/main/ipc/ai.ts | 78 +++----------- electron/preload/apis/ai.ts | 5 +- electron/preload/index.d.ts | 3 + packages/node-runtime/src/ai/index.ts | 4 + packages/node-runtime/src/ai/llm-stream.ts | 119 +++++++++++++++++++++ packages/node-runtime/src/index.ts | 3 +- packages/server/src/http/routes/ai.ts | 46 ++++++++ packages/server/src/http/routes/web.ts | 49 +++++++++ src/components/analysis/SQLLabTab.vue | 79 +++++++++++--- src/i18n/locales/en-US/ai.json | 3 + src/i18n/locales/ja-JP/ai.json | 3 + src/i18n/locales/zh-CN/ai.json | 3 + src/i18n/locales/zh-TW/ai.json | 3 + src/services/registry.ts | 68 +++++++++++- 14 files changed, 384 insertions(+), 82 deletions(-) create mode 100644 packages/node-runtime/src/ai/llm-stream.ts diff --git a/electron/main/ipc/ai.ts b/electron/main/ipc/ai.ts index 1c5c33e8..c1b8c0e9 100644 --- a/electron/main/ipc/ai.ts +++ b/electron/main/ipc/ai.ts @@ -22,7 +22,7 @@ import { type CompressionConfig, type CompressionLlmAdapter, completeSimple, - streamSimple, + runSimpleLlmStream, type PiMessage, type PiTextContent, } from '@openchatlab/node-runtime' @@ -764,76 +764,22 @@ export function registerAIHandlers({ win }: IpcContext): void { return { success: false, error: t('llm.notConfigured') } } const piModel = buildPiModel(activeConfig) - const now = Date.now() - const systemMsg = messages.find((m) => m.role === 'system') - const nonSystemMsgs = messages.filter((m) => m.role !== 'system') - const eventStream = streamSimple( - piModel, - { - systemPrompt: systemMsg?.content, - messages: toPiSimpleMessages(nonSystemMsgs, now), - }, - { + ;(async () => { + await runSimpleLlmStream({ + messages, apiKey: activeConfig.apiKey, + piModel, temperature: options?.temperature, maxTokens: options?.maxTokens, - } - ) - - // 异步消费流，通过事件发送 chunks - ;(async () => { - let hasTerminalChunk = false - try { - for await (const event of eventStream) { - if (event.type === 'text_delta') { - win.webContents.send('llm:streamChunk', { - requestId, - chunk: { content: event.delta, isFinished: false }, - }) - continue + onChunk: (chunk) => { + if (chunk.error) { + win.webContents.send('llm:streamChunk', { requestId, error: chunk.error, chunk }) + } else { + win.webContents.send('llm:streamChunk', { requestId, chunk }) } - - if (event.type === 'done') { - hasTerminalChunk = true - win.webContents.send('llm:streamChunk', { - requestId, - chunk: { content: '', isFinished: true, finishReason: event.reason === 'length' ? 'length' : 'stop' }, - }) - return - } - - if (event.type === 'error') { - hasTerminalChunk = true - const errorMsg = - event.error?.errorMessage || formatAIError(event.error, activeConfig.provider) || 'Unknown LLM error' - aiLogger.error('IPC', 'llm:chatStream LLM error', { requestId, error: errorMsg }) - win.webContents.send('llm:streamChunk', { - requestId, - error: errorMsg, - chunk: { content: '', isFinished: true, finishReason: 'error' }, - }) - return - } - } - - if (!hasTerminalChunk) { - win.webContents.send('llm:streamChunk', { - requestId, - chunk: { content: '', isFinished: true, finishReason: 'stop' }, - }) - } - } catch (error) { - if (!hasTerminalChunk) { - const friendlyError = formatAIError(error, activeConfig.provider) - aiLogger.error('IPC', 'llm:chatStream stream error', { requestId, error: String(error) }) - win.webContents.send('llm:streamChunk', { - requestId, - error: friendlyError, - chunk: { content: '', isFinished: true, finishReason: 'error' }, - }) - } - } + }, + }) })() return { success: true } diff --git a/electron/preload/apis/ai.ts b/electron/preload/apis/ai.ts index a751499b..a8084bfc 100644 --- a/electron/preload/apis/ai.ts +++ b/electron/preload/apis/ai.ts @@ -88,6 +88,9 @@ export interface ChatStreamChunk { content: string isFinished: boolean finishReason?: 'stop' | 'length' | 'error' + error?: string + thinking?: string + thinkingDone?: boolean } // Agent API 类型 — 从 shared/types 统一导入 @@ -792,7 +795,7 @@ export const llmApi = { if (data.error) { console.log('[preload] chatStream 收到错误:', data.error) if (onChunk) { - onChunk({ content: '', isFinished: true, finishReason: 'error' }) + onChunk({ content: '', isFinished: true, finishReason: 'error', error: data.error }) } ipcRenderer.removeListener('llm:streamChunk', handler) resolve({ success: false, error: data.error }) diff --git a/electron/preload/index.d.ts b/electron/preload/index.d.ts index 30e2ddf8..22673e46 100644 --- a/electron/preload/index.d.ts +++ b/electron/preload/index.d.ts @@ -558,6 +558,9 @@ interface LLMChatStreamChunk { content: string isFinished: boolean finishReason?: 'stop' | 'length' | 'error' + error?: string + thinking?: string + thinkingDone?: boolean } interface LlmApi { diff --git a/packages/node-runtime/src/ai/index.ts b/packages/node-runtime/src/ai/index.ts index 2a50345e..cd4590b5 100644 --- a/packages/node-runtime/src/ai/index.ts +++ b/packages/node-runtime/src/ai/index.ts @@ -132,6 +132,10 @@ export type { RemoteModel, FetchRemoteModelsResult, RemoteApiOptions } from './r // Re-exports from @mariozechner/pi-agent-core export type { AgentTool, AgentToolResult } from '@mariozechner/pi-agent-core' +// LLM simple streaming +export { runSimpleLlmStream } from './llm-stream' +export type { LlmStreamChunk, RunSimpleLlmStreamOptions } from './llm-stream' + // Re-exports from @mariozechner/pi-ai export { Type, completeSimple, streamSimple } from '@mariozechner/pi-ai' export type { diff --git a/packages/node-runtime/src/ai/llm-stream.ts b/packages/node-runtime/src/ai/llm-stream.ts new file mode 100644 index 00000000..8d9f3a8c --- /dev/null +++ b/packages/node-runtime/src/ai/llm-stream.ts @@ -0,0 +1,119 @@ +/** + * Shared LLM simple streaming — used by both Electron and Server. + * + * Wraps pi-ai's streamSimple with message preprocessing, + * event mapping, and error handling. + */ + +import { streamSimple } from '@mariozechner/pi-ai' +import type { Model as PiModel, Api as PiApi, Message as PiMessage } from '@mariozechner/pi-ai' + +export interface LlmStreamChunk { + content: string + isFinished: boolean + finishReason?: 'stop' | 'length' | 'error' + error?: string + /** When present, this chunk carries thinking/reasoning content rather than final text. */ + thinking?: string + /** Signals the end of a thinking block (thinking is empty string, thinkingDone is true). */ + thinkingDone?: boolean +} + +export interface RunSimpleLlmStreamOptions { + messages: Array<{ role: string; content: string }> + apiKey: string + piModel: PiModel + temperature?: number + maxTokens?: number + onChunk: (chunk: LlmStreamChunk) => void + abortSignal?: AbortSignal +} + +function toPiMessages(messages: Array<{ role: string; content: string }>, timestamp: number): PiMessage[] { + return messages.map((m) => ({ + role: m.role as 'user' | 'assistant', + content: m.content, + timestamp, + })) as unknown as PiMessage[] +} + +/** + * Run a simple LLM streaming call. + * Separates system message, streams text deltas via onChunk callback. + */ +export async function runSimpleLlmStream(options: RunSimpleLlmStreamOptions): Promise { + const { messages, apiKey, piModel, temperature, maxTokens, onChunk, abortSignal } = options + + const systemMsg = messages.find((m) => m.role === 'system') + const nonSystemMsgs = messages.filter((m) => m.role !== 'system') + const now = Date.now() + + const eventStream = streamSimple( + piModel, + { + systemPrompt: systemMsg?.content, + messages: toPiMessages(nonSystemMsgs, now), + }, + { apiKey, temperature, maxTokens, signal: abortSignal } + ) + + let hasTerminalChunk = false + + try { + for await (const event of eventStream) { + if (abortSignal?.aborted) { + if (!hasTerminalChunk) { + hasTerminalChunk = true + onChunk({ content: '', isFinished: true, finishReason: 'stop' }) + } + return + } + + if (event.type === 'thinking_start') { + onChunk({ content: '', isFinished: false, thinking: '' }) + continue + } + + if (event.type === 'thinking_delta') { + onChunk({ content: '', isFinished: false, thinking: event.delta }) + continue + } + + if (event.type === 'thinking_end') { + onChunk({ content: '', isFinished: false, thinking: '', thinkingDone: true }) + continue + } + + if (event.type === 'text_delta') { + onChunk({ content: event.delta, isFinished: false }) + continue + } + + if (event.type === 'done') { + hasTerminalChunk = true + onChunk({ content: '', isFinished: true, finishReason: event.reason === 'length' ? 'length' : 'stop' }) + return + } + + if (event.type === 'error') { + hasTerminalChunk = true + const errorMsg = + event.error?.content + ?.filter((c) => c.type === 'text') + .map((c) => ('text' in c ? c.text : '')) + .join('') || 'Unknown LLM error' + onChunk({ content: '', isFinished: true, finishReason: 'error', error: errorMsg }) + return + } + } + + if (!hasTerminalChunk) { + onChunk({ content: '', isFinished: true, finishReason: 'stop' }) + } + } catch (error) { + if (!hasTerminalChunk) { + const msg = error instanceof Error ? error.message : String(error) + onChunk({ content: '', isFinished: true, finishReason: 'error', error: msg }) + } + } +} diff --git a/packages/node-runtime/src/index.ts b/packages/node-runtime/src/index.ts index 5fba925f..474709e2 100644 --- a/packages/node-runtime/src/index.ts +++ b/packages/node-runtime/src/index.ts @@ -240,7 +240,8 @@ export type { TempDbMeta, ExportedSession } from './merger/temp-db' // Re-exports: @mariozechner/pi-agent-core & @mariozechner/pi-ai export type { AgentTool, AgentToolResult } from './ai' -export { Type, completeSimple, streamSimple } from './ai' +export { Type, completeSimple, streamSimple, runSimpleLlmStream } from './ai' +export type { LlmStreamChunk, RunSimpleLlmStreamOptions } from './ai' export type { PiModel, PiApi, PiMessage, PiUsage, PiTextContent, PiAssistantMessage } from './ai' // ==================== RAG ==================== diff --git a/packages/server/src/http/routes/ai.ts b/packages/server/src/http/routes/ai.ts index f402bd50..e14d4013 100644 --- a/packages/server/src/http/routes/ai.ts +++ b/packages/server/src/http/routes/ai.ts @@ -23,6 +23,7 @@ import { getChatOverview, } from '@openchatlab/core' import type { DataSnapshot, OwnerInfo, MentionedMember } from '@openchatlab/node-runtime' +import { runSimpleLlmStream } from '@openchatlab/node-runtime' import { AGENT_TOOL_REGISTRY } from '@openchatlab/tools' import { adaptToolsForAgent } from '../../ai/tool-adapter' import { getDefaultAssistantConfig, buildPiModel } from '../../ai/llm-config' @@ -399,6 +400,51 @@ export function registerAiRoutes( return fetchRemoteModels(provider, apiKey, baseUrl, apiFormat) }) + // ==================== LLM Simple Chat Stream ==================== + + server.post<{ + Body: { + messages: Array<{ role: string; content: string }> + options?: { temperature?: number; maxTokens?: number } + } + }>('/_web/ai/llm/chat-stream', async (request, reply) => { + const { messages, options } = request.body + + const aiDataDir = getAiDir(dbManager) + const llmConfig = getDefaultAssistantConfig(aiDataDir) + if (!llmConfig) { + return reply.code(400).send({ success: false, error: 'LLM service not configured' }) + } + + const piModel = buildPiModel(llmConfig) + + reply.raw.writeHead(200, { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + }) + + const sendChunk = (data: unknown) => { + reply.raw.write(`event: chunk\ndata: ${JSON.stringify(data)}\n\n`) + } + + try { + await runSimpleLlmStream({ + messages, + apiKey: llmConfig.apiKey, + piModel, + temperature: options?.temperature, + maxTokens: options?.maxTokens, + onChunk: sendChunk, + }) + } catch (error) { + const msg = error instanceof Error ? error.message : String(error) + sendChunk({ content: '', isFinished: true, finishReason: 'error', error: msg }) + } + + reply.raw.end() + }) + // ==================== Tool Catalog ==================== server.get('/_web/ai/tools/catalog', async () => { diff --git a/packages/server/src/http/routes/web.ts b/packages/server/src/http/routes/web.ts index 7d964830..c6762560 100644 --- a/packages/server/src/http/routes/web.ts +++ b/packages/server/src/http/routes/web.ts @@ -1205,4 +1205,53 @@ export function registerWebRoutes( return { success: true, handles } }) } + + // ==================== Cache / File Operations ==================== + + const downloadsDir = options?.pathProvider?.getDownloadsDir() || path.join(os.homedir(), 'Downloads') + + server.post<{ + Body: { filename: string; dataUrl: string } + }>('/_web/cache/save-to-downloads', async (request) => { + const { filename, dataUrl } = request.body + if (!filename || !dataUrl) { + return { success: false, error: 'filename and dataUrl are required' } + } + + const base64Prefix = dataUrl.indexOf(',') + const base64Data = base64Prefix >= 0 ? dataUrl.slice(base64Prefix + 1) : dataUrl + const buffer = Buffer.from(base64Data, 'base64') + + if (!fs.existsSync(downloadsDir)) { + fs.mkdirSync(downloadsDir, { recursive: true }) + } + + const filePath = path.join(downloadsDir, filename) + fs.writeFileSync(filePath, buffer) + + return { success: true, filePath } + }) + + server.post<{ + Body: { filePath: string } + }>('/_web/cache/show-in-folder', async (request) => { + const { filePath } = request.body + if (!filePath) { + return { success: false, error: 'filePath is required' } + } + + const { exec } = await import('child_process') + const platform = process.platform + const dir = path.dirname(filePath) + + if (platform === 'darwin') { + exec(`open -R "${filePath}"`) + } else if (platform === 'win32') { + exec(`explorer /select,"${filePath}"`) + } else { + exec(`xdg-open "${dir}"`) + } + + return { success: true } + }) } diff --git a/src/components/analysis/SQLLabTab.vue b/src/components/analysis/SQLLabTab.vue index 4dc1c456..b8d9721a 100644 --- a/src/components/analysis/SQLLabTab.vue +++ b/src/components/analysis/SQLLabTab.vue @@ -1,5 +1,5 @@