Spaces:
Sleeping
Sleeping
File size: 5,229 Bytes
3be03dd | 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 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | """
Public facade for the memory system.
Import surface (backward-compatible):
from memory.memory_manager import MemoryManager
from memory.memory_manager import ShortTermMemory # used by eval_agent.py
from memory.memory_manager import WorkingMemory # used by eval_agent.py
from memory.memory_manager import LongTermMemory # used by eval_agent.py
from memory.memory_manager import EpisodicMemory # used by eval_agent.py
"""
from memory.backends import _using_supabase
from memory.stores import (
ShortTermMemory,
WorkingMemory,
LongTermMemory,
EpisodicMemory,
)
# Re-exported so existing callers (eval_agent.py) can still import directly
# from memory.memory_manager without touching their import lines.
__all__ = [
"MemoryManager",
"ShortTermMemory",
"WorkingMemory",
"LongTermMemory",
"EpisodicMemory",
]
class MemoryManager:
"""
The only class the orchestrator talks to.
The orchestrator calls three methods:
build_memory_context() - before analysis starts
save_completed_analysis() - after analysis finishes
reset_short_term() - at the start of each new run
Everything else (which DB, which embedding model, fallback logic)
is hidden inside the four memory classes in memory.stores.
This is the stable-interface / swappable-implementation pattern.
The orchestrator has never changed even as the memory backend went
from SQLite to ChromaDB to Supabase.
"""
def __init__(self):
self.short_term = ShortTermMemory()
self.long_term = LongTermMemory()
self.episodic = EpisodicMemory()
self.working = WorkingMemory()
if _using_supabase():
print("[memory] Backend: Supabase (persistent)")
else:
print("[memory] Backend: SQLite local fallback "
"(set SUPABASE_URL + SUPABASE_KEY for persistence)")
def build_memory_context(self, ticker: str, user_id: str) -> str:
"""
Assemble the full memory block injected into every agent's system prompt.
Order of priority (most -> least specific):
1. User's investment profile (preferences they've set)
2. History of THIS ticker specifically
3. Snippet from the last analysis of this ticker
4. Semantically similar past analyses (cross-ticker)
5. Current conversation context (short-term)
"""
parts = []
profile = self.long_term.get_user_profile(user_id)
if profile:
parts.append(f"## User investment profile:\n{profile}")
history = self.episodic.get_ticker_history(user_id, ticker)
if history:
parts.append(history)
past = self.long_term.recall_past_analyses(ticker, user_id, n=1)
if past:
parts.append(f"## Last analysis of {ticker}:\n{past[0]}")
similar = self.long_term.semantic_search(
f"stock analysis {ticker} {self._sector_hint(ticker)}", user_id, n=2
)
if similar:
parts.append("## Similar past analyses:\n" + "\n---\n".join(similar))
conv = self.short_term.get_context()
if conv:
parts.append(conv)
if not parts:
return ""
return (
"## MEMORY CONTEXT\n"
+ "\n\n".join(parts)
+ "\n\nUse this context to personalise analysis. "
"Reference past calls by date when relevant.\n---\n"
)
def save_completed_analysis(self, ticker: str, user_id: str,
report: str, recommendation: str,
confidence: int, price: float):
"""Persist a completed, human-approved analysis to both memory stores."""
self.long_term.store_analysis(
ticker, user_id, report, recommendation, confidence
)
self.episodic.record(
user_id, ticker, recommendation, confidence, price, report
)
def reset_short_term(self):
"""Call at the start of each new analysis run."""
self.short_term.clear()
self.working.reset()
def _sector_hint(self, ticker: str) -> str:
"""Quick sector lookup for better semantic search queries."""
try:
import yfinance as yf
return yf.Ticker(ticker).info.get("sector", "")
except Exception:
return ""
# Convenience pass-throughs -----------------------------------------------
def store_preference(self, user_id: str, preference: str):
"""Save a user preference e.g. 'risk averse', 'prefers UK stocks'"""
self.long_term.store_preference(user_id, preference)
def get_all_history(self, user_id: str) -> list[dict]:
"""For the Streamlit sidebar - all past recommendations."""
return self.episodic.get_all_history(user_id)
def update_outcome(self, episode_id: int, outcome: str):
"""Record what actually happened after a call - builds track record."""
self.episodic.update_outcome(episode_id, outcome)
def is_using_supabase(self) -> bool:
"""Let the UI know which backend is active."""
return _using_supabase()
|