AI Assistant commited on
Commit
0039ed6
·
1 Parent(s): 56cf8f7

feat: add multi-document vectorless RAG, session isolation, and memory services

Browse files
app/core/dependencies.py CHANGED
@@ -23,3 +23,10 @@ def get_rag_pipeline() -> VectorlessRAG:
23
 
24
  def get_storage() -> MemoryStorage:
25
  return storage
 
 
 
 
 
 
 
 
23
 
24
  def get_storage() -> MemoryStorage:
25
  return storage
26
+
27
+
28
+ @lru_cache()
29
+ def get_memory_service() -> 'MemoryService':
30
+ from services.memory_service import MemoryService
31
+ return MemoryService(get_llm_service())
32
+
app/core/persistent_db.py CHANGED
@@ -63,6 +63,40 @@ CREATE TABLE IF NOT EXISTS doc_meta (
63
  created_at REAL NOT NULL,
64
  updated_at REAL NOT NULL
65
  );
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
66
  """
67
 
68
 
@@ -109,6 +143,13 @@ class PersistentDB:
109
  if col not in existing_cols:
110
  conn.execute(f"ALTER TABLE doc_meta ADD COLUMN {col} {col_type} DEFAULT {default_val}")
111
 
 
 
 
 
 
 
 
112
  # ── chat CRUD ───────────────────────────────────────────────────────────
113
  def create_chat(self, chat_id: Optional[str] = None, title: Optional[str] = None) -> str:
114
  cid = chat_id or str(uuid.uuid4())
@@ -255,8 +296,201 @@ class PersistentDB:
255
  with self._tx() as conn:
256
  cur = conn.execute("DELETE FROM doc_meta WHERE doc_id=?", (doc_id,))
257
  conn.execute("UPDATE chats SET doc_id=NULL, doc_name=NULL WHERE doc_id=?", (doc_id,))
 
 
258
  return cur.rowcount > 0
259
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
260
 
261
 
262
  # Global singleton
 
63
  created_at REAL NOT NULL,
64
  updated_at REAL NOT NULL
65
  );
66
+
67
+ CREATE TABLE IF NOT EXISTS chat_documents (
68
+ chat_id TEXT NOT NULL REFERENCES chats(chat_id) ON DELETE CASCADE,
69
+ doc_id TEXT NOT NULL REFERENCES doc_meta(doc_id) ON DELETE CASCADE,
70
+ doc_name TEXT,
71
+ attached_at REAL NOT NULL,
72
+ PRIMARY KEY (chat_id, doc_id)
73
+ );
74
+ CREATE INDEX IF NOT EXISTS idx_chat_docs ON chat_documents(chat_id);
75
+
76
+ CREATE TABLE IF NOT EXISTS chat_summaries (
77
+ chat_id TEXT PRIMARY KEY REFERENCES chats(chat_id) ON DELETE CASCADE,
78
+ summary TEXT NOT NULL,
79
+ turn_count INTEGER DEFAULT 0,
80
+ updated_at REAL NOT NULL
81
+ );
82
+
83
+ CREATE TABLE IF NOT EXISTS knowledge_memory (
84
+ doc_id TEXT PRIMARY KEY REFERENCES doc_meta(doc_id) ON DELETE CASCADE,
85
+ filename TEXT NOT NULL,
86
+ executive_summary TEXT,
87
+ key_entities TEXT,
88
+ key_findings TEXT,
89
+ topics TEXT,
90
+ updated_at REAL NOT NULL
91
+ );
92
+
93
+ CREATE TABLE IF NOT EXISTS global_knowledge_summary (
94
+ id TEXT PRIMARY KEY,
95
+ collective_summary TEXT NOT NULL,
96
+ total_documents INTEGER DEFAULT 0,
97
+ key_insights TEXT,
98
+ updated_at REAL NOT NULL
99
+ );
100
  """
101
 
102
 
 
143
  if col not in existing_cols:
144
  conn.execute(f"ALTER TABLE doc_meta ADD COLUMN {col} {col_type} DEFAULT {default_val}")
145
 
146
+ # Migrate existing chats.doc_id into chat_documents table
147
+ conn.execute("""
148
+ INSERT OR IGNORE INTO chat_documents(chat_id, doc_id, doc_name, attached_at)
149
+ SELECT chat_id, doc_id, doc_name, created_at FROM chats
150
+ WHERE doc_id IS NOT NULL AND doc_id != ''
151
+ """)
152
+
153
  # ── chat CRUD ───────────────────────────────────────────────────────────
154
  def create_chat(self, chat_id: Optional[str] = None, title: Optional[str] = None) -> str:
155
  cid = chat_id or str(uuid.uuid4())
 
296
  with self._tx() as conn:
297
  cur = conn.execute("DELETE FROM doc_meta WHERE doc_id=?", (doc_id,))
298
  conn.execute("UPDATE chats SET doc_id=NULL, doc_name=NULL WHERE doc_id=?", (doc_id,))
299
+ conn.execute("DELETE FROM chat_documents WHERE doc_id=?", (doc_id,))
300
+ conn.execute("DELETE FROM knowledge_memory WHERE doc_id=?", (doc_id,))
301
  return cur.rowcount > 0
302
 
303
+ # ── multi-doc session associations ──────────────────────────────────────
304
+ def add_chat_document(self, chat_id: str, doc_id: str, doc_name: Optional[str] = None) -> None:
305
+ now = time.time()
306
+ with self._tx() as conn:
307
+ conn.execute(
308
+ """
309
+ INSERT INTO chat_documents(chat_id, doc_id, doc_name, attached_at)
310
+ VALUES(?,?,?,?)
311
+ ON CONFLICT(chat_id, doc_id) DO UPDATE SET
312
+ doc_name=COALESCE(excluded.doc_name, chat_documents.doc_name),
313
+ attached_at=excluded.attached_at
314
+ """,
315
+ (chat_id, doc_id, doc_name, now),
316
+ )
317
+ # Maintain backward compatibility on chats table with latest doc
318
+ conn.execute(
319
+ "UPDATE chats SET doc_id=?, doc_name=?, updated_at=? WHERE chat_id=?",
320
+ (doc_id, doc_name, now, chat_id),
321
+ )
322
+
323
+ def remove_chat_document(self, chat_id: str, doc_id: str) -> None:
324
+ now = time.time()
325
+ with self._tx() as conn:
326
+ conn.execute("DELETE FROM chat_documents WHERE chat_id=? AND doc_id=?", (chat_id, doc_id))
327
+ rem = conn.execute(
328
+ "SELECT doc_id, doc_name FROM chat_documents WHERE chat_id=? ORDER BY attached_at DESC LIMIT 1",
329
+ (chat_id,),
330
+ ).fetchone()
331
+ if rem:
332
+ conn.execute(
333
+ "UPDATE chats SET doc_id=?, doc_name=?, updated_at=? WHERE chat_id=?",
334
+ (rem["doc_id"], rem["doc_name"], now, chat_id),
335
+ )
336
+ else:
337
+ conn.execute(
338
+ "UPDATE chats SET doc_id=NULL, doc_name=NULL, updated_at=? WHERE chat_id=?",
339
+ (now, chat_id),
340
+ )
341
+
342
+ def get_chat_documents(self, chat_id: str) -> List[Dict[str, Any]]:
343
+ rows = self._conn().execute(
344
+ """
345
+ SELECT cd.chat_id, cd.doc_id, cd.doc_name, cd.attached_at,
346
+ dm.status, dm.page_count, dm.section_count, dm.filename
347
+ FROM chat_documents cd
348
+ LEFT JOIN doc_meta dm ON cd.doc_id = dm.doc_id
349
+ WHERE cd.chat_id=?
350
+ ORDER BY cd.attached_at ASC
351
+ """,
352
+ (chat_id,),
353
+ ).fetchall()
354
+ return [dict(r) for r in rows]
355
+
356
+ def clear_chat_documents(self, chat_id: str) -> None:
357
+ now = time.time()
358
+ with self._tx() as conn:
359
+ conn.execute("DELETE FROM chat_documents WHERE chat_id=?", (chat_id,))
360
+ conn.execute("UPDATE chats SET doc_id=NULL, doc_name=NULL, updated_at=? WHERE chat_id=?", (now, chat_id))
361
+
362
+ # ── session long memory & rolling summaries ─────────────────────────────
363
+ def set_chat_summary(self, chat_id: str, summary: str, turn_count: int = 0) -> None:
364
+ now = time.time()
365
+ with self._tx() as conn:
366
+ conn.execute(
367
+ """
368
+ INSERT INTO chat_summaries(chat_id, summary, turn_count, updated_at)
369
+ VALUES(?,?,?,?)
370
+ ON CONFLICT(chat_id) DO UPDATE SET
371
+ summary=excluded.summary,
372
+ turn_count=excluded.turn_count,
373
+ updated_at=excluded.updated_at
374
+ """,
375
+ (chat_id, summary, turn_count, now),
376
+ )
377
+
378
+ def get_chat_summary(self, chat_id: str) -> Optional[Dict[str, Any]]:
379
+ row = self._conn().execute(
380
+ "SELECT * FROM chat_summaries WHERE chat_id=?", (chat_id,)
381
+ ).fetchone()
382
+ return dict(row) if row else None
383
+
384
+ # ── global knowledge memory & document dossiers ────────────────────────
385
+ def save_knowledge_dossier(
386
+ self,
387
+ doc_id: str,
388
+ filename: str,
389
+ executive_summary: str,
390
+ key_entities: Optional[List[str]] = None,
391
+ key_findings: Optional[List[str]] = None,
392
+ topics: Optional[List[str]] = None,
393
+ ) -> None:
394
+ now = time.time()
395
+ with self._tx() as conn:
396
+ conn.execute(
397
+ """
398
+ INSERT INTO knowledge_memory(
399
+ doc_id, filename, executive_summary, key_entities, key_findings, topics, updated_at
400
+ )
401
+ VALUES(?,?,?,?,?,?,?)
402
+ ON CONFLICT(doc_id) DO UPDATE SET
403
+ filename=excluded.filename,
404
+ executive_summary=excluded.executive_summary,
405
+ key_entities=excluded.key_entities,
406
+ key_findings=excluded.key_findings,
407
+ topics=excluded.topics,
408
+ updated_at=excluded.updated_at
409
+ """,
410
+ (
411
+ doc_id,
412
+ filename,
413
+ executive_summary,
414
+ json.dumps(key_entities or []),
415
+ json.dumps(key_findings or []),
416
+ json.dumps(topics or []),
417
+ now,
418
+ ),
419
+ )
420
+
421
+ def get_knowledge_dossier(self, doc_id: str) -> Optional[Dict[str, Any]]:
422
+ row = self._conn().execute(
423
+ "SELECT * FROM knowledge_memory WHERE doc_id=?", (doc_id,)
424
+ ).fetchone()
425
+ if not row:
426
+ return None
427
+ res = dict(row)
428
+ for k in ("key_entities", "key_findings", "topics"):
429
+ if res.get(k) and isinstance(res[k], str):
430
+ try:
431
+ res[k] = json.loads(res[k])
432
+ except Exception:
433
+ res[k] = []
434
+ return res
435
+
436
+ def list_knowledge_dossiers(self) -> List[Dict[str, Any]]:
437
+ rows = self._conn().execute(
438
+ "SELECT * FROM knowledge_memory ORDER BY updated_at DESC"
439
+ ).fetchall()
440
+ result = []
441
+ for r in rows:
442
+ item = dict(r)
443
+ for k in ("key_entities", "key_findings", "topics"):
444
+ if item.get(k) and isinstance(item[k], str):
445
+ try:
446
+ item[k] = json.loads(item[k])
447
+ except Exception:
448
+ item[k] = []
449
+ result.append(item)
450
+ return result
451
+
452
+ def delete_knowledge_dossier(self, doc_id: str) -> None:
453
+ with self._tx() as conn:
454
+ conn.execute("DELETE FROM knowledge_memory WHERE doc_id=?", (doc_id,))
455
+
456
+ def save_global_knowledge_summary(
457
+ self,
458
+ collective_summary: str,
459
+ total_documents: int,
460
+ key_insights: Optional[List[str]] = None,
461
+ ) -> None:
462
+ now = time.time()
463
+ gid = "global"
464
+ with self._tx() as conn:
465
+ conn.execute(
466
+ """
467
+ INSERT INTO global_knowledge_summary(
468
+ id, collective_summary, total_documents, key_insights, updated_at
469
+ )
470
+ VALUES(?,?,?,?,?)
471
+ ON CONFLICT(id) DO UPDATE SET
472
+ collective_summary=excluded.collective_summary,
473
+ total_documents=excluded.total_documents,
474
+ key_insights=excluded.key_insights,
475
+ updated_at=excluded.updated_at
476
+ """,
477
+ (gid, collective_summary, total_documents, json.dumps(key_insights or []), now),
478
+ )
479
+
480
+ def get_global_knowledge_summary(self) -> Optional[Dict[str, Any]]:
481
+ row = self._conn().execute(
482
+ "SELECT * FROM global_knowledge_summary WHERE id='global'"
483
+ ).fetchone()
484
+ if not row:
485
+ return None
486
+ res = dict(row)
487
+ if res.get("key_insights") and isinstance(res["key_insights"], str):
488
+ try:
489
+ res["key_insights"] = json.loads(res["key_insights"])
490
+ except Exception:
491
+ res["key_insights"] = []
492
+ return res
493
+
494
 
