feat(desktop): Premium LLM SSE 스트리밍 — 토큰 단위 실시간 응답
- sse-parser.ts: Anthropic SSE 파서 (content_block_delta.text 추출) - CloudSyncService.invokeFunctionStream(): raw fetch + SSE ReadableStream - PremiumLLMService.chatStream(): stream=true, 토큰 단위 yield - SSE 실패 시 자동 비스트리밍 fallback - AbortController 연결 (취소 지원)
This commit is contained in:
parent
667c09242b
commit
7467bd6ce9
3 changed files with 172 additions and 11 deletions
|
|
@ -425,6 +425,62 @@ class CloudSyncService extends EventEmitter {
|
||||||
return { data, error: null }
|
return { data, error: null }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Edge Function 스트리밍 호출 — SSE 응답의 ReadableStream 반환.
|
||||||
|
* Supabase JS의 functions.invoke()는 JSON 자동 파싱하므로 SSE에 부적합.
|
||||||
|
* raw fetch + auth 토큰으로 직접 호출.
|
||||||
|
*/
|
||||||
|
async invokeFunctionStream(
|
||||||
|
name: string,
|
||||||
|
body: Record<string, unknown>,
|
||||||
|
signal?: AbortSignal,
|
||||||
|
): Promise<{ stream: ReadableStream<Uint8Array> | 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 {
|
getState(): CloudSyncState {
|
||||||
return {
|
return {
|
||||||
authenticated: this.isAuthenticated(),
|
authenticated: this.isAuthenticated(),
|
||||||
|
|
|
||||||
|
|
@ -15,6 +15,7 @@ import { getCloudSyncService } from './CloudSyncService'
|
||||||
import { D3ROError, ErrorCode } from '@d3ro/core/errors'
|
import { D3ROError, ErrorCode } from '@d3ro/core/errors'
|
||||||
import type { LLMAction } from '@d3ro/core/types'
|
import type { LLMAction } from '@d3ro/core/types'
|
||||||
import { resolveSystemPrompt } from './llm-prompts'
|
import { resolveSystemPrompt } from './llm-prompts'
|
||||||
|
import { parseAnthropicSSE } from '../utils/sse-parser'
|
||||||
|
|
||||||
const logger = getLogger('PremiumLLMService')
|
const logger = getLogger('PremiumLLMService')
|
||||||
|
|
||||||
|
|
@ -169,8 +170,7 @@ class PremiumLLMService extends EventEmitter {
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 스트리밍 대화 (Voice Conversation용) — LocalLLMService.chatStream 시그니처 동일.
|
* 스트리밍 대화 (Voice Conversation용) — LocalLLMService.chatStream 시그니처 동일.
|
||||||
* Phase 3.2 MVP: 비스트리밍으로 받은 응답을 한 번에 yield.
|
* SSE 스트리밍: llm-proxy에 stream=true로 요청, Anthropic SSE를 토큰 단위 yield.
|
||||||
* 추후 SSE 스트림 파싱으로 실시간 토큰 yield로 업그레이드 예정.
|
|
||||||
*/
|
*/
|
||||||
async *chatStream(
|
async *chatStream(
|
||||||
messages: Array<{ role: string; content: string }>,
|
messages: Array<{ role: string; content: string }>,
|
||||||
|
|
@ -178,7 +178,6 @@ class PremiumLLMService extends EventEmitter {
|
||||||
): AsyncGenerator<string, string> {
|
): AsyncGenerator<string, string> {
|
||||||
this._ensureAuth()
|
this._ensureAuth()
|
||||||
|
|
||||||
// Ollama 메시지 형식 → Claude Messages 형식으로 정규화
|
|
||||||
const claudeMessages: ChatMessage[] = messages
|
const claudeMessages: ChatMessage[] = messages
|
||||||
.filter((m) => m.role === 'user' || m.role === 'assistant')
|
.filter((m) => m.role === 'user' || m.role === 'assistant')
|
||||||
.map((m) => ({ role: m.role as 'user' | 'assistant', content: m.content }))
|
.map((m) => ({ role: m.role as 'user' | 'assistant', content: m.content }))
|
||||||
|
|
@ -187,20 +186,49 @@ class PremiumLLMService extends EventEmitter {
|
||||||
messages: claudeMessages,
|
messages: claudeMessages,
|
||||||
model: options?.model,
|
model: options?.model,
|
||||||
max_tokens: 2048,
|
max_tokens: 2048,
|
||||||
stream: false,
|
stream: true,
|
||||||
}
|
}
|
||||||
|
|
||||||
const response = await this._invokeProxy(body)
|
this._abortController = new AbortController()
|
||||||
|
const cloud = getCloudSyncService()
|
||||||
|
|
||||||
|
const { stream, error } = await cloud.invokeFunctionStream(
|
||||||
|
'llm-proxy',
|
||||||
|
body as unknown as Record<string, unknown>,
|
||||||
|
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 firstBlock = response.content?.[0]
|
||||||
const text = firstBlock?.type === 'text' ? firstBlock.text : ''
|
const text = firstBlock?.type === 'text' ? firstBlock.text : ''
|
||||||
|
if (text.length > 0) yield text
|
||||||
// 단일 chunk yield — 추후 SSE 스트리밍으로 업그레이드 시 여러 번 yield.
|
|
||||||
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 {
|
cancelGeneration(): void {
|
||||||
if (this._abortController) {
|
if (this._abortController) {
|
||||||
this._abortController.abort()
|
this._abortController.abort()
|
||||||
|
|
|
||||||
77
apps/desktop/src/main/utils/sse-parser.ts
Normal file
77
apps/desktop/src/main/utils/sse-parser.ts
Normal file
|
|
@ -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<Uint8Array>을 파싱하여 텍스트 토큰을 yield.
|
||||||
|
* Anthropic SSE 형식: "data: {json}\n\n" 라인 단위.
|
||||||
|
* content_block_delta.delta.text 추출, message_stop 또는 [DONE] 시 종료.
|
||||||
|
*/
|
||||||
|
export async function* parseAnthropicSSE(
|
||||||
|
stream: ReadableStream<Uint8Array>,
|
||||||
|
): AsyncGenerator<string> {
|
||||||
|
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()
|
||||||
|
}
|
||||||
|
}
|
||||||
Loading…
Add table
Add a link
Reference in a new issue