From 951ceaa05bb4df1438b82d49c19b3a5457120e66 Mon Sep 17 00:00:00 2001 From: Shivering Ocelot Date: Wed, 25 Jun 2025 20:40:11 -0500 Subject: [PATCH] Fix various agent-chat bugs, add qdrant dependency, WIP memory processing for conversation agent --- .gitignore | 5 +++- backend/agent.py | 50 +++++++++++++++++++++++++++++----------- backend/app.py | 43 +++++++++++++++++++--------------- backend/chat.py | 2 +- backend/requirements.txt | 2 +- testing/suite.js | 2 +- 6 files changed, 68 insertions(+), 36 deletions(-) diff --git a/.gitignore b/.gitignore index 809d61c..162c071 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,7 @@ .venv .env __pycache__ -horde.db \ No newline at end of file +horde.db + +/test-data +/local-data \ No newline at end of file diff --git a/backend/agent.py b/backend/agent.py index a05cfb0..370478c 100644 --- a/backend/agent.py +++ b/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) - self.process_thread(old_thread) + old_thread = chat_view.get_thread(saved_context.current_thread_id) + 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 - pass + # TODO: somewhere here it must be known if the thread was seen before: the same memory should not be stored multiple times + 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): diff --git a/backend/app.py b/backend/app.py index 3733adc..b4159cd 100644 --- a/backend/app.py +++ b/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 - if sender != 'user': - chat_owner = session.execute(select(db.Chat.user_id).where(db.Chat.id == chat_id)).scalar() - self.notify_message_to_user(chat_owner, message) + # Notify chat owner + if sender != 'user': + chat_owner = session.execute(select(db.Chat.user_id).where(db.Chat.id == chat_id)).scalar() + self.notify_message_to_user(chat_owner, message) - # Notify participant agents - agent_ids = session.execute(select(db.Membership.agent_id).where(db.Membership.chat_id == chat_id)).scalars().all() - for agent_id in agent_ids: - if agent_id != sender: - self.notify_message_to_agent(agent_id, message) + # Notify participant agents + agent_ids = session.execute(select(db.Membership.agent_id).where(db.Membership.chat_id == chat_id)).scalars().all() + for agent_id in agent_ids: + if agent_id != sender: + self.notify_message_to_agent(agent_id, message) - return { - 'id': message.id, - 'chatId': message.chat_id, - 'time': message.time, - 'sender': message.sender, - 'kind': message.kind, - 'content': message.content, - } + return { + 'id': message.id, + 'chatId': message.chat_id, + 'time': message.time, + 'sender': message.sender, + 'kind': message.kind, + 'content': message.content, + } def get_chat_context(self, chat_id): return ChatContext(self.get_chat_messages(chat_id)) diff --git a/backend/chat.py b/backend/chat.py index 4867f1e..9c343ce 100644 --- a/backend/chat.py +++ b/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) diff --git a/backend/requirements.txt b/backend/requirements.txt index 1b9ac4b..d9edbb5 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -4,4 +4,4 @@ quart-cors sqlalchemy nest_asyncio replicate -chromadb \ No newline at end of file +qdrant-client[fastembed] \ No newline at end of file diff --git a/testing/suite.js b/testing/suite.js index fa667c5..3ea62aa 100644 --- a/testing/suite.js +++ b/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`);