495
 
496
  # Global singleton
app/core/rag_pipeline.py CHANGED
@@ -1,67 +1,148 @@
 
1
  import time
2
  from typing import Optional, List, Dict, Any
3
  from utils.logger import app_logger
4
 
 
5
  class VectorlessRAG:
6
  def __init__(self, llm_service):
7
  self.llm = llm_service
8
 
9
- async def answer_query_stream(
10
  self,
11
  query: str,
12
- tree: list,
13
  chat_history: Optional[List[Dict[str, str]]] = None,
14
- doc_name: Optional[str] = None,
15
- node_map: Optional[Dict[str, Any]] = None,
16
- flat_sections: Optional[List[Dict[str, Any]]] = None,
17
  deep_analysis: bool = False,
18
  ):
 
 
 
 
 
 
 
 
 
 
 
19
  try:
20
  t_retrieval_start = time.monotonic()
21
- search_result = await self.llm.llm_tree_search(
22
- query, tree, chat_history=chat_history, flat_sections=flat_sections
23
- )
24
- node_ids = search_result.get("node_list", [])
25
- nodes = self.llm.find_nodes_by_ids(tree, node_ids, node_map=node_map)
 
 
 
 
 
 
 
 
 
26
 
27
- # If short / referential or overview query returned no explicit node matches, select top overview sections
28
- if not nodes:
29
- overview_triggers = ["دا", "ده", "دي", "لخص", "ملخص", "اشرح", "بحث", "summary", "summarize", "overview", "what is this", "about"]
30
- normalized_q = query.strip().lower()
31
- is_short_or_overview = len(normalized_q) <= 8 or any(t in normalized_q for t in overview_triggers)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
32
 
33
- if is_short_or_overview:
34
- app_logger.info(f"Query '{query}' resolved as overview/referential query with 0 matched nodes. Defaulting to leading document sections.")
35
- top_nodes = []
36
- def collect_leading(items):
37
- for item in items:
38
- if isinstance(item, dict):
39
- top_nodes.append(item)
40
- if len(top_nodes) >= 4:
41
- return
42
- if item.get("nodes"):
43
- collect_leading(item["nodes"])
44
- collect_leading(tree)
45
- nodes = top_nodes
 
 
 
 
 
46
 
47
  t_retrieval = time.monotonic() - t_retrieval_start
 
 
 
48
  app_logger.info(
49
- f"[PERF] rag_pipeline retrieval_time={t_retrieval:.2f}s "
50
- f"nodes_matched={len(nodes)} deep_analysis={deep_analysis}"
51
  )
52
 
53
- # Signal retrieval completion to frontend immediately
54
- yield f'<!--STAGE:{{"stage":"retrieved","duration":{t_retrieval:.2f},"nodes":{len(nodes)}}}-->\n'
55
 
 
56
  async for chunk in self.llm.generate_answer_stream(
57
- query, nodes, chat_history=chat_history, doc_name=doc_name, deep_analysis=deep_analysis
 
 
 
 
 
 
58
  ):
59
  yield chunk
60
 
61
  except Exception as e:
62
- app_logger.error(f"Error in RAG pipeline: {str(e)}", exc_info=True)
63
  is_arabic = any("\u0600" <= c <= "\u06ff" for c in query)
64
  if is_arabic:
65
- yield "حدث خطأ أثناء استرجاع المعلومات من المستند المفهرس. يرجى إعادة المحاولة."
66
  else:
67
- yield "An error occurred while retrieving information from the indexed document. Please try again."
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import asyncio
2
  import time
3
  from typing import Optional, List, Dict, Any
4
  from utils.logger import app_logger
5
 
6
+
7
  class VectorlessRAG:
8
  def __init__(self, llm_service):
9
  self.llm = llm_service
10
 
11
+ async def answer_multi_doc_query_stream(
12
  self,
13
  query: str,
14
+ docs: List[Dict[str, Any]],
15
  chat_history: Optional[List[Dict[str, str]]] = None,
16
+ session_summary: Optional[str] = None,
17
+ global_knowledge: Optional[str] = None,
 
18
  deep_analysis: bool = False,
19
  ):
20
+ """
21
+ Executes Vectorless RAG across multiple documents concurrently.
22
+ Each item in `docs`:
23
+ {
24
+ "doc_id": str,
25
+ "filename": str,
26
+ "tree": list,
27
+ "node_map": dict,
28
+ "flat_sections": list
29
+ }
30
+ """
31
  try:
32
  t_retrieval_start = time.monotonic()
33
+ all_selected_nodes = []
34
+
35
+ # Search across each document tree concurrently
36
+ async def search_single_doc(doc):
37
+ tree = doc.get("tree") or []
38
+ node_map = doc.get("node_map")
39
+ flat_sections = doc.get("flat_sections")
40
+ doc_name = doc.get("filename") or "Document"
41
+
42
+ search_res = await self.llm.llm_tree_search(
43
+ query, tree, chat_history=chat_history, flat_sections=flat_sections
44
+ )
45
+ node_ids = search_res.get("node_list", [])
46
+ nodes = self.llm.find_nodes_by_ids(tree, node_ids, node_map=node_map)
47
 
48
+ # Fallback to leading sections for overview/comparative queries if no nodes matched
49
+ if not nodes:
50
+ overview_triggers = [
51
+ "دا", "ده", "دي", "لخص", "ملخص", "اشرح", "بحث", "مقارنة", "قارن",
52
+ "summary", "summarize", "overview", "compare", "comparison", "what is this", "about"
53
+ ]
54
+ normalized_q = query.strip().lower()
55
+ is_short_or_overview = len(normalized_q) <= 10 or any(t in normalized_q for t in overview_triggers)
56
+ if is_short_or_overview:
57
+ top_nodes = []
58
+ def collect_leading(items):
59
+ for item in items:
60
+ if isinstance(item, dict):
61
+ top_nodes.append(item)
62
+ if len(top_nodes) >= 3:
63
+ return
64
+ if item.get("nodes"):
65
+ collect_leading(item["nodes"])
66
+ collect_leading(tree)
67
+ nodes = top_nodes
68
 
69
+ tagged_nodes = []
70
+ for n in nodes:
71
+ n_copy = dict(n)
72
+ n_copy["doc_name"] = doc_name
73
+ n_copy["filename"] = doc_name
74
+ n_copy["doc_id"] = doc.get("doc_id")
75
+ tagged_nodes.append(n_copy)
76
+ return tagged_nodes
77
+
78
+ # Run searches concurrently across all attached documents
79
+ search_tasks = [search_single_doc(d) for d in docs]
80
+ results = await asyncio.gather(*search_tasks, return_exceptions=True)
81
+
82
+ for r in results:
83
+ if isinstance(r, list):
84
+ all_selected_nodes.extend(r)
85
+ elif isinstance(r, Exception):
86
+ app_logger.error(f"[MULTI-RAG] Document tree search error: {r}")
87
 
