Merge branch 'pr/703'

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.

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,6 +127,7 @@ export const POST = async (req: Request) => {
       [],
     );
 
+    if (!body.stream) {
       return new Promise(
         (
           resolve: (value: Response) => void,
@@ -133,7 +136,7 @@ export const POST = async (req: Request) => {
           let message = '';
           let sources: any[] = [];
 
-        emitter.on('data', (data) => {
+          emitter.on('data', (data: string) => {
             try {
               const parsedData = JSON.parse(data);
               if (parsedData.type === 'response') {
@@ -143,7 +146,10 @@ export const POST = async (req: Request) => {
               }
             } catch (error) {
               reject(
-              Response.json({ message: 'Error parsing data' }, { status: 500 }),
+                Response.json(
+                  { message: 'Error parsing data' },
+                  { status: 500 },
+                ),
               );
             }
           });
@@ -152,13 +158,106 @@ export const POST = async (req: Request) => {
             resolve(Response.json({ message, sources }, { status: 200 }));
           });
 
-        emitter.on('error', (error) => {
+          emitter.on('error', (error: any) => {
             reject(
-            Response.json({ message: 'Search error', error }, { status: 500 }),
+              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[] = [];
+
+        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') {
+              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) {
+            controller.error(error);
+          }
+        });
+
+        emitter.on('end', () => {
+          if (signal.aborted) return;
+
+          controller.enqueue(
+            encoder.encode(
+              JSON.stringify({
+                type: 'done',
+              }) + '\n',
+            ),
+          );
+          controller.close();
+        });
+
+        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(