diff --git a/apps/desktop/src/main/services/CloudSyncService.ts b/apps/desktop/src/main/services/CloudSyncService.ts index c8ebb64..d0a9816 100644 --- a/apps/desktop/src/main/services/CloudSyncService.ts +++ b/apps/desktop/src/main/services/CloudSyncService.ts @@ -425,6 +425,62 @@ class CloudSyncService extends EventEmitter { return { data, error: null } } + /** + * Edge Function 스트리밍 호출 — SSE 응답의 ReadableStream 반환. + * Supabase JS의 functions.invoke()는 JSON 자동 파싱하므로 SSE에 부적합. + * raw fetch + auth 토큰으로 직접 호출. + */ + async invokeFunctionStream( + name: string, + body: Record, + signal?: AbortSignal, + ): Promise<{ stream: ReadableStream | null; error: { message: string } | null }> { + if (!this._client) { + return { stream: null, error: { message: 'Supabase client not initialized' } } + } + + const { data: sessionData } = await this._client.auth.getSession() + const token = sessionData.session?.access_token + if (!token) { + return { stream: null, error: { message: 'No active session — 로그인 필요' } } + } + + const url = configGet('supabaseUrl') as string | undefined + const anonKey = configGet('supabaseAnonKey') as string | undefined + if (!url || !anonKey) { + return { stream: null, error: { message: 'Supabase URL/key not configured' } } + } + + try { + const response = await fetch(`${url}/functions/v1/${name}`, { + method: 'POST', + headers: { + Authorization: `Bearer ${token}`, + apikey: anonKey, + 'Content-Type': 'application/json', + }, + body: JSON.stringify(body), + signal, + }) + + if (!response.ok) { + const text = await response.text() + return { stream: null, error: { message: `${response.status}: ${text}` } } + } + + if (!response.body) { + return { stream: null, error: { message: 'No response body' } } + } + + return { stream: response.body, error: null } + } catch (err) { + if ((err as Error).name === 'AbortError') { + return { stream: null, error: { message: 'Request aborted' } } + } + return { stream: null, error: { message: err instanceof Error ? err.message : String(err) } } + } + } + getState(): CloudSyncState { return { authenticated: this.isAuthenticated(), diff --git a/apps/desktop/src/main/services/PremiumLLMService.ts b/apps/desktop/src/main/services/PremiumLLMService.ts index 2f80324..9240e04 100644 --- a/apps/desktop/src/main/services/PremiumLLMService.ts +++ b/apps/desktop/src/main/services/PremiumLLMService.ts @@ -15,6 +15,7 @@ import { getCloudSyncService } from './CloudSyncService' import { D3ROError, ErrorCode } from '@d3ro/core/errors' import type { LLMAction } from '@d3ro/core/types' import { resolveSystemPrompt } from './llm-prompts' +import { parseAnthropicSSE } from '../utils/sse-parser' const logger = getLogger('PremiumLLMService') @@ -169,8 +170,7 @@ class PremiumLLMService extends EventEmitter { /** * 스트리밍 대화 (Voice Conversation용) — LocalLLMService.chatStream 시그니처 동일. - * Phase 3.2 MVP: 비스트리밍으로 받은 응답을 한 번에 yield. - * 추후 SSE 스트림 파싱으로 실시간 토큰 yield로 업그레이드 예정. + * SSE 스트리밍: llm-proxy에 stream=true로 요청, Anthropic SSE를 토큰 단위 yield. */ async *chatStream( messages: Array<{ role: string; content: string }>, @@ -178,7 +178,6 @@ class PremiumLLMService extends EventEmitter { ): AsyncGenerator { this._ensureAuth() - // Ollama 메시지 형식 → Claude Messages 형식으로 정규화 const claudeMessages: ChatMessage[] = messages .filter((m) => m.role === 'user' || m.role === 'assistant') .map((m) => ({ role: m.role as 'user' | 'assistant', content: m.content })) @@ -187,18 +186,47 @@ class PremiumLLMService extends EventEmitter { messages: claudeMessages, model: options?.model, max_tokens: 2048, - stream: false, + stream: true, } - const response = await this._invokeProxy(body) - const firstBlock = response.content?.[0] - const text = firstBlock?.type === 'text' ? firstBlock.text : '' + this._abortController = new AbortController() + const cloud = getCloudSyncService() - // 단일 chunk yield — 추후 SSE 스트리밍으로 업그레이드 시 여러 번 yield. - if (text.length > 0) { - yield text + const { stream, error } = await cloud.invokeFunctionStream( + 'llm-proxy', + body as unknown as Record, + this._abortController.signal, + ) + + if (error || !stream) { + const msg = error?.message ?? 'Stream unavailable' + logger.error(`SSE stream failed: ${msg}`) + + // SSE 실패 시 비스트리밍 fallback + logger.info('Falling back to non-streaming Premium LLM') + const fallbackBody = { ...body, stream: false } + const response = await this._invokeProxy(fallbackBody) + const firstBlock = response.content?.[0] + const text = firstBlock?.type === 'text' ? firstBlock.text : '' + if (text.length > 0) yield text + return text } - return text + + let accumulated = '' + try { + for await (const token of parseAnthropicSSE(stream)) { + accumulated += token + yield token + } + } catch (err) { + if ((err as Error).name !== 'AbortError') { + logger.warn(`SSE parse error: ${err instanceof Error ? err.message : String(err)}`) + } + } finally { + this._abortController = null + } + + return accumulated } cancelGeneration(): void { diff --git a/apps/desktop/src/main/utils/sse-parser.ts b/apps/desktop/src/main/utils/sse-parser.ts new file mode 100644 index 0000000..dab2eff --- /dev/null +++ b/apps/desktop/src/main/utils/sse-parser.ts @@ -0,0 +1,77 @@ +// src/main/utils/sse-parser.ts +// Anthropic Claude Messages API SSE 스트림 파서 +// content_block_delta 이벤트에서 텍스트 토큰을 추출하는 AsyncGenerator + +interface ContentBlockDelta { + type: 'content_block_delta' + delta: { + type: 'text_delta' + text: string + } +} + +interface SSEEvent { + type: string + [key: string]: unknown +} + +/** + * ReadableStream을 파싱하여 텍스트 토큰을 yield. + * Anthropic SSE 형식: "data: {json}\n\n" 라인 단위. + * content_block_delta.delta.text 추출, message_stop 또는 [DONE] 시 종료. + */ +export async function* parseAnthropicSSE( + stream: ReadableStream, +): AsyncGenerator { + const reader = stream.getReader() + const decoder = new TextDecoder() + let buffer = '' + + try { + 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() + + // 빈 줄 또는 이벤트 타입 라인 (event:) 건너뜀 + if (!trimmed || trimmed.startsWith('event:')) continue + + // "data: [DONE]" — 종료 시그널 + if (trimmed === 'data: [DONE]') return + + // "data: {...}" — JSON 파싱 + if (trimmed.startsWith('data: ')) { + const json = trimmed.substring(6) + try { + const evt = JSON.parse(json) as SSEEvent + + // message_stop → 스트림 종료 + if (evt.type === 'message_stop') return + + // content_block_delta → 텍스트 토큰 yield + if (evt.type === 'content_block_delta') { + const delta = evt as unknown as ContentBlockDelta + if (delta.delta?.type === 'text_delta' && delta.delta.text) { + yield delta.delta.text + } + } + // 그 외 이벤트 (message_start, content_block_start 등)는 건너뜀 + } catch { + // JSON 파싱 실패 — 건너뜀 + } + } + } + } + } finally { + reader.releaseLock() + } +}