Spaces:
Running
Running
File size: 3,352 Bytes
f2eba97 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | 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}"
)
|