From 1af644e4abe203127014e3553e7fd0896902827d Mon Sep 17 00:00:00 2001 From: digua Date: Mon, 1 Jun 2026 20:52:13 +0800 Subject: [PATCH] refactor(ai): migrate streaming to shared HTTP SSE Squash Phase 2-2 streaming migration: frontend stream service, LLM SSE route, Agent SSE route, and IPC/preload cleanup. --- apps/cli/src/ai/agent-stream-runner.ts | 129 ++++++ apps/cli/src/http/index.ts | 5 +- apps/cli/src/http/routes/ai.ts | 257 ------------ apps/cli/src/http/routes/web/index.ts | 3 + apps/desktop/main/ai/agent-stream-runner.ts | 248 +++++++++++ apps/desktop/main/internal-api.ts | 2 + apps/desktop/main/ipc/ai.ts | 388 +----------------- apps/desktop/preload/apis/ai.ts | 260 +----------- apps/desktop/preload/index.d.ts | 97 +---- apps/desktop/preload/index.ts | 8 +- packages/http-routes/src/context.ts | 38 ++ packages/http-routes/src/index.ts | 2 +- packages/http-routes/src/register.ts | 4 + .../src/routes/web/ai-agent-stream.ts | 69 ++++ .../src/routes/web/ai-llm-stream.ts | 50 +++ .../analysis/SQLLab/ResultTable.vue | 3 +- src/components/analysis/SQLLabTab.vue | 3 +- src/services/ai-stream/service.ts | 156 +++++++ src/services/ai-stream/types.ts | 96 +++++ src/services/registry.ts | 181 +------- src/stores/aiChat.ts | 100 +++-- 21 files changed, 888 insertions(+), 1211 deletions(-) create mode 100644 apps/cli/src/ai/agent-stream-runner.ts delete mode 100644 apps/cli/src/http/routes/ai.ts create mode 100644 apps/desktop/main/ai/agent-stream-runner.ts create mode 100644 packages/http-routes/src/routes/web/ai-agent-stream.ts create mode 100644 packages/http-routes/src/routes/web/ai-llm-stream.ts create mode 100644 src/services/ai-stream/service.ts create mode 100644 src/services/ai-stream/types.ts diff --git a/apps/cli/src/ai/agent-stream-runner.ts b/apps/cli/src/ai/agent-stream-runner.ts new file mode 100644 index 00000000..d19f3c8d --- /dev/null +++ b/apps/cli/src/ai/agent-stream-runner.ts @@ -0,0 +1,129 @@ +/** + * CLI Agent stream runner — provides runAgentStream implementation + * for the shared HTTP route context. + */ + +import type { DatabaseManager, AIConversationManager, AgentStreamChunk } from '@openchatlab/node-runtime' +import { SkillManager, createActivateSkillTool } from '@openchatlab/node-runtime' +import { getChatOverview } from '@openchatlab/core' +import type { DataSnapshot } from '@openchatlab/node-runtime' +import type { AgentStreamRequest } from '@openchatlab/http-routes' +import { AGENT_TOOL_REGISTRY } from '@openchatlab/tools' +import { adaptToolsForAgent } from './tool-adapter' +import { getDefaultAssistantConfig, buildPiModel } from './llm-config' +import { loadAssistantConfig } from './assistant-loader' +import { runServerAgent } from './agent' + +function getAiDir(dbManager: DatabaseManager): string { + const pathProvider = (dbManager as any)['pathProvider'] + if (!pathProvider) { + throw Object.assign(new Error('PathProvider not available'), { statusCode: 500 }) + } + return pathProvider.getAiDataDir() +} + +export function createCliRunAgentStream( + dbManager: DatabaseManager, + convManager: AIConversationManager +): (params: AgentStreamRequest, onEvent: (chunk: AgentStreamChunk) => void, abortSignal: AbortSignal) => Promise { + return async (params, onEvent, abortSignal) => { + const { + userMessage, + conversationId, + historyLeafMessageId, + sessionId, + chatType, + locale, + assistantId, + compressionConfig, + ownerInfo, + mentionedMembers, + thinkingLevel, + } = params + + const aiDataDir = getAiDir(dbManager) + + let assistantSystemPrompt: string | undefined + if (assistantId) { + const assistantConfig = loadAssistantConfig(aiDataDir, assistantId) + if (assistantConfig?.systemPrompt) { + assistantSystemPrompt = assistantConfig.systemPrompt + } + } + + const llmConfig = getDefaultAssistantConfig(aiDataDir) + const maxToolResultPercent = compressionConfig?.maxToolResultPercent ?? 50 + const contextWindow = llmConfig ? (buildPiModel(llmConfig).contextWindow ?? 128000) : 128000 + const maxToolResultTokens = Math.floor(contextWindow * (maxToolResultPercent / 100)) + + const db = (dbManager as any).open?.(sessionId) + const agentTools = db + ? adaptToolsForAgent(AGENT_TOOL_REGISTRY, () => ({ db, sessionId, locale }), { maxToolResultTokens }) + : [] + + const skillManager = new SkillManager(aiDataDir) + skillManager.init() + const toolNames = agentTools.map((t: { name: string }) => t.name) + const skillMenu = skillManager.getSkillMenu(chatType ?? 'group', toolNames) + + if (skillMenu) { + const activateSkillTool = createActivateSkillTool({ + chatType: chatType ?? 'group', + allowedTools: toolNames, + locale, + getSkillConfig: (id) => skillManager.getSkillConfig(id), + }) + agentTools.push(activateSkillTool as any) + } + + const resolvedCompression = compressionConfig?.enabled + ? { + enabled: true as const, + tokenThresholdPercent: compressionConfig.tokenThresholdPercent ?? 75, + bufferSizePercent: compressionConfig.bufferSizePercent ?? 20, + maxToolResultPercent: compressionConfig.maxToolResultPercent, + } + : undefined + + let dataSnapshot: DataSnapshot | undefined + if (db) { + try { + const overview = getChatOverview(db, 5) + if (overview) { + dataSnapshot = { + name: overview.name, + platform: overview.platform, + type: overview.type, + totalMessages: overview.totalMessages, + totalMembers: overview.totalMembers, + firstMessageTs: overview.firstMessageTs, + lastMessageTs: overview.lastMessageTs, + capturedAt: Math.floor(Date.now() / 1000), + } + } + } catch { + // non-fatal + } + } + + await runServerAgent({ + userMessage, + conversationId, + historyLeafMessageId, + chatType, + locale, + assistantSystemPrompt, + skillMenu, + compressionConfig: resolvedCompression, + tools: agentTools, + aiDataDir, + convManager, + onEvent, + abortSignal, + ownerInfo, + mentionedMembers, + dataSnapshot, + thinkingLevel, + }) + } +} diff --git a/apps/cli/src/http/index.ts b/apps/cli/src/http/index.ts index 2238e495..ea5ee0d8 100644 --- a/apps/cli/src/http/index.ts +++ b/apps/cli/src/http/index.ts @@ -24,10 +24,10 @@ import type { ConfigStorage } from '@openchatlab/node-runtime' import { createServer } from './server' import { setAuthToken, setRequireAuth } from '@openchatlab/http-routes' import { registerWebRoutes } from './routes/web' -import { registerAiRoutes } from './routes/ai' import { registerProxyRoutes } from './routes/proxy' import { initServerAiLogger, closeServerAiLogger } from '../ai/logger' import { getAssistantManager, getSkillManagerCore } from '../ai/manager-factory' +import { createCliRunAgentStream } from '../ai/agent-stream-runner' import { initSync, cleanupSync } from '../sync' import { resolveCliPath } from '../paths' import { resolveApiKey, writeAuthProfile } from '@openchatlab/config' @@ -164,7 +164,6 @@ export async function startHttpServer(options?: HttpServerOptions): Promise<{ limits: { fileSize: 1024 * 1024 * 1024 }, // 1GB }) - registerAiRoutes(server, dbManager, convManager) registerWebRoutes(server, dbManager, { pathProvider, nativeBinding, @@ -176,6 +175,7 @@ export async function startHttpServer(options?: HttpServerOptions): Promise<{ llmConfigStore, customProviderStore: new CustomProviderStore(createFileConfigStorage(aiDataDir)), customModelStore: new CustomModelStore(createFileConfigStorage(aiDataDir)), + runAgentStream: createCliRunAgentStream(dbManager, convManager), }, }) @@ -229,4 +229,3 @@ export async function stopHttpServer(): Promise { export { createServer } from './server' export { registerWebRoutes } from './routes/web' -export { registerAiRoutes } from './routes/ai' diff --git a/apps/cli/src/http/routes/ai.ts b/apps/cli/src/http/routes/ai.ts deleted file mode 100644 index 42faafbd..00000000 --- a/apps/cli/src/http/routes/ai.ts +++ /dev/null @@ -1,257 +0,0 @@ -/** - * AI Web API — CLI-specific routes - * - * Routes that remain CLI-specific (agent streaming, tool catalog, chat stream). - * CRUD routes for assistants, skills, LLM config, conversations, and summaries - * are now handled by @openchatlab/http-routes shared package. - */ - -import type { FastifyInstance } from 'fastify' -import type { DatabaseManager, AIConversationManager } from '@openchatlab/node-runtime' -import { SkillManager, createActivateSkillTool } from '@openchatlab/node-runtime' -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' -import { loadAssistantConfig } from '../../ai/assistant-loader' -import { runServerAgent, type AgentStreamChunk } from '../../ai/agent' - -function getAiDir(dbManager: DatabaseManager): string { - const pathProvider = (dbManager as any)['pathProvider'] - if (!pathProvider) { - throw Object.assign(new Error('PathProvider not available'), { statusCode: 500 }) - } - return pathProvider.getAiDataDir() -} - -export function registerAiRoutes( - server: FastifyInstance, - dbManager: DatabaseManager, - convManager?: AIConversationManager -): void { - // ==================== 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() - }) - - if (!convManager) return - - // ==================== Agent SSE Stream ==================== - - const activeAgentAborts = new Map() - - server.post<{ - Body: { - userMessage: string - conversationId: string - historyLeafMessageId?: string | null - sessionId: string - chatType?: 'group' | 'private' - locale?: string - assistantId?: string - compressionConfig?: { - enabled: boolean - tokenThresholdPercent?: number - bufferSizePercent?: number - maxToolResultPercent?: number - } - ownerInfo?: OwnerInfo - mentionedMembers?: MentionedMember[] - thinkingLevel?: string - } - }>('/_web/ai/agent/stream', async (request, reply) => { - const { - userMessage, - conversationId, - historyLeafMessageId, - sessionId, - chatType, - locale, - assistantId, - compressionConfig, - ownerInfo, - mentionedMembers, - thinkingLevel, - } = request.body - - const requestId = `req_${Date.now()}_${Math.random().toString(36).slice(2, 8)}` - const abortController = new AbortController() - activeAgentAborts.set(requestId, abortController) - - reply.raw.writeHead(200, { - 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', - Connection: 'keep-alive', - 'X-Request-Id': requestId, - }) - - const sendSSE = (event: string, data: unknown) => { - reply.raw.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`) - } - - sendSSE('meta', { requestId }) - - const aiDataDir = getAiDir(dbManager) - - let assistantSystemPrompt: string | undefined - if (assistantId) { - const assistantConfig = loadAssistantConfig(aiDataDir, assistantId) - if (assistantConfig?.systemPrompt) { - assistantSystemPrompt = assistantConfig.systemPrompt - } - } - - const llmConfig = getDefaultAssistantConfig(aiDataDir) - const maxToolResultPercent = compressionConfig?.maxToolResultPercent ?? 50 - const contextWindow = llmConfig ? (buildPiModel(llmConfig).contextWindow ?? 128000) : 128000 - const maxToolResultTokens = Math.floor(contextWindow * (maxToolResultPercent / 100)) - - const db = (dbManager as any).open?.(sessionId) - const agentTools = db - ? adaptToolsForAgent(AGENT_TOOL_REGISTRY, () => ({ db, sessionId, locale }), { maxToolResultTokens }) - : [] - - const skillManager = new SkillManager(aiDataDir) - skillManager.init() - const toolNames = agentTools.map((t: { name: string }) => t.name) - const skillMenu = skillManager.getSkillMenu(chatType ?? 'group', toolNames) - - if (skillMenu) { - const activateSkillTool = createActivateSkillTool({ - chatType: chatType ?? 'group', - allowedTools: toolNames, - locale, - getSkillConfig: (id) => skillManager.getSkillConfig(id), - }) - agentTools.push(activateSkillTool as any) - } - - const onEvent = (event: AgentStreamChunk) => { - sendSSE(event.type, event) - if (event.type === 'done') { - activeAgentAborts.delete(requestId) - reply.raw.end() - } - } - - reply.raw.on('close', () => { - if (!abortController.signal.aborted) { - abortController.abort() - } - activeAgentAborts.delete(requestId) - }) - - try { - const resolvedCompression = compressionConfig?.enabled - ? { - enabled: true as const, - tokenThresholdPercent: compressionConfig.tokenThresholdPercent ?? 75, - bufferSizePercent: compressionConfig.bufferSizePercent ?? 20, - maxToolResultPercent: compressionConfig.maxToolResultPercent, - } - : undefined - - let dataSnapshot: DataSnapshot | undefined - if (db) { - try { - const overview = getChatOverview(db, 5) - if (overview) { - dataSnapshot = { - name: overview.name, - platform: overview.platform, - type: overview.type, - totalMessages: overview.totalMessages, - totalMembers: overview.totalMembers, - firstMessageTs: overview.firstMessageTs, - lastMessageTs: overview.lastMessageTs, - capturedAt: Math.floor(Date.now() / 1000), - } - } - } catch { - // non-fatal: proceed without snapshot - } - } - - await runServerAgent({ - userMessage, - conversationId, - historyLeafMessageId, - chatType, - locale, - assistantSystemPrompt, - skillMenu, - compressionConfig: resolvedCompression, - tools: agentTools, - aiDataDir, - convManager, - onEvent, - abortSignal: abortController.signal, - ownerInfo, - mentionedMembers, - dataSnapshot, - thinkingLevel, - }) - } catch (error) { - const msg = error instanceof Error ? error.message : String(error) - sendSSE('error', { type: 'error', error: { name: 'ServerError', message: msg } }) - sendSSE('done', { type: 'done', isFinished: true }) - activeAgentAborts.delete(requestId) - reply.raw.end() - } - }) - - server.post<{ - Body: { requestId: string } - }>('/_web/ai/agent/abort', async (request) => { - const { requestId } = request.body - const controller = activeAgentAborts.get(requestId) - if (controller) { - controller.abort() - activeAgentAborts.delete(requestId) - return { success: true } - } - return { success: false } - }) -} diff --git a/apps/cli/src/http/routes/web/index.ts b/apps/cli/src/http/routes/web/index.ts index ef128429..7540d20a 100644 --- a/apps/cli/src/http/routes/web/index.ts +++ b/apps/cli/src/http/routes/web/index.ts @@ -32,6 +32,7 @@ import type { } from '@openchatlab/node-runtime' import { createDatabaseManagerAdapter } from '@openchatlab/node-runtime' import { registerSharedRoutes } from '@openchatlab/http-routes' +import type { HttpRouteContext } from '@openchatlab/http-routes' import { MergeSessionCache } from '../../../merger/merge-cache' import { registerImportRoutes } from './import' import { openDirectoryPath, showPathInFolder } from './cache' @@ -46,6 +47,7 @@ export interface AiContextOptions { llmConfigStore: LLMConfigStore customProviderStore: CustomProviderStore customModelStore: CustomModelStore + runAgentStream?: HttpRouteContext['runAgentStream'] } /** @@ -130,6 +132,7 @@ export function registerWebRoutes( llmConfigStore: ai.llmConfigStore, customProviderStore: ai.customProviderStore, customModelStore: ai.customModelStore, + runAgentStream: ai.runAgentStream, }), }, ai ? { requireAi: true } : undefined diff --git a/apps/desktop/main/ai/agent-stream-runner.ts b/apps/desktop/main/ai/agent-stream-runner.ts new file mode 100644 index 00000000..918bd536 --- /dev/null +++ b/apps/desktop/main/ai/agent-stream-runner.ts @@ -0,0 +1,248 @@ +/** + * Electron Agent stream runner — provides runAgentStream implementation + * for the shared HTTP route context. + * + * Mirrors the logic from ipc/ai.ts `agent:runStream` handler, + * but outputs events via callback instead of IPC. + */ + +import type { AgentStreamChunk as SharedAgentStreamChunk } from '@openchatlab/node-runtime' +import type { AgentStreamRequest } from '@openchatlab/http-routes' +import { checkAndCompress, createCompressionLlmAdapter, formatAIError } from '@openchatlab/node-runtime' +import type { CompressionConfig, CompressionLlmAdapter, AgentRuntimeStatus } from '@openchatlab/node-runtime' +import { Agent, type AgentStreamChunk, type SkillContext } from './agent' +import type { ToolContext } from './tools/types' +import { getDefaultAssistantConfig, buildPiModel, findModelDefinition } from './llm' +import type { AIServiceConfig } from './llm/types' +import { getDefaultGeneralAssistantId } from './assistant/defaultGeneral' +import * as assistantManager from './assistant' +import type { AssistantConfig } from './assistant/types' +import * as skillManager from './skills' +import { aiLogger } from './logger' +import { serializeError } from './serialize-error' +import { getManager as getConversationManager } from './conversations' +import { t } from '../i18n' +import * as workerManager from '../worker/workerManager' +import { getProviderInfo, type LLMProvider } from './llm' + +const DEFAULT_CONTEXT_WINDOW = 128000 + +function resolveProviderName(provider?: LLMProvider): string { + if (provider === 'openai-compatible') return t('llm.genericProviderName') + return provider ? getProviderInfo(provider)?.name || provider : t('llm.genericProviderName') +} + +function buildCompressionAdapter(activeAIConfig: AIServiceConfig, onCompressing?: () => void): CompressionLlmAdapter { + const modelDef = findModelDefinition(activeAIConfig.provider, activeAIConfig.model || '') + return createCompressionLlmAdapter({ + piModel: buildPiModel(activeAIConfig), + apiKey: activeAIConfig.apiKey, + contextWindow: modelDef?.contextWindow ?? DEFAULT_CONTEXT_WINDOW, + onCompressing, + onError: (error) => aiLogger.warn('Compression', 'LLM compression attempt failed', { error: String(error) }), + }) +} + +const compressionLogger = { + info: (cat: string, msg: string, extra?: Record) => aiLogger.info(cat, msg, extra), + warn: (cat: string, msg: string, extra?: Record) => aiLogger.warn(cat, msg, extra), + error: (cat: string, msg: string, extra?: Record) => aiLogger.error(cat, msg, extra), +} + +export function createElectronRunAgentStream(): ( + params: AgentStreamRequest, + onEvent: (chunk: SharedAgentStreamChunk) => void, + abortSignal: AbortSignal +) => Promise { + return async (params, onEvent, abortSignal) => { + const { + userMessage, + conversationId, + historyLeafMessageId, + sessionId, + chatType, + locale, + assistantId, + skillId, + enableAutoSkill, + compressionConfig, + ownerInfo, + mentionedMembers, + thinkingLevel, + } = params + + const requestId = `internal_${Date.now()}` + + aiLogger.info('AgentStream', `Agent stream request: ${requestId}`, { + userMessage: userMessage.slice(0, 100), + sessionId, + conversationId, + chatType: chatType ?? 'group', + assistantId: assistantId ?? '(none)', + skillId: skillId ?? '(none)', + enableAutoSkill: enableAutoSkill ?? false, + }) + + const activeAIConfig = getDefaultAssistantConfig() + if (!activeAIConfig) { + onEvent({ type: 'error', error: { name: 'ConfigError', message: t('llm.notConfigured') } }) + onEvent({ type: 'done', isFinished: true }) + return + } + const piModel = buildPiModel(activeAIConfig) + + if (compressionConfig?.enabled && conversationId && historyLeafMessageId === undefined) { + try { + const tempAssistantConfig = assistantId + ? (assistantManager.getAssistantConfig(assistantId) ?? undefined) + : undefined + const systemPromptForCompression = tempAssistantConfig?.systemPrompt || '' + + const compressionResult = await checkAndCompress( + conversationId, + compressionConfig as CompressionConfig, + systemPromptForCompression, + buildCompressionAdapter(activeAIConfig, () => { + onEvent({ + type: 'status', + status: { + phase: 'compressing', + round: 0, + toolsUsed: 0, + contextTokens: 0, + totalUsage: { promptTokens: 0, completionTokens: 0, totalTokens: 0 }, + updatedAt: Date.now(), + } satisfies AgentRuntimeStatus, + }) + }), + getConversationManager(), + compressionLogger + ) + + if (compressionResult.compressed && compressionResult.summaryContent) { + onEvent({ + type: 'compression_done', + compressionResult: { + summaryContent: compressionResult.summaryContent, + tokensBefore: compressionResult.tokensBefore ?? 0, + tokensAfter: compressionResult.tokensAfter ?? 0, + timestamp: Date.now(), + }, + }) + } + } catch (error) { + aiLogger.error('AgentStream', `Compression failed: ${requestId}`, { error: String(error) }) + } + } + + const defaultAssistantId = getDefaultGeneralAssistantId(locale) + let resolvedAssistantId = assistantId || defaultAssistantId + let assistantConfig: AssistantConfig | undefined = + assistantManager.getAssistantConfig(resolvedAssistantId) ?? undefined + if (!assistantConfig && resolvedAssistantId !== defaultAssistantId) { + resolvedAssistantId = defaultAssistantId + assistantConfig = assistantManager.getAssistantConfig(defaultAssistantId) ?? undefined + } + + let skillCtx: SkillContext | undefined + if (skillId) { + const skillDef = skillManager.getSkillConfig(skillId) ?? undefined + if (skillDef) { + skillCtx = { skillDef } + } + } else if (enableAutoSkill) { + const effectiveChatType = chatType ?? 'group' + const allowedTools = assistantConfig?.allowedBuiltinTools + const menu = skillManager.getSkillMenu(effectiveChatType, allowedTools) + if (menu) { + skillCtx = { skillMenu: menu } + } + } + + const maxToolResultPercent = compressionConfig?.maxToolResultPercent ?? 50 + const modelDef = findModelDefinition(activeAIConfig.provider, activeAIConfig.model || '') + const resolvedContextWindow = modelDef?.contextWindow || DEFAULT_CONTEXT_WINDOW + const maxToolResultTokens = Math.floor(resolvedContextWindow * (maxToolResultPercent / 100)) + + let dataSnapshot: ToolContext['dataSnapshot'] | undefined + try { + const overview = await workerManager.getChatOverview(sessionId, 5) + if (overview) { + dataSnapshot = { + name: overview.name, + platform: overview.platform, + type: overview.type, + totalMessages: overview.totalMessages, + totalMembers: overview.totalMembers, + firstMessageTs: overview.firstMessageTs, + lastMessageTs: overview.lastMessageTs, + capturedAt: Math.floor(Date.now() / 1000), + } + } + } catch (error) { + aiLogger.warn('AgentStream', `Failed to load data snapshot: ${requestId}`, { error: String(error) }) + } + + const context: ToolContext = { + sessionId, + conversationId, + historyLeafMessageId: historyLeafMessageId ?? undefined, + timeFilter: params.timeFilter, + maxMessagesLimit: params.maxMessagesLimit, + ownerInfo, + mentionedMembers: mentionedMembers as ToolContext['mentionedMembers'], + preprocessConfig: params.preprocessConfig as ToolContext['preprocessConfig'], + maxToolResultTokens, + dataSnapshot, + } + + const agent = new Agent( + context, + piModel, + activeAIConfig.apiKey, + { + abortSignal, + thinkingLevel: thinkingLevel as import('@openchatlab/core').ThinkingLevel | undefined, + }, + chatType ?? 'group', + locale ?? 'zh-CN', + assistantConfig, + skillCtx + ) + + try { + const result = await agent.executeStream(userMessage, (chunk: AgentStreamChunk) => { + if (abortSignal.aborted) return + onEvent(chunk as SharedAgentStreamChunk) + }) + + if (abortSignal.aborted) { + onEvent({ + type: 'done', + isFinished: true, + usage: result.totalUsage, + }) + return + } + + onEvent({ + type: 'done', + isFinished: true, + usage: result.totalUsage, + }) + } catch (error) { + if (error instanceof Error && error.name === 'AbortError') { + onEvent({ type: 'done', isFinished: true }) + return + } + const serializedError = serializeError(error, activeAIConfig.provider) + serializedError.friendlyMessage = formatAIError(error, { + providerName: resolveProviderName(activeAIConfig.provider), + rawErrorLabel: t('llm.rawErrorLabel'), + }) + aiLogger.error('AgentStream', `Agent execution error: ${requestId}`, serializedError) + onEvent({ type: 'error', error: serializedError, isFinished: true }) + onEvent({ type: 'done', isFinished: true }) + } + } +} diff --git a/apps/desktop/main/internal-api.ts b/apps/desktop/main/internal-api.ts index f37e6b35..185d9317 100644 --- a/apps/desktop/main/internal-api.ts +++ b/apps/desktop/main/internal-api.ts @@ -34,6 +34,7 @@ import { resolveApiKey, writeAuthProfile } from '@openchatlab/config' import { getManager as getConversationManager } from './ai/conversations' import { getManager as getAssistantManager } from './ai/assistant/manager' import { getManager as getSkillManager } from './ai/skills/manager' +import { createElectronRunAgentStream } from './ai/agent-stream-runner' export interface InternalEndpoint { baseUrl: string @@ -151,6 +152,7 @@ export async function startInternalServer(pathProvider: PathProvider): Promise) => aiLogger.info(cat, msg, extra), - warn: (cat: string, msg: string, extra?: Record) => aiLogger.warn(cat, msg, extra), - error: (cat: string, msg: string, extra?: Record) => aiLogger.error(cat, msg, extra), -} - -function buildCompressionAdapter(activeAIConfig: AIServiceConfig, onCompressing?: () => void): CompressionLlmAdapter { - const modelDef = findModelDefinition(activeAIConfig.provider, activeAIConfig.model || '') - return createCompressionLlmAdapter({ - piModel: buildPiModel(activeAIConfig), - apiKey: activeAIConfig.apiKey, - contextWindow: modelDef?.contextWindow ?? DEFAULT_CONTEXT_WINDOW, - onCompressing, - onError: (error) => aiLogger.warn('Compression', 'LLM compression attempt failed', { error: String(error) }), - }) -} - function toPiSimpleMessages(messages: Array<{ role: string; content: string }>, timestamp: number): PiMessage[] { return messages.map((message) => ({ role: message.role as 'user' | 'assistant', @@ -61,15 +33,7 @@ function toPiSimpleMessages(messages: Array<{ role: string; content: string }>, })) as unknown as PiMessage[] } -// ==================== AI Agent 请求追踪 ==================== -const activeAgentRequests = new Map() - -function resolveProviderName(provider?: llm.LLMProvider): string { - if (provider === 'openai-compatible') return t('llm.genericProviderName') - return provider ? llm.getProviderInfo(provider)?.name || provider : t('llm.genericProviderName') -} - -export function registerAIHandlers({ win }: IpcContext): void { +export function registerAIHandlers(_ctx: IpcContext): void { console.log('[IPC] Registering AI handlers...') try { @@ -185,46 +149,6 @@ export function registerAIHandlers({ win }: IpcContext): void { } ) - ipcMain.handle( - 'llm:chatStream', - async ( - _, - requestId: string, - messages: Array<{ role: string; content: string }>, - options?: { temperature?: number; maxTokens?: number } - ) => { - try { - const activeConfig = getDefaultAssistantConfig() - if (!activeConfig) { - return { success: false, error: t('llm.notConfigured') } - } - const piModel = buildPiModel(activeConfig) - - ;(async () => { - await runSimpleLlmStream({ - messages, - apiKey: activeConfig.apiKey, - piModel, - temperature: options?.temperature, - maxTokens: options?.maxTokens, - onChunk: (chunk) => { - if (chunk.error) { - win.webContents.send('llm:streamChunk', { requestId, error: chunk.error, chunk }) - } else { - win.webContents.send('llm:streamChunk', { requestId, chunk }) - } - }, - }) - })() - - return { success: true } - } catch (error) { - aiLogger.error('IPC', 'llm:chatStream error', { error: String(error) }) - return { success: false, error: String(error) } - } - } - ) - // ==================== 工具测试 API ==================== const activeToolTests = new Map() @@ -313,291 +237,7 @@ export function registerAIHandlers({ win }: IpcContext): void { return { success: false } }) - // ==================== AI Agent API ==================== - - ipcMain.handle( - 'agent:runStream', - async ( - _, - requestId: string, - userMessage: string, - context: ToolContext, - chatType?: 'group' | 'private', - locale?: string, - assistantId?: string, - skillId?: string | null, - enableAutoSkill?: boolean, - compressionConfig?: CompressionConfig, - thinkingLevel?: string - ) => { - aiLogger.info('IPC', `Agent stream request received: ${requestId}`, { - userMessage: userMessage.slice(0, 100), - sessionId: context.sessionId, - conversationId: context.conversationId, - chatType: chatType ?? 'group', - assistantId: assistantId ?? '(none)', - skillId: skillId ?? '(none)', - enableAutoSkill: enableAutoSkill ?? false, - }) - - try { - const abortController = new AbortController() - activeAgentRequests.set(requestId, abortController) - - const activeAIConfig = getDefaultAssistantConfig() - if (!activeAIConfig) { - return { success: false, error: t('llm.notConfigured') } - } - const piModel = buildPiModel(activeAIConfig) - - if (compressionConfig?.enabled && context.conversationId && context.historyLeafMessageId === undefined) { - try { - const tempAssistantConfig = assistantId - ? (assistantManager.getAssistantConfig(assistantId) ?? undefined) - : undefined - const systemPromptForCompression = tempAssistantConfig?.systemPrompt || '' - - const compressionResult = await checkAndCompress( - context.conversationId, - compressionConfig, - systemPromptForCompression, - buildCompressionAdapter(activeAIConfig, () => { - win.webContents.send('agent:streamChunk', { - requestId, - chunk: { - type: 'status', - status: { - phase: 'compressing', - round: 0, - toolsUsed: 0, - contextTokens: 0, - totalUsage: { promptTokens: 0, completionTokens: 0, totalTokens: 0 }, - updatedAt: Date.now(), - } satisfies import('@electron/shared/types').AgentRuntimeStatus, - }, - }) - }), - getConversationManager(), - compressionLogger - ) - - aiLogger.info('IPC', `Compression result for ${requestId}`, compressionResult) - - if (compressionResult.compressed && compressionResult.summaryContent) { - win.webContents.send('agent:streamChunk', { - requestId, - chunk: { - type: 'compression_done', - compressionResult: { - summaryContent: compressionResult.summaryContent, - tokensBefore: compressionResult.tokensBefore ?? 0, - tokensAfter: compressionResult.tokensAfter ?? 0, - timestamp: Date.now(), - }, - }, - }) - } - } catch (error) { - aiLogger.error('IPC', `Compression failed for ${requestId}, continuing without compression`, { - error: String(error), - }) - } - } else if (compressionConfig?.enabled && context.historyLeafMessageId !== undefined) { - aiLogger.info('IPC', `Skipping compression for edited branch request: ${requestId}`, { - conversationId: context.conversationId, - historyLeafMessageId: context.historyLeafMessageId, - }) - } - - const pp = context.preprocessConfig - aiLogger.info('IPC', `Agent context: ${requestId}`, { - model: activeAIConfig.model, - provider: activeAIConfig.provider, - baseUrl: activeAIConfig.baseUrl || '(default)', - maxMessagesLimit: context.maxMessagesLimit, - hasTimeFilter: !!context.timeFilter, - mentionedMembersCount: context.mentionedMembers?.length ?? 0, - preprocess: pp - ? { - dataCleaning: pp.dataCleaning ?? true, - mergeConsecutive: pp.mergeConsecutive, - denoise: pp.denoise, - desensitize: pp.desensitize, - anonymizeNames: pp.anonymizeNames, - } - : '(disabled)', - }) - - const defaultAssistantId = getDefaultGeneralAssistantId(locale) - let resolvedAssistantId = assistantId || defaultAssistantId - let assistantConfig: AssistantConfig | undefined = - assistantManager.getAssistantConfig(resolvedAssistantId) ?? undefined - if (!assistantConfig && resolvedAssistantId !== defaultAssistantId) { - aiLogger.warn('IPC', `Assistant not found: ${resolvedAssistantId}, falling back to ${defaultAssistantId}`, { - requestedAssistantId: assistantId ?? null, - locale: locale ?? null, - }) - resolvedAssistantId = defaultAssistantId - assistantConfig = assistantManager.getAssistantConfig(defaultAssistantId) ?? undefined - } - - let skillCtx: SkillContext | undefined - if (skillId) { - const skillDef = skillManager.getSkillConfig(skillId) ?? undefined - if (skillDef) { - skillCtx = { skillDef } - } else { - aiLogger.warn('IPC', `Skill not found: ${skillId}`) - } - } else if (enableAutoSkill) { - const effectiveChatType = chatType ?? 'group' - const allowedTools = assistantConfig?.allowedBuiltinTools - const menu = skillManager.getSkillMenu(effectiveChatType, allowedTools) - if (menu) { - skillCtx = { skillMenu: menu } - } - } - - const maxToolResultPercent = compressionConfig?.maxToolResultPercent ?? 50 - const modelDef = llm.findModelDefinition(activeAIConfig.provider, activeAIConfig.model || '') - const resolvedContextWindow = modelDef?.contextWindow || 128000 - const maxToolResultTokens = Math.floor(resolvedContextWindow * (maxToolResultPercent / 100)) - let dataSnapshot: ToolContext['dataSnapshot'] | undefined = context.dataSnapshot - try { - const overview = await workerManager.getChatOverview(context.sessionId, 5) - if (overview) { - dataSnapshot = { - name: overview.name, - platform: overview.platform, - type: overview.type, - totalMessages: overview.totalMessages, - totalMembers: overview.totalMembers, - firstMessageTs: overview.firstMessageTs, - lastMessageTs: overview.lastMessageTs, - capturedAt: Math.floor(Date.now() / 1000), - } - } - } catch (error) { - aiLogger.warn('IPC', `Failed to load agent data snapshot: ${requestId}`, { error: String(error) }) - } - const enrichedContext: ToolContext = { ...context, maxToolResultTokens, dataSnapshot } - - const agent = new Agent( - enrichedContext, - piModel, - activeAIConfig.apiKey, - { - abortSignal: abortController.signal, - thinkingLevel: thinkingLevel as import('@openchatlab/core').ThinkingLevel | undefined, - }, - chatType ?? 'group', - locale ?? 'zh-CN', - assistantConfig, - skillCtx - ) - - ;(async () => { - try { - const result = await agent.executeStream(userMessage, (chunk: AgentStreamChunk) => { - if (abortController.signal.aborted) { - return - } - if (chunk.type === 'tool_start') { - aiLogger.info('IPC', `Tool call: ${chunk.toolName}`, chunk.toolParams) - } - win.webContents.send('agent:streamChunk', { requestId, chunk }) - }) - - if (abortController.signal.aborted) { - aiLogger.info('IPC', `Agent aborted: ${requestId}`) - win.webContents.send('agent:complete', { - requestId, - result: { - content: result.content, - toolsUsed: result.toolsUsed, - toolRounds: result.toolRounds, - totalUsage: result.totalUsage, - aborted: true, - }, - }) - return - } - - win.webContents.send('agent:complete', { - requestId, - result: { - content: result.content, - toolsUsed: result.toolsUsed, - toolRounds: result.toolRounds, - totalUsage: result.totalUsage, - }, - }) - - aiLogger.info('IPC', `Agent execution completed: ${requestId}`, { - toolsUsed: result.toolsUsed, - toolRounds: result.toolRounds, - contentLength: result.content.length, - totalUsage: result.totalUsage, - }) - } catch (error) { - if (error instanceof Error && error.name === 'AbortError') { - aiLogger.info('IPC', `Agent request aborted (error): ${requestId}`) - win.webContents.send('agent:complete', { - requestId, - result: { content: '', toolsUsed: [], toolRounds: 0, aborted: true }, - }) - return - } - const serializedError = serializeError(error, activeAIConfig.provider) - serializedError.friendlyMessage = formatAIError(error, { - providerName: resolveProviderName(activeAIConfig.provider), - rawErrorLabel: t('llm.rawErrorLabel'), - }) - if (!serializedError.url && activeAIConfig.baseUrl) { - serializedError.url = activeAIConfig.baseUrl - } - aiLogger.error('IPC', `Agent execution error: ${requestId}`, serializedError) - win.webContents.send('agent:streamChunk', { - requestId, - chunk: { type: 'error', error: serializedError, isFinished: true }, - }) - win.webContents.send('agent:complete', { - requestId, - result: { - content: '', - toolsUsed: [], - toolRounds: 0, - totalUsage: { promptTokens: 0, completionTokens: 0, totalTokens: 0 }, - error: serializedError, - }, - }) - } finally { - activeAgentRequests.delete(requestId) - } - })() - - return { success: true } - } catch (error) { - aiLogger.error('IPC', `Failed to create Agent request: ${requestId}`, { error: String(error) }) - return { success: false, error: String(error) } - } - } - ) - - ipcMain.handle('agent:abort', async (_, requestId: string) => { - aiLogger.info('IPC', `Abort request received: ${requestId}`) - - const abortController = activeAgentRequests.get(requestId) - if (abortController) { - abortController.abort() - activeAgentRequests.delete(requestId) - aiLogger.info('IPC', `Agent request aborted: ${requestId}`) - return { success: true } - } else { - aiLogger.warn('IPC', `Agent request not found: ${requestId}`) - return { success: false, error: 'Request not found' } - } - }) + // Agent streaming has been migrated to shared SSE route (ai-agent-stream.ts) // ==================== 上下文 token 估算 ==================== diff --git a/apps/desktop/preload/apis/ai.ts b/apps/desktop/preload/apis/ai.ts index 5b8c2d61..9d6fd40b 100644 --- a/apps/desktop/preload/apis/ai.ts +++ b/apps/desktop/preload/apis/ai.ts @@ -1,17 +1,14 @@ /** * AI 相关 API — 仅保留 IPC 必须的能力 * - * Conversation/message CRUD 和 debug 已迁移到 HTTP 共享路由（FetchAIAdapter）， + * Conversation/message CRUD 已迁移到 HTTP 共享路由（FetchAIAdapter）。 + * LLM/Agent streaming 已迁移到 SSE 共享路由（useAgentStreamService/useLlmStreamService）。 * 此处只保留需要 worker、native shell、工具注册表等 IPC 才能提供的功能。 */ import { ipcRenderer } from 'electron' import type { ExportProgress } from '../../../../src/types/base' -// Agent API 类型 — 从 shared/types 统一导入 -export type { TokenUsage, AgentRuntimeStatus } from '../../shared/types' -import type { TokenUsage, AgentRuntimeStatus, SerializedErrorInfo } from '../../shared/types' - -export type { SerializedErrorInfo } from '../../shared/types' +export type { TokenUsage, AgentRuntimeStatus, SerializedErrorInfo } from '../../shared/types' // ==================== 类型定义 ==================== @@ -25,42 +22,6 @@ export interface ChatOptions { maxTokens?: number } -export interface ChatStreamChunk { - content: string - isFinished: boolean - finishReason?: 'stop' | 'length' | 'error' - error?: string - thinking?: string - thinkingDone?: boolean -} - -export interface AgentStreamChunk { - type: 'content' | 'think' | 'tool_start' | 'tool_result' | 'status' | 'compression_done' | 'done' | 'error' - content?: string - thinkTag?: string - thinkDurationMs?: number - toolName?: string - toolParams?: Record - toolResult?: unknown - status?: AgentRuntimeStatus - error?: SerializedErrorInfo - isFinished?: boolean - compressionResult?: { - summaryContent: string - tokensBefore: number - tokensAfter: number - timestamp: number - } - usage?: TokenUsage -} - -export interface AgentResult { - content: string - toolsUsed: string[] - toolRounds: number - error?: SerializedErrorInfo -} - export interface DesensitizeRule { id: string label: string @@ -98,24 +59,6 @@ export interface PreprocessConfig { anonymizeNames: boolean } -export interface ToolContext { - sessionId: string - conversationId?: string - historyLeafMessageId?: string | null - timeFilter?: { startTs: number; endTs: number } - maxMessagesLimit?: number - ownerInfo?: { platformId: string; displayName: string } - mentionedMembers?: Array<{ - memberId: number - platformId: string - displayName: string - aliases: string[] - mentionText: string - }> - locale?: string - preprocessConfig?: PreprocessConfig -} - // ==================== AI API (IPC-only subset) ==================== export const aiApi = { @@ -283,8 +226,7 @@ export const aiApi = { }, } -// ==================== LLM API ==================== - +// LLM chat (non-streaming) still uses IPC for Electron-specific config resolution export const llmApi = { chat: ( messages: ChatMessage[], @@ -292,198 +234,4 @@ export const llmApi = { ): Promise<{ success: boolean; content?: string; error?: string }> => { return ipcRenderer.invoke('llm:chat', messages, options) }, - - chatStream: ( - messages: ChatMessage[], - options?: ChatOptions, - onChunk?: (chunk: ChatStreamChunk) => void - ): Promise<{ success: boolean; error?: string }> => { - return new Promise((resolve) => { - const requestId = `llm_${Date.now()}_${Math.random().toString(36).slice(2, 8)}` - console.log('[preload] chatStream 开始，requestId:', requestId) - - const handler = ( - _event: Electron.IpcRendererEvent, - data: { requestId: string; chunk: ChatStreamChunk; error?: string } - ) => { - if (data.requestId === requestId) { - if (data.error) { - console.log('[preload] chatStream 收到错误:', data.error) - if (onChunk) { - onChunk({ content: '', isFinished: true, finishReason: 'error', error: data.error }) - } - ipcRenderer.removeListener('llm:streamChunk', handler) - resolve({ success: false, error: data.error }) - } else { - if (onChunk) { - onChunk(data.chunk) - } - - if (data.chunk.isFinished) { - console.log('[preload] chatStream 完成，requestId:', requestId) - ipcRenderer.removeListener('llm:streamChunk', handler) - resolve({ success: true }) - } - } - } - } - - ipcRenderer.on('llm:streamChunk', handler) - - ipcRenderer - .invoke('llm:chatStream', requestId, messages, options) - .then((result) => { - console.log('[preload] chatStream invoke 返回:', result) - if (!result.success) { - ipcRenderer.removeListener('llm:streamChunk', handler) - resolve(result) - } - }) - .catch((error) => { - console.error('[preload] chatStream invoke 错误:', error) - ipcRenderer.removeListener('llm:streamChunk', handler) - resolve({ success: false, error: String(error) }) - }) - }) - }, -} - -// ==================== Agent API ==================== - -export const agentApi = { - runStream: ( - userMessage: string, - context: ToolContext, - onChunk?: (chunk: AgentStreamChunk) => void, - chatType?: 'group' | 'private', - locale?: string, - assistantId?: string, - skillId?: string | null, - enableAutoSkill?: boolean, - compressionConfig?: { - enabled: boolean - tokenThresholdPercent: number - bufferSizePercent: number - maxToolResultPercent?: number - }, - thinkingLevel?: string - ): { - requestId: string - promise: Promise<{ success: boolean; result?: AgentResult; error?: SerializedErrorInfo }> - } => { - const sanitizedContext: ToolContext = { - sessionId: context.sessionId, - conversationId: context.conversationId, - historyLeafMessageId: context.historyLeafMessageId, - timeFilter: context.timeFilter - ? { - startTs: context.timeFilter.startTs, - endTs: context.timeFilter.endTs, - } - : undefined, - maxMessagesLimit: context.maxMessagesLimit, - ownerInfo: context.ownerInfo - ? { - platformId: context.ownerInfo.platformId, - displayName: context.ownerInfo.displayName, - } - : undefined, - mentionedMembers: context.mentionedMembers - ? context.mentionedMembers.map((member) => ({ - memberId: member.memberId, - platformId: member.platformId, - displayName: member.displayName, - aliases: [...member.aliases], - mentionText: member.mentionText, - })) - : undefined, - locale: context.locale, - preprocessConfig: context.preprocessConfig, - } - - const requestId = `agent_${Date.now()}_${Math.random().toString(36).slice(2, 8)}` - console.log( - '[preload] Agent runStream 开始，requestId:', - requestId, - 'conversationId:', - sanitizedContext.conversationId ?? 'none', - 'chatType:', - chatType ?? 'group' - ) - - const promise = new Promise<{ success: boolean; result?: AgentResult; error?: SerializedErrorInfo }>((resolve) => { - const chunkHandler = ( - _event: Electron.IpcRendererEvent, - data: { requestId: string; chunk: AgentStreamChunk } - ) => { - if (data.requestId === requestId) { - if (onChunk) { - onChunk(data.chunk) - } - } - } - - const completeHandler = ( - _event: Electron.IpcRendererEvent, - data: { requestId: string; result: AgentResult & { error?: SerializedErrorInfo } } - ) => { - if (data.requestId === requestId) { - console.log('[preload] Agent 完成，requestId:', requestId, 'hasError:', !!data.result?.error) - ipcRenderer.removeListener('agent:streamChunk', chunkHandler) - ipcRenderer.removeListener('agent:complete', completeHandler) - if (data.result?.error) { - resolve({ success: false, error: data.result.error }) - } else { - resolve({ success: true, result: data.result }) - } - } - } - - ipcRenderer.on('agent:streamChunk', chunkHandler) - ipcRenderer.on('agent:complete', completeHandler) - - ipcRenderer - .invoke( - 'agent:runStream', - requestId, - userMessage, - sanitizedContext, - chatType, - locale, - assistantId, - skillId, - enableAutoSkill, - compressionConfig, - thinkingLevel - ) - .then((result) => { - console.log('[preload] Agent invoke 返回:', result) - if (!result.success) { - ipcRenderer.removeListener('agent:streamChunk', chunkHandler) - ipcRenderer.removeListener('agent:complete', completeHandler) - resolve(result) - } - }) - .catch((error) => { - console.error('[preload] Agent invoke 错误:', error) - ipcRenderer.removeListener('agent:streamChunk', chunkHandler) - ipcRenderer.removeListener('agent:complete', completeHandler) - resolve({ - success: false, - error: { - name: error instanceof Error ? error.name : null, - message: error instanceof Error ? error.message : String(error), - stack: error instanceof Error ? (error.stack ?? null) : null, - }, - }) - }) - }) - - return { requestId, promise } - }, - - abort: (requestId: string): Promise<{ success: boolean; error?: string }> => { - console.log('[preload] Agent abort 请求，requestId:', requestId) - return ipcRenderer.invoke('agent:abort', requestId) - }, } diff --git a/apps/desktop/preload/index.d.ts b/apps/desktop/preload/index.d.ts index e211e681..f202e731 100644 --- a/apps/desktop/preload/index.d.ts +++ b/apps/desktop/preload/index.d.ts @@ -254,63 +254,16 @@ interface LLMChatOptions { maxTokens?: number } -interface LLMChatStreamChunk { - content: string - isFinished: boolean - finishReason?: 'stop' | 'length' | 'error' - error?: string - thinking?: string - thinkingDone?: boolean -} - /** * LLM CRUD (config, provider, model, validate, etc.) has been migrated to - * HTTP service layer via FetchLLMAdapter. Only streaming / non-streaming - * chat APIs remain on IPC. + * HTTP service layer via FetchLLMAdapter. Streaming chat uses shared SSE route. + * Only non-streaming chat remains on IPC. */ interface LlmApi { chat: ( messages: LLMChatMessage[], options?: LLMChatOptions ) => Promise<{ success: boolean; content?: string; error?: string }> - chatStream: ( - messages: LLMChatMessage[], - options?: LLMChatOptions, - onChunk?: (chunk: LLMChatStreamChunk) => void - ) => Promise<{ success: boolean; error?: string }> -} - -// TokenUsage & AgentRuntimeStatus — imported from electron/shared/types.ts - -// Agent 相关类型 -interface AgentStreamChunk { - type: 'content' | 'think' | 'tool_start' | 'tool_result' | 'status' | 'compression_done' | 'done' | 'error' - content?: string - thinkTag?: string - thinkDurationMs?: number - toolName?: string - toolParams?: Record - toolResult?: unknown - status?: AgentRuntimeStatus - error?: SerializedErrorInfo - isFinished?: boolean - compressionResult?: { - summaryContent: string - tokensBefore: number - tokensAfter: number - timestamp: number - } - /** Token 使用量（type=done 时返回累计值） */ - usage?: TokenUsage -} - -interface AgentResult { - content: string - toolsUsed: string[] - toolRounds: number - /** 总 Token 使用量（累计所有 LLM 调用） */ - totalUsage?: TokenUsage - error?: SerializedErrorInfo } /** Owner 信息（当前用户在对话中的身份） */ @@ -362,50 +315,7 @@ interface PreprocessConfig { anonymizeNames: boolean } -interface ToolContext { - sessionId: string - conversationId?: string - historyLeafMessageId?: string | null - timeFilter?: { startTs: number; endTs: number } - /** 用户配置：每次发送给 AI 的最大消息条数 */ - maxMessagesLimit?: number - /** Owner 信息（当前用户在对话中的身份） */ - ownerInfo?: OwnerInfo - /** 本轮显式 @ 的成员 */ - mentionedMembers?: Array<{ - memberId: number - platformId: string - displayName: string - aliases: string[] - mentionText: string - }> - /** 语言环境 */ - locale?: string - /** 聊天记录预处理配置 */ - preprocessConfig?: PreprocessConfig -} - -interface AgentApi { - runStream: ( - userMessage: string, - context: ToolContext, - onChunk?: (chunk: AgentStreamChunk) => void, - chatType?: 'group' | 'private', - locale?: string, - assistantId?: string, - skillId?: string | null, - enableAutoSkill?: boolean, - compressionConfig?: { - enabled: boolean - tokenThresholdPercent: number - bufferSizePercent: number - maxToolResultPercent?: number - }, - thinkingLevel?: string - ) => { requestId: string; promise: Promise<{ success: boolean; result?: AgentResult; error?: SerializedErrorInfo }> } - abort: (requestId: string) => Promise<{ success: boolean; error?: string }> -} - +// Agent streaming migrated to shared SSE route (useAgentStreamService) // Assistant CRUD migrated to HTTP service layer (FetchAssistantAdapter) // Skill CRUD migrated to HTTP service layer (FetchSkillAdapter) @@ -615,7 +525,6 @@ declare global { chatApi: ChatApi aiApi: AiApi llmApi: LlmApi - agentApi: AgentApi cacheApi: CacheApi networkApi: NetworkApi sessionApi: SessionApi diff --git a/apps/desktop/preload/index.ts b/apps/desktop/preload/index.ts index bf8bd428..2dd80d05 100644 --- a/apps/desktop/preload/index.ts +++ b/apps/desktop/preload/index.ts @@ -8,7 +8,7 @@ import { electronAPI } from '@electron-toolkit/preload' // 从拆分的模块导入 API import { extendedApi } from './apis/core' import { chatApi } from './apis/chat' -import { aiApi, llmApi, agentApi } from './apis/ai' +import { aiApi } from './apis/ai' import { networkApi, cacheApi, sessionApi } from './apis/utils' import { apiServerApi } from './apis/api-server' import { internalApi } from './apis/internal-api' @@ -37,8 +37,6 @@ if (process.contextIsolated) { contextBridge.exposeInMainWorld('api', extendedApi) contextBridge.exposeInMainWorld('chatApi', chatApi) contextBridge.exposeInMainWorld('aiApi', aiApi) - contextBridge.exposeInMainWorld('llmApi', llmApi) - contextBridge.exposeInMainWorld('agentApi', agentApi) contextBridge.exposeInMainWorld('cacheApi', cacheApi) contextBridge.exposeInMainWorld('networkApi', networkApi) contextBridge.exposeInMainWorld('sessionApi', sessionApi) @@ -57,10 +55,6 @@ if (process.contextIsolated) { // @ts-ignore (define in dts) window.aiApi = aiApi // @ts-ignore (define in dts) - window.llmApi = llmApi - // @ts-ignore (define in dts) - window.agentApi = agentApi - // @ts-ignore (define in dts) window.cacheApi = cacheApi // @ts-ignore (define in dts) window.networkApi = networkApi diff --git a/packages/http-routes/src/context.ts b/packages/http-routes/src/context.ts index 14ba0e15..ec8ed94b 100644 --- a/packages/http-routes/src/context.ts +++ b/packages/http-routes/src/context.ts @@ -18,6 +18,7 @@ import type { CustomProviderStore, CustomModelStore, MergeSessionCache, + AgentStreamChunk, } from '@openchatlab/node-runtime' export interface HttpRouteContext { @@ -55,4 +56,41 @@ export interface HttpRouteContext { downloadsDir?: string defaultUserDataDir?: string isCustomDataDir?: boolean + + /** Agent streaming — platform-specific execution (optional) */ + runAgentStream?: ( + params: AgentStreamRequest, + onEvent: (chunk: AgentStreamChunk) => void, + abortSignal: AbortSignal + ) => Promise +} + +export interface AgentStreamRequest { + userMessage: string + conversationId: string + historyLeafMessageId?: string | null + sessionId: string + chatType?: 'group' | 'private' + locale?: string + assistantId?: string + skillId?: string | null + enableAutoSkill?: boolean + compressionConfig?: { + enabled: boolean + tokenThresholdPercent?: number + bufferSizePercent?: number + maxToolResultPercent?: number + } + ownerInfo?: { platformId: string; displayName: string } + mentionedMembers?: Array<{ + memberId: number + platformId: string + displayName: string + aliases: string[] + mentionText: string + }> + thinkingLevel?: string + timeFilter?: { startTs: number; endTs: number } + maxMessagesLimit?: number + preprocessConfig?: Record } diff --git a/packages/http-routes/src/index.ts b/packages/http-routes/src/index.ts index 6d5e1877..df773158 100644 --- a/packages/http-routes/src/index.ts +++ b/packages/http-routes/src/index.ts @@ -1,4 +1,4 @@ -export type { HttpRouteContext } from './context' +export type { HttpRouteContext, AgentStreamRequest } from './context' export { registerSharedRoutes } from './register' export type { SharedRouteOptions } from './register' export { setAuthToken, setRequireAuth, authHook } from './auth' diff --git a/packages/http-routes/src/register.ts b/packages/http-routes/src/register.ts index c7b57ab4..c1acee40 100644 --- a/packages/http-routes/src/register.ts +++ b/packages/http-routes/src/register.ts @@ -20,6 +20,8 @@ import { registerNlpRoutes } from './routes/web/nlp' import { registerAiAssistantRoutes } from './routes/web/ai-assistants' import { registerAiSkillRoutes } from './routes/web/ai-skills' import { registerAiLlmRoutes } from './routes/web/ai-llm' +import { registerAiLlmStreamRoutes } from './routes/web/ai-llm-stream' +import { registerAiAgentStreamRoutes } from './routes/web/ai-agent-stream' import { registerAiConversationRoutes } from './routes/web/ai-conversations' import { registerAiSummaryRoutes } from './routes/web/ai-summaries' import { registerMergeRoutes } from './routes/web/merge' @@ -66,6 +68,8 @@ export function registerSharedRoutes( registerAiAssistantRoutes(server, ctx) registerAiSkillRoutes(server, ctx) registerAiLlmRoutes(server, ctx) + registerAiLlmStreamRoutes(server, ctx) + registerAiAgentStreamRoutes(server, ctx) registerAiConversationRoutes(server, ctx) registerAiSummaryRoutes(server, ctx) diff --git a/packages/http-routes/src/routes/web/ai-agent-stream.ts b/packages/http-routes/src/routes/web/ai-agent-stream.ts new file mode 100644 index 00000000..c0647c46 --- /dev/null +++ b/packages/http-routes/src/routes/web/ai-agent-stream.ts @@ -0,0 +1,69 @@ +import type { FastifyInstance } from 'fastify' +import type { HttpRouteContext, AgentStreamRequest } from '../../context' + +const activeAgentAborts = new Map() + +export function registerAiAgentStreamRoutes(server: FastifyInstance, ctx: HttpRouteContext): void { + if (!ctx.runAgentStream) return + + const runAgentStream = ctx.runAgentStream + + server.post<{ Body: AgentStreamRequest }>('/_web/ai/agent/stream', async (request, reply) => { + const requestId = `req_${Date.now()}_${Math.random().toString(36).slice(2, 8)}` + const abortController = new AbortController() + activeAgentAborts.set(requestId, abortController) + + reply.raw.writeHead(200, { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + 'X-Request-Id': requestId, + }) + + const sendSSE = (event: string, data: unknown) => { + reply.raw.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`) + } + + sendSSE('meta', { requestId }) + + reply.raw.on('close', () => { + if (!abortController.signal.aborted) { + abortController.abort() + } + activeAgentAborts.delete(requestId) + }) + + try { + await runAgentStream( + request.body, + (chunk) => { + sendSSE(chunk.type, chunk) + if (chunk.type === 'done') { + activeAgentAborts.delete(requestId) + reply.raw.end() + } + }, + abortController.signal + ) + } catch (error) { + const msg = error instanceof Error ? error.message : String(error) + sendSSE('error', { type: 'error', error: { name: 'ServerError', message: msg } }) + sendSSE('done', { type: 'done', isFinished: true }) + activeAgentAborts.delete(requestId) + reply.raw.end() + } + }) + + server.post<{ + Body: { requestId: string } + }>('/_web/ai/agent/abort', async (request) => { + const { requestId } = request.body + const controller = activeAgentAborts.get(requestId) + if (controller) { + controller.abort() + activeAgentAborts.delete(requestId) + return { success: true } + } + return { success: false } + }) +} diff --git a/packages/http-routes/src/routes/web/ai-llm-stream.ts b/packages/http-routes/src/routes/web/ai-llm-stream.ts new file mode 100644 index 00000000..e4b19ffd --- /dev/null +++ b/packages/http-routes/src/routes/web/ai-llm-stream.ts @@ -0,0 +1,50 @@ +import type { FastifyInstance } from 'fastify' +import type { HttpRouteContext } from '../../context' +import { buildPiModel, runSimpleLlmStream } from '@openchatlab/node-runtime' + +export function registerAiLlmStreamRoutes(server: FastifyInstance, ctx: HttpRouteContext): void { + const store = ctx.llmConfigStore + if (!store) return + + 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 llmConfig = store.getDefaultAssistantConfig() + 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() + }) +} diff --git a/src/components/analysis/SQLLab/ResultTable.vue b/src/components/analysis/SQLLab/ResultTable.vue index 4cca75df..9839d7b1 100644 --- a/src/components/analysis/SQLLab/ResultTable.vue +++ b/src/components/analysis/SQLLab/ResultTable.vue @@ -12,6 +12,7 @@ import { useLayoutStore } from '@/stores/layout' import { exportSQLResult, type SQLExportFormat } from '@/utils/sqlExport' import { useLLMService } from '@/services' import { useCacheService } from '@/services/cache/service' +import { useLlmStreamService } from '@/services/ai-stream/service' const { t, locale } = useI18n() const toast = useToast() @@ -339,7 +340,7 @@ ${resultSummary} 2. 关键发现（2-4 个要点） 3. 如有明显的趋势或异常，请指出` - const result = await window.llmApi.chatStream( + const result = await useLlmStreamService().chatStream( [ { role: 'system', diff --git a/src/components/analysis/SQLLabTab.vue b/src/components/analysis/SQLLabTab.vue index 44f11198..1ad3a53b 100644 --- a/src/components/analysis/SQLLabTab.vue +++ b/src/components/analysis/SQLLabTab.vue @@ -4,6 +4,7 @@ import { useI18n } from 'vue-i18n' import { SchemaPanel, AIHistoryModal, ResultTable, getTableLabel, getColumnLabel } from './SQLLab' import type { AIHistory, SQLResult, TableSchema } from './SQLLab' import { useDataService, useLLMService } from '@/services' +import { useLlmStreamService } from '@/services/ai-stream/service' const { t, locale } = useI18n() @@ -237,7 +238,7 @@ async function generateAndRunSQL() { try { const fullPrompt = buildAIPrompt(prompt, schemaList) - const result = await window.llmApi.chatStream( + const result = await useLlmStreamService().chatStream( [ { role: 'system', content: '你是一个 SQLite 专家，请以 JSON 格式输出 sql 和 explanation 字段。' }, { role: 'user', content: fullPrompt }, diff --git a/src/services/ai-stream/service.ts b/src/services/ai-stream/service.ts new file mode 100644 index 00000000..6ed76e42 --- /dev/null +++ b/src/services/ai-stream/service.ts @@ -0,0 +1,156 @@ +import type { LlmStreamChunk, AgentStreamChunk, AgentStreamResult, AgentStreamParams } from './types' +import { fetchSSE } from '../utils/sse' + +/** + * LLM stream service — uses fetchSSE to consume the shared + * `/_web/ai/llm/chat-stream` SSE endpoint. + */ +export function useLlmStreamService() { + return { + async chatStream( + messages: Array<{ role: string; content: string }>, + options?: { temperature?: number; maxTokens?: number }, + onChunk?: (chunk: LlmStreamChunk) => void + ): Promise<{ success: boolean; error?: string }> { + try { + await fetchSSE({ + url: '/_web/ai/llm/chat-stream', + body: { messages, options }, + onEvent: ({ data }) => { + if (!onChunk) return + try { + const parsed = JSON.parse(data) as LlmStreamChunk + onChunk(parsed) + } catch { + // skip malformed JSON + } + }, + }) + return { success: true } + } catch (error) { + const msg = error instanceof Error ? error.message : String(error) + return { success: false, error: msg } + } + }, + } +} + +/** + * Agent stream service — uses fetchSSE to consume the shared + * `/_web/ai/agent/stream` SSE endpoint. + */ + +const requestIdMap = new Map() + +export function useAgentStreamService() { + return { + runStream( + params: AgentStreamParams, + onChunk?: (chunk: AgentStreamChunk) => void + ): { requestId: string; promise: Promise } { + const abortController = new AbortController() + const localRequestId = `local_${Date.now()}_${Math.random().toString(36).slice(2, 8)}` + requestIdMap.set(localRequestId, { serverId: '', abortController }) + + const promise = new Promise((resolve) => { + let resultContent = '' + const toolsUsed: string[] = [] + let toolRounds = 0 + let lastUsage: AgentStreamResult['result'] extends { totalUsage?: infer U } ? U : undefined + let hasError = false + let lastError: AgentStreamResult['error'] + + fetchSSE({ + url: '/_web/ai/agent/stream', + body: params, + signal: abortController.signal, + onEvent: ({ event, data }) => { + try { + const parsed = JSON.parse(data) + + if (event === 'meta') { + const entry = requestIdMap.get(localRequestId) + if (entry) entry.serverId = parsed.requestId ?? '' + return + } + + const chunk = parsed as AgentStreamChunk + onChunk?.(chunk) + + switch (chunk.type) { + case 'content': + if (chunk.content) resultContent += chunk.content + break + case 'tool_start': + if (chunk.toolName && !toolsUsed.includes(chunk.toolName)) { + toolsUsed.push(chunk.toolName) + } + break + case 'status': + if (chunk.status) toolRounds = chunk.status.round + break + case 'done': + if (chunk.usage) lastUsage = chunk.usage + break + case 'error': + hasError = true + lastError = chunk.error as AgentStreamResult['error'] + break + } + } catch { + // skip malformed JSON + } + }, + }) + .then(() => { + requestIdMap.delete(localRequestId) + resolve({ + success: !hasError, + result: { + content: resultContent, + toolsUsed, + toolRounds, + totalUsage: lastUsage, + }, + ...(hasError && lastError ? { error: lastError } : {}), + }) + }) + .catch((error) => { + requestIdMap.delete(localRequestId) + if (abortController.signal.aborted) { + resolve({ + success: false, + result: { content: resultContent, toolsUsed, toolRounds, aborted: true }, + }) + } else { + resolve({ + success: false, + error: { name: 'FetchError', message: error.message ?? String(error), stack: null }, + }) + } + }) + }) + + return { requestId: localRequestId, promise } + }, + + async abort(requestId: string): Promise<{ success: boolean; error?: string }> { + const entry = requestIdMap.get(requestId) + if (!entry) return { success: false, error: 'Request not found' } + + entry.abortController.abort() + + if (entry.serverId) { + try { + const { post } = await import('../utils/http') + await post<{ success: boolean }>('/ai/agent/abort', { requestId: entry.serverId }) + } catch { + // server-side abort is best-effort; client already aborted the fetch + } + } + + requestIdMap.delete(requestId) + return { success: true } + }, + } +} diff --git a/src/services/ai-stream/types.ts b/src/services/ai-stream/types.ts new file mode 100644 index 00000000..ed147a17 --- /dev/null +++ b/src/services/ai-stream/types.ts @@ -0,0 +1,96 @@ +export interface LlmStreamChunk { + content: string + isFinished: boolean + finishReason?: 'stop' | 'length' | 'error' + error?: string + thinking?: string + thinkingDone?: boolean +} + +export interface AgentStreamChunk { + type: 'content' | 'think' | 'tool_start' | 'tool_result' | 'status' | 'compression_done' | 'done' | 'error' + content?: string + thinkTag?: string + thinkDurationMs?: number + toolName?: string + toolParams?: Record + toolResult?: unknown + error?: unknown + isFinished?: boolean + usage?: TokenUsage + status?: AgentRuntimeStatus + compressionResult?: { + summaryContent: string + tokensBefore: number + tokensAfter: number + timestamp: number + } +} + +export interface TokenUsage { + promptTokens: number + completionTokens: number + totalTokens: number +} + +export interface AgentRuntimeStatus { + phase: 'compressing' | 'preparing' | 'thinking' | 'tool_running' | 'responding' | 'completed' | 'aborted' | 'error' + round: number + toolsUsed: number + currentTool?: string + contextTokens: number + totalUsage: TokenUsage + updatedAt: number +} + +export interface SerializedErrorInfo { + name: string | null + message: string | null + stack?: string | null + friendlyMessage?: string | null + url?: string | null +} + +export interface AgentStreamResult { + success: boolean + result?: { + content: string + toolsUsed: string[] + toolRounds: number + totalUsage?: TokenUsage + aborted?: boolean + } + error?: SerializedErrorInfo | Record +} + +export interface CompressionConfig { + enabled: boolean + tokenThresholdPercent: number + bufferSizePercent: number + maxToolResultPercent?: number +} + +export interface AgentStreamParams { + userMessage: string + sessionId: string + conversationId?: string + historyLeafMessageId?: string | null + chatType?: 'group' | 'private' + locale?: string + assistantId?: string + skillId?: string | null + enableAutoSkill?: boolean + compressionConfig?: CompressionConfig + ownerInfo?: { platformId: string; displayName: string } + mentionedMembers?: Array<{ + memberId: number + platformId: string + displayName: string + aliases: string[] + mentionText: string + }> + thinkingLevel?: string + timeFilter?: { startTs: number; endTs: number } + maxMessagesLimit?: number + preprocessConfig?: Record +} diff --git a/src/services/registry.ts b/src/services/registry.ts index 71d512ff..3fecb878 100644 --- a/src/services/registry.ts +++ b/src/services/registry.ts @@ -135,186 +135,15 @@ async function initWebServeAdapters(): Promise { const { FetchCacheAdapter } = await import('./cache/fetch') registerAdapter('cache', new FetchCacheAdapter()) - await installAiApiShims() + await installWebServeShims() } /** - * Install remaining window shims for SSE streaming APIs that the - * service-layer adapters do not (yet) cover. + * Install remaining window shims for web-serve mode. + * AI streaming shims have been removed — the service layer now + * uses fetchSSE directly via useAgentStreamService/useLlmStreamService. */ -async function installAiApiShims(): Promise { - ;(window as any).llmApi = { - chatStream: async ( - messages: Array<{ role: string; content: string }>, - options?: { temperature?: number; maxTokens?: number }, - onChunk?: (chunk: { - content: string - isFinished: boolean - finishReason?: string - error?: string - thinking?: string - thinkingDone?: boolean - }) => void - ): Promise<{ success: boolean; error?: string }> => { - try { - const { fetchSSE: fetchSSEStream } = await import('./utils/sse') - let streamError = '' - await fetchSSEStream({ - url: '/_web/ai/llm/chat-stream', - method: 'POST', - body: { messages, options }, - onEvent: ({ data }) => { - try { - const chunk = JSON.parse(data) - if (chunk.error) streamError = chunk.error - if (onChunk) onChunk(chunk) - } catch { - // skip malformed JSON - } - }, - }) - return streamError ? { success: false, error: streamError } : { success: true } - } catch (error) { - return { success: false, error: error instanceof Error ? error.message : String(error) } - } - }, - } - const { fetchSSE } = await import('./utils/sse') - const { post: httpPost } = await import('./utils/http') - - const agentApiImpl = { - runStream: ( - userMessage: string, - context: Record, - onChunk?: (chunk: Record) => void, - chatType?: string, - locale?: string, - assistantId?: string, - _skillId?: string | null, - _enableAutoSkill?: boolean, - _compressionConfig?: Record, - thinkingLevel?: string - ): { - requestId: string - promise: Promise<{ success: boolean; result?: Record; error?: Record }> - } => { - const localRequestId = `agent_${Date.now()}_${Math.random().toString(36).slice(2, 8)}` - let serverRequestId = localRequestId - const abortController = new AbortController() - - const promise = new Promise<{ - success: boolean - result?: Record - error?: Record - }>((resolve) => { - let content = '' - const toolsUsed: string[] = [] - let toolRounds = 0 - let lastUsage: Record | undefined - let hasError = false - let streamError: Record | undefined - - fetchSSE({ - url: '/_web/ai/agent/stream', - method: 'POST', - body: { - userMessage, - conversationId: context.conversationId, - historyLeafMessageId: context.historyLeafMessageId, - sessionId: context.sessionId, - chatType: chatType || 'group', - locale: locale || 'zh-CN', - assistantId, - compressionConfig: _compressionConfig, - ownerInfo: context.ownerInfo, - mentionedMembers: context.mentionedMembers, - thinkingLevel, - }, - signal: abortController.signal, - onEvent: ({ event, data }) => { - try { - const parsed = JSON.parse(data) - - if (event === 'meta') { - serverRequestId = parsed.requestId || localRequestId - return - } - - if (onChunk) onChunk(parsed) - - switch (event) { - case 'content': - if (parsed.content) content += parsed.content - break - case 'tool_start': - if (parsed.toolName) toolsUsed.push(parsed.toolName) - break - case 'tool_result': - break - case 'status': - if (parsed.status?.round !== undefined) toolRounds = parsed.status.round - break - case 'error': - hasError = true - streamError = parsed.error - break - case 'done': - if (parsed.usage) lastUsage = parsed.usage - break - } - } catch { - // skip malformed - } - }, - }) - .then(() => { - if (hasError) { - resolve({ success: false, error: streamError }) - } else { - resolve({ - success: true, - result: { content, toolsUsed, toolRounds, totalUsage: lastUsage }, - }) - } - }) - .catch((err) => { - if (abortController.signal.aborted) { - resolve({ success: false, error: { name: 'AbortError', message: 'Aborted' } }) - } else { - resolve({ success: false, error: { name: 'NetworkError', message: err?.message || String(err) } }) - } - }) - }) - - // Store abort controller for external abort - ;(agentApiImpl as any)._abortControllers = (agentApiImpl as any)._abortControllers || new Map() - ;(agentApiImpl as any)._abortControllers.set(localRequestId, { - abortController, - serverRequestId: () => serverRequestId, - }) - - return { requestId: localRequestId, promise } - }, - - abort: async (requestId: string) => { - const controllers = (agentApiImpl as any)._abortControllers as - | Map string }> - | undefined - const entry = controllers?.get(requestId) - if (entry) { - entry.abortController.abort() - controllers!.delete(requestId) - try { - await httpPost('/ai/agent/abort', { requestId: entry.serverRequestId() }) - } catch { - // best-effort - } - } - }, - } - - ;(window as any).agentApi = agentApiImpl - +async function installWebServeShims(): Promise { installMergeShims('web-serve') } diff --git a/src/stores/aiChat.ts b/src/stores/aiChat.ts index 9abd5b07..aa8ffda9 100644 --- a/src/stores/aiChat.ts +++ b/src/stores/aiChat.ts @@ -16,6 +16,7 @@ import { useAssistantStore } from '@/stores/assistant' import { useSkillStore } from '@/stores/skill' import { useLLMStore } from '@/stores/llm' import type { TokenUsage, AgentRuntimeStatus, SerializedErrorInfo } from '@electron/shared/types' +import { useAgentStreamService } from '@/services/ai-stream/service' // 工具调用记录 export interface ToolCallRecord { @@ -792,9 +793,33 @@ export const useAIChatStore = defineStore('aiChatRuntime', () => { searchContextAfter: aiGlobalSettings.value.searchContextAfter, } - const { requestId: agentReqId, promise: agentPromise } = window.agentApi.runStream( - content, - context, + const { requestId: agentReqId, promise: agentPromise } = useAgentStreamService().runStream( + { + userMessage: content, + sessionId: state.sessionId, + conversationId: resolvedConversationId, + timeFilter: context.timeFilter, + maxMessagesLimit: context.maxMessagesLimit, + ownerInfo: context.ownerInfo, + mentionedMembers: context.mentionedMembers, + preprocessConfig: context.preprocessConfig, + chatType: state.chatType, + locale: state.locale, + assistantId: currentAssistantId, + skillId: currentSkillId, + enableAutoSkill: !currentSkillId ? autoSkillEnabled : undefined, + compressionConfig: { + enabled: aiGlobalSettings.value.contextCompression?.enabled ?? false, + tokenThresholdPercent: aiGlobalSettings.value.contextCompression?.tokenThresholdPercent ?? 75, + bufferSizePercent: aiGlobalSettings.value.contextCompression?.bufferSizePercent ?? 20, + maxToolResultPercent: aiGlobalSettings.value.contextCompression?.maxToolResultPercent ?? 50, + }, + thinkingLevel: (() => { + const cfg = llmStore.defaultAssistant + if (!cfg?.configId || !cfg?.modelId) return undefined + return promptStore.getThinkingLevel(cfg.configId, cfg.modelId) + })(), + }, (chunk) => { if (state.isAborted || thisRequestId !== state.currentRequestId) { return @@ -895,23 +920,7 @@ export const useAIChatStore = defineStore('aiChatRuntime', () => { setAgentPhase(state, 'error') break } - }, - state.chatType, - state.locale, - currentAssistantId, - currentSkillId, - !currentSkillId ? autoSkillEnabled : undefined, - { - enabled: aiGlobalSettings.value.contextCompression?.enabled ?? false, - tokenThresholdPercent: aiGlobalSettings.value.contextCompression?.tokenThresholdPercent ?? 75, - bufferSizePercent: aiGlobalSettings.value.contextCompression?.bufferSizePercent ?? 20, - maxToolResultPercent: aiGlobalSettings.value.contextCompression?.maxToolResultPercent ?? 50, - }, - (() => { - const cfg = llmStore.defaultAssistant - if (!cfg?.configId || !cfg?.modelId) return undefined - return promptStore.getThinkingLevel(cfg.configId, cfg.modelId) - })() + } ) state.currentAgentRequestId = agentReqId @@ -1333,9 +1342,34 @@ export const useAIChatStore = defineStore('aiChatRuntime', () => { searchContextAfter: aiGlobalSettings.value.searchContextAfter, } - const { requestId: agentReqId, promise: agentPromise } = window.agentApi.runStream( - content, - context, + const { requestId: agentReqId, promise: agentPromise } = useAgentStreamService().runStream( + { + userMessage: content, + sessionId: state.sessionId, + conversationId: state.currentConversationId!, + historyLeafMessageId: originalMessage.parentId ?? null, + timeFilter: context.timeFilter, + maxMessagesLimit: context.maxMessagesLimit, + ownerInfo: context.ownerInfo, + mentionedMembers: context.mentionedMembers, + preprocessConfig: context.preprocessConfig, + chatType: state.chatType, + locale: state.locale, + assistantId: currentAssistantId, + skillId: currentSkillId, + enableAutoSkill: !currentSkillId ? (aiGlobalSettings.value.enableAutoSkill ?? true) : undefined, + compressionConfig: { + enabled: aiGlobalSettings.value.contextCompression?.enabled ?? false, + tokenThresholdPercent: aiGlobalSettings.value.contextCompression?.tokenThresholdPercent ?? 75, + bufferSizePercent: aiGlobalSettings.value.contextCompression?.bufferSizePercent ?? 20, + maxToolResultPercent: aiGlobalSettings.value.contextCompression?.maxToolResultPercent ?? 50, + }, + thinkingLevel: (() => { + const cfg = llmStore.defaultAssistant + if (!cfg?.configId || !cfg?.modelId) return undefined + return promptStore.getThinkingLevel(cfg.configId, cfg.modelId) + })(), + }, (chunk) => { if (state.isAborted || thisRequestId !== state.currentRequestId) return switch (chunk.type) { @@ -1380,23 +1414,7 @@ export const useAIChatStore = defineStore('aiChatRuntime', () => { break } } - }, - state.chatType, - state.locale, - currentAssistantId, - currentSkillId, - !currentSkillId ? (aiGlobalSettings.value.enableAutoSkill ?? true) : undefined, - { - enabled: aiGlobalSettings.value.contextCompression?.enabled ?? false, - tokenThresholdPercent: aiGlobalSettings.value.contextCompression?.tokenThresholdPercent ?? 75, - bufferSizePercent: aiGlobalSettings.value.contextCompression?.bufferSizePercent ?? 20, - maxToolResultPercent: aiGlobalSettings.value.contextCompression?.maxToolResultPercent ?? 50, - }, - (() => { - const cfg = llmStore.defaultAssistant - if (!cfg?.configId || !cfg?.modelId) return undefined - return promptStore.getThinkingLevel(cfg.configId, cfg.modelId) - })() + } ) state.currentAgentRequestId = agentReqId @@ -1521,7 +1539,7 @@ export const useAIChatStore = defineStore('aiChatRuntime', () => { if (state.currentAgentRequestId) { try { - await window.agentApi.abort(state.currentAgentRequestId) + await useAgentStreamService().abort(state.currentAgentRequestId) } catch (error) { console.error('[AI] 中止 Agent 请求失败:', error) }