88
  t_retrieval = time.monotonic() - t_retrieval_start
89
+ doc_names_list = [d.get("filename", "Doc") for d in docs]
90
+ doc_names_str = ", ".join(doc_names_list)
91
+
92
  app_logger.info(
93
+ f"[PERF] multi_rag retrieval_time={t_retrieval:.2f}s "
94
+ f"docs_count={len(docs)} nodes_matched={len(all_selected_nodes)} deep_analysis={deep_analysis}"
95
  )
96
 
97
+ # Signal retrieval completion stage to frontend
98
+ yield f'<!--STAGE:{{"stage":"retrieved","duration":{t_retrieval:.2f},"nodes":{len(all_selected_nodes)},"docs":{len(docs)}}}-->\n'
99
 
100
+ # Stream final synthesis with multi-document attribution
101
  async for chunk in self.llm.generate_answer_stream(
102
+ query=query,
103
+ nodes=all_selected_nodes,
104
+ chat_history=chat_history,
105
+ doc_name=doc_names_str,
106
+ session_summary=session_summary,
107
+ global_knowledge=global_knowledge,
108
+ deep_analysis=deep_analysis,
109
  ):
110
  yield chunk
111
 
112
  except Exception as e:
113
+ app_logger.error(f"Error in multi-doc RAG pipeline: {str(e)}", exc_info=True)
114
  is_arabic = any("\u0600" <= c <= "\u06ff" for c in query)
115
  if is_arabic:
116
+ yield "حدث خطأ أثناء استرجاع المعلومات من المستندات المفهرسة. يرجى إعادة المحاولة."
117
  else:
118
+ yield "An error occurred while retrieving information from the indexed documents. Please try again."
119
+
120
+ async def answer_query_stream(
121
+ self,
122
+ query: str,
123
+ tree: list,
124
+ chat_history: Optional[List[Dict[str, str]]] = None,
125
+ doc_name: Optional[str] = None,
126
+ node_map: Optional[Dict[str, Any]] = None,
127
+ flat_sections: Optional[List[Dict[str, Any]]] = None,
128
+ deep_analysis: bool = False,
129
+ session_summary: Optional[str] = None,
130
+ global_knowledge: Optional[str] = None,
131
+ ):
132
+ """Single-document backward-compatible wrapper."""
133
+ doc_obj = {
134
+ "doc_id": "single",
135
+ "filename": doc_name or "Uploaded Document",
136
+ "tree": tree,
137
+ "node_map": node_map,
138
+ "flat_sections": flat_sections,
139
+ }
140
+ async for chunk in self.answer_multi_doc_query_stream(
141
+ query=query,
142
+ docs=[doc_obj],
143
+ chat_history=chat_history,
144
+ session_summary=session_summary,
145
+ global_knowledge=global_knowledge,
146
+ deep_analysis=deep_analysis,
147
+ ):
148
+ yield chunk
app/core/storage.py CHANGED
@@ -485,13 +485,35 @@ class MemoryStorage:
485
  doc = self._documents.get(doc_id)
486
  doc_name = doc.get("filename") if doc else None
487
  persistent_db.set_chat_doc(chat_id=chat_id, doc_id=doc_id, doc_name=doc_name)
 
 
 
 
 
 
 
 
 
488
 
489
  def get_chat_doc(self, chat_id: str) -> Optional[str]:
 
490
  meta = persistent_db.get_chat(chat_id)
491
- return meta["doc_id"] if meta else None
 
 
 
 
 
 
492
 
493
- def remove_chat_doc(self, chat_id: str) -> None:
494
- persistent_db.remove_chat_doc(chat_id)
 
 
 
 
 
 
495
 
496
  def delete_chat(self, chat_id: str) -> None:
497
  persistent_db.delete_chat(chat_id)
@@ -502,6 +524,56 @@ class MemoryStorage:
502
  def update_chat_title(self, chat_id: str, title: str) -> None:
503
  persistent_db.update_chat_title(chat_id=chat_id, title=title)
504
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
505
 
506
  # Global singleton storage instance
507
  storage = MemoryStorage()
 
485
  doc = self._documents.get(doc_id)
486
  doc_name = doc.get("filename") if doc else None
487
  persistent_db.set_chat_doc(chat_id=chat_id, doc_id=doc_id, doc_name=doc_name)
488
+ persistent_db.add_chat_document(chat_id=chat_id, doc_id=doc_id, doc_name=doc_name)
489
+
490
+ def add_chat_doc(self, chat_id: str, doc_id: str, doc_name: Optional[str] = None) -> None:
491
+ if not self.chat_exists(chat_id):
492
+ persistent_db.create_chat(chat_id=chat_id)
493
+ if doc_name is None:
494
+ doc = self._documents.get(doc_id)
495
+ doc_name = doc.get("filename") if doc else None
496
+ persistent_db.add_chat_document(chat_id=chat_id, doc_id=doc_id, doc_name=doc_name)
497
 
498
  def get_chat_doc(self, chat_id: str) -> Optional[str]:
499
+ # Return primary attached doc or first multi-doc
500
  meta = persistent_db.get_chat(chat_id)
501
+ if meta and meta.get("doc_id"):
502
+ return meta["doc_id"]
503
+ docs = persistent_db.get_chat_documents(chat_id)
504
+ return docs[0]["doc_id"] if docs else None
505
+
506
+ def get_chat_docs(self, chat_id: str) -> List[Dict[str, Any]]:
507
+ return persistent_db.get_chat_documents(chat_id)
508
 
509
+ def remove_chat_doc(self, chat_id: str, doc_id: Optional[str] = None) -> None:
510
+ if doc_id:
511
+ persistent_db.remove_chat_document(chat_id, doc_id)
512
+ else:
513
+ persistent_db.remove_chat_doc(chat_id)
514
+
515
+ def clear_chat_docs(self, chat_id: str) -> None:
516
+ persistent_db.clear_chat_documents(chat_id)
517
 
518
  def delete_chat(self, chat_id: str) -> None:
519
  persistent_db.delete_chat(chat_id)
 
524
  def update_chat_title(self, chat_id: str, title: str) -> None:
525
  persistent_db.update_chat_title(chat_id=chat_id, title=title)
526
 
527
+ # ── session long memory & rolling summaries ─────────────────────────────
528
+ def set_chat_summary(self, chat_id: str, summary: str, turn_count: int = 0) -> None:
529
+ persistent_db.set_chat_summary(chat_id=chat_id, summary=summary, turn_count=turn_count)
530
+
531
+ def get_chat_summary(self, chat_id: str) -> Optional[Dict[str, Any]]:
532
+ return persistent_db.get_chat_summary(chat_id=chat_id)
533
+
534
+ # ── global knowledge memory & document dossiers ────────────────────────
535
+ def save_knowledge_dossier(
536
+ self,
537
+ doc_id: str,
538
+ filename: str,
539
+ executive_summary: str,
540
+ key_entities: Optional[List[str]] = None,
541
+ key_findings: Optional[List[str]] = None,
542
+ topics: Optional[List[str]] = None,
543
+ ) -> None:
544
+ persistent_db.save_knowledge_dossier(
545
+ doc_id=doc_id,
546
+ filename=filename,
547
+ executive_summary=executive_summary,
548
+ key_entities=key_entities,
549
+ key_findings=key_findings,
550
+ topics=topics,
551
+ )
552
+
553
+ def get_knowledge_dossier(self, doc_id: str) -> Optional[Dict[str, Any]]:
554
+ return persistent_db.get_knowledge_dossier(doc_id)
555
+
556
+ def list_knowledge_dossiers(self) -> List[Dict[str, Any]]:
557
+ return persistent_db.list_knowledge_dossiers()
558
+
559
+ def delete_knowledge_dossier(self, doc_id: str) -> None:
560
+ persistent_db.delete_knowledge_dossier(doc_id)
561
+
562
+ def save_global_knowledge_summary(
563
+ self,
564
+ collective_summary: str,
565
+ total_documents: int,
566
+ key_insights: Optional[List[str]] = None,
567
+ ) -> None:
568
+ persistent_db.save_global_knowledge_summary(
569
+ collective_summary=collective_summary,
570
+ total_documents=total_documents,
571
+ key_insights=key_insights,
572
+ )
573
+
574
+ def get_global_knowledge_summary(self) -> Optional[Dict[str, Any]]:
575
+ return persistent_db.get_global_knowledge_summary()
576
+
577
 
578
  # Global singleton storage instance
579
  storage = MemoryStorage()
app/routers/__init__.py CHANGED
@@ -1,9 +1,11 @@
1
  from fastapi import APIRouter
2
  from routers.documents import router as documents_router
3
  from routers.chat import router as chat_router
 
4
 
5
  api_router = APIRouter()
6
  api_router.include_router(documents_router)
7
  api_router.include_router(chat_router)
 
8
 
9
- __all__ = ["api_router", "documents_router", "chat_router"]
 
1
  from fastapi import APIRouter
2
  from routers.documents import router as documents_router
3
  from routers.chat import router as chat_router
4
+ from routers.memory import router as memory_router
5
 
6
  api_router = APIRouter()
7
  api_router.include_router(documents_router)
8
  api_router.include_router(chat_router)
9
+ api_router.include_router(memory_router)
10
 
11
+ __all__ = ["api_router", "documents_router", "chat_router", "memory_router"]
app/routers/chat.py CHANGED
@@ -1,22 +1,32 @@
 
 
 
1
  from fastapi import APIRouter, HTTPException, Depends
2
  from fastapi.responses import StreamingResponse
3
  from pydantic import BaseModel
4
 
5
- from schemas.schemas import AskRequest, NewChatResponse, ChatHistoryResponse
 
 
 
 
 
 
6
  from core.storage import MemoryStorage
7
- from core.dependencies import get_llm_service, get_rag_pipeline, get_storage
 
 
 
 
 
8
  from core.rag_pipeline import VectorlessRAG
9
  from services.llm_service import LLMService
 
