diff --git a/electron/harness/adapters/agnes-ai.adapter.ts b/electron/harness/adapters/agnes-ai.adapter.ts index 98fcc81..057463a 100644 --- a/electron/harness/adapters/agnes-ai.adapter.ts +++ b/electron/harness/adapters/agnes-ai.adapter.ts @@ -1,48 +1,136 @@ /** * Agnes AI Provider Adapter * - * OpenAI 兼容 API，免费使用。 - * 支持 Tool Calling、Thinking 模式、512K 上下文、多模态（图片 base64）。 + * OpenAI 兼容 API。支持 Tool Calling、Thinking 模式、多模态（图片）。 * - * 差异于 DeepSeek： + * 独立继承 BaseAdapter，通过 shared/openai-format 和 shared/sse-stream 复用 + * OpenAI 兼容格式构建和 SSE 流式解析逻辑。不与其他 Provider Adapter 耦合。 + * + * 与 DeepSeek 的差异： * - Thinking 模式使用 chat_template_kwargs（非 thinking 字段） - * - 图片使用 base64 格式（与 Ollama 一致，通过 images 字段传递） - * - 512K 上下文，65.5K 最大输出 + * - 默认 max_tokens 更大（65536 vs 8192） * * @see apis/agnes-ai-api-docs-20260625.html */ -import { DeepSeekAdapter } from './deepseek.adapter'; -import type { MetonaRequest } from '../types'; +import { BaseAdapter } from './base-adapter'; +import type { MetonaRequest, MetonaResponse, MetonaStreamEvent } from '../types'; +import { MetonaFinishReason } from '../types'; +import { buildOpenAICompatibleMessages, buildOpenAICompatibleTools } from './shared/openai-format'; +import { parseSSEStream, parseOpenAICompatibleResponse } from './shared/sse-stream'; -export class AgnesAdapter extends DeepSeekAdapter { +export class AgnesAdapter extends BaseAdapter { override readonly provider: string = 'agnes'; readonly supportedModels = ['agnes-2.0-flash']; readonly supportsToolCalling = true; readonly supportsThinking = true; + // ===== POST /chat/completions (非流式) ===== + + async chat(request: MetonaRequest): Promise { + const body = this.toNativeRequest(request, false); + + const response = await fetch(`${this.config.baseURL}/chat/completions`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + Authorization: `Bearer ${this.config.apiKey}`, + ...this.config.headers, + }, + body: JSON.stringify(body), + signal: AbortSignal.timeout(this.config.timeoutMs ?? 300_000), + }); + + if (!response.ok) { + const errorBody = await response.text().catch(() => ''); + throw new Error(`Agnes AI API error: ${response.status} ${response.statusText} - ${errorBody}`); + } + + const data = await response.json() as Record; + const parsed = parseOpenAICompatibleResponse(data, request.meta.requestId, this.provider, this.config.defaultModel); + + return { + meta: { + requestId: request.meta.requestId, + provider: this.provider, + model: (data.model as string) ?? this.config.defaultModel, + latencyMs: 0, + timestamp: Date.now(), + }, + content: parsed.content, + reasoningContent: parsed.reasoningContent, + toolCalls: parsed.toolCalls, + usage: parsed.usage, + finishReason: parsed.finishReason as MetonaFinishReason, + }; + } + + // ===== POST /chat/completions (流式) ===== + + async *chatStream(request: MetonaRequest): AsyncIterable { + const body = this.toNativeRequest(request, true); + + const response = await fetch(`${this.config.baseURL}/chat/completions`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + Authorization: `Bearer ${this.config.apiKey}`, + ...this.config.headers, + }, + body: JSON.stringify(body), + }); + + if (!response.ok || !response.body) { + throw new Error(`Agnes AI stream error: ${response.status}`); + } + + yield* parseSSEStream( + response.body, + request.meta.requestId, + request.meta.sessionId, + request.meta.iteration, + ); + } + + // ========== 私有方法 ========== + /** - * 覆盖原生请求构建 + * 构建 Agnes AI 原生请求体 * - * 差异： - * 1. Thinking 模式使用 chat_template_kwargs - * 2. 图片处理由基类 DeepSeekAdapter.toNativeRequest 完成（OpenAI 多模态格式） + * Agnes AI 特有参数： + * - chat_template_kwargs: { enable_thinking: true } — 启用思考模式（非 thinking 字段） + * - 默认 max_tokens: 65536（512K 上下文） */ - protected override toNativeRequest(request: MetonaRequest): Record { - const base = super.toNativeRequest(request) as Record; + private toNativeRequest(request: MetonaRequest, stream: boolean): Record { + const messages = buildOpenAICompatibleMessages(request); + const tools = buildOpenAICompatibleTools(request.tools); - // Agnes Thinking 模式：使用 chat_template_kwargs - if (request.params?.thinkingEnabled) { - delete base.thinking; - delete base.reasoning_effort; - base.chat_template_kwargs = { enable_thinking: true }; + const body: Record = { + model: this.config.defaultModel, + messages, + temperature: request.params.temperature ?? 0, + max_tokens: request.params.maxTokens ?? 65536, + stream, + }; + + if (stream) { + body.stream_options = { include_usage: true }; } - // Agnes 默认 max_tokens 更大 - if (!request.params.maxTokens) { - base.max_tokens = 65536; + if (tools) { + body.tools = tools; } - return base; + // Thinking 模式：Agnes 使用 chat_template_kwargs 而非 thinking + if (request.params.thinkingEnabled) { + body.chat_template_kwargs = { enable_thinking: true }; + } + + // 停止序列 + if (request.params.stopSequences?.length) { + body.stop = request.params.stopSequences; + } + + return body; } } diff --git a/electron/harness/adapters/deepseek.adapter.ts b/electron/harness/adapters/deepseek.adapter.ts index fe010b8..2957458 100644 --- a/electron/harness/adapters/deepseek.adapter.ts +++ b/electron/harness/adapters/deepseek.adapter.ts @@ -1,26 +1,19 @@ /** * DeepSeek Provider Adapter * - * 基于 OpenAI 兼容 API 的 DeepSeek 适配器。 - * 支持 Tool Calling、Thinking 模式、流式输出、JSON 结构化输出。 + * 基于 OpenAI 兼容 API。支持 Tool Calling、Thinking 模式、流式输出。 * - * 完整实现 DeepSeek API 文档中所有参数： - * - model, messages, thinking, reasoning_effort - * - stream, stream_options - * - tools, tool_choice - * - response_format (JSON) - * - stop, max_tokens, temperature, top_p - * - logprobs, top_logprobs - * - prompt_cache_hit/miss_tokens, reasoning_tokens - * - GET /models, GET /user/balance + * 独立继承 BaseAdapter，通过 shared/openai-format 和 shared/sse-stream 复用 + * OpenAI 兼容格式构建和 SSE 流式解析逻辑。不与其他 Provider Adapter 耦合。 * * @see apis/deepseek-api-docs-20260518.html */ import { BaseAdapter } from './base-adapter'; -import { nanoid } from 'nanoid'; import type { MetonaRequest, MetonaResponse, MetonaStreamEvent } from '../types'; -import { MetonaFinishReason, MetonaStreamEventType, MetonaErrorCode } from '../types'; +import { MetonaFinishReason, MetonaErrorCode } from '../types'; +import { buildOpenAICompatibleMessages, buildOpenAICompatibleTools } from './shared/openai-format'; +import { parseSSEStream, parseOpenAICompatibleResponse } from './shared/sse-stream'; export class DeepSeekAdapter extends BaseAdapter { override readonly provider: string = 'deepseek'; @@ -28,8 +21,10 @@ export class DeepSeekAdapter extends BaseAdapter { readonly supportsToolCalling = true; readonly supportsThinking = true; + // ===== POST /chat/completions (非流式) ===== + async chat(request: MetonaRequest): Promise { - const nativeRequest = this.toNativeRequest(request); + const body = this.toNativeRequest(request, false); const response = await fetch(`${this.config.baseURL}/chat/completions`, { method: 'POST', @@ -38,7 +33,7 @@ export class DeepSeekAdapter extends BaseAdapter { Authorization: `Bearer ${this.config.apiKey}`, ...this.config.headers, }, - body: JSON.stringify(nativeRequest), + body: JSON.stringify(body), signal: AbortSignal.timeout(this.config.timeoutMs ?? 120_000), }); @@ -47,12 +42,29 @@ export class DeepSeekAdapter extends BaseAdapter { throw new Error(`DeepSeek API error: ${response.status} ${response.statusText} - ${errorBody}`); } - const data = await response.json(); - return this.toMetonaResponse(data, request.meta.requestId); + const data = await response.json() as Record; + const parsed = parseOpenAICompatibleResponse(data, request.meta.requestId, this.provider, this.config.defaultModel); + + return { + meta: { + requestId: request.meta.requestId, + provider: this.provider, + model: (data.model as string) ?? this.config.defaultModel, + latencyMs: 0, + timestamp: Date.now(), + }, + content: parsed.content, + reasoningContent: parsed.reasoningContent, + toolCalls: parsed.toolCalls, + usage: parsed.usage, + finishReason: parsed.finishReason as MetonaFinishReason, + }; } + // ===== POST /chat/completions (流式) ===== + async *chatStream(request: MetonaRequest): AsyncIterable { - const nativeRequest = { ...this.toNativeRequest(request), stream: true, stream_options: { include_usage: true } }; + const body = this.toNativeRequest(request, true); const response = await fetch(`${this.config.baseURL}/chat/completions`, { method: 'POST', @@ -61,155 +73,23 @@ export class DeepSeekAdapter extends BaseAdapter { Authorization: `Bearer ${this.config.apiKey}`, ...this.config.headers, }, - body: JSON.stringify(nativeRequest), + body: JSON.stringify(body), }); if (!response.ok || !response.body) { throw new Error(`DeepSeek stream error: ${response.status}`); } - const reader = response.body.getReader(); - const decoder = new TextDecoder(); - let seq = 0; - let buffer = ''; - - // 工具调用缓冲区：index → { name, argsBuffer } - const toolCallsBuffer = new Map(); - - while (true) { - const { done, value } = await reader.read(); - if (done) break; - - buffer += decoder.decode(value, { stream: true }); - const lines = buffer.split('\n'); - buffer = lines.pop() ?? ''; - - for (const line of lines) { - const trimmed = line.trim(); - if (!trimmed || !trimmed.startsWith('data: ')) continue; - const data = trimmed.slice(6); - if (data === '[DONE]') { - // 流结束前，将缓冲区中的工具调用转为 TOOL_CALL_COMPLETE - for (const [index, buf] of toolCallsBuffer) { - try { - yield { - type: MetonaStreamEventType.TOOL_CALL_COMPLETE, - requestId: request.meta.requestId, - sessionId: request.meta.sessionId, - iteration: request.meta.iteration, - seq: seq++, - timestamp: Date.now(), - toolCall: { - id: `tc_${nanoid(8)}`, - name: buf.name, - args: buf.argsBuffer ? JSON.parse(buf.argsBuffer) : {}, - iteration: request.meta.iteration, - timestamp: Date.now(), - }, - }; - } catch { - // JSON 解析失败，跳过 - } - } - toolCallsBuffer.clear(); - - yield { - type: MetonaStreamEventType.DONE, - requestId: request.meta.requestId, - sessionId: request.meta.sessionId, - iteration: request.meta.iteration, - seq: seq++, - timestamp: Date.now(), - }; - return; - } - - try { - const chunk = JSON.parse(data); - const delta = chunk.choices?.[0]?.delta; - - // 文本内容增量 - if (delta?.content) { - yield { - type: MetonaStreamEventType.TEXT_DELTA, - requestId: request.meta.requestId, - sessionId: request.meta.sessionId, - iteration: request.meta.iteration, - seq: seq++, - timestamp: Date.now(), - delta: delta.content, - }; - } - - // 推理内容增量（Thinking 模式） - if (delta?.reasoning_content) { - yield { - type: MetonaStreamEventType.REASONING_DELTA, - requestId: request.meta.requestId, - sessionId: request.meta.sessionId, - iteration: request.meta.iteration, - seq: seq++, - timestamp: Date.now(), - delta: delta.reasoning_content, - }; - } - - // 工具调用增量 — 缓冲拼接 - if (delta?.tool_calls) { - for (const tc of delta.tool_calls) { - const idx = tc.index ?? 0; - if (!toolCallsBuffer.has(idx)) { - toolCallsBuffer.set(idx, { name: tc.function?.name ?? '', argsBuffer: '' }); - } - const buf = toolCallsBuffer.get(idx)!; - if (tc.function?.name) buf.name = tc.function.name; - if (tc.function?.arguments) buf.argsBuffer += tc.function.arguments; - - yield { - type: MetonaStreamEventType.TOOL_CALL_DELTA, - requestId: request.meta.requestId, - sessionId: request.meta.sessionId, - iteration: request.meta.iteration, - seq: seq++, - timestamp: Date.now(), - toolCallDelta: { - index: idx, - name: tc.function?.name, - argsDelta: tc.function?.arguments, - }, - }; - } - } - - // 流结束时的 usage 信息 - if (chunk.usage) { - yield { - type: MetonaStreamEventType.USAGE, - requestId: request.meta.requestId, - sessionId: request.meta.sessionId, - iteration: request.meta.iteration, - seq: seq++, - timestamp: Date.now(), - usage: { - inputTokens: chunk.usage.prompt_tokens ?? 0, - outputTokens: chunk.usage.completion_tokens ?? 0, - totalTokens: chunk.usage.total_tokens ?? 0, - reasoningTokens: chunk.usage.completion_tokens_details?.reasoning_tokens, - cacheHitTokens: chunk.usage.prompt_cache_hit_tokens, - cacheMissTokens: chunk.usage.prompt_cache_miss_tokens, - }, - }; - } - } catch { - // 跳过解析失败的行 - } - } - } + yield* parseSSEStream( + response.body, + request.meta.requestId, + request.meta.sessionId, + request.meta.iteration, + ); } - /** - * 列出可用模型 - */ + // ===== GET /models ===== + async listModels(): Promise { try { const response = await fetch(`${this.config.baseURL}/models`, { @@ -224,17 +104,24 @@ export class DeepSeekAdapter extends BaseAdapter { } } - /** - * 查询账户余额 - */ - async getBalance(): Promise<{ currency: string; totalBalance: string; grantedBalance: string; toppedUpBalance: string } | null> { + // ===== GET /user/balance ===== + + async getBalance(): Promise<{ + currency: string; + totalBalance: string; + grantedBalance: string; + toppedUpBalance: string; + } | null> { try { const response = await fetch(`${this.config.baseURL}/user/balance`, { headers: { Authorization: `Bearer ${this.config.apiKey}` }, signal: AbortSignal.timeout(10_000), }); if (!response.ok) return null; - const data = await response.json() as { currency?: string; total_balance?: string; granted_balance?: string; topped_up_balance?: string }; + const data = await response.json() as { + currency?: string; total_balance?: string; + granted_balance?: string; topped_up_balance?: string; + }; return { currency: data.currency ?? 'CNY', totalBalance: data.total_balance ?? '0', @@ -246,89 +133,42 @@ export class DeepSeekAdapter extends BaseAdapter { } } - // ========== 私有转换方法 ========== + // ========== 私有方法 ========== /** - * 将 MetonaRequest 转换为 OpenAI 兼容的原生请求格式 + * 构建 DeepSeek 原生请求体 + * + * DeepSeek 特有参数： + * - thinking: { type: "enabled" } — 启用思考模式 + * - reasoning_effort — 思考强度映射 + * - stream_options: { include_usage: true } — 流式返回 usage */ - protected toNativeRequest(request: MetonaRequest): Record { - const messages = [ - { - role: 'system', - content: [ - request.systemPrompt.roleDefinition, - request.systemPrompt.outputConstraints, - request.systemPrompt.safetyGuidelines, - request.systemPrompt.dynamicReminders, - ].filter(Boolean).join('\n\n'), - }, - ...request.messages.filter((m) => m.role !== 'system').map((m) => { - const msg: Record = { role: m.role }; - - // 多模态图片处理：将 images 转为 OpenAI content 数组格式 - // [{"type":"text","text":"..."}, {"type":"image_url","image_url":{"url":"data:..."}}] - if (m.images && m.images.length > 0) { - const contentParts: Array> = []; - if (m.content) { - contentParts.push({ type: 'text', text: m.content }); - } - for (const img of m.images) { - contentParts.push({ - type: 'image_url', - image_url: { url: img.url, detail: img.detail ?? 'auto' }, - }); - } - msg.content = contentParts; - } else { - msg.content = m.content; - } - - if (m.role === 'assistant' && m.toolCalls?.length) { - msg.tool_calls = m.toolCalls.map((tc) => ({ - id: tc.id, - type: 'function', - function: { name: tc.name, arguments: JSON.stringify(tc.args) }, - })); - // 工具调用轮次的 reasoning_content 必须携带回上下文 - if (m.reasoningContent) { - msg.reasoning_content = m.reasoningContent; - } - } - if (m.role === 'tool' && m.toolResult) { - msg.tool_call_id = m.toolResult.toolCallId; - msg.content = typeof m.toolResult.result === 'string' - ? m.toolResult.result - : JSON.stringify(m.toolResult.result); - } - return msg; - }), - ]; + private toNativeRequest(request: MetonaRequest, stream: boolean): Record { + const messages = buildOpenAICompatibleMessages(request); + const tools = buildOpenAICompatibleTools(request.tools); const body: Record = { model: this.config.defaultModel, messages, temperature: request.params.temperature ?? 0, max_tokens: request.params.maxTokens, - stream: request.params.stream ?? false, + stream, }; - // Tool Calling - if (request.tools?.length) { - body.tools = request.tools.map((t) => ({ - type: 'function', - function: { - name: t.name, - description: t.description, - parameters: t.parameters, - }, - })); + if (stream) { + body.stream_options = { include_usage: true }; + } + + if (tools) { + body.tools = tools; } // Thinking 模式 if (request.params.thinkingEnabled) { body.thinking = { type: 'enabled' }; - // reasoning_effort 映射 - const effortMap: Record = { low: 'high', medium: 'high', high: 'high', max: 'max' }; + const effortMap: Record = { + low: 'high', medium: 'high', high: 'high', max: 'max', + }; body.reasoning_effort = effortMap[request.params.thinkingEffort ?? 'high'] ?? 'high'; } @@ -337,59 +177,6 @@ export class DeepSeekAdapter extends BaseAdapter { body.stop = request.params.stopSequences; } - // 安全约束 - if (request.constraints?.allowedTools?.length) { - body.tool_choice = 'auto'; - } - return body; } - - private toMetonaResponse(data: Record, requestId: string): MetonaResponse { - const choice = (data.choices as Array>)?.[0]; - const message = choice?.message as Record | undefined; - const usage = data.usage as Record | undefined; - const toolCalls = message?.tool_calls as Array> | undefined; - - return { - meta: { - requestId, - provider: this.provider, - model: (data.model as string) ?? this.config.defaultModel, - latencyMs: 0, - timestamp: Date.now(), - }, - content: (message?.content as string) ?? '', - reasoningContent: message?.reasoning_content as string | undefined, - toolCalls: toolCalls?.map((tc) => { - const fn = tc.function as Record; - return { - id: tc.id as string, - name: fn.name as string, - args: JSON.parse(fn.arguments as string), - iteration: 0, - timestamp: Date.now(), - }; - }), - usage: { - inputTokens: (usage?.prompt_tokens as number) ?? 0, - outputTokens: (usage?.completion_tokens as number) ?? 0, - totalTokens: (usage?.total_tokens as number) ?? 0, - reasoningTokens: (usage?.completion_tokens_details as Record)?.reasoning_tokens as number | undefined, - cacheHitTokens: usage?.prompt_cache_hit_tokens as number | undefined, - cacheMissTokens: usage?.prompt_cache_miss_tokens as number | undefined, - }, - finishReason: this.mapFinishReason(choice?.finish_reason as string), - }; - } - - private mapFinishReason(reason: string): MetonaFinishReason { - switch (reason) { - case 'stop': return MetonaFinishReason.STOP; - case 'length': return MetonaFinishReason.LENGTH; - case 'tool_calls': return MetonaFinishReason.TOOL_CALLS; - case 'content_filter': return MetonaFinishReason.CONTENT_FILTER; - default: return MetonaFinishReason.STOP; - } - } } diff --git a/electron/harness/adapters/index.ts b/electron/harness/adapters/index.ts index 5da5e69..213dee4 100644 --- a/electron/harness/adapters/index.ts +++ b/electron/harness/adapters/index.ts @@ -1,3 +1,16 @@ +/** + * Provider Adapter 导出 + * + * 三种 Provider 各自独立继承 BaseAdapter，无耦合关系： + * - DeepSeekAdapter — OpenAI 兼容 + DeepSeek 特有参数 + * - AgnesAdapter — OpenAI 兼容 + Agnes 特有参数 + * - OllamaAdapter — Ollama 原生 API + * + * 共享工具（仅供 OpenAI 兼容 Adapter 使用）： + * - shared/openai-format — 消息/工具格式构建 + * - shared/sse-stream — SSE 流式解析 + */ + export { BaseAdapter } from './base-adapter'; export { DeepSeekAdapter } from './deepseek.adapter'; export { AgnesAdapter } from './agnes-ai.adapter'; diff --git a/electron/harness/adapters/shared/openai-format.ts b/electron/harness/adapters/shared/openai-format.ts new file mode 100644 index 0000000..bcaf498 --- /dev/null +++ b/electron/harness/adapters/shared/openai-format.ts @@ -0,0 +1,100 @@ +/** + * OpenAI 兼容 API 格式构建工具 + * + * 将 MetonaRequest 转换为 OpenAI /chat/completions 兼容的原生请求格式。 + * DeepSeek 和 Agnes AI 共享此工具，各自 Adapter 只需处理 Provider 特有的差异参数。 + * + * @see electron/harness/types/metona-request.ts — MetonaRequest 定义 + * @see apis/deepseek-api-docs-20260518.html + * @see apis/agnes-ai-api-docs-20260625.html + */ + +import type { MetonaRequest, MetonaToolDef } from '../../types'; + +/** + * 构建 OpenAI 兼容的 messages 数组 + * + * 处理： + * - System Prompt 拼接（静态区 + 动态区 + 安全准则） + * - 图片 → 多模态 content 数组 [{type:"text"}, {type:"image_url"}] + * - 工具调用历史保留（reasoning_content + tool_calls） + * - 工具结果注入（tool_call_id + content） + */ +export function buildOpenAICompatibleMessages( + request: MetonaRequest, +): Array> { + const systemContent = [ + request.systemPrompt.roleDefinition, + request.systemPrompt.outputConstraints, + request.systemPrompt.safetyGuidelines, + request.systemPrompt.dynamicReminders, + ] + .filter(Boolean) + .join('\n\n'); + + const nonSystemMessages = request.messages + .filter((m) => m.role !== 'system') + .map((m) => { + const msg: Record = { role: m.role }; + + // === 多模态图片处理 === + if (m.images && m.images.length > 0) { + const contentParts: Array> = []; + if (m.content) { + contentParts.push({ type: 'text', text: m.content }); + } + for (const img of m.images) { + contentParts.push({ + type: 'image_url', + image_url: { url: img.url, detail: img.detail ?? 'auto' }, + }); + } + msg.content = contentParts; + } else { + msg.content = m.content; + } + + // === Assistant 工具调用历史 === + if (m.role === 'assistant' && m.toolCalls?.length) { + msg.tool_calls = m.toolCalls.map((tc) => ({ + id: tc.id, + type: 'function', + function: { name: tc.name, arguments: JSON.stringify(tc.args) }, + })); + // 推理内容必须带回上下文（否则 LLM 丢失思考链） + if (m.reasoningContent) { + msg.reasoning_content = m.reasoningContent; + } + } + + // === 工具执行结果 === + if (m.role === 'tool' && m.toolResult) { + msg.tool_call_id = m.toolResult.toolCallId; + msg.content = + typeof m.toolResult.result === 'string' + ? m.toolResult.result + : JSON.stringify(m.toolResult.result); + } + + return msg; + }); + + return [{ role: 'system', content: systemContent }, ...nonSystemMessages]; +} + +/** + * 构建 OpenAI 兼容的 tools 数组 + */ +export function buildOpenAICompatibleTools( + tools?: MetonaToolDef[], +): Array> | undefined { + if (!tools?.length) return undefined; + return tools.map((t) => ({ + type: 'function', + function: { + name: t.name, + description: t.description, + parameters: t.parameters, + }, + })); +} diff --git a/electron/harness/adapters/shared/sse-stream.ts b/electron/harness/adapters/shared/sse-stream.ts new file mode 100644 index 0000000..4e0cb1c --- /dev/null +++ b/electron/harness/adapters/shared/sse-stream.ts @@ -0,0 +1,226 @@ +/** + * SSE 流式解析工具 + * + * 解析 OpenAI 兼容的 Server-Sent Events (SSE) 流式响应， + * 产出 MetonaStreamEvent。DeepSeek 和 Agnes AI 共享此工具。 + * + * SSE 格式：data: {json}\n\n + * 结束标记：data: [DONE] + */ + +import { nanoid } from 'nanoid'; +import type { MetonaStreamEvent, MetonaTokenUsage } from '../../types'; +import { MetonaStreamEventType } from '../../types'; + +/** + * 解析 OpenAI 兼容 SSE 流式响应 + * + * @param responseBody - fetch Response.body (ReadableStream) + * @param requestId - 对应的请求 ID + * @param sessionId - 会话 ID + * @param iteration - 当前迭代轮次 + * @yields MetonaStreamEvent + */ +export async function* parseSSEStream( + responseBody: ReadableStream, + requestId: string, + sessionId: string, + iteration: number, +): AsyncGenerator { + const reader = responseBody.getReader(); + const decoder = new TextDecoder(); + let seq = 0; + let buffer = ''; + + // 工具调用缓冲区：index → { name, argsBuffer } + const toolCallsBuffer = new Map(); + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split('\n'); + buffer = lines.pop() ?? ''; + + for (const line of lines) { + const trimmed = line.trim(); + if (!trimmed || !trimmed.startsWith('data: ')) continue; + const data = trimmed.slice(6); + + // 流结束 + if (data === '[DONE]') { + // 将缓冲区中未完成拼接的工具调用发送 + for (const [index, buf] of toolCallsBuffer) { + try { + yield { + type: MetonaStreamEventType.TOOL_CALL_COMPLETE, + requestId, + sessionId, + iteration, + seq: seq++, + timestamp: Date.now(), + toolCall: { + id: `tc_${nanoid(8)}`, + name: buf.name, + args: buf.argsBuffer ? JSON.parse(buf.argsBuffer) : {}, + iteration, + timestamp: Date.now(), + }, + }; + } catch { + // JSON 解析失败，跳过 + } + } + toolCallsBuffer.clear(); + + yield { + type: MetonaStreamEventType.DONE, + requestId, + sessionId, + iteration, + seq: seq++, + timestamp: Date.now(), + }; + return; + } + + try { + const chunk = JSON.parse(data); + const delta = chunk.choices?.[0]?.delta; + + // 文本内容增量 + if (delta?.content) { + yield { + type: MetonaStreamEventType.TEXT_DELTA, + requestId, + sessionId, + iteration, + seq: seq++, + timestamp: Date.now(), + delta: delta.content, + }; + } + + // 推理内容增量（Thinking 模式） + if (delta?.reasoning_content) { + yield { + type: MetonaStreamEventType.REASONING_DELTA, + requestId, + sessionId, + iteration, + seq: seq++, + timestamp: Date.now(), + delta: delta.reasoning_content, + }; + } + + // 工具调用增量 — 缓冲拼接 + if (delta?.tool_calls) { + for (const tc of delta.tool_calls) { + const idx = tc.index ?? 0; + if (!toolCallsBuffer.has(idx)) { + toolCallsBuffer.set(idx, { name: tc.function?.name ?? '', argsBuffer: '' }); + } + const buf = toolCallsBuffer.get(idx)!; + if (tc.function?.name) buf.name = tc.function.name; + if (tc.function?.arguments) buf.argsBuffer += tc.function.arguments; + + yield { + type: MetonaStreamEventType.TOOL_CALL_DELTA, + requestId, + sessionId, + iteration, + seq: seq++, + timestamp: Date.now(), + toolCallDelta: { + index: idx, + name: tc.function?.name, + argsDelta: tc.function?.arguments, + }, + }; + } + } + + // Token 使用统计 + if (chunk.usage) { + const usage: MetonaTokenUsage = { + inputTokens: chunk.usage.prompt_tokens ?? 0, + outputTokens: chunk.usage.completion_tokens ?? 0, + totalTokens: chunk.usage.total_tokens ?? 0, + reasoningTokens: chunk.usage.completion_tokens_details?.reasoning_tokens, + cacheHitTokens: chunk.usage.prompt_cache_hit_tokens, + cacheMissTokens: chunk.usage.prompt_cache_miss_tokens, + }; + + yield { + type: MetonaStreamEventType.USAGE, + requestId, + sessionId, + iteration, + seq: seq++, + timestamp: Date.now(), + usage, + }; + } + } catch { + // 跳过解析失败的行 + } + } + } +} + +/** + * 解析 OpenAI 兼容的非流式 JSON 响应 → MetonaResponse + */ +export function parseOpenAICompatibleResponse( + data: Record, + requestId: string, + provider: string, + defaultModel: string, +): { + content: string; + reasoningContent?: string; + toolCalls?: Array<{ id: string; name: string; args: Record; iteration: number; timestamp: number }>; + finishReason: string; + usage: MetonaTokenUsage; +} { + const choice = (data.choices as Array>)?.[0]; + const message = choice?.message as Record | undefined; + const usage = data.usage as Record | undefined; + const rawToolCalls = message?.tool_calls as Array> | undefined; + + return { + content: (message?.content as string) ?? '', + reasoningContent: message?.reasoning_content as string | undefined, + toolCalls: rawToolCalls?.map((tc) => { + const fn = tc.function as Record; + return { + id: tc.id as string, + name: fn.name as string, + args: JSON.parse(fn.arguments as string), + iteration: 0, + timestamp: Date.now(), + }; + }), + finishReason: mapOpenAIFinishReason(choice?.finish_reason as string), + usage: { + inputTokens: (usage?.prompt_tokens as number) ?? 0, + outputTokens: (usage?.completion_tokens as number) ?? 0, + totalTokens: (usage?.total_tokens as number) ?? 0, + reasoningTokens: (usage?.completion_tokens_details as Record)?.reasoning_tokens as number | undefined, + cacheHitTokens: usage?.prompt_cache_hit_tokens as number | undefined, + cacheMissTokens: usage?.prompt_cache_miss_tokens as number | undefined, + }, + }; +} + +function mapOpenAIFinishReason(reason: string): string { + switch (reason) { + case 'stop': return 'stop'; + case 'length': return 'length'; + case 'tool_calls': return 'tool_calls'; + case 'content_filter': return 'content_filter'; + default: return 'stop'; + } +}