Spaces:
Sleeping
Sleeping
Create memory.py
Browse files- modules/memory.py +59 -0
modules/memory.py
ADDED
@@ -0,0 +1,59 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
1 |
+
# modules/memory.py
|
2 |
+
|
3 |
+
from datetime import datetime
|
4 |
+
from modules.data_helpers import smart_label_converter
|
5 |
+
from modules.task_management import roadmap_unlock
|
6 |
+
from modules.analysis import render_text_roadmap
|
7 |
+
from utils.constants import course_suggestions
|
8 |
+
from utils.api_clients import pine_index
|
9 |
+
from utils.embedding_model import initialize_embedding_model
|
10 |
+
|
11 |
+
embed_model = initialize_embedding_model()
|
12 |
+
|
13 |
+
def save_to_memory(user_id, goal, summary, steps, courses):
|
14 |
+
try:
|
15 |
+
text_blob = f"Goal: {goal}\nSummary: {summary}\nSteps: {' | '.join(steps)}\nCourses: {' | '.join([c[0] for c in courses])}"
|
16 |
+
embedding = embed_model.embed_query(text_blob)
|
17 |
+
metadata = {
|
18 |
+
"user_id": user_id,
|
19 |
+
"goal": goal,
|
20 |
+
"summary": summary,
|
21 |
+
"steps": steps,
|
22 |
+
"courses": [f"{c[0]} | {c[1]}" for c in courses],
|
23 |
+
"timestamp": datetime.utcnow().isoformat()
|
24 |
+
}
|
25 |
+
pine_index.upsert([(user_id + ":" + goal.replace(" ", "_"), embedding, metadata)])
|
26 |
+
return True
|
27 |
+
except Exception as e:
|
28 |
+
print(f"β Failed to save memory: {e}")
|
29 |
+
return False
|
30 |
+
|
31 |
+
|
32 |
+
def recall_from_memory(user_id, goal):
|
33 |
+
try:
|
34 |
+
query = user_id + ":" + goal.replace(" ", "_")
|
35 |
+
result = pine_index.fetch([query])
|
36 |
+
if query not in result.vectors:
|
37 |
+
return "β No saved plan found for this goal."
|
38 |
+
|
39 |
+
metadata = result.vectors[query].get("metadata", {})
|
40 |
+
steps = metadata.get("steps", [])
|
41 |
+
steps = [smart_label_converter(s) for s in steps if isinstance(s, str) and len(s.strip()) > 1]
|
42 |
+
summary = metadata.get("summary", "")
|
43 |
+
courses = metadata.get("courses", [])
|
44 |
+
diagram = render_text_roadmap(goal, steps)
|
45 |
+
|
46 |
+
course_section = "\n\n### π Recommended Courses\n" + "\n".join(
|
47 |
+
[f"- [{c['name']}]({c['url']})" for c in courses if isinstance(c, dict) and 'name' in c and 'url' in c]
|
48 |
+
) if courses else ""
|
49 |
+
|
50 |
+
return f"""### π Recalled Plan for {goal}
|
51 |
+
|
52 |
+
{diagram}
|
53 |
+
|
54 |
+
{summary}{course_section}
|
55 |
+
|
56 |
+
**π Book your weekly study check-in:** [Click here](https://calendly.com)
|
57 |
+
"""
|
58 |
+
except Exception as e:
|
59 |
+
return f"β Error recalling memory: {e}"
|