10
  from utils.logger import app_logger
11
- import re
12
 
13
  router = APIRouter(tags=["Chat"])
14
 
15
 
16
- class AttachDocRequest(BaseModel):
17
- doc_id: str
18
-
19
-
20
  @router.post("/chat/new", response_model=NewChatResponse)
21
  async def create_chat(storage: MemoryStorage = Depends(get_storage)):
22
  chat_id = storage.create_chat()
@@ -30,23 +40,75 @@ async def attach_document_to_chat(
30
  body: AttachDocRequest,
31
  storage: MemoryStorage = Depends(get_storage),
32
  ):
33
- """Attach an indexed document to an active conversation."""
34
  if not storage.tree_exists(body.doc_id):
35
  raise HTTPException(status_code=404, detail="Document ID not found in memory")
36
- storage.set_chat_doc(chat_id, body.doc_id)
37
  app_logger.info(f"Attached document {body.doc_id} to chat session {chat_id}")
38
  return {"message": "Document attached successfully", "chat_id": chat_id, "doc_id": body.doc_id}
39
 
40
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
41
  @router.delete("/chat/{chat_id}/document")
42
  async def detach_document_from_chat(
43
  chat_id: str,
44
  storage: MemoryStorage = Depends(get_storage),
45
  ):
46
- """Detach document from active chat session."""
 
47
  storage.remove_chat_doc(chat_id)
48
- app_logger.info(f"Detached document from chat session {chat_id}")
49
- return {"message": "Document detached successfully", "chat_id": chat_id}
50
 
51
 
52
  @router.post("/ask")
@@ -55,96 +117,134 @@ async def ask_question(
55
  storage: MemoryStorage = Depends(get_storage),
56
  llm_service: LLMService = Depends(get_llm_service),
57
  rag_pipeline: VectorlessRAG = Depends(get_rag_pipeline),
 
58
  ):
59
- chat_history = storage.ensure_chat(request.chat_id)
60
-
61
- # 1. Resolve active document identity for this chat session
62
- effective_doc_id = request.doc_id or storage.get_chat_doc(request.chat_id)
63
- doc_state = storage.get_document(effective_doc_id) if effective_doc_id else None
64
-
65
- # If request passed doc_id, ensure session association is persisted
66
- if request.doc_id and effective_doc_id:
67
- storage.set_chat_doc(request.chat_id, effective_doc_id)
68
-
69
- # 2. Deterministic routing decision
70
- error_detail = None
71
- if not effective_doc_id:
72
- selected_mode = "conversational"
73
- elif doc_state is None:
74
- selected_mode = "error"
75
- error_detail = "Document session not found or expired"
76
- elif doc_state.get("status") == "indexing":
77
- selected_mode = "indexing"
78
- elif doc_state.get("status") == "ready" and doc_state.get("tree"):
79
- selected_mode = "pageindex"
80
  else:
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
81
  selected_mode = "error"
82
- error_detail = doc_state.get("error") or "Document tree unavailable"
 
 
83
 
84
  # Router Audit Logging
85
  app_logger.info(
86
- f"[CHAT ROUTER] conversation_id={request.chat_id} document_id={effective_doc_id} "
87
- f"document_status={doc_state.get('status') if doc_state else 'none'} "
88
- f"tree_ready={bool(doc_state and doc_state.get('tree'))} mode={selected_mode} "
89
- f"deep_analysis={bool(request.deep_analysis)}"
90
  )
91
 
92
- # Use concise recent history directly to preserve TTFT without blocking LLM calls
93
- effective_history = chat_history[-6:] if len(chat_history) > 6 else chat_history
 
 
 
 
 
94
 
95
  is_arabic = any("\u0600" <= c <= "\u06ff" for c in request.query)
96
- doc_name = (doc_state.get("filename") if doc_state else None) or request.doc_name
97
 
98
  async def stream_generator():
99
  full_answer = ""
100
  try:
101
- if selected_mode == "pageindex":
102
- # Mode A: Deterministic PageIndex tree retrieval & grounded synthesis
103
- tree = doc_state["tree"]
104
- node_map = storage.get_node_map(effective_doc_id)
105
- flat_sections = storage.get_flat_sections(effective_doc_id)
106
- if node_map is None or flat_sections is None:
107
- node_map, flat_sections = llm_service.build_index_maps(tree)
108
- storage.set_node_map(effective_doc_id, node_map)
109
- storage.set_flat_sections(effective_doc_id, flat_sections)
110
-
111
- async for chunk in rag_pipeline.answer_query_stream(
112
- request.query,
113
- tree,
114
- chat_history=effective_history,
115
- doc_name=doc_name,
116
- node_map=node_map,
117
- flat_sections=flat_sections,
118
  deep_analysis=bool(request.deep_analysis),
119
  ):
120
  full_answer += chunk
121
  yield chunk
122
 
123
  elif selected_mode == "indexing":
124
- # Mode B: Document attached but still indexing - inform user explicitly
125
- filename_display = doc_name or "PDF"
126
  if is_arabic:
127
- msg = f"المستند ({filename_display}) قيد المعالجة والفهرسة حالياً. يرجى الانتظار لحظة حتى تكتمل الفهرسة."
128
  else:
129
- msg = f"Your document ({filename_display}) is currently being indexed. Please wait a moment."
130
  full_answer = msg
131
  yield msg
132
 
133
  elif selected_mode == "error":
134
- # Mode C: Document failed or unavailable - do NOT fall back to conversational
135
- filename_display = doc_name or "PDF"
136
- detail_str = error_detail or "File unavailable"
137
  if is_arabic:
138
- msg = f"تعذر الوصول إلى المستند المطلوب ({detail_str}). يرجى إعادة رفع الملف."
139
  else:
140
- msg = f"Could not access the requested document ({detail_str}). Please try re-uploading the file."
141
  full_answer = msg
142
  yield msg
143
 
144
  else:
145
- # Mode D: Pure conversational AI assistant (strictly when no document is attached)
146
  async for chunk in llm_service.generate_conversational_stream(
147
- request.query, chat_history
 
 
 
148
  ):
149
  full_answer += chunk
150
  yield chunk
@@ -156,6 +256,13 @@ async def ask_question(
156
  storage.add_message(
157
  request.chat_id, role="assistant", content=clean_answer or full_answer
158
  )
 
 
 
 
 
 
 
159
 
160
  return StreamingResponse(stream_generator(), media_type="text/event-stream")
161
 
@@ -171,7 +278,7 @@ async def get_chat_history(
171
  return ChatHistoryResponse(history=history)
172
 
173
 
174
- # ── Long Memory Endpoints ─────────────────────────────────────────────────
175
 
176
  @router.get("/chats")
177
  async def list_all_chats(storage: MemoryStorage = Depends(get_storage)):
 
1
+ import asyncio
2
+ import re
3
+ from typing import List, Optional
4
  from fastapi import APIRouter, HTTPException, Depends
5
  from fastapi.responses import StreamingResponse
6
  from pydantic import BaseModel
7
 
8
+ from schemas.schemas import (
9
+ AskRequest,
10
+ NewChatResponse,
11
+ ChatHistoryResponse,
12
+ AttachDocRequest,
13
+ AttachMultipleDocsRequest,
14
+ )
15
  from core.storage import MemoryStorage
16
+ from core.dependencies import (
17
+ get_llm_service,
18
+ get_rag_pipeline,
19
+ get_storage,
20
+ get_memory_service,
21
+ )
22
  from core.rag_pipeline import VectorlessRAG
23
  from services.llm_service import LLMService
24
+ from services.memory_service import MemoryService
25
  from utils.logger import app_logger
 
26
 
27
  router = APIRouter(tags=["Chat"])
28
 
29
 
 
 
 
 
30
  @router.post("/chat/new", response_model=NewChatResponse)
31
  async def create_chat(storage: MemoryStorage = Depends(get_storage)):
32
  chat_id = storage.create_chat()
 
40
  body: AttachDocRequest,
41
  storage: MemoryStorage = Depends(get_storage),
42
  ):
43
+ """Attach an indexed document to an active conversation session."""
44
  if not storage.tree_exists(body.doc_id):
45
  raise HTTPException(status_code=404, detail="Document ID not found in memory")
46
+ storage.add_chat_doc(chat_id, body.doc_id, doc_name=body.doc_name)
47
  app_logger.info(f"Attached document {body.doc_id} to chat session {chat_id}")
48
  return {"message": "Document attached successfully", "chat_id": chat_id, "doc_id": body.doc_id}
49
 
50
 
51
+ @router.post("/chat/{chat_id}/attach-multiple")
52
+ async def attach_multiple_documents_to_chat(
53
+ chat_id: str,
54
+ body: AttachMultipleDocsRequest,
55
+ storage: MemoryStorage = Depends(get_storage),
56
+ ):
57
+ """Attach multiple indexed documents to an active conversation session."""
58
+ attached = []
59
+ for doc_id in body.doc_ids:
60
+ if storage.tree_exists(doc_id):
61
+ storage.add_chat_doc(chat_id, doc_id)
62
+ attached.append(doc_id)
63
+ app_logger.info(f"Attached {len(attached)} documents to chat session {chat_id}")
64
+ return {"message": f"Attached {len(attached)} documents successfully", "chat_id": chat_id, "doc_ids": attached}
65
+
66
+
67
+ @router.get("/chat/{chat_id}/documents")
68
+ async def get_chat_documents(
69
+ chat_id: str,
70
+ storage: MemoryStorage = Depends(get_storage),
71
+ ):
72
+ """Return all attached documents for this session with status and page counts."""
73
+ docs = storage.get_chat_docs(chat_id)
74
+ detailed_docs = []
75
+ for d in docs:
76
+ doc_info = storage.get_document(d["doc_id"])
77
+ if doc_info:
78
+ detailed_docs.append({
79
+ "doc_id": d["doc_id"],
80
+ "filename": doc_info.get("filename", d.get("doc_name", "document.pdf")),
81
+ "status": doc_info.get("status", "ready"),
82
+ "page_count": doc_info.get("page_count", 1),
83
+ "section_count": doc_info.get("section_count", 0),
84
+ })
85
+ else:
86
+ detailed_docs.append(d)
87
+ return {"chat_id": chat_id, "documents": detailed_docs}
88
+
89
+
90
+ @router.delete("/chat/{chat_id}/document/{doc_id}")
91
+ async def detach_specific_document_from_chat(
92
+ chat_id: str,
93
+ doc_id: str,
94
+ storage: MemoryStorage = Depends(get_storage),
95
+ ):
96
+ """Detach a specific document from the active chat session."""
97
+ storage.remove_chat_doc(chat_id, doc_id=doc_id)
98
+ app_logger.info(f"Detached document {doc_id} from chat session {chat_id}")
99
+ return {"message": "Document detached successfully", "chat_id": chat_id, "doc_id": doc_id}
100
+
101
+
102
  @router.delete("/chat/{chat_id}/document")
103
  async def detach_document_from_chat(
104
  chat_id: str,
105
  storage: MemoryStorage = Depends(get_storage),
106
  ):
107
+ """Detach all documents from active chat session."""
108
+ storage.clear_chat_docs(chat_id)
109
  storage.remove_chat_doc(chat_id)
110
+ app_logger.info(f"Cleared all documents from chat session {chat_id}")
111
+ return {"message": "All documents detached successfully", "chat_id": chat_id}
112
 
113
 
114
  @router.post("/ask")
 
117
  storage: MemoryStorage = Depends(get_storage),
118
  llm_service: LLMService = Depends(get_llm_service),
119
  rag_pipeline: VectorlessRAG = Depends(get_rag_pipeline),
120
+ memory_service: MemoryService = Depends(get_memory_service),
121
  ):
