/** * 自建模型网关(OpenAI 兼容)。 * * 职责: * 1. 持有上游 baseUrl / apiKey(永不进入渲染进程) * 2. 按配置地址选择 Chat Completions / Responses + SSE 流式 * 3. 把上游的增量 delta 组装成「文本增量」与「完整工具调用」两类输出 * * 设计约束:网关只做协议适配与组装,不含业务策略; * 模型路由、上下文裁剪、工具编排由上层(agent/loop.ts)负责。 */ import type { GatewayConfig, TokenUsage, ToolSpec } from '@shared/types' import { resolveOpenAIEndpoint } from '@shared/openai-endpoint' import { parseSse } from './sse' import { buildResponsesRequest, parseResponsesStream } from './responses' /** 上游 wire 格式的消息 */ export interface WireMessage { role: 'system' | 'user' | 'assistant' | 'tool' content: string | null | Array<{ type: 'text'; text: string } | { type: 'image_url'; image_url: { url: string } }> tool_calls?: WireToolCall[] tool_call_id?: string } export interface WireToolCall { id: string type: 'function' function: { name: string; arguments: string } } /** 网关输出的增量事件 */ export type GatewayDelta = | { type: 'text'; text: string } | { type: 'reasoning'; text: string } | { type: 'tool_calls'; calls: WireToolCall[] } | { type: 'usage'; usage: TokenUsage } | { type: 'finish'; reason: string | null } export interface GatewayStreamRequest { messages: WireMessage[] tools?: ToolSpec[] signal?: AbortSignal } export class GatewayError extends Error { constructor( message: string, readonly status?: number, readonly body?: string ) { super(message) this.name = 'GatewayError' } } export class Gateway { constructor(private readonly config: GatewayConfig) {} async *stream(req: GatewayStreamRequest): AsyncGenerator { if (!this.config.apiKey && this.config.authMode !== 'none') { throw new GatewayError('未配置 API Key,请先在设置中填写') } if (!this.config.baseUrl) { throw new GatewayError('未配置上游地址(baseUrl)') } let endpoint: ReturnType try { endpoint = resolveOpenAIEndpoint(this.config.baseUrl) } catch { throw new GatewayError('上游地址格式无效') } const responses = endpoint.api === 'openai-responses' const body: Record = responses ? buildResponsesRequest(this.config, req) : { model: this.config.model, messages: req.messages, stream: true, temperature: this.config.temperature } if (!responses && this.config.maxOutputTokens) body.max_tokens = this.config.maxOutputTokens if (!responses && req.tools && req.tools.length > 0) { body.tools = req.tools body.tool_choice = 'auto' } let res: Response try { res = await fetch(endpoint.endpoint, { method: 'POST', headers: { 'content-type': 'application/json', accept: 'text/event-stream', ...(this.config.apiKey ? { authorization: `Bearer ${this.config.apiKey}` } : {}) }, body: JSON.stringify(body), signal: req.signal }) } catch (err) { if ((err as Error).name === 'AbortError') throw err throw new GatewayError(responses ? '无法连接 Responses 上游' : `无法连接上游:${(err as Error).message}`) } if (!res.ok || !res.body) { // A Responses provider can echo request credentials in arbitrary error bodies. const text = responses ? undefined : await safeReadText(res) if (responses) await res.body?.cancel().catch(() => {}) throw new GatewayError( responses ? `Responses 上游返回 HTTP ${res.status}` : `上游返回 ${res.status} ${res.statusText}`.trim(), res.status, text ) } if (responses) { yield* parseResponsesStream(res.body, (message) => new GatewayError(message), req.signal) return } // 累加器:按 index 归并流式工具调用分片 const acc = new Map() let finishReason: string | null = null for await (const evt of parseSse(res.body)) { if (evt.data === '[DONE]') break let payload: any try { payload = JSON.parse(evt.data) } catch { continue // 非 JSON 的心跳/注释行直接跳过 } if (payload?.error) { throw new GatewayError( `上游错误:${payload.error.message ?? JSON.stringify(payload.error)}` ) } const choice = payload?.choices?.[0] if (!choice) continue const delta = choice.delta ?? {} if (typeof delta.content === 'string' && delta.content.length > 0) { yield { type: 'text', text: delta.content } } if (typeof delta.reasoning_content === 'string' && delta.reasoning_content.length > 0) { yield { type: 'reasoning', text: delta.reasoning_content } } if (Array.isArray(delta.tool_calls)) { for (const piece of delta.tool_calls) { const idx = typeof piece.index === 'number' ? piece.index : 0 const slot = acc.get(idx) ?? { id: '', name: '', args: '' } if (piece.id) slot.id = piece.id if (piece.function?.name) slot.name = piece.function.name if (piece.function?.arguments) slot.args += piece.function.arguments acc.set(idx, slot) } } if (choice.finish_reason) finishReason = choice.finish_reason } const calls: WireToolCall[] = [...acc.entries()] .sort((a, b) => a[0] - b[0]) .map(([idx, slot]) => ({ id: slot.id || `call_${idx}_${Date.now()}`, type: 'function' as const, function: { name: slot.name, arguments: slot.args || '{}' } })) if (calls.length > 0) yield { type: 'tool_calls', calls } yield { type: 'finish', reason: finishReason } } } async function safeReadText(res: Response): Promise { try { return (await res.text()).slice(0, 2000) } catch { return '' } }