Fix various agent-chat bugs, add qdrant dependency, WIP memory processing for conversation agent

.gitignore

@@ -1,4 +1,7 @@
 .venv
 .env
 __pycache__
-horde.db
+horde.db
+
+/test-data
+/local-data

backend/agent.py

@@ -96,31 +96,30 @@ class ConversationAgent(Agent):
         return response
 
     async def received_message(self, message):
-        def sender_name(message):
-            sender = message.sender
-            is_digit = isinstance(sender, (int, str)) and str(sender).isdigit()
-            is_self = is_digit and str(sender) == str(self.id)
-            return self.name if is_self else sender
 
         # TODO: before retrieving the chat context, acquire a lock for it
 
         saved_context = ThreadContext(self.app.get_agent_resource(self.id, 'last-thread-context', message.chat_id))
         chat_view = self.app.get_chat_view(message.chat_id)
+        print("ChatView:\n" + chat_view.debug_string())
 
         if message.thread_id != saved_context.current_thread_id:
             # the incoming message is not from the same thread that the agent was last aware of
             # store details about the old thread in the agent's long term memory
-            old_thread = chat_view.thread_map.get(message.thread_id)
+            old_thread = chat_view.get_thread(saved_context.current_thread_id)
-            self.process_thread(old_thread)
+            await self.process_thread(old_thread)
 
             # TODO: This is a good place to store the new context, its fine that the agent hasnt responded
 
-        current_thread = chat_view.thread_map.get(message.thread_id)
+        current_thread = chat_view.get_thread(message.thread_id)
 
-        chat_summary = "\n".join(f"[{sender_name(message)}]: {message.content}" for message in current_thread.messages)
+        # REMOVE THIS. Only for testing, while the function only prints...
+        await self.process_thread(current_thread)
+
+        chat_summary = self.serialize_thread_history(current_thread)
         messages = [
             LlamaMessage("system", f"{self.system_prompt}\nYou are given the context of the conversation, respond with your next message in the conversation."),
-            LlamaMessage("user", f"Recent Messages:\n{chat_summary}")
+            LlamaMessage("user", f"Conversation History:\n{chat_summary}")
         ]
 
         response = await replicate_llama_completion(self.app, messages)
@@ -128,15 +127,39 @@ class ConversationAgent(Agent):
             chat_id=message.chat_id,
             sender=self.id,
             content=response,
+            thread_id=current_thread.id,
         )
 
         # TODO: drop the chat context lock
+
+    # Encode a thread's messages as a string to pass to the agent
+    def serialize_thread_history(self, thread):
+        def sender_name(message):
+            sender = message.sender
+            is_digit = isinstance(sender, (int, str)) and str(sender).isdigit()
+            is_self = is_digit and str(sender) == str(self.id)
+            return self.name if is_self else sender
+        
+        return "\n".join(f"[{sender_name(message)}]: {message.content}" for message in thread.messages)
     
     # Compute thread summnary and concrete memories, then save the progress done
-    def process_thread(self, thread):
+    async def process_thread(self, thread):
         # TODO: extract memories from the thread messages, then store them in the agent's vector memory db
-        # TODO: somewhere here it must be known if the thread was seen before, the same memory should not be stored multiple times
+        # TODO: somewhere here it must be known if the thread was seen before: the same memory should not be stored multiple times
-        pass
+        print(f"PROCESSING THREAD: {thread.id}")
+
+        instruction = "Your task is to identify facts from a conversation that its participants would remember in the long term. Each line in your response is a different memory, if there's nothing relevant to remember, simply give an empty response."
+        intro = f"Identify the facts that {self.name} will remember long term after the conversation. Following the line break is the personality prompt for {self.name}."
+
+        chat_history = self.serialize_thread_history(thread)
+        messages = [
+            LlamaMessage("system", f"{instruction}\n{intro}\n\n{self.system_prompt}"),
+            LlamaMessage("user", f"Conversation History:\n{chat_history}")
+        ]
+        response = await replicate_llama_completion(self.app, messages)
+        memories = [line for line in response.split('\n') if line.strip()]
+        
+        print("New agent memories:", memories)
 
 class ToolUsingAgent:
     identifier = "tool-using"