122
+ # Ensure chat session exists in persistent storage
123
+ storage.ensure_chat(request.chat_id)
124
+
125
+ # 1. Resolve target document IDs for this query session
126
+ target_doc_ids: List[str] = []
127
+ if request.doc_ids:
128
+ target_doc_ids.extend(request.doc_ids)
129
+ elif request.doc_id:
130
+ target_doc_ids.append(request.doc_id)
131
+
132
+ if target_doc_ids:
133
+ # Associate newly requested docs with this session
134
+ for did in target_doc_ids:
135
+ storage.add_chat_doc(request.chat_id, did)
 
 
 
 
 
 
 
136
  else:
137
+ # Check existing docs attached to this session
138
+ session_docs = storage.get_chat_docs(request.chat_id)
139
+ if session_docs:
140
+ target_doc_ids = [d["doc_id"] for d in session_docs]
141
+ else:
142
+ primary_id = storage.get_chat_doc(request.chat_id)
143
+ if primary_id:
144
+ target_doc_ids = [primary_id]
145
+
146
+ # Deduplicate while preserving order
147
+ target_doc_ids = list(dict.fromkeys(target_doc_ids))
148
+
149
+ # 2. Inspect document states
150
+ ready_docs = []
151
+ indexing_docs = []
152
+ error_docs = []
153
+
154
+ for did in target_doc_ids:
155
+ doc_state = storage.get_document(did)
156
+ if not doc_state:
157
+ error_docs.append({"doc_id": did, "filename": did, "error": "Document not found or expired"})
158
+ elif doc_state.get("status") == "ready" and doc_state.get("tree"):
159
+ node_map = storage.get_node_map(did)
160
+ flat_sections = storage.get_flat_sections(did)
161
+ if node_map is None or flat_sections is None:
162
+ node_map, flat_sections = llm_service.build_index_maps(doc_state["tree"])
163
+ storage.set_node_map(did, node_map)
164
+ storage.set_flat_sections(did, flat_sections)
165
+ ready_docs.append({
166
+ "doc_id": did,
167
+ "filename": doc_state.get("filename", "document.pdf"),
168
+ "tree": doc_state["tree"],
169
+ "node_map": node_map,
170
+ "flat_sections": flat_sections,
171
+ })
172
+ elif doc_state.get("status") == "indexing":
173
+ indexing_docs.append(doc_state)
174
+ else:
175
+ error_docs.append(doc_state)
176
+
177
+ # 3. Determine execution mode
178
+ if ready_docs:
179
+ selected_mode = "multi_doc_pageindex"
180
+ elif indexing_docs:
181
+ selected_mode = "indexing"
182
+ elif error_docs and not ready_docs:
183
  selected_mode = "error"
184
+ else:
185
+ # Pure conversational AI assistant with workspace global knowledge
186
+ selected_mode = "conversational"
187
 
188
  # Router Audit Logging
189
  app_logger.info(
190
+ f"[CHAT ROUTER] conversation_id={request.chat_id} target_docs={len(target_doc_ids)} "
191
+ f"ready_docs={len(ready_docs)} indexing_docs={len(indexing_docs)} "
192
+ f"mode={selected_mode} deep_analysis={bool(request.deep_analysis)}"
 
193
  )
194
 
195
+ # 4. Retrieve session memory (isolated short-term window + long-term rolling summary)
196
+ mem_context = memory_service.get_session_context(request.chat_id)
197
+ recent_turns = mem_context.get("recent_turns", [])
198
+ session_summary = mem_context.get("session_summary")
199
+
200
+ # 5. Retrieve global workspace knowledge summary
201
+ global_knowledge = memory_service.get_global_knowledge()
202
 
203
  is_arabic = any("\u0600" <= c <= "\u06ff" for c in request.query)
 
204
 
205
  async def stream_generator():
206
  full_answer = ""
207
  try:
208
+ if selected_mode == "multi_doc_pageindex":
209
+ # Mode A: Deterministic Vectorless multi-doc tree retrieval & grounded synthesis
210
+ async for chunk in rag_pipeline.answer_multi_doc_query_stream(
211
+ query=request.query,
212
+ docs=ready_docs,
213
+ chat_history=recent_turns,
214
+ session_summary=session_summary,
215
+ global_knowledge=global_knowledge,
 
 
 
 
 
 
 
 
 
216
  deep_analysis=bool(request.deep_analysis),
217
  ):
218
  full_answer += chunk
219
  yield chunk
220
 
221
  elif selected_mode == "indexing":
222
+ # Mode B: Document(s) attached but still indexing
223
+ doc_names = ", ".join([d.get("filename", "PDF") for d in indexing_docs])
224
  if is_arabic:
225
+ msg = f"المستندات التالية قيد الفهرسة حالياً ({doc_names}). يرجى الانتظار لحظة حتى تكتمل المعالجة."
226
  else:
227
+ msg = f"The following documents are currently being indexed ({doc_names}). Please wait a moment."
228
  full_answer = msg
229
  yield msg
230
 
231
  elif selected_mode == "error":
232
+ # Mode C: Requested documents failed or unavailable
233
+ err_names = ", ".join([d.get("filename", "PDF") for d in error_docs])
 
234
  if is_arabic:
235
+ msg = f"تعذر استرجاع المستندات التالية ({err_names}). يرجى إعادة رفع الملفات."
236
  else:
237
+ msg = f"Could not access the requested documents ({err_names}). Please try re-uploading."
238
  full_answer = msg
239
  yield msg
240
 
241
  else:
242
+ # Mode D: Pure conversational AI assistant with workspace collective knowledge injected
243
  async for chunk in llm_service.generate_conversational_stream(
244
+ query=request.query,
245
+ chat_history=recent_turns,
246
+ session_summary=session_summary,
247
+ global_knowledge=global_knowledge,
248
  ):
249
  full_answer += chunk
250
  yield chunk
 
256
  storage.add_message(
257
  request.chat_id, role="assistant", content=clean_answer or full_answer
258
  )
259
+ # Asynchronously update long-term rolling summary if conversation threshold reached
260
+ try:
261
+ asyncio.create_task(
262
+ memory_service.update_session_summary_if_needed(request.chat_id)
263
+ )
264
+ except Exception:
265
+ pass
266
 
267
  return StreamingResponse(stream_generator(), media_type="text/event-stream")
268
 
 
278
  return ChatHistoryResponse(history=history)
279
 
280
 
281
+ # ── Long Memory & Conversation Management Endpoints ────────────────────────
282
 
283
  @router.get("/chats")
284
  async def list_all_chats(storage: MemoryStorage = Depends(get_storage)):
app/routers/documents.py CHANGED
@@ -75,6 +75,21 @@ async def run_background_indexing(
75
  )
76
  storage.set_hash_mapping(content_hash, doc_id)
77
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
78
  app_logger.info(
79
  f"[INDEX] document_id={doc_id} stage=ready duration={total_duration:.2f}s "
80
  f"method={indexing_method} pages={page_count} sections={section_count}"
 
75
  )
76
  storage.set_hash_mapping(content_hash, doc_id)
77
 
