diff --git a/frontend/src/api/endpoints.ts b/frontend/src/api/endpoints.ts index 573b58a8..e0dc7742 100644 --- a/frontend/src/api/endpoints.ts +++ b/frontend/src/api/endpoints.ts @@ -77,6 +77,10 @@ const endpoints = { WORKFLOWS: '/api/workflows', WORKFLOW: (id: string) => `/api/workflows/${id}`, }, + V1: { + CHAT_COMPLETIONS: '/v1/chat/completions', + MODELS: '/v1/models', + }, CONVERSATION: { ANSWER: '/api/answer', ANSWER_STREAMING: '/stream', diff --git a/frontend/src/api/services/conversationService.ts b/frontend/src/api/services/conversationService.ts index 853a6863..a79370aa 100644 --- a/frontend/src/api/services/conversationService.ts +++ b/frontend/src/api/services/conversationService.ts @@ -54,6 +54,18 @@ const conversationService = { apiClient.get(endpoints.CONVERSATION.DELETE_ALL, token, {}), update: (data: any, token: string | null): Promise => apiClient.post(endpoints.CONVERSATION.UPDATE, data, token, {}), + chatCompletions: ( + data: any, + agentApiKey: string, + signal: AbortSignal, + ): Promise => + apiClient.post( + endpoints.V1.CHAT_COMPLETIONS, + data, + null, + { Authorization: `Bearer ${agentApiKey}` }, + signal, + ), }; export default conversationService; diff --git a/frontend/src/conversation/conversationHandlers.ts b/frontend/src/conversation/conversationHandlers.ts index 9bed754f..c7fcdf44 100644 --- a/frontend/src/conversation/conversationHandlers.ts +++ b/frontend/src/conversation/conversationHandlers.ts @@ -254,6 +254,190 @@ export function handleSubmitToolActions( }); } +/** + * Stream a chat completion via the /v1/chat/completions endpoint. + * + * Translates the standard streaming format (choices[0].delta) back into + * the internal DocsGPT event shape so the existing Redux reducers can + * consume the events without any changes. + */ +export function handleV1ChatCompletionStreaming( + question: string, + signal: AbortSignal, + agentApiKey: string, + history: { prompt: string; response: string }[], + onEvent: (event: MessageEvent) => void, + tools?: any[], + attachments?: string[], +): Promise { + // Build messages array from history + current question + const messages: any[] = []; + for (const h of history) { + messages.push({ role: 'user', content: h.prompt }); + messages.push({ role: 'assistant', content: h.response }); + } + messages.push({ role: 'user', content: question }); + + const payload: any = { + messages, + stream: true, + }; + if (tools && tools.length > 0) { + payload.tools = tools; + } + if (attachments && attachments.length > 0) { + payload.docsgpt = { attachments }; + } + + return new Promise((resolve, reject) => { + conversationService + .chatCompletions(payload, agentApiKey, signal) + .then((response) => { + if (!response.body) throw Error('No response body'); + + let buffer = ''; + const reader = response.body.getReader(); + const decoder = new TextDecoder('utf-8'); + + const processStream = ({ + done, + value, + }: ReadableStreamReadResult) => { + if (done) return; + + const chunk = decoder.decode(value); + buffer += chunk; + + const events = buffer.split('\n\n'); + buffer = events.pop() ?? ''; + + for (const event of events) { + if (!event.trim().startsWith('data:')) continue; + + const dataLine = event + .split('\n') + .map((line: string) => line.replace(/^data:\s?/, '')) + .join(''); + + const trimmed = dataLine.trim(); + + // Handle [DONE] sentinel + if (trimmed === '[DONE]') { + onEvent( + new MessageEvent('message', { + data: JSON.stringify({ type: 'end' }), + }), + ); + continue; + } + + try { + const parsed = JSON.parse(trimmed); + // Translate standard format to DocsGPT internal events + const translated = translateV1ChunkToInternalEvents(parsed); + for (const evt of translated) { + onEvent( + new MessageEvent('message', { + data: JSON.stringify(evt), + }), + ); + } + } catch { + // Skip unparseable chunks + } + } + + reader.read().then(processStream).catch(reject); + }; + + reader.read().then(processStream).catch(reject); + }) + .catch((error) => { + console.error('V1 chat completion stream failed:', error); + reject(error); + }); + }); +} + +/** + * Translate a single v1 streaming chunk to internal DocsGPT event(s). + * + * Standard format: {"choices": [{"delta": {"content": "chunk"}, ...}]} + * Extension format: {"docsgpt": {"type": "source", ...}} + */ +function translateV1ChunkToInternalEvents( + chunk: any, +): { type: string; [key: string]: any }[] { + const events: { type: string; [key: string]: any }[] = []; + + // DocsGPT extension chunks + if (chunk.docsgpt) { + const ext = chunk.docsgpt; + if (ext.type === 'source') { + events.push({ type: 'source', source: ext.sources }); + } else if (ext.type === 'tool_call') { + events.push({ type: 'tool_call', data: ext.data }); + } else if (ext.type === 'tool_calls_pending') { + events.push({ + type: 'tool_calls_pending', + data: { pending_tool_calls: ext.pending_tool_calls }, + }); + } else if (ext.type === 'id') { + events.push({ type: 'id', id: ext.conversation_id }); + } + return events; + } + + // Error chunks + if (chunk.error) { + events.push({ type: 'error', error: chunk.error.message || 'Error' }); + return events; + } + + // Standard choices chunks + const choice = chunk.choices?.[0]; + if (!choice) return events; + + const delta = choice.delta || {}; + const finishReason = choice.finish_reason; + + if (delta.content) { + events.push({ type: 'answer', answer: delta.content }); + } + + if (delta.reasoning_content) { + events.push({ type: 'thought', thought: delta.reasoning_content }); + } + + if (delta.tool_calls) { + for (const tc of delta.tool_calls) { + events.push({ + type: 'tool_call', + data: { + call_id: tc.id, + action_name: tc.function?.name || '', + tool_name: tc.function?.name || '', + arguments: tc.function?.arguments + ? JSON.parse(tc.function.arguments) + : {}, + status: 'requires_client_execution', + }, + }); + } + } + + if (finishReason === 'stop') { + events.push({ type: 'end' }); + } else if (finishReason === 'tool_calls') { + events.push({ + type: 'tool_calls_pending', + data: { pending_tool_calls: [] }, + }); + } + + return events; +} + export function handleSearch( question: string, token: string | null, diff --git a/frontend/src/conversation/conversationSlice.ts b/frontend/src/conversation/conversationSlice.ts index 76f3beb7..6181e1da 100644 --- a/frontend/src/conversation/conversationSlice.ts +++ b/frontend/src/conversation/conversationSlice.ts @@ -11,6 +11,7 @@ import { handleFetchAnswer, handleFetchAnswerSteaming, handleSubmitToolActions, + handleV1ChatCompletionStreaming, } from './conversationHandlers'; import { Answer, @@ -28,6 +29,7 @@ const initialState: ConversationState = { }; const API_STREAMING = import.meta.env.VITE_API_STREAMING === 'true'; +const USE_V1_API = import.meta.env.VITE_USE_V1_API === 'true'; let abortController: AbortController | null = null; export function handleAbort() { @@ -61,7 +63,102 @@ export const fetchAnswer = createAsyncThunk< state.preference.selectedModel?.id; if (state.preference) { - if (API_STREAMING) { + const agentKey = state.preference.selectedAgent?.key; + if (USE_V1_API && agentKey) { + // Build history from prior queries for v1 format + const v1History = state.conversation.queries + .filter((q) => q.response) + .map((q) => ({ prompt: q.prompt, response: q.response || '' })); + + await handleV1ChatCompletionStreaming( + question, + signal, + agentKey, + v1History, + (event) => { + const data = JSON.parse(event.data); + const targetIndex = indx ?? state.conversation.queries.length - 1; + + if (currentConversationId === state.conversation.conversationId) { + if (data.type === 'end') { + dispatch(conversationSlice.actions.setStatus('idle')); + getConversations(state.preference.token) + .then((fetchedConversations) => { + dispatch(setConversations(fetchedConversations)); + }) + .catch((error) => { + console.error('Failed to fetch conversations: ', error); + }); + if (!isSourceUpdated) { + dispatch( + updateStreamingSource({ + conversationId: currentConversationId, + index: targetIndex, + query: { sources: [] }, + }), + ); + } + } else if (data.type === 'id') { + const currentState = getState() as RootState; + if (currentState.conversation.conversationId === null) { + dispatch( + updateConversationId({ + query: { conversationId: data.id }, + }), + ); + } + } else if (data.type === 'thought') { + dispatch( + updateThought({ + conversationId: currentConversationId, + index: targetIndex, + query: { thought: data.thought }, + }), + ); + } else if (data.type === 'source') { + isSourceUpdated = true; + dispatch( + updateStreamingSource({ + conversationId: currentConversationId, + index: targetIndex, + query: { sources: data.source ?? [] }, + }), + ); + } else if (data.type === 'tool_call') { + dispatch( + updateToolCall({ + index: targetIndex, + tool_call: data.data as ToolCallsType, + }), + ); + } else if (data.type === 'tool_calls_pending') { + dispatch( + conversationSlice.actions.setStatus('awaiting_tool_actions'), + ); + } else if (data.type === 'error') { + dispatch(conversationSlice.actions.setStatus('failed')); + dispatch( + conversationSlice.actions.raiseError({ + conversationId: currentConversationId, + index: targetIndex, + message: data.error, + }), + ); + } else { + dispatch( + updateStreamingQuery({ + conversationId: currentConversationId, + index: targetIndex, + query: { response: data.answer }, + }), + ); + } + } + }, + undefined, + attachmentIds.length > 0 ? attachmentIds : undefined, + ); + } else if (API_STREAMING) { await handleFetchAnswerSteaming( question, signal,