diff --git a/src/components/AikitPlayground/AikitPlayground.tsx b/src/components/AikitPlayground/AikitPlayground.tsx index 0b3fddba5199..233784114271 100644 --- a/src/components/AikitPlayground/AikitPlayground.tsx +++ b/src/components/AikitPlayground/AikitPlayground.tsx @@ -1,4 +1,4 @@ -import {ChatContainer} from '@gravity-ui/aikit'; +import {ChatContainer, useOpenAIStreamAdapter} from '@gravity-ui/aikit'; import type { ChatStatus, TAssistantMessage, @@ -7,12 +7,13 @@ import type { TUserMessage, } from '@gravity-ui/aikit'; import {useTranslation} from 'next-i18next'; -import React, {memo, useCallback, useState} from 'react'; +import React, {memo, useCallback, useEffect, useState} from 'react'; import {block} from '../../utils'; import {PlaygroundWrap} from '../PlaygroundWrap'; import './AikitPlayground.scss'; +import {createMockMcpStream, isMockMcpEnabled} from './mockMcpStream'; const b = block('aikit-playground'); @@ -24,61 +25,14 @@ type ApiMessage = { content: string; }; -/** - * Responses API event type - */ -type ResponseEvent = { - event: string; - data: { - type: string; - delta?: string; - text?: string; - // eslint-disable-next-line @typescript-eslint/no-explicit-any - [key: string]: any; - }; - error?: string; +/** Stream options for useOpenAIStreamAdapter: initialMessages and assistantMessageId are set when the request starts. */ +type StreamOptions = { + initialMessages: TChatMessage[]; + assistantMessageId: string; }; -/** - * Parses data from SSE stream - */ -function parseStreamData(line: string): ResponseEvent | null { - try { - return JSON.parse(line.slice(6)) as ResponseEvent; - } catch { - return null; - } -} - -/** - * Extracts text from Responses API event - */ -function extractTextFromEvent(event: ResponseEvent): string | null { - // Handle text delta events - const isOutputTextDelta = - event.event === 'response.output_text.delta' || - event.data?.type === 'response.output_text.delta'; - - if (isOutputTextDelta) { - return event.data?.delta || null; - } - - // Handle content part events - const isContentPartDelta = - event.event === 'response.content_part.delta' || - event.data?.type === 'response.content_part.delta'; - - if (isContentPartDelta) { - return event.data?.delta || null; - } - - // Fallback for legacy format - if (event.event === 'content' && event.data?.content) { - return event.data.content; - } - - return null; -} +/** Stream source: fetch Response or mock AsyncIterable (e.g. ?mockMcp=1 for MCP testing). */ +type StreamSource = Response | AsyncIterable> | null; /** * Extracts text content from chat message @@ -88,7 +42,6 @@ function getMessageContent(message: TChatMessage): string { return message.content; } - // For assistant messages, content can be of different types const {content} = message; if (typeof content === 'string') { @@ -114,48 +67,96 @@ function getMessageContent(message: TChatMessage): string { return ''; } +const getChatStatus = (hasResponse: boolean, isFetching: boolean, status: string): ChatStatus => { + if (!hasResponse) { + return isFetching ? 'submitted' : 'ready'; + } + if (status === 'streaming') return 'streaming'; + if (status === 'error') return 'error'; + return 'ready'; +}; + /** * AIKit Playground component - * Uses /api/chat endpoint for OpenAI interaction + * Uses /api/chat endpoint for OpenAI interaction and useOpenAIStreamAdapter for response handling */ export const AikitPlayground = memo(() => { const {t} = useTranslation('aikit'); const [messages, setMessages] = useState([]); - const [status, setStatus] = useState('ready'); const [controller, setController] = useState(null); + const [isFetching, setIsFetching] = useState(false); + const [streamSource, setStreamSource] = useState(null); + const [streamOptions, setStreamOptions] = useState(null); + + const streamResult = useOpenAIStreamAdapter(streamSource, { + initialMessages: streamOptions?.initialMessages ?? [], + assistantMessageId: streamOptions?.assistantMessageId ?? 'assistant-idle', + }); + + const hasResponse = Boolean(streamSource); + const displayMessages = hasResponse ? streamResult.messages : messages; + const chatStatus = getChatStatus(hasResponse, isFetching, streamResult.status); + + // Commit stream result to messages when stream ends (done or error) + useEffect(() => { + if ( + streamSource === null || + (streamResult.status !== 'done' && streamResult.status !== 'error') + ) { + return; + } + // Drop assistant messages with empty content (stream adapter may leave a placeholder) + const committed = streamResult.messages.filter( + (msg) => msg.role !== 'assistant' || getMessageContent(msg) !== '', + ); + setMessages(committed); + setStreamSource(null); + setStreamOptions(null); + }, [streamSource, streamResult.status, streamResult.messages]); /** * Message send handler - * Supports streaming responses from API + * Passes fetch Response to useOpenAIStreamAdapter; all stream parsing is done in the hook */ const handleSendMessage = useCallback( async (data: TSubmitData) => { - // Add user message const userMessage: TUserMessage = { id: Date.now().toString(), role: 'user', content: data.content, }; - setMessages((prev) => [...prev, userMessage]); + const initialMessages: TChatMessage[] = [...messages, userMessage]; + const assistantMessageId = (Date.now() + 1).toString(); + + setMessages(initialMessages); + setIsFetching(true); - // Start streaming - setStatus('submitted'); const abortController = new AbortController(); setController(abortController); + // Mock stream with MCP events for testing when real API has no MCP (?mockMcp=1) + if (isMockMcpEnabled()) { + setStreamSource(createMockMcpStream()); + setStreamOptions({initialMessages, assistantMessageId}); + setIsFetching(false); + setController(null); + return; + } + try { - // Build message history for API const apiMessages: ApiMessage[] = [ { role: 'system', content: 'You are a helpful AI assistant. Respond in the same language as the user message.', }, - ...messages.map((msg) => ({ - role: msg.role as 'user' | 'assistant', - content: getMessageContent(msg), - })), + ...messages + .filter((msg) => msg.role !== 'assistant' || getMessageContent(msg) !== '') + .map((msg) => ({ + role: msg.role as 'user' | 'assistant', + content: getMessageContent(msg), + })), {role: 'user' as const, content: data.content}, ]; @@ -173,117 +174,44 @@ export const AikitPlayground = memo(() => { throw new Error(`API error: ${response.status}`); } - const reader = response.body?.getReader(); - if (!reader) { + if (!response.body) { throw new Error('No response body'); } - const decoder = new TextDecoder(); - - // Assistant message ID for updates - const assistantMessageId = (Date.now() + 1).toString(); - - // Add empty assistant message - setMessages((prev) => [ - ...prev, - { - id: assistantMessageId, - role: 'assistant', - content: '', - } as TAssistantMessage, - ]); - - let buffer = ''; - // Local variable for text accumulation - let accumulatedContent = ''; - - setStatus('streaming'); - // Read data stream - // eslint-disable-next-line no-constant-condition - 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 trimmedLine = line.trim(); - - if (trimmedLine === 'data: [DONE]') { - continue; - } - - if (!trimmedLine.startsWith('data: ')) { - continue; - } - - const parsedEvent = parseStreamData(trimmedLine); - if (!parsedEvent) continue; - - if (parsedEvent.error) { - throw new Error(parsedEvent.error); - } - - const textContent = extractTextFromEvent(parsedEvent); - if (textContent) { - // Accumulate text in local variable - accumulatedContent += textContent; - // Copy value for closure - const newContent = accumulatedContent; - - setMessages((prev) => { - return prev.map((msg): TChatMessage => { - if (msg.id === assistantMessageId && msg.role === 'assistant') { - return { - ...msg, - content: newContent, - } as TAssistantMessage; - } - return msg; - }); - }); - } - } - } + // Set stream and options in one tick so the hook never sees new options with stream=null + setStreamSource(response); + setStreamOptions({initialMessages, assistantMessageId}); } catch (error) { if ((error as Error).name !== 'AbortError') { // eslint-disable-next-line no-console console.error('Chat error:', error); - // Add error message const errorMessage: TAssistantMessage = { id: (Date.now() + 2).toString(), role: 'assistant', content: t('error.apiError'), }; - setMessages((prev) => { - // Remove empty assistant message if exists const filtered = prev.filter( (msg) => msg.role !== 'assistant' || msg.content !== '', ); return [...filtered, errorMessage]; }); } + setStreamSource(null); + setStreamOptions(null); } finally { - setStatus('ready'); + setIsFetching(false); setController(null); } }, [messages, t], ); - /** - * Request cancellation handler - */ const handleCancel = useCallback(async () => { controller?.abort(); - setStatus('ready'); + setStreamSource(null); + setStreamOptions(null); }, [controller]); return ( @@ -298,10 +226,10 @@ export const AikitPlayground = memo(() => { description: t('welcome.description'), }} showActionsOnHover={true} - messages={messages} + messages={displayMessages} onSendMessage={handleSendMessage} onCancel={handleCancel} - status={status} + status={chatStatus} i18nConfig={{ header: { defaultTitle: t('chat.title'), diff --git a/src/components/AikitPlayground/mockMcpStream.ts b/src/components/AikitPlayground/mockMcpStream.ts new file mode 100644 index 000000000000..159e8d9aa4bc --- /dev/null +++ b/src/components/AikitPlayground/mockMcpStream.ts @@ -0,0 +1,244 @@ +/** + * Mock OpenAI Responses API stream that includes all event types supported by useOpenAIStreamAdapter. + * Use for manual testing when the real API does not return MCP (e.g. in AikitPlayground). Enable with ?mockMcp=1. + * + * Covered events (aligned with aikit useOpenAIStreamAdapter): + * - Text: response.output_text.delta, response.content_part.delta + * - Reasoning: response.output_item.added (reasoning), response.reasoning_text.delta/done, response.output_item.done (reasoning) + * - MCP: response.output_item.added (mcp_call), response.mcp_call.in_progress/completed/failed, response.output_item.done (mcp_call) + * - MCP approval: response.output_item.added (mcp_approval_request) + * - File search: response.output_item.added (file_search_call), response.file_search_call.searching/completed, response.output_item.done + * - Function call: response.output_item.added (function_call), response.output_item.done (function_call) + * - End: response.done (or response.completed / response.failed / error via endWith option) + * Optional end variants (use createMockMcpStream({ endWith: 'failed' }) or endWith: 'error'): response.failed, error + */ + +// Event types compatible with useOpenAIStreamAdapter (OpenAIStreamEventLike) +type MockStreamEvent = + | {type: 'response.output_text.delta'; delta: string} + | {type: 'response.content_part.delta'; delta: string} + | { + type: 'response.output_item.added'; + item?: { + type: string; + id?: string; + call_id?: string; + name?: string; + server_label?: string; + arguments?: string; + }; + } + | {type: 'response.mcp_call.in_progress'; item_id: string} + | {type: 'response.mcp_call.completed'; item_id: string} + | {type: 'response.mcp_call.failed'; item_id: string; error?: string} + | {type: 'response.file_search_call.searching'; item_id: string} + | {type: 'response.file_search_call.in_progress'; item_id: string} + | {type: 'response.file_search_call.completed'; item_id: string} + | {type: 'response.reasoning_text.delta'; item_id: string; delta: string} + | {type: 'response.reasoning_text.done'; item_id: string; text: string} + | { + type: 'response.output_item.done'; + item?: { + type: string; + id?: string; + call_id?: string; + name?: string; + status?: string; + output?: string; + result?: string; + error?: string; + content?: Array<{type: string; text?: string}>; + summary?: Array<{type: string; text?: string}>; + }; + } + | {type: 'response.done'} + | {type: 'response.completed'} + | {type: 'response.failed'} + | {type: 'error'; error?: string}; + +/** Optional delay between events (ms) so streaming is visible in UI. Set 0 to disable. */ +const EVENT_DELAY_MS = 80; + +function delay(ms: number): Promise { + return ms > 0 ? new Promise((r) => setTimeout(r, ms)) : Promise.resolve(); +} + +export type MockStreamEndWith = 'done' | 'completed' | 'failed' | 'error'; + +export async function* createMockMcpStream(options?: { + delayMs?: number; + /** End stream with response.done (default), response.completed, response.failed, or error event. */ + endWith?: MockStreamEndWith; +}): AsyncIterable { + const delayMs = options?.delayMs ?? EVENT_DELAY_MS; + const endWith = options?.endWith ?? 'done'; + + // 1) Initial text + yield {type: 'response.output_text.delta', delta: 'Checking files... '}; + await delay(delayMs); + + // 2) Reasoning (thinking) block + yield { + type: 'response.output_item.added', + item: {type: 'reasoning', id: 'reasoning-mock-1'}, + }; + await delay(delayMs); + yield { + type: 'response.reasoning_text.delta', + item_id: 'reasoning-mock-1', + delta: 'Analyzing request.\n', + }; + await delay(delayMs); + yield { + type: 'response.reasoning_text.delta', + item_id: 'reasoning-mock-1', + delta: 'Searching for config.', + }; + await delay(delayMs); + yield { + type: 'response.reasoning_text.done', + item_id: 'reasoning-mock-1', + text: 'Analyzing request.\nSearching for config.', + }; + await delay(delayMs); + yield { + type: 'response.output_item.done', + item: { + type: 'reasoning', + id: 'reasoning-mock-1', + content: [{type: 'reasoning_text', text: 'Analyzing request.\nSearching for config.'}], + }, + }; + await delay(delayMs); + + // 3) MCP call (success) + yield { + type: 'response.output_item.added', + item: { + type: 'mcp_call', + id: 'mcp-mock-1', + name: 'file_search', + server_label: 'Files', + }, + }; + await delay(delayMs); + yield {type: 'response.mcp_call.in_progress', item_id: 'mcp-mock-1'}; + await delay(delayMs); + yield {type: 'response.mcp_call.completed', item_id: 'mcp-mock-1'}; + await delay(delayMs); + yield { + type: 'response.output_item.done', + item: { + type: 'mcp_call', + id: 'mcp-mock-1', + status: 'completed', + output: '{"path":"/project/config.json"}', + }, + }; + await delay(delayMs); + + yield {type: 'response.output_text.delta', delta: 'Found config. '}; + await delay(delayMs); + // Alternative text delta (adapter supports both output_text.delta and content_part.delta) + yield {type: 'response.content_part.delta', delta: '(via content_part)\n'}; + await delay(delayMs); + + // 4) File search call (tool type like web_search_call, code_interpreter_call, etc.) + yield { + type: 'response.output_item.added', + item: {type: 'file_search_call', id: 'fs-mock-1'}, + }; + await delay(delayMs); + yield {type: 'response.file_search_call.searching', item_id: 'fs-mock-1'}; + await delay(delayMs); + yield {type: 'response.file_search_call.completed', item_id: 'fs-mock-1'}; + await delay(delayMs); + yield { + type: 'response.output_item.done', + item: {type: 'file_search_call', id: 'fs-mock-1', status: 'completed', output: '[]'}, + }; + await delay(delayMs); + + // 5) MCP approval request (waiting confirmation) + yield { + type: 'response.output_item.added', + item: { + type: 'mcp_approval_request', + id: 'mcp-approval-mock-1', + name: 'roll', + server_label: 'Dice', + arguments: '{"diceRollExpression":"2d4+1"}', + }, + }; + await delay(delayMs); + + // 6) MCP call (failed) + yield { + type: 'response.output_item.added', + item: { + type: 'mcp_call', + id: 'mcp-mock-failed', + name: 'failing_tool', + server_label: 'Test', + }, + }; + await delay(delayMs); + yield {type: 'response.mcp_call.failed', item_id: 'mcp-mock-failed', error: 'Tool unavailable'}; + await delay(delayMs); + yield { + type: 'response.output_item.done', + item: { + type: 'mcp_call', + id: 'mcp-mock-failed', + status: 'failed', + error: 'Tool unavailable', + }, + }; + await delay(delayMs); + + // 7) Function call (non-MCP tool) + yield { + type: 'response.output_item.added', + item: { + type: 'function_call', + call_id: 'fn-mock-1', + name: 'get_weather', + arguments: '{"city":"Moscow"}', + }, + }; + await delay(delayMs); + yield { + type: 'response.output_item.done', + item: { + type: 'function_call', + call_id: 'fn-mock-1', + name: 'get_weather', + status: 'completed', + output: '{"temp":5,"unit":"celsius"}', + }, + }; + await delay(delayMs); + + yield { + type: 'response.output_text.delta', + delta: 'Summary: config found, one tool failed, weather fetched.', + }; + await delay(delayMs); + + // 8) End event + if (endWith === 'error') { + yield {type: 'error', error: 'Mock stream error'}; + } else if (endWith === 'failed') { + yield {type: 'response.failed'}; + } else if (endWith === 'completed') { + yield {type: 'response.completed'}; + } else { + yield {type: 'response.done'}; + } +} + +/** Returns true when mock MCP stream should be used (e.g. ?mockMcp=1). */ +export function isMockMcpEnabled(): boolean { + if (typeof window === 'undefined') return false; + return new URLSearchParams(window.location.search).get('mockMcp') === '1'; +}