78
+ # Trigger background knowledge dossier generation and global synthesis
79
+ try:
80
+ from core.dependencies import get_memory_service
81
+ mem_service = get_memory_service()
82
+ asyncio.create_task(
83
+ mem_service.generate_document_dossier(
84
+ doc_id=doc_id,
85
+ filename=filename,
86
+ tree=tree,
87
+ flat_sections=flat_sections,
88
+ )
89
+ )
90
+ except Exception as mex:
91
+ app_logger.warning(f"Failed to launch dossier generation for {doc_id}: {mex}")
92
+
93
  app_logger.info(
94
  f"[INDEX] document_id={doc_id} stage=ready duration={total_duration:.2f}s "
95
  f"method={indexing_method} pages={page_count} sections={section_count}"
app/routers/memory.py ADDED
@@ -0,0 +1,98 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import time
2
+ from fastapi import APIRouter, HTTPException, Depends
3
+ from typing import Optional, List, Dict, Any
4
+ from pydantic import BaseModel
5
+
6
+ from core.dependencies import get_memory_service, get_storage
7
+ from core.storage import MemoryStorage
8
+ from services.memory_service import MemoryService
9
+ from utils.logger import app_logger
10
+
11
+ router = APIRouter(prefix="/memory", tags=["Memory & Knowledge"])
12
+
13
+
14
+ @router.get("/knowledge")
15
+ async def get_global_knowledge(
16
+ memory_service: MemoryService = Depends(get_memory_service),
17
+ ):
18
+ """
19
+ Retrieve workspace collective knowledge summary and all per-document knowledge dossiers.
20
+ Provides cross-document intelligence and long-term memory across sessions.
21
+ """
22
+ return memory_service.get_global_knowledge()
23
+
24
+
25
+ @router.post("/knowledge/refresh")
26
+ async def refresh_global_knowledge(
27
+ memory_service: MemoryService = Depends(get_memory_service),
28
+ ):
29
+ """
30
+ Synthesize and rebuild the global workspace knowledge memory across all indexed documents.
31
+ """
32
+ app_logger.info("Triggered manual refresh of global knowledge memory")
33
+ result = await memory_service.synthesize_global_knowledge()
34
+ dossiers = memory_service.list_knowledge_dossiers()
35
+ return {
36
+ "message": "Global knowledge memory refreshed successfully",
37
+ "collective_summary": result.get("collective_summary"),
38
+ "key_insights": result.get("key_insights", []),
39
+ "total_documents": result.get("total_documents", len(dossiers)),
40
+ "dossiers": dossiers,
41
+ }
42
+
43
+
44
+ @router.get("/dossier/{doc_id}")
45
+ async def get_document_dossier(
46
+ doc_id: str,
47
+ memory_service: MemoryService = Depends(get_memory_service),
48
+ ):
49
+ """
50
+ Retrieve structured knowledge dossier for an individual document.
51
+ """
52
+ dossier = memory_service.get_document_dossier(doc_id)
53
+ if not dossier:
54
+ raise HTTPException(status_code=404, detail="Knowledge dossier not found for this document")
55
+ return dossier
56
+
57
+
58
+ @router.get("/chat/{chat_id}")
59
+ async def get_chat_memory_context(
60
+ chat_id: str,
61
+ memory_service: MemoryService = Depends(get_memory_service),
62
+ storage: MemoryStorage = Depends(get_storage),
63
+ ):
64
+ """
65
+ Inspect the short-term and long-term isolated memory for a specific chat session.
66
+ """
67
+ if not storage.chat_exists(chat_id):
68
+ raise HTTPException(status_code=404, detail="Chat session not found")
69
+
70
+ context = memory_service.get_session_context(chat_id)
71
+ attached_docs = storage.get_chat_docs(chat_id)
72
+ return {
73
+ "chat_id": chat_id,
74
+ "session_summary": context.get("session_summary"),
75
+ "recent_turns_count": len(context.get("recent_turns", [])),
76
+ "recent_turns": context.get("recent_turns", []),
77
+ "attached_documents": attached_docs,
78
+ }
79
+
80
+
81
+ @router.post("/chat/{chat_id}/summarize")
82
+ async def force_summarize_chat(
83
+ chat_id: str,
84
+ memory_service: MemoryService = Depends(get_memory_service),
85
+ storage: MemoryStorage = Depends(get_storage),
86
+ ):
87
+ """
88
+ Force-generate or update the rolling long-term memory summary for this chat session.
89
+ """
90
+ if not storage.chat_exists(chat_id):
91
+ raise HTTPException(status_code=404, detail="Chat session not found")
92
+
93
+ summary = await memory_service.update_session_summary_if_needed(chat_id, force=True)
94
+ return {
95
+ "message": "Chat session summary updated",
96
+ "chat_id": chat_id,
97
+ "summary": summary,
98
+ }
app/schemas/schemas.py CHANGED
@@ -5,9 +5,17 @@ class AskRequest(BaseModel):
5
  chat_id: str = Field(..., description="Active chat session ID")
6
  query: str = Field(..., description="User query / question")
7
  doc_id: Optional[str] = Field(None, description="Optional document ID if attached")
 
8
  doc_name: Optional[str] = Field(None, description="Optional document name if known")
9
  deep_analysis: Optional[bool] = Field(False, description="Enable deep analysis mode")
10
 
 
 
 
 
 
 
 
11
  class UploadResponse(BaseModel):
12
  message: str = "Success"
13
  doc_id: str
 
5
  chat_id: str = Field(..., description="Active chat session ID")
6
  query: str = Field(..., description="User query / question")
7
  doc_id: Optional[str] = Field(None, description="Optional document ID if attached")
8
+ doc_ids: Optional[List[str]] = Field(None, description="Optional list of document IDs for multi-doc search")
9
  doc_name: Optional[str] = Field(None, description="Optional document name if known")
10
  deep_analysis: Optional[bool] = Field(False, description="Enable deep analysis mode")
11
 
12
+ class AttachDocRequest(BaseModel):
13
+ doc_id: str
14
+ doc_name: Optional[str] = None
15
+
16
+ class AttachMultipleDocsRequest(BaseModel):
17
+ doc_ids: List[str]
18
+
19
  class UploadResponse(BaseModel):
20
  message: str = "Success"
21
  doc_id: str
app/services/llm_service.py CHANGED
@@ -296,6 +296,8 @@ CONVERSATION HISTORY
296
 
297
  return ""
298
 
 
 
299
  async def llm_tree_search(
300
  self,
301
  query: str,
@@ -693,6 +695,8 @@ If there is insufficient evidence:
693
  chat_history: list = None,
694
  doc_name: Optional[str] = None,
695
  model: Optional[str] = None,
 
 
696
  deep_analysis: bool = False,
697
  ):
698
  """Chat & synthesis model generates streaming answer with multimodal citations."""
@@ -717,10 +721,13 @@ If there is insufficient evidence:
717
  text = (node.get("text") or node.get("content") or "Content not available.").strip()
718
  if len(text) > max_node_chars:
719
  text = text[:max_node_chars].rstrip() + "\n[...section continues...]"
720
- part = f"[Section: '{title}' | Page {page}]\n{text}{image_markers}"
 
 
 
721
  if current_total_chars + len(part) > max_total_chars and context_parts:
722
  remaining_budget = max(500, max_total_chars - current_total_chars)
723
- part = f"[Section: '{title}' | Page {page}]\n{text[:remaining_budget].rstrip()}\n[...truncated for context limit...]{image_markers}"
724
  context_parts.append(part)
725
  break
726
  context_parts.append(part)
@@ -730,6 +737,14 @@ If there is insufficient evidence:
730
 
731
  effective_doc_name = doc_name or "Uploaded Document"
732
 
 
 
 
 
 
 
 
 
733
  system_prompt = f"""You are Atlas, a production-grade document intelligence assistant operating inside a Retrieval-Augmented Generation (RAG) system.
734
  Your name is Atlas. When asked about your identity, name, or purpose, introduce yourself clearly as Atlas, the Document Intelligence research assistant.
735
 
@@ -1041,7 +1056,7 @@ CHAT HISTORY = CONTEXT
1041
 
1042
  Never treat an assumption from chat history as a document fact.
1043
 
1044
- ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
1045
  DOCUMENT CONTEXT
1046
  ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
1047
 
@@ -1168,9 +1183,12 @@ Return only the final user-facing answer."""
1168
  query: str,
1169
  chat_history: list = None,
1170
  model: Optional[str] = None,
 
 
1171
  ):
1172
- """Conversational stream when no document is attached yet.
1173
  Uses the smaller/faster retrieval model for low-latency chat, with automatic Groq failover.
 
1174
  """
1175
  # Use chat_model (Qwen2.5-7B from HF_CHAT_MODEL env) — fast and capable for conversational use
1176
  target_model = model or self.chat_model
@@ -1192,6 +1210,12 @@ Always match the user's language.
1192
  Be concise and accurate. Never invent facts.
1193
  If the user asks about a document and none is uploaded, tell them to upload a PDF.
1194
  """
 
 
 
 
 
 
1195
  if chat_history_text:
1196
  system_prompt += f"\n[Recent conversation]\n{chat_history_text}\n"
1197
 
 
296
 
297
  return ""
298
 
299
+ summarize_conversation = summarize_history
300
+
301
  async def llm_tree_search(
302
  self,
303
  query: str,
 
695
  chat_history: list = None,
696
  doc_name: Optional[str] = None,
697
  model: Optional[str] = None,
698
+ session_summary: Optional[str] = None,
699
+ global_knowledge: Optional[str] = None,
700
  deep_analysis: bool = False,
701
  ):
702
  """Chat & synthesis model generates streaming answer with multimodal citations."""
 
721
  text = (node.get("text") or node.get("content") or "Content not available.").strip()
722
  if len(text) > max_node_chars:
723
  text = text[:max_node_chars].rstrip() + "\n[...section continues...]"
724
+
725
+ doc_label = node.get("doc_name") or node.get("filename")
726
+ doc_prefix = f"Document: '{doc_label}' | " if doc_label else ""
727
+ part = f"[{doc_prefix}Section: '{title}' | Page {page}]\n{text}{image_markers}"
728
  if current_total_chars + len(part) > max_total_chars and context_parts:
729
  remaining_budget = max(500, max_total_chars - current_total_chars)
730
+ part = f"[{doc_prefix}Section: '{title}' | Page {page}]\n{text[:remaining_budget].rstrip()}\n[...truncated for context limit...]{image_markers}"
731
  context_parts.append(part)
732
  break
733
  context_parts.append(part)
 
737
 
738
  effective_doc_name = doc_name or "Uploaded Document"
739
 
740
+ session_summary_str = ""
741
+ if session_summary:
742
+ session_summary_str = f"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\nSESSION LONG-TERM MEMORY\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n{session_summary}\n\n"
743
+
744
+ global_knowledge_str = ""
745
+ if global_knowledge:
746
+ global_knowledge_str = f"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\nGLOBAL KNOWLEDGE MEMORY (All workspace documents)\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n{global_knowledge}\n\n"
747
+
748
  system_prompt = f"""You are Atlas, a production-grade document intelligence assistant operating inside a Retrieval-Augmented Generation (RAG) system.
749
  Your name is Atlas. When asked about your identity, name, or purpose, introduce yourself clearly as Atlas, the Document Intelligence research assistant.
750
 
 
1056
 
1057
  Never treat an assumption from chat history as a document fact.
1058
 
1059
+ {session_summary_str}{global_knowledge_str}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
1060
  DOCUMENT CONTEXT
