Iterate

api/src/agents/channels.txt

@@ -2,4 +2,5 @@ stopChat :chatId
 agentStream:chatId:chunk chunk
 agentStream:chatId:end 1
 agentStream:chatId:update payload
-OK hset subscriptions chatId payload
+hset subscriptions chatId payload
+indexStream:indexId

api/src/controllers/discovery.js

@@ -75,23 +75,22 @@ export const chat = async (req, res, next) => {
   }
 };
 
-const handleMessage = async (query, message, channel) => {
+const handleMessage = async (query, payload) => {
   try {
     const chatRequest = {
       prompt: `
-      For the given information, determine whether the new field might be relevant to the user's intent based on the question. You are a personal assistant whose task is to classify whether the new information is relevant or not.
-      If the information field might be NOT relevant, respond with "NOT_RELEVANT", and don't output anything else. If the information field might be relevant and new to the question, output an update in assistant tone for the user based on the information field.
-      You can use a conversational format (hypertext, links, etc.) in your message to extend the metadata of the information.
+      Think step by step.
+      For the given information, determine whether the new field might be relevant to the user's intent based on the question. You are a relevancy extractor whose task is to classify whether the new information is relevant or not.
+      If the information field might be NOT relevant, respond with "NOT_RELEVANT", and don't output anything else. If the information field might be relevant and new to the question, summarize the relevancy reason based on the information field.
       Do not add any HTML tags or title fields. Use links when useful. "Cast" means Farcaster Cast, so write your message accordingly.
-      Also, try to use a personal assistant tone, don't sell.
       ----------------
       QUESTION: {query}
       ----------------
-      INFORMATION: {message}
+      INFORMATION: {information}
       `,
-      input: {
+      inputs: {
         query,
-        message,
+        information: JSON.stringify(payload),
       },
     };
 
@@ -117,7 +116,7 @@ const handleMessage = async (query, message, channel) => {
 
 export const updates = async (req, res, next) => {
   const definition = req.app.get("runtimeDefinition");
-  const { query, sources } = req.body;
+  const { query, sources } = req.query;
 
   const session = await DIDSession.fromSession(req.query.session);
   if (!session || !session.isAuthorized()) {
@@ -132,13 +131,18 @@ export const updates = async (req, res, next) => {
   const reqIndexIds = await flattenSources(sources, didService);
   const reqIndexChannels = reqIndexIds.map((id) => `indexStream:${id}`);
 
-  await pubSubClient.subscribe(reqIndexChannels, async (message, channel) => {
-    const response = await handleMessage(query, message, channel);
-
+  await pubSubClient.subscribe(reqIndexChannels, async (payload, channel) => {
+    const response = await handleMessage(query, payload);
+    const indexId = channel.replace(`indexStream:`, "");
     if (response) {
-      const channelType = channel.replace(`indexStream:${reqIndexIds[0]}:`, "");
       res.write(
-        `data: ${JSON.stringify({ channel: channelType, data: JSON.parse(response) })}\n\n`,
+        `data: ${JSON.stringify({
+          indexId,
+          data: {
+            relevance: response,
+            node: JSON.parse(payload),
+          },
+        })}\n\n`,
       );
     }
   });

api/src/libs/indexer.js

@@ -304,7 +304,7 @@ class Indexer {
 
     if (embedding.index.controllerDID.id !== embedding.controllerDID.id) {
       logger.warn(
-        `Step [0]: Embedding is unauthorized to index: ${JSON.stringify(indexItem)}`,
+        `Step [0]: Embedding is unauthorized to embedding: ${JSON.stringify(embedding)}`,
       );
       return;
     }
@@ -339,6 +339,11 @@ class Indexer {
         payload,
       );
 
+      await redis.publish(
+        `indexStream:${embedding.index.id}`,
+        JSON.stringify(itemStream.content),
+      );
+
       // todo send fluence as well.
       logger.info(
         `Step [3]: Index ${embedding.indexId} with Item ${embedding.itemId} indexed with it's content and embeddings`,

api/src/packages/api.js

@@ -380,7 +380,7 @@ app.post(
 
 app.get(
   "/discovery/updates",
-  validator.body(
+  validator.query(
     Joi.object({
       sources: Joi.array().items(Joi.string()).required(),
       query: Joi.string().min(1).optional(),

sdk/js/README.md

@@ -119,3 +119,32 @@ const stopListening = indexClient.listenToConversationUpdates(
   handleError,
 );
 ```
+
+### Listening to Index Updates
+
+The Index Client SDK allows you to listen for updates to miltiple indexes in real-time. This is useful for applications that need to react to new data events, using natural language.
+
+Here is an example of how you can use the `listenToIndexUpdates` method to handle real-time updates in a conversation:
+
+```typescript
+const sources = ["did:pkh:eip155:1:0x1b9Aceb609a62bae0c0a9682A9268138Faff4F5f"];
+
+const query = "if it is relevant to decentralized AI";
+
+const handleMessage = (data: any) => {
+  console.log("New event received:", data);
+  // Handle the new message data
+};
+
+const handleError = (error: any) => {
+  console.error("Error receiving updates:", error);
+  // Handle the error
+};
+
+const stopListening = indexClient.listenToIndexUpdates(
+  sources,
+  query
+  handleMessage,
+  handleError,
+);
+```

sdk/js/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@indexnetwork/sdk",
-  "version": "0.1.3",
+  "version": "0.1.4",
   "main": "dist/indexclient.cjs.js",
   "module": "dist/indexclient.es.js",
   "types": "dist/index.d.ts",

sdk/js/src/index.ts

@@ -394,6 +394,40 @@ export default class IndexClient {
       },
     );
   }
+
+  public listenToIndexUpdates(
+    sources: string[],
+    query: string,
+    handleMessage: (data: any) => void,
+    handleError: (error: any) => void,
+  ) {
+    const queryParams = new URLSearchParams();
+
+    const params = { query, sources, session: this.session };
+    for (const key in params) {
+      if (params.hasOwnProperty(key)) {
+        queryParams.append(key, params[key]);
+      }
+    }
+
+    const eventUrl = `${this.baseUrl}/discovery/updates${queryParams.toString()}`;
+    const eventSource = new EventSource(eventUrl);
+
+    eventSource.onmessage = (event) => {
+      console.log("Received message from server", event.data);
+      handleMessage(event.data);
+    };
+
+    eventSource.onerror = (err) => {
+      console.error("EventSource failed:", err);
+      handleError(err);
+      eventSource.close();
+    };
+
+    return () => {
+      eventSource.close();
+    };
+  }
 }
 
 export { IndexVectorStore, IndexConfig };