@@ -308,6 +331,7 @@ class Llama8bAgent(Agent):
             chat_id=message.chat_id,
             sender=self.id,
             content=response,
+            thread_id=message.thread_id,
         )
 
 class BasicAgent(Llama8bAgent):

backend/app.py

@@ -7,12 +7,13 @@ import os
 from sqlalchemy.orm import Session
 from sqlalchemy.sql.expression import select
 from quart import Quart, Response, request, send_from_directory, jsonify, websocket
+from qdrant_client import QdrantClient
 import replicate
 
 import database as db
 import agent as agent_mod
 from server import start_server
-from chat import ChatView
+from chat import ChatView, Message
 
 MESSAGE_PAGE_SIZE = 32
 
@@ -24,11 +25,14 @@ class ChatContext:
 class App:
     def __init__(self, data_directory, replicate_api_token=None):
         self.data_directory = data_directory
-        self.sql_engine = db.create_sql_engine(self.data_directory)
         self.user_connections = {}
         self.remote_connections = {}
         self.streams = {}
 
+        # Initialize Clients
+        self.sql_engine = db.create_sql_engine(self.data_directory)
+        self.qdrant_client = QdrantClient(path=f"{self.data_directory}/qdrant_data")
+
         if replicate_api_token:
             self.replicate_client = replicate.Client(api_token=replicate_api_token)
         else:
@@ -246,26 +250,27 @@ class App:
 
             session.add(message)
             session.commit()
+            session.refresh(message)
 
-            # Notify chat owner
+        # Notify chat owner
-            if sender != 'user':
+        if sender != 'user':
-                chat_owner = session.execute(select(db.Chat.user_id).where(db.Chat.id == chat_id)).scalar()
+            chat_owner = session.execute(select(db.Chat.user_id).where(db.Chat.id == chat_id)).scalar()
-                self.notify_message_to_user(chat_owner, message)
+            self.notify_message_to_user(chat_owner, message)
 
-            # Notify participant agents
+        # Notify participant agents
-            agent_ids = session.execute(select(db.Membership.agent_id).where(db.Membership.chat_id == chat_id)).scalars().all()
+        agent_ids = session.execute(select(db.Membership.agent_id).where(db.Membership.chat_id == chat_id)).scalars().all()
-            for agent_id in agent_ids:
+        for agent_id in agent_ids:
-                if agent_id != sender:
+            if agent_id != sender:
-                    self.notify_message_to_agent(agent_id, message)
+                self.notify_message_to_agent(agent_id, message)
 
-            return {
+        return {
-                'id': message.id,
+            'id': message.id,
-                'chatId': message.chat_id,
+            'chatId': message.chat_id,
-                'time': message.time,
+            'time': message.time,
-                'sender': message.sender,
+            'sender': message.sender,
-                'kind': message.kind,
+            'kind': message.kind,
-                'content': message.content,
+            'content': message.content,
-            }
+        }
 
     def get_chat_context(self, chat_id):
         return ChatContext(self.get_chat_messages(chat_id))

backend/chat.py

@@ -87,7 +87,7 @@ class ChatView:
                         thread = Thread(record.thread_id, self.chat_id)
                         self.thread_map[thread.id] = thread
                         self.threads.insert(0, thread)
-                # This is the oldest message processed, add it to the start of the chat
+                # This is the oldest message so far, add it to the start of the chat
                 message = Message(record.id, record.content, record.sender, record.time)
                 thread.messages.insert(0, message)
             

backend/requirements.txt

@@ -4,4 +4,4 @@ quart-cors
 sqlalchemy
 nest_asyncio
 replicate
-chromadb
+qdrant-client[fastembed]

testing/suite.js

@@ -1,6 +1,6 @@
 const WebSocket = require('ws');
 
-const USE_LLM = false;
+const USE_LLM = true;
 
 const hostname = 'localhost:5000';
 const ws = new WebSocket(`ws://${hostname}/ws`);