1061
  ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
1062
 
 
1183
  query: str,
1184
  chat_history: list = None,
1185
  model: Optional[str] = None,
1186
+ session_summary: Optional[str] = None,
1187
+ global_knowledge: Optional[str] = None,
1188
  ):
1189
+ """Conversational stream when no specific document is attached yet.
1190
  Uses the smaller/faster retrieval model for low-latency chat, with automatic Groq failover.
1191
+ Integrates Session Long-Term Memory and Global Knowledge Memory across all documents.
1192
  """
1193
  # Use chat_model (Qwen2.5-7B from HF_CHAT_MODEL env) — fast and capable for conversational use
1194
  target_model = model or self.chat_model
 
1210
  Be concise and accurate. Never invent facts.
1211
  If the user asks about a document and none is uploaded, tell them to upload a PDF.
1212
  """
1213
+ if session_summary:
1214
+ system_prompt += f"\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\nSESSION LONG-TERM MEMORY (Summary of earlier turns)\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n{session_summary}\n"
1215
+
1216
+ if global_knowledge:
1217
+ system_prompt += f"\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━���━━━\nGLOBAL KNOWLEDGE MEMORY (All workspace documents)\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n{global_knowledge}\nYou can answer questions about the collective documents in the workspace using the Global Knowledge Memory above.\n"
1218
+
1219
  if chat_history_text:
1220
  system_prompt += f"\n[Recent conversation]\n{chat_history_text}\n"
1221
 
app/services/memory_service.py ADDED
@@ -0,0 +1,271 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """
2
+ memory_service.py
3
+ ─────────────────
4
+ Manages Multi-Tier Memory for Atlas Document Intelligence:
5
+ 1. Short-Term Session Memory: Strict per-session sliding context window.
6
+ 2. Long-Term Session Memory: Rolling conversation summaries for long chats.
7
+ 3. Document Knowledge Dossiers: Synthesized core findings & entities per document.
8
+ 4. Global Knowledge Memory: Collective synthesis across all ingested documents.
9
+ """
10
+
11
+ import json
12
+ import time
13
+ from typing import Any, Dict, List, Optional
14
+
15
+ from core.storage import storage
16
+ from utils.logger import app_logger
17
+
18
+
19
+ class MemoryService:
20
+ def __init__(self, llm_service) -> None:
21
+ self.llm = llm_service
22
+
23
+ # ── 1. Short-Term & Long-Term Session Memory ────────────────────────────
24
+ def get_session_context(
25
+ self, chat_id: str, max_recent_turns: int = 6
26
+ ) -> Dict[str, Any]:
27
+ """
28
+ Retrieves strictly isolated session memory:
29
+ - Recent turns (short-term memory window)
30
+ - Rolling session summary (long-term memory)
31
+ """
32
+ full_history = storage.get_chat_history(chat_id) or []
33
+ summary_record = storage.get_chat_summary(chat_id)
34
+ session_summary = summary_record.get("summary") if summary_record else None
35
+
36
+ recent_turns = full_history[-max_recent_turns:] if len(full_history) > max_recent_turns else full_history
37
+
38
+ return {
39
+ "chat_id": chat_id,
40
+ "session_summary": session_summary,
41
+ "recent_turns": recent_turns,
42
+ "total_messages": len(full_history),
43
+ }
44
+
45
+ async def update_session_summary_if_needed(self, chat_id: str) -> None:
46
+ """Trigger rolling summarization if conversation length exceeds threshold."""
47
+ try:
48
+ full_history = storage.get_chat_history(chat_id) or []
49
+ if len(full_history) < 8:
50
+ return
51
+
52
+ last_summary = storage.get_chat_summary(chat_id)
53
+ last_count = last_summary.get("turn_count", 0) if last_summary else 0
54
+
55
+ # Only re-summarize if at least 4 new messages occurred
56
+ if (len(full_history) - last_count) < 4:
57
+ return
58
+
59
+ app_logger.info(f"[MEMORY] Updating rolling summary for chat_id={chat_id}")
60
+ summary = await self.llm.summarize_conversation(full_history)
61
+ if summary:
62
+ storage.set_chat_summary(chat_id, summary, turn_count=len(full_history))
63
+ app_logger.info(f"[MEMORY] Successfully updated summary for chat_id={chat_id}")
64
+ except Exception as e:
65
+ app_logger.error(f"[MEMORY] Error updating session summary: {e}")
66
+
67
+ # ── 2. Document Knowledge Dossier ──────────────────────────────────────
68
+ async def generate_document_dossier(
69
+ self, doc_id: str, filename: str, tree: list, flat_sections: Optional[list] = None
70
+ ) -> Optional[Dict[str, Any]]:
71
+ """
72
+ Analyzes a newly indexed document to generate an executive dossier:
73
+ - Executive summary
74
+ - Key entities
75
+ - Key findings
76
+ - Topics
77
+ """
78
+ try:
79
+ # Build representative text preview from first few sections or flat sections
80
+ sections_to_read = flat_sections[:8] if flat_sections else []
81
+ preview_texts = []
82
+ for s in sections_to_read:
83
+ title = s.get("title", "Section")
84
+ text = (s.get("text") or "")[:800]
85
+ preview_texts.append(f"### {title}\n{text}")
86
+
87
+ if not preview_texts:
88
+ # Fallback to tree node text
89
+ for node in tree[:5]:
90
+ if isinstance(node, dict):
91
+ t = node.get("title", "")
92
+ c = (node.get("text") or "")[:800]
93
+ preview_texts.append(f"### {t}\n{c}")
94
+
95
+ combined_preview = "\n\n".join(preview_texts)[:4000]
96
+ if not combined_preview.strip():
97
+ combined_preview = f"Document: {filename} with {len(tree)} outline nodes."
98
+
99
+ prompt = f"""You are the Document Intelligence Synthesizer for Atlas.
100
+ Analyze the following document preview and extract a structured knowledge dossier.
101
+
102
+ DOCUMENT: {filename}
103
+ CONTENT PREVIEW:
104
+ {combined_preview}
105
+
106
+ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
107
+ OUTPUT REQUIREMENTS
108
+ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
109
+ Return ONLY valid JSON matching this schema:
110
+ {{
111
+ "executive_summary": "A 2-3 paragraph comprehensive summary of the document purpose, findings, and themes.",
112
+ "key_entities": ["Entity 1", "Entity 2", "Entity 3"],
113
+ "key_findings": ["Key finding 1", "Key finding 2", "Key finding 3"],
114
+ "topics": ["Topic 1", "Topic 2", "Topic 3", "Topic 4"]
115
+ }}
116
+ """
117
+ raw_result = ""
118
+ if not self.llm.hf_exhausted:
119
+ try:
120
+ resp = await self.llm.client.chat_completion(
121
+ messages=[{"role": "user", "content": prompt}],
122
+ model=self.llm.chat_model,
123
+ temperature=0.2,
124
+ max_tokens=800,
125
+ )
126
+ raw_result = resp.choices[0].message.content or ""
127
+ except Exception as e:
128
+ self.llm._is_hf_exhausted(e)
129
+ app_logger.warning(f"HF dossier generation failed ({e}). Checking Groq fallback...")
130
+
131
+ if (not raw_result or self.llm.hf_exhausted) and self.llm.groq_client:
132
+ groq_resp = await self.llm.groq_client.chat.completions.create(
133
+ model=self.llm.groq_chat_model,
134
+ messages=[{"role": "user", "content": prompt}],
135
+ temperature=0.2,
136
+ max_tokens=800,
137
+ )
138
+ raw_result = groq_resp.choices[0].message.content or ""
139
+
140
+ data = self.llm._extract_json(raw_result)
141
+ summary = data.get("executive_summary") or f"Knowledge dossier for {filename}."
142
+ entities = data.get("key_entities") or []
143
+ findings = data.get("key_findings") or []
144
+ topics = data.get("topics") or []
145
+
146
+ storage.save_knowledge_dossier(
147
+ doc_id=doc_id,
148
+ filename=filename,
149
+ executive_summary=summary,
150
+ key_entities=entities,
151
+ key_findings=findings,
152
+ topics=topics,
153
+ )
154
+ app_logger.info(f"[MEMORY] Saved knowledge dossier for doc_id={doc_id} ({filename})")
155
+
156
+ # Asynchronously update global knowledge summary
157
+ await self.synthesize_global_knowledge()
158
+
159
+ return {
160
+ "doc_id": doc_id,
161
+ "filename": filename,
162
+ "executive_summary": summary,
163
+ "key_entities": entities,
164
+ "key_findings": findings,
165
+ "topics": topics,
166
+ }
167
+ except Exception as e:
168
+ app_logger.error(f"[MEMORY] Failed to generate document dossier for {doc_id}: {e}", exc_info=True)
169
+ return None
170
+
171
+ # ── 3. Global Knowledge Summary Across All Data ────────────────────────
172
+ async def synthesize_global_knowledge(self) -> Dict[str, Any]:
173
+ """
174
+ Synthesizes a collective master knowledge summary of all indexed documents in the workspace.
175
+ """
176
+ dossiers = storage.list_knowledge_dossiers()
177
+ total_docs = len(dossiers)
178
+
179
+ if not dossiers:
180
+ # Check if any documents exist in doc_meta without dossiers yet
181
+ all_docs = storage.list_documents()
182
+ if not all_docs:
183
+ default_summary = "No documents currently indexed in the knowledge base."
184
+ storage.save_global_knowledge_summary(default_summary, 0, [])
185
+ return {"collective_summary": default_summary, "total_documents": 0, "key_insights": []}
186
+
187
+ # Build combined summary representation of all dossiers
188
+ dossier_summaries = []
189
+ for d in dossiers:
190
+ fname = d.get("filename", "Doc")
191
+ summ = d.get("executive_summary", "")[:600]
192
+ findings = "; ".join(d.get("key_findings", [])[:3])
193
+ topics = ", ".join(d.get("topics", [])[:4])
194
+ dossier_summaries.append(
195
+ f"- **{fname}**\n Summary: {summ}\n Topics: {topics}\n Findings: {findings}"
196
+ )
197
+
198
+ combined_input = "\n\n".join(dossier_summaries)
199
+
200
+ prompt = f"""You are the Chief Knowledge Architect for Atlas Document Intelligence.
201
+ Synthesize the collective knowledge across all {total_docs} documents ingested into the platform.
202
+
203
+ INDEXED DOCUMENTS DOSSIERS:
204
+ {combined_input}
205
+
206
+ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
207
+ OUTPUT REQUIREMENTS
208
+ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
209
+ Return ONLY valid JSON with this exact schema:
210
+ {{
211
+ "collective_summary": "A cohesive 2-4 paragraph overview of the global knowledge contained in all documents, highlighting overarching themes, areas of synergy, and domain expertise.",
212
+ "key_insights": [
213
+ "Cross-cutting insight 1",
214
+ "Cross-cutting insight 2",
215
+ "Cross-cutting insight 3"
216
+ ]
217
+ }}
218
+ """
219
+ raw_result = ""
220
+ if not self.llm.hf_exhausted:
221
+ try:
222
+ resp = await self.llm.client.chat_completion(
223
+ messages=[{"role": "user", "content": prompt}],
224
+ model=self.llm.chat_model,
225
+ temperature=0.3,
226
+ max_tokens=1000,
227
+ )
228
+ raw_result = resp.choices[0].message.content or ""
229
+ except Exception as e:
230
+ self.llm._is_hf_exhausted(e)
231
+ app_logger.warning(f"HF global knowledge synthesis failed ({e}). Checking Groq fallback...")
232
+
233
+ if (not raw_result or self.llm.hf_exhausted) and self.llm.groq_client:
234
+ try:
235
+ groq_resp = await self.llm.groq_client.chat.completions.create(
236
+ model=self.llm.groq_chat_model,
237
+ messages=[{"role": "user", "content": prompt}],
238
+ temperature=0.3,
239
+ max_tokens=1000,
240
+ )
241
+ raw_result = groq_resp.choices[0].message.content or ""
242
+ except Exception as ge:
243
+ app_logger.error(f"Groq global knowledge synthesis failed: {ge}")
244
+
245
+ data = self.llm._extract_json(raw_result)
246
+ collective = data.get("collective_summary") or "Atlas Global Knowledge Base synthesis complete."
247
+ insights = data.get("key_insights") or []
248
+
249
+ storage.save_global_knowledge_summary(collective, total_docs, insights)
250
+ app_logger.info(f"[MEMORY] Successfully synthesized global knowledge across {total_docs} documents.")
251
+
252
+ return {
253
+ "collective_summary": collective,
254
+ "total_documents": total_docs,
255
+ "key_insights": insights,
256
+ "updated_at": time.time(),
257
+ }
258
+
259
+ def get_global_knowledge(self) -> Dict[str, Any]:
260
+ """Fetch current global knowledge summary and document dossiers."""
261
+ global_rec = storage.get_global_knowledge_summary() or {}
262
+ dossiers = storage.list_knowledge_dossiers()
263
+ return {
264
+ "collective_summary": global_rec.get(
265
+ "collective_summary", "No collective summary available yet. Ingest documents to build knowledge."
266
+ ),
267
+ "total_documents": global_rec.get("total_documents", len(dossiers)),
268
+ "key_insights": global_rec.get("key_insights", []),
269
+ "updated_at": global_rec.get("updated_at"),
270
+ "dossiers": dossiers,
271
+ }
test_memory_multidoc.py ADDED
@@ -0,0 +1,122 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import os
2
+ import sys
3
+
4
+ # Add app to path
5
+ app_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "app"))
6
+ if app_dir not in sys.path:
7
+ sys.path.insert(0, app_dir)
8
+
9
+ from fastapi.testclient import TestClient
10
+ from app import app
11
+ from core.storage import storage
12
+ from core.persistent_db import persistent_db
13
+
14
+ client = TestClient(app)
15
+
16
+ def test_full_memory_and_multidoc_isolation():
17
+ print("\n--- 1. Testing Session Creation & Isolation ---")
18
+ res_a = client.post("/chat/new")
19
+ assert res_a.status_code == 200
20
+ chat_a = res_a.json()["chat_id"]
21
+
22
+ res_b = client.post("/chat/new")
23
+ assert res_b.status_code == 200
24
+ chat_b = res_b.json()["chat_b" if "chat_b" in res_b.json() else "chat_id"]
25
+
26
+ print(f"Chat A ID: {chat_a}")
27
+ print(f"Chat B ID: {chat_b}")
28
+ assert chat_a != chat_b
29
+
30
+ # Create dummy mock docs in storage
31
+ doc1_id = "mock-doc-1"
32
+ doc2_id = "mock-doc-2"
33
+ mock_tree_1 = [
34
+ {"title": "Section 1: Architecture", "page": 1, "text": "Atlas uses tree-based retrieval."},
35
+ {"title": "Section 2: Memory", "page": 2, "text": "Session memory isolates short and long context."},
36
+ ]
37
+ mock_tree_2 = [
38
+ {"title": "Overview: Financials", "page": 1, "text": "Revenue grew by 24% in Q3."},
39
+ ]
40
+
41
+ storage.set_document_ready(doc1_id, tree=mock_tree_1, filename="Architecture_Guide.pdf", page_count=2, section_count=2)
42
+ storage.set_document_ready(doc2_id, tree=mock_tree_2, filename="Financial_Report_Q3.pdf", page_count=1, section_count=1)
43
+
44
+ print("\n--- 2. Testing Multi-Document Attachment to Chat A ---")
45
+ attach_res = client.post(f"/chat/{chat_a}/attach-multiple", json={"doc_ids": [doc1_id, doc2_id]})
46
+ assert attach_res.status_code == 200
47
+ assert len(attach_res.json()["doc_ids"]) == 2
48
+
49
+ docs_a_res = client.get(f"/chat/{chat_a}/documents")
50
+ assert docs_a_res.status_code == 200
51
+ docs_a = docs_a_res.json()["documents"]
52
+ print(f"Chat A attached documents count: {len(docs_a)}")
53
+ assert len(docs_a) == 2
54
+ assert {d["doc_id"] for d in docs_a} == {doc1_id, doc2_id}
55
+
56
+ # Verify Chat B has zero attached documents (Strict Session Isolation)
57
+ docs_b_res = client.get(f"/chat/{chat_b}/documents")
58
+ assert docs_b_res.status_code == 200
59
+ docs_b = docs_b_res.json()["documents"]
60
+ print(f"Chat B attached documents count: {len(docs_b)}")
61
+ assert len(docs_b) == 0, "Violation of session isolation: Chat B should have 0 docs"
62
+
63
+ print("\n--- 3. Testing Message Isolation Between Sessions ---")
64
+ storage.add_message(chat_a, "user", "What is Atlas?")
65
+ storage.add_message(chat_a, "assistant", "Atlas is a document intelligence platform.")
66
+
67
+ storage.add_message(chat_b, "user", "Hello from user B")
68
+ storage.add_message(chat_b, "assistant", "Greetings, this is session B.")
69
+
70
+ hist_a = client.get(f"/chat/{chat_a}/history").json()["history"]
71
+ hist_b = client.get(f"/chat/{chat_b}/history").json()["history"]
72
+ assert len(hist_a) == 2
73
+ assert hist_a[0]["content"] == "What is Atlas?"
74
+ assert len(hist_b) == 2
75
+ assert hist_b[0]["content"] == "Hello from user B"
76
+ print("Session history isolation verified!")
77
+
78
+ print("\n--- 4. Testing Session Memory Diagnostic Endpoint ---")
79
+ mem_res = client.get(f"/memory/chat/{chat_a}")
80
+ assert mem_res.status_code == 200
81
+ mem_data = mem_res.json()
82
+ assert mem_data["chat_id"] == chat_a
83
+ assert mem_data["recent_turns_count"] == 2
84
+ print(f"Memory diagnostic for Chat A: {mem_data['recent_turns_count']} turns recorded.")
85
+
86
+ print("\n--- 5. Testing Detach Single Document ---")
87
+ detach_res = client.delete(f"/chat/{chat_a}/document/{doc1_id}")
88
+ assert detach_res.status_code == 200
89
+ remaining_docs = client.get(f"/chat/{chat_a}/documents").json()["documents"]
90
+ assert len(remaining_docs) == 1
91
+ assert remaining_docs[0]["doc_id"] == doc2_id
92
+ print(f"Remaining document in Chat A: {remaining_docs[0]['filename']}")
93
+
94
+ print("\n--- 6. Testing Global Knowledge & Dossier Storage ---")
95
+ # Save a test dossier
96
+ persistent_db.save_knowledge_dossier(
97
+ doc_id=doc1_id,
98
+ filename="Architecture_Guide.pdf",
99
+ executive_summary="Explains Atlas tree-based vectorless RAG architecture.",
100
+ key_entities=["Atlas", "PageIndex", "Groq", "FastAPI"],
101
+ key_findings=["Tree search outperforms embeddings on dense tabular texts."],
102
+ topics=["Architecture", "Document Intelligence", "FastAPI"]
103
+ )
104
+
105
+ persistent_db.save_global_knowledge_summary(
106
+ collective_summary="Atlas workspace contains technical documentation on tree retrieval and quarterly financial reports.",
107
+ total_documents=2,
108
+ key_insights=["Tree retrieval avoids embedding drift", "Multi-doc synthesis enables cross-sectional queries"]
109
+ )
110
+
111
+ g_res = client.get("/memory/knowledge")
112
+ assert g_res.status_code == 200
113
+ g_data = g_res.json()
114
+ print(f"Global Knowledge Summary: {g_data['collective_summary']}")
115
+ print(f"Dossiers count: {len(g_data['dossiers'])}")
116
+ assert len(g_data["dossiers"]) >= 1
117
+ assert "Atlas" in g_data["collective_summary"]
118
+
119
+ print("\n ALL BACKEND MULTI-DOC & MEMORY TESTS PASSED PERFECTLY!")
120
+
121
+ if __name__ == "__main__":
122
+ test_full_memory_and_multidoc_isolation()