diff --git a/docs/API/SEARCH.md b/docs/API/SEARCH.md index 3007901..3a28a78 100644 --- a/docs/API/SEARCH.md +++ b/docs/API/SEARCH.md @@ -32,7 +32,8 @@ The API accepts a JSON object in the request body, where you define the focus mo "history": [ ["human", "Hi, how are you?"], ["assistant", "I am doing well, how can I help you today?"] - ] + ], + "stream": false } ``` @@ -71,11 +72,13 @@ The API accepts a JSON object in the request body, where you define the focus mo ] ``` +- **`stream`** (boolean, optional): When set to `true`, enables streaming responses. Default is `false`. + ### Response The response from the API includes both the final message and the sources used to generate that message. -#### Example Response +#### Standard Response (stream: false) ```json { @@ -100,6 +103,28 @@ The response from the API includes both the final message and the sources used t } ``` +#### Streaming Response (stream: true) + +When streaming is enabled, the API returns a stream of newline-delimited JSON objects. Each line contains a complete, valid JSON object. The response has Content-Type: application/json. + +Example of streamed response objects: + +``` +{"type":"init","data":"Stream connected"} +{"type":"sources","data":[{"pageContent":"...","metadata":{"title":"...","url":"..."}},...]} +{"type":"response","data":"Perplexica is an "} +{"type":"response","data":"innovative, open-source "} +{"type":"response","data":"AI-powered search engine..."} +{"type":"done"} +``` + +Clients should process each line as a separate JSON object. The different message types include: + +- **`init`**: Initial connection message +- **`sources`**: All sources used for the response +- **`response`**: Chunks of the generated answer text +- **`done`**: Indicates the stream is complete + ### Fields in the Response - **`message`** (string): The search result, generated based on the query and focus mode. diff --git a/src/app/api/search/route.ts b/src/app/api/search/route.ts index b980623..d3e98ca 100644 --- a/src/app/api/search/route.ts +++ b/src/app/api/search/route.ts @@ -33,6 +33,7 @@ interface ChatRequestBody { embeddingModel?: embeddingModel; query: string; history: Array<[string, string]>; + stream?: boolean; } export const POST = async (req: Request) => { @@ -48,6 +49,7 @@ export const POST = async (req: Request) => { body.history = body.history || []; body.optimizationMode = body.optimizationMode || 'balanced'; + body.stream = body.stream || false; const history: BaseMessage[] = body.history.map((msg) => { return msg[0] === 'human' @@ -125,40 +127,137 @@ export const POST = async (req: Request) => { [], ); - return new Promise( - ( - resolve: (value: Response) => void, - reject: (value: Response) => void, - ) => { - let message = ''; + if (!body.stream) { + return new Promise( + ( + resolve: (value: Response) => void, + reject: (value: Response) => void, + ) => { + let message = ''; + let sources: any[] = []; + + emitter.on('data', (data: string) => { + try { + const parsedData = JSON.parse(data); + if (parsedData.type === 'response') { + message += parsedData.data; + } else if (parsedData.type === 'sources') { + sources = parsedData.data; + } + } catch (error) { + reject( + Response.json( + { message: 'Error parsing data' }, + { status: 500 }, + ), + ); + } + }); + + emitter.on('end', () => { + resolve(Response.json({ message, sources }, { status: 200 })); + }); + + emitter.on('error', (error: any) => { + reject( + Response.json( + { message: 'Search error', error }, + { status: 500 }, + ), + ); + }); + }, + ); + } + + const encoder = new TextEncoder(); + + const abortController = new AbortController(); + const { signal } = abortController; + + const stream = new ReadableStream({ + start(controller) { let sources: any[] = []; - emitter.on('data', (data) => { + controller.enqueue( + encoder.encode( + JSON.stringify({ + type: 'init', + data: 'Stream connected', + }) + '\n', + ), + ); + + signal.addEventListener('abort', () => { + emitter.removeAllListeners(); + + try { + controller.close(); + } catch (error) {} + }); + + emitter.on('data', (data: string) => { + if (signal.aborted) return; + try { const parsedData = JSON.parse(data); + if (parsedData.type === 'response') { - message += parsedData.data; + controller.enqueue( + encoder.encode( + JSON.stringify({ + type: 'response', + data: parsedData.data, + }) + '\n', + ), + ); } else if (parsedData.type === 'sources') { sources = parsedData.data; + controller.enqueue( + encoder.encode( + JSON.stringify({ + type: 'sources', + data: sources, + }) + '\n', + ), + ); } } catch (error) { - reject( - Response.json({ message: 'Error parsing data' }, { status: 500 }), - ); + controller.error(error); } }); emitter.on('end', () => { - resolve(Response.json({ message, sources }, { status: 200 })); + if (signal.aborted) return; + + controller.enqueue( + encoder.encode( + JSON.stringify({ + type: 'done', + }) + '\n', + ), + ); + controller.close(); }); - emitter.on('error', (error) => { - reject( - Response.json({ message: 'Search error', error }, { status: 500 }), - ); + emitter.on('error', (error: any) => { + if (signal.aborted) return; + + controller.error(error); }); }, - ); + cancel() { + abortController.abort(); + }, + }); + + return new Response(stream, { + headers: { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache, no-transform', + Connection: 'keep-alive', + }, + }); } catch (err: any) { console.error(`Error in getting search results: ${err.message}`); return Response.json(