Spaces:
Sleeping
Sleeping
| from typing import List, Dict | |
| from groq import Groq | |
| from agents.base_agent import BaseAgent | |
| from memory.sqlite_memory import load_history, get_session_summary, count_turns | |
| import config | |
| from utils.logger import get_logger | |
| logger = get_logger(__name__) | |
| class MemoryAgent(BaseAgent): | |
| SYSTEM_PROMPT = """You are a helpful assistant with access to the conversation history below. | |
| The user is asking about something that was previously discussed. | |
| Your job: | |
| - Recall relevant details from the conversation history | |
| - Provide a clear, accurate summary or answer based on what was discussed | |
| - If a topic was NOT discussed, say so clearly: "We haven't discussed that yet." | |
| - Be conversational and natural — you remember the user personally | |
| - Format your answer with markdown. IMPORTANT: Always include a space after bolding (e.g., **Hello!** nice) to ensure the UI renders correctly. | |
| Conversation History: | |
| {history}""" | |
| def __init__(self): | |
| super().__init__(name="Memory Agent") | |
| self._client = Groq(api_key=config.GROQ_API_KEY) | |
| logger.info("MemoryAgent ready.") | |
| def run( | |
| self, | |
| query: str, | |
| context: str = "", | |
| history: List[Dict[str, str]] = None, | |
| session_id: str = "", | |
| ) -> str: | |
| logger.info(f"MemoryAgent processing: '{query[:80]}'") | |
| total_turns = count_turns(session_id) if session_id else 0 | |
| if total_turns == 0 and not history: | |
| logger.warning("MemoryAgent: No conversation history found.") | |
| return ( | |
| "🧠 **No conversation history yet.**\n\n" | |
| "We haven't talked about anything yet in this session. " | |
| "Ask me a question and I'll remember it for you!" | |
| ) | |
| db_history_text = get_session_summary(session_id) if session_id else "" | |
| in_memory_text = "" | |
| if history: | |
| lines = [f"{t['role'].capitalize()}: {t['content']}" for t in history] | |
| in_memory_text = "\n".join(lines) | |
| full_history = db_history_text or in_memory_text | |
| if db_history_text and in_memory_text and db_history_text != in_memory_text: | |
| full_history = f"{db_history_text}\n\n[Recent turns:]\n{in_memory_text}" | |
| logger.info(f" → Loaded {total_turns} turns from SQLite for session '{session_id}'.") | |
| system_msg = self.SYSTEM_PROMPT.format(history=full_history) | |
| messages = [ | |
| {"role": "system", "content": system_msg}, | |
| {"role": "user", "content": query}, | |
| ] | |
| try: | |
| response = self._client.chat.completions.create( | |
| model=config.GROQ_MODEL_NAME, | |
| messages=messages, | |
| temperature=0.3, | |
| max_tokens=config.GROQ_MAX_TOKENS, | |
| ) | |
| answer = response.choices[0].message.content.strip() | |
| logger.info(" → Memory recall response generated.") | |
| answer += ( | |
| f"\n\n---\n" | |
| f"*🧠 I remember {total_turns} message(s) from our conversation.*" | |
| ) | |
| return answer | |
| except Exception as exc: | |
| logger.error(f"MemoryAgent LLM call failed: {exc}") | |
| return ( | |
| f"I encountered an error recalling our conversation. " | |
| f"Please try again.\n\nError: {exc}" | |
| ) | |