chore: update requirements and refactor benchmark methods to use synchronous recommendations

benchmarks/benchmark.py

@@ -66,7 +66,7 @@ def benchmark_full_recommendation(recommender: BookRecommender, n_runs: int = 30
     for query in TEST_QUERIES:
         for _ in range(n_runs // len(TEST_QUERIES)):
             start = time.perf_counter()
-            recommender.get_recommendations(query, category="All", tone="All")
+            recommender.get_recommendations_sync(query, category="All", tone="All")
             latencies.append((time.perf_counter() - start) * 1000)
     
     return {
@@ -88,7 +88,7 @@ def benchmark_throughput(recommender: BookRecommender, duration_sec: int = 10) -
     query_idx = 0
     
     while (time.perf_counter() - start) < duration_sec:
-        recommender.get_recommendations(
+        recommender.get_recommendations_sync(
             TEST_QUERIES[query_idx % len(TEST_QUERIES)],
             category="All",
             tone="All"

config/router.json

@@ -0,0 +1,12 @@
+{
+  "detail_keywords": [
+    "twist", "ending", "spoiler", "readers", "felt", "cried", "hated", "loved",
+    "review", "opinion", "think", "unreliable", "narrator", "realize", "find out"
+  ],
+  "freshness_keywords": [
+    "new", "newest", "latest", "recent", "modern", "contemporary", "current"
+  ],
+  "strong_freshness_keywords": [
+    "newest", "latest"
+  ]
+}

docs/TECHNICAL_REPORT.md

@@ -316,6 +316,16 @@ Feature importance (v2.6.0 LGBMRanker, representative subset):
 | Reranking | Cross-Encoder | LLM reranking | 400ms vs 2s latency; proven accuracy |
 | Chunking | Sentence-level (Small-to-Big) | Fixed 512 tokens | Semantic integrity; detail-level matching |
 | SFT Data | Self-Instruct | Manual annotation | Scalable; leverages existing reviews |
+| Freshness fallback writes | Staging store (`online_books.db`) | Append to `books_processed.csv` | Data: training CSV stays frozen. perf: main `books.db` read-only; no write lock contention |
+
+### 7.1 Staging Store for Online Writes
+
+When `freshness_fallback` fetches books from Google Books, they are written to a **separate** `online_books.db` SQLite file instead of the main store. This decouples:
+
+1. **Data risk**: `books_processed.csv` and `books.db` remain frozen for training; no distribution shift.
+2. **Performance**: Main `books.db` is read-only during serving; writes go only to `online_books.db`, avoiding lock contention on high-concurrency reads.
+
+Lookup: `metadata_store.get_book_metadata()` checks main first, then `online_books_store`. FTS5 search merges results from both indices.
 
 ---
 
@@ -351,7 +361,10 @@ src/
 │   ├── router.py              # Agentic Query Router
 │   ├── reranker.py            # Cross-Encoder Reranking
 │   ├── temporal.py            # Recency Boosting
-│   └── context_compressor.py  # Chat History Compression
+│   ├── context_compressor.py  # Chat History Compression
+│   ├── diversity_reranker.py  # P0: MMR + popularity penalty + category constraint
+│   ├── diversity_metrics.py  # P3: Category Coverage, ILSD
+│   └── online_books_store.py  # Staging store for freshness_fallback (separate DB)
 ├── recall/
 │   ├── itemcf.py              # ItemCF Recall (direction-weighted)
 │   ├── usercf.py              # UserCF Recall
@@ -373,6 +386,18 @@ src/
 
 ---
 
+## 9.1 P0–P3 Optimizations (Post-v2.6)
+
+| Priority | Optimization | Location | Description |
+|:---|:---|:---|:---|
+| **P0** | Diversity Rerank | `DiversityReranker`, `RecommendationService` | MMR (λ=0.75), popularity penalty, max 3 per category in top-k |
+| **P1** | Real-time Sequence | `SASRecRecall`, `DINRanker`, `FeatureEngineer`, `RecommendationService` | `real_time_sequence` merges session ISBNs into recall/ranking |
+| **P2** | Hard/Random Ratio | `train_ranker.py`, `train_din_ranker.py` | `--hard_ratio 0.5` for half hard half random negatives |
+| **P3** | Diversity Metrics | `evaluate.py`, `diversity_metrics.py` | Category Coverage@10, ILSD@10 reported |
+| **P3** | Hard Neg Filter | `train_ranker.py --filter_similar` | Exclude hard negs with embedding sim > 0.9 to positive |
+
+---
+
 ## 10. Limitations
 
 - **Single-dataset evaluation**: All RecSys metrics are on Amazon Books 200K; no cross-domain or external validation.

docs/build_guide.md

@@ -85,17 +85,22 @@ Place in `data/raw/`:
 - `books_data.csv` - Book metadata (title, author, description, categories)
 - `Books_rating.csv` - User ratings (User_id, Id, review/score, review/time, review/text)
 
-### 2.2 Data Processing Scripts
+### 2.2 Pipeline DAG (Execution Order)
 
-| Order | Script | Purpose | Output |
+**Recommended**: Use `make data-pipeline` or `python scripts/run_pipeline.py` — it defines the full DAG.
+
+| Stage | Script | Purpose | Output |
 |:---:|:---|:---|:---|
-| 0 | `clean_data.py` | HTML/encoding/whitespace cleanup | books_processed.csv (cleaned) |
-| 1 | `build_books_basic_info.py` | Extract basic book info | books_basic_info.csv |
-| 2 | `generate_emotions.py` | Sentiment analysis (5 emotions) | +joy,sadness,fear,anger,surprise |
-| 3 | `generate_tags.py` | TF-IDF keyword extraction | +tags column |
-| 4 | `split_rec_data.py` | Leave-Last-Out time split | rec/train,val,test.csv |
-| 5 | `build_sequences.py` | User history → sequences | rec/user_sequences.pkl |
+| 1 | `build_books_basic_info.py` | Merge raw books + ratings | books_basic_info.csv |
+| 2 | *books_processed.csv* | From HuggingFace or manual merge of basic_info + review_highlights | books_processed.csv |
+| 3 | `clean_data.py` | HTML/encoding/whitespace cleanup | books_processed.csv (cleaned) |
+| 4 | `generate_emotions.py` | Sentiment analysis (5 emotions) | +joy,sadness,fear,anger,surprise |
+| 5 | `generate_tags.py` | TF-IDF keyword extraction | +tags column |
 | 6 | `chunk_reviews.py` | Reviews → sentences | review_chunks.jsonl |
+| 7 | `split_rec_data.py` | Leave-Last-Out time split | rec/train,val,test.csv |
+| 8 | `build_sequences.py` | User history → sequences | rec/user_sequences.pkl |
+
+**Note**: `books_processed.csv` may be pre-downloaded from HuggingFace. If building from scratch, merge `books_basic_info.csv` with review data and run `extract_review_sentences.py` first.
 
 ### 2.3 Script Details
 
@@ -126,6 +131,8 @@ python scripts/data/split_rec_data.py
 python scripts/data/build_sequences.py
 ```
 
+**Script conventions**: Use `config.data_config` for paths; `scripts.utils.setup_script_logger()` for logging.
+
 ---
 
 ## Phase 3: Index Building

docs/interview_guide.md

@@ -73,7 +73,145 @@
 
 > "在 `src/model/sasrec.py` 中，你使用了 Transformer。在推理（Inference）阶段，如果用户每点一本书我们都要刷新推荐，SASRec 的计算成本是很高的。你如何缓存用户的 Embedding 状态以避免每次从头计算整个序列？"
 > *(考察点：对深度学习模型线上推理（Inference）优化的理解。关键在于 KV Cache 或者增量计算)*
+
+
+
+**Q4. metadata_store 的 SQLite 高并发改造：**
+
+> "在 recommender.py 中，你提到了 'Zero-RAM mode' 并从 SQLite 读取元数据。在高并发场景下（QPS > 1000），SQLite 的磁盘 I/O 会成为致命瓶颈。**如果现在系统 QPS 暴涨 100 倍，除了加机器，你会怎么改造 metadata_store 的读写架构？**"
+> *(考察点：对存储层 scaling 的理解。评议：通常会用 Redis/Memcached 做热数据缓存，或使用 Cassandra/HBase 列式存储)*
+
+**建议回答**:
+
+> "我会分阶段改造 metadata_store：
+>
+> 1. **短期**：在 SQLite 前加 Redis 读缓存，对 ISBN 做 key-value 缓存。metadata 是静态/准静态数据，热门书籍命中率可到 80%+，SQLite 压力可下降一个数量级。
+> 2. **中期**：抽象 MetadataStore 接口，实现 `CachedMetadataStore`（Redis + SQLite fallback），并新增 `get_book_metadata_batch()` 批量查询，减少 N 次往返变成 1 次。
+> 3. **长期**：若仍不足，可将 metadata 迁移到 PostgreSQL 或 Cassandra，Redis 做热数据缓存。SQLite 退化为冷备份或离线数据源。
+>
+> 核心思路：把 SQLite 从 '唯一真相源' 降级为 '冷数据源'，高频读写交给 Redis 或分布式存储。"
+>
+> **补充：Staging 写入**：freshness_fallback 的在线爬取写入 `online_books.db`（独立 SQLite），不污染 `books_processed.csv` 和主 `books.db`。既解耦训练数据污染，又避免写锁阻塞读（主库只读）。
+>
+
+---
+
+## 🔬 深度技术问题 (Advanced Technical Q&A)
+
+### Q5. 负采样 (Negative Sampling)
+
+**问题**：你在 TECHNICAL_REPORT 中使用了 "Hard negative sampling from recall results"。这样做会不会导致 **False Negative** 问题（即把用户其实喜欢但没点击的物品当成了负样本）？在训练 DIN 或 LGBMRanker 时，你是如何平衡 Random Negatives 和 Hard Negatives 的比例的？这对模型收敛有什么影响？
+
+**考察点**：对推荐系统训练数据构造的理解，以及负采样策略的 trade-off。
+
+**建议回答**：
+
+> **False Negative 风险**：存在。Hard negatives 来自 Recall 的 top-50 中「不是正样本」的 item。这些 item 很可能是用户会喜欢但尚未交互的（未曝光、未点击、或未来会点击）。若被标成负样本，就会形成 False Negative。Leave-Last-Out 下，正样本是用户最后一次交互；Recall 中其他 item 可能是「未来正样本」，却被当作负样本训练。
+>
+> **比例策略**：当前实现是「hard 优先，random 补齐」。`neg_ratio=4` 表示每个正样本 4 个负样本；先用 recall 中非正样本填满，不足时用 random 补齐。没有显式比例（如 2 hard + 2 random）。
+>
+> **收敛影响**：Hard negatives 梯度更有信息量，但 False Negative 会误导模型。可考虑 Curriculum Learning（先 random 后 hard）、或显式控制 hard:random 比例做实验。
+
+---
+
+### Q6. 实时性 (Real-time / Near-line)
+
+**问题**：SASRec 主要是离线训练的。在 Spotify 场景下，如果用户刚刚连续听了 3 首 "Heavy Metal"，我们希望下一首推荐立刻跟上这个兴趣变化。在目前的架构下，如何将用户的**实时交互序列**（还没落库到 CSV）注入到 SASRec 或 DIN 的推理过程中？需要在 `RecommendationService` 里增加什么逻辑？
+
+**考察点**：对离线训练 / 在线推理架构的理解，以及 session-level 实时反馈的工程实现。
+
+**建议回答**：
+
+> **当前架构**：SASRec 的 `user_seq_emb` 和 DIN 的 `user_sequences` 都来自预计算的 pkl 文件，无法利用 session 内实时交互。
+>
+> **需要增加的逻辑**：
+>
+> 1. **SASRecRecall**：新增 `recommend(user_id, ..., real_time_seq=None)`。当 `real_time_seq` 非空时，将 `effective_seq = (离线序列 + real_time_seq)[-max_len:]` 送入 SASRec 做一次 forward，得到新 `u_emb`，再查 Faiss。
+> 2. **DINRanker**：`predict(..., override_hist=None)`，用 `override_hist` 覆盖 `user_sequences.get(user_id)`。
+> 3. **FeatureEngineer**：`generate_features_batch(..., override_seq=None)`，用 override 序列计算 `sasrec_score`、`sim_max` 等。
+> 4. **RecommendationService**：`get_recommendations(..., real_time_sequence=None)`，收到 session 内最近交互的 ISBN 列表，合并后传给上述各模块。
 >
+> **注意**：新 item 不在 `item_map` 时需 fallback；SASRec forward 有计算开销，可对 session 做短时缓存（如 5 分钟内相同 seq 复用 embedding）。
+
+---
+
+### Q7. 评估指标：Diversity 与 Serendipity
+
+**问题**：目前关注的是 HR@10 和 NDCG。作为内容平台，发现推荐列表里全是热门书（Harry Potter 效应）。如果要求在不显著降低 Accuracy 的前提下，提升推荐结果的 **Diversity（多样性）** 和 **Serendipity（惊喜感）**，你会如何在 Ranking 阶段或 Rerank 阶段修改目标函数或逻辑？
+
+**考察点**：对推荐系统多目标优化、trade-off 的理解，以及常见 diversity / serendipity 手段。
+
+**建议回答**：
+
+> **Rerank 阶段（推荐优先）**：
+>
+> 1. **MMR（Maximal Marginal Relevance）**：`score = λ * relevance - (1-λ) * max_sim(candidate, already_selected)`，用 category 或 embedding 相似度，λ 控制 accuracy vs diversity。
+> 2. **Category 多样性约束**：限制 top-k 中同一 category 最多 N 本（如 2–3 本）。
+> 3. **Popularity 惩罚**：对高 `i_cnt` 的 item 降权，`score_adj = score / (1 + γ * log(1 + item_cnt))`。
+>
+> **Ranking 阶段**：
+>
+> - 增加 diversity 相关特征（如 `category_coverage`、`popularity_penalty`）。
+> - 多目标优化：`loss = NDCG_loss + α * (-diversity_score)`。
+>
+> **Serendipity**：惩罚与用户历史过度相似的 item（如 `sim_max` 上限）；或引入「意外但合理」的 item（同大类不同子类、同一作者不同风格）。
+>
+> **评估**：补充 ILSD、Category Coverage、Gini 等 diversity 指标，做 accuracy–diversity Pareto 曲线。
+
+---
+
+## 📋 已知限制与改进方向 (Known Limitations & Improvement)
+
+### Q6. "Research" 风格的代码残留
+
+**现象**：代码库在向 production 演进过程中，仍保留了一些研究原型风格的痕迹。
+
+#### 6.1 注释掉的代码与 print 语句
+
+| 位置 | 问题 | 建议 |
+|------|------|------|
+| `scripts/model/evaluate.py:38-40` | 注释掉的 `service.ranker_loaded = False` 和 debug logger | 删除或移至 `if DEBUG` 分支 |
+| `src/ranking/features.py:470` | `if __name__` 中的 `print(df_feats.head())` | 改为 `logger.debug` 或删除 |
+| `src/services/recommend_service.py:282-286` | `if __name__` 中的硬编码 print | 保留（仅主程序入口），可改为 `logger.info` |
+| `src/recall/fusion.py`, `itemcf.py`, `usercf.py`, `item2vec.py` | 各模块 `if __name__` 中的 test print | 统一改为 `logger.info` 或移入测试脚本 |
+
+**原则**：调试输出应受 `DEBUG` 控制，或仅在 `__main__` 下使用 `logger`，避免裸 `print`。
+
+#### 6.2 混合范式：Dict vs Pydantic / DataFrame
+
+**问题**：API 层使用 Pydantic 模型（`BookResponse`, `RecommendationResponse`），但内部大量传递 `Dict[str, Any]`，导致：
+
+- IDE 无法自动补全字段
+- 类型检查失效，易出现 `KeyError`（如 `meta.get("title")` 拼写错误难以发现）
+- 与 pandas 脚本式风格混用（`df['user_id'].iloc[0]` 直接取数据）
+
+**典型分布**：
+
+| 层级 | 当前形态 | 涉及文件 |
+|------|----------|----------|
+| API 入/出 | Pydantic ✅ | `main.py`: `BookResponse`, `RecommendationResponse` |
+| 内部传递 | `Dict[str, Any]` | `recommendation_orchestrator`, `response_formatter`, `metadata_store`, `fallback_provider`, `reranker` |
+| 数据层 | `pd.DataFrame` + `iloc` | `recommend_service`, `recall/fusion`, `ranking/features` |
+
+**改进方向**：
+
+1. **定义领域模型**：为书籍元数据、推荐结果引入 Pydantic 或 TypedDict：
+   ```python
+   class BookMetadata(BaseModel):
+       isbn: str
+       title: str
+       authors: str
+       description: str
+       thumbnail: Optional[str] = None
+       average_rating: float = 0.0
+       # ...
+   ```
+2. **内层使用强类型**：`format_book_response(meta: BookMetadata, ...)` 替代 `meta: Dict[str, Any]`。
+3. **`__main__` 入口**：用 `BookMetadata.model_validate(row)` 或显式构造，避免 `df.iloc[0]` 直接当 dict 用。
+
+**面试话术**：
+
+> "项目从研究原型迭代而来，内部仍有 `Dict[str, Any]` 和 pandas 脚本式写法。若继续演进，我会在核心推荐流向 Pydantic 或 TypedDict 迁移，减少 KeyError 并提升 IDE 支持；同时将 `__main__` 中的 print 统一为受 DEBUG 控制的 logger。"
 
 ---
 

requirements.txt

@@ -14,6 +14,7 @@ python-dotenv
 # LangChain components
 langchain
 langchain-community
+langgraph>=0.2.0
 langchain-text-splitters
 langchain-chroma
 langchain-huggingface

scripts/model/evaluate.py

@@ -7,10 +7,17 @@ import numpy as np
 import logging
 from tqdm import tqdm
 from src.services.recommend_service import RecommendationService
+from src.core.metadata_store import metadata_store
+from src.core.diversity_metrics import compute_diversity_metrics
 
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
 logger = logging.getLogger(__name__)
 
+
+def _get_category(isbn: str) -> str:
+    meta = metadata_store.get_book_metadata(str(isbn))
+    return (meta.get("simple_categories", "") or "Unknown").strip()
+
 def evaluate_baseline(sample_n=1000):
     logger.info("Initializing Evaluation...")
     
@@ -28,10 +35,6 @@ def evaluate_baseline(sample_n=1000):
     # 2. Init Service
     service = RecommendationService()
     service.load_resources()
-    # FORCE DISABLE RANKER for debugging - ENABLED NOW
-    # service.ranker_loaded = False
-    # logger.info("DEBUG: Ranker DISABLED to test Recall performance.")
-    
     # Load ISBN -> Title map for evaluation
     isbn_to_title = {}
     try:
@@ -46,10 +49,11 @@ def evaluate_baseline(sample_n=1000):
     k = 10
     hits = 0
     mrr_sum = 0.0
-    
-    # Cache for speed analysis
-    total_time = 0
-    
+    # P3: Diversity metrics (aggregate over all users)
+    diversity_cov_sum = 0.0
+    diversity_ilsd_sum = 0.0
+    diversity_count = 0
+
     results = []
     
     for idx, (_, row) in tqdm(enumerate(eval_df.iterrows()), total=len(eval_df), desc="Evaluating"):
@@ -59,8 +63,9 @@ def evaluate_baseline(sample_n=1000):
         # Get Recs
         try:
             # We disable favorite filtering for evaluation to handle potential data leakage in test set splits
-            recs = service.get_recommendations(user_id, top_k=50, filter_favorites=False) 
-            
+            recs = service.get_recommendations(user_id, top_k=50, filter_favorites=False)
+            # P3: Optional A/B test diversity: enable_diversity_rerank=True by default
+
             if not recs:
                 if idx < 5: 
                     logger.warning(f"Empty recs for user {user_id}")
@@ -89,6 +94,13 @@ def evaluate_baseline(sample_n=1000):
                            # logger.info(f"Title Match! Target: {target_isbn} ({target_title}) matches Rec: {r_isbn}")
                            break
             
+            # P3: Diversity metrics on top-10
+            if rec_isbns:
+                d = compute_diversity_metrics(rec_isbns, _get_category, top_k=10)
+                diversity_cov_sum += d["category_coverage"]
+                diversity_ilsd_sum += d["ilsd"]
+                diversity_count += 1
+
             if hit:
                 # HR@10
                 if rank < 10:
@@ -96,7 +108,7 @@ def evaluate_baseline(sample_n=1000):
 
                 # MRR (consider top 50)
                 # MRR@5 (Strict)
-                if (rank + 1) <= 5: # Check if rank is within top 5 (1-indexed)
+                if (rank + 1) <= 5:  # Check if rank is within top 5 (1-indexed)
                     mrr_sum += 1.0 / (rank + 1)
             else:
                 if idx < 5:
@@ -110,14 +122,16 @@ def evaluate_baseline(sample_n=1000):
 
     # 4. Report
     hr_10 = hits / len(eval_df)
-    mean_mrr = mrr_sum / len(eval_df) # Changed from mrr to mrr_sum
-    
+    mean_mrr = mrr_sum / len(eval_df)
+    div_n = max(diversity_count, 1)
     logger.info("==============================")
-    logger.info("  EVALUATION RESULTS (Strict)") # Changed title
+    logger.info("  EVALUATION RESULTS (Strict)")
     logger.info("==============================")
     logger.info(f"Users Evaluated: {len(eval_df)}")
     logger.info(f"Hit Rate@10:   {hr_10:.4f}")
-    logger.info(f"MRR@5:         {mean_mrr:.4f}") # Changed MRR@50 to MRR@5
+    logger.info(f"MRR@5:         {mean_mrr:.4f}")
+    logger.info(f"P3 Category Coverage@10: {diversity_cov_sum / div_n:.4f}")
+    logger.info(f"P3 ILSD@10:              {diversity_ilsd_sum / div_n:.4f}")
     logger.info("==============================")
 
 if __name__ == "__main__":

scripts/model/evaluate_rag.py

@@ -92,7 +92,7 @@ def evaluate_rag(
 
     for query, relevant_isbns in golden.items():
         try:
-            recs = recommender.get_recommendations(query, top_k=top_k * 2)
+            recs = recommender.get_recommendations_sync(query, category="All")
             rec_isbns = [r.get("isbn") or r.get("isbn13") for r in recs if r]
             rec_isbns = [str(x).replace(".0", "") for x in rec_isbns if pd.notna(x)]
             rec_top = rec_isbns[:top_k]

scripts/model/train_din_ranker.py

@@ -49,6 +49,7 @@ def build_din_data(
     data_dir: str = "data/rec",
     model_dir: str = "data/model/recall",
     neg_ratio: int = 4,
+    hard_ratio: float = 1.0,
     max_samples: int = 20000,
 ) -> tuple[pd.DataFrame, dict, dict]:
     """
@@ -77,9 +78,10 @@ def build_din_data(
 
         user_rows = [{"user_id": user_id, "isbn": pos_isbn, "label": 1}]
 
+        n_hard_max = max(0, int(neg_ratio * hard_ratio))
         try:
             recall_items = fusion.get_recall_items(user_id, k=50)
-            hard_negs = [item for item, _ in recall_items if item != pos_isbn][:neg_ratio]
+            hard_negs = [item for item, _ in recall_items if item != pos_isbn][:n_hard_max]
         except Exception:
             hard_negs = []
 
@@ -153,6 +155,7 @@ def train_din(
     model_dir: str = "data/model",
     recall_dir: str = "data/model/recall",
     max_samples: int = 20000,
+    hard_ratio: float = 1.0,
     max_hist_len: int = 50,
     embed_dim: int = 64,
     epochs: int = 10,
@@ -164,7 +167,7 @@ def train_din(
     rank_dir.mkdir(parents=True, exist_ok=True)
 
     df, user_sequences, item_map = build_din_data(
-        data_dir, recall_dir, neg_ratio=4, max_samples=max_samples
+        data_dir, recall_dir, neg_ratio=4, hard_ratio=hard_ratio, max_samples=max_samples
     )
     num_items = len(item_map)
 
@@ -254,10 +257,12 @@ if __name__ == "__main__":
     parser.add_argument("--epochs", type=int, default=10)
     parser.add_argument("--batch_size", type=int, default=256)
     parser.add_argument("--aux", action="store_true", help="Use aux features from FeatureEngineer")
+    parser.add_argument("--hard_ratio", type=float, default=1.0, help="P2: Fraction of negatives that are hard")
     args = parser.parse_args()
 
     train_din(
         max_samples=args.max_samples,
+        hard_ratio=args.hard_ratio,
         epochs=args.epochs,
         batch_size=args.batch_size,
         use_aux=args.aux,

scripts/model/train_ranker.py

@@ -21,9 +21,12 @@ TIME-SPLIT (no leakage):
     - sasrec_score and user_seq_emb come from train-only SASRec.
     - Pipeline order: split -> build_sequences(train-only) -> recall(train) -> ranker(val).
 
-Negative Sampling Strategy:
-    - Hard negatives: items from recall results that are NOT the positive
-    - Random negatives: fill remaining slots if recall returns too few
+Negative Sampling Strategy (P2 configurable):
+    - hard_ratio: fraction of neg_ratio that should be hard (e.g. 0.5 = 2 hard + 2 random).
+    - Hard negatives: from recall results, capped at int(neg_ratio * hard_ratio).
+    - Random negatives: fill remaining slots.
+    - P3 filter_similar_to_positive: exclude hard negs with embedding sim > threshold (reduce FN).
+    - P3 Curriculum Learning: use lower hard_ratio (e.g. 0.5) for more stable convergence.
 """
 
 import sys
@@ -48,14 +51,59 @@ logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(level
 logger = logging.getLogger(__name__)
 
 
-def build_ranker_data(data_dir='data/rec', model_dir='data/model/recall', neg_ratio=4, max_samples=20000):
+def _filter_similar_to_positive(hard_negs, pos_isbn, fusion, sim_threshold):
+    """P3: Exclude hard negs with embedding cosine similarity > threshold to positive."""
+    try:
+        sasrec = fusion.sasrec
+        if not hasattr(sasrec, "item_emb") or sasrec.item_emb is None:
+            return hard_negs
+        item_map = getattr(sasrec, "item_map", {})
+        emb = sasrec.item_emb
+        pos_idx = item_map.get(str(pos_isbn), 0)
+        if pos_idx <= 0:
+            return hard_negs
+        pos_emb = emb[pos_idx]
+        pos_norm = np.linalg.norm(pos_emb)
+        if pos_norm < 1e-9:
+            return hard_negs
+        filtered = []
+        for neg in hard_negs:
+            neg_idx = item_map.get(str(neg), 0)
+            if neg_idx <= 0:
+                filtered.append(neg)
+                continue
+            neg_emb = emb[neg_idx]
+            sim = np.dot(pos_emb, neg_emb) / (pos_norm * np.linalg.norm(neg_emb) + 1e-9)
+            if sim <= sim_threshold:
+                filtered.append(neg)
+        return filtered
+    except Exception as e:
+        logger.warning(f"Could not filter similar to positive: {e}")
+        return hard_negs
+
+
+def build_ranker_data(
+    data_dir='data/rec',
+    model_dir='data/model/recall',
+    neg_ratio=4,
+    hard_ratio=1.0,
+    max_samples=20000,
+    filter_similar_to_positive: bool = False,
+    sim_threshold: float = 0.9,
+):
     """
     Construct training data with hard negative sampling.
 
     For each user in val.csv (sampled to max_samples for speed):
         - Positive: the actual item from val.csv (label=1)
-        - Hard negatives: top items recalled by the system but NOT the positive
-        - Random negatives: fill if recall gives fewer than neg_ratio candidates
+        - Hard negatives: up to int(neg_ratio * hard_ratio) from recall (P2)
+        - Random negatives: fill remaining to total neg_ratio
+
+    Args:
+        hard_ratio: Fraction of neg_ratio for hard negatives. 1.0=all hard (fill random);
+            0.5=half hard half random; 0.0=all random.
+        filter_similar_to_positive: P3 - Exclude hard negs with embedding sim > threshold to pos.
+        sim_threshold: Cosine similarity threshold for filtering (default 0.9).
 
     Returns:
         train_data: DataFrame [user_id, isbn, label]
@@ -85,18 +133,23 @@ def build_ranker_data(data_dir='data/rec', model_dir='data/model/recall', neg_ra
         # 1. Positive
         user_rows = [{'user_id': user_id, 'isbn': pos_isbn, 'label': 1}]
 
-        # 2. Hard negatives from recall
+        # 2. Hard negatives from recall (P2: cap by hard_ratio; P3: filter too-similar)
+        n_hard_max = max(0, int(neg_ratio * hard_ratio))
         try:
             recall_items = fusion.get_recall_items(user_id, k=50)
             hard_negs = [item for item, _ in recall_items if item != pos_isbn]
-            hard_negs = hard_negs[:neg_ratio]
+            if filter_similar_to_positive and hard_negs:
+                hard_negs = _filter_similar_to_positive(
+                    hard_negs, pos_isbn, fusion, sim_threshold
+                )
+            hard_negs = hard_negs[:n_hard_max]
         except Exception:
             hard_negs = []
 
         for neg_isbn in hard_negs:
             user_rows.append({'user_id': user_id, 'isbn': neg_isbn, 'label': 0})
 
-        # 3. Fill with random negatives if not enough
+        # 3. Fill with random negatives to reach neg_ratio
         n_remaining = neg_ratio - len(hard_negs)
         if n_remaining > 0:
             random_negs = np.random.choice(all_items, size=n_remaining, replace=False)
@@ -111,14 +164,25 @@ def build_ranker_data(data_dir='data/rec', model_dir='data/model/recall', neg_ra
     return train_data, group
 
 
-def train_ranker(max_samples=20000):
+def train_ranker(
+    max_samples=20000,
+    hard_ratio=1.0,
+    filter_similar_to_positive=False,
+    sim_threshold=0.9,
+):
     data_dir = Path('data/rec')
     model_dir = Path('data/model/ranking')
     model_dir.mkdir(parents=True, exist_ok=True)
 
     # 1. Prepare Data
     train_samples, group = build_ranker_data(
-        str(data_dir), model_dir='data/model/recall', neg_ratio=4, max_samples=max_samples
+        str(data_dir),
+        model_dir='data/model/recall',
+        neg_ratio=4,
+        hard_ratio=hard_ratio,
+        max_samples=max_samples,
+        filter_similar_to_positive=filter_similar_to_positive,
+        sim_threshold=sim_threshold,
     )
     logger.info(f"Training samples: {len(train_samples)}, groups: {len(group)}")
 
@@ -159,7 +223,12 @@ def train_ranker(max_samples=20000):
         logger.info(f"Feature {features[i]}: {score}")
 
 
-def train_stacking(max_samples=20000):
+def train_stacking(
+    max_samples=20000,
+    hard_ratio=1.0,
+    filter_similar_to_positive=False,
+    sim_threshold=0.9,
+):
     """
     Train Level-1 models (LGBMRanker + XGBClassifier) via GroupKFold CV
     to produce out-of-fold (OOF) predictions, then train Level-2 meta-learner
@@ -177,7 +246,13 @@ def train_stacking(max_samples=20000):
     # 1. Prepare Data (reuse existing build_ranker_data)
     # =========================================================================
     train_samples, group = build_ranker_data(
-        str(data_dir), model_dir='data/model/recall', neg_ratio=4, max_samples=max_samples
+        str(data_dir),
+        model_dir='data/model/recall',
+        neg_ratio=4,
+        hard_ratio=hard_ratio,
+        max_samples=max_samples,
+        filter_similar_to_positive=filter_similar_to_positive,
+        sim_threshold=sim_threshold,
     )
     logger.info(f"Stacking training samples: {len(train_samples)}, groups: {len(group)}")
 
@@ -341,9 +416,21 @@ if __name__ == "__main__":
                         help='Train with model stacking (LGB + XGB + Meta-Learner)')
     parser.add_argument('--max_samples', type=int, default=20000,
                         help='Number of samples used for training (default=20000)')
+    parser.add_argument('--hard_ratio', type=float, default=1.0,
+                        help='P2: Fraction of negatives that are hard. 0.5=half hard half random')
+    parser.add_argument('--filter_similar', action='store_true',
+                        help='P3: Exclude hard negs with embedding sim > threshold to positive')
+    parser.add_argument('--sim_threshold', type=float, default=0.9,
+                        help='P3: Cosine sim threshold for filter_similar (default 0.9)')
     args = parser.parse_args()
 
+    kwargs = dict(
+        max_samples=args.max_samples,
+        hard_ratio=args.hard_ratio,
+        filter_similar_to_positive=args.filter_similar,
+        sim_threshold=args.sim_threshold,
+    )
     if args.stacking:
-        train_stacking(max_samples=args.max_samples)
+        train_stacking(**kwargs)
     else:
-        train_ranker(max_samples=args.max_samples)
+        train_ranker(**kwargs)

scripts/utils.py

@@ -0,0 +1,63 @@
+"""
+Shared utilities for scripts/. Reduces duplication across data/model scripts.
+"""
+from __future__ import annotations
+
+import logging
+import sys
+from pathlib import Path
+
+# Ensure project root on path for config imports
+_PROJECT_ROOT = Path(__file__).resolve().parent.parent
+if str(_PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(_PROJECT_ROOT))
+
+
+def get_project_root() -> Path:
+    """Project root directory."""
+    return _PROJECT_ROOT
+
+
+def get_data_dir() -> Path:
+    """Data directory (data/)."""
+    return _PROJECT_ROOT / "data"
+
+
+def setup_script_logger(
+    name: str,
+    level: int = logging.INFO,
+    format_str: str = "%(asctime)s | %(levelname)s | %(name)s | %(message)s",
+    datefmt: str = "%H:%M:%S",
+) -> logging.Logger:
+    """
+    Configure logging for a script. Use instead of ad-hoc logging.basicConfig.
+    """
+    logger = logging.getLogger(name)
+    if not logger.handlers:
+        handler = logging.StreamHandler()
+        handler.setFormatter(logging.Formatter(format_str, datefmt=datefmt))
+        logger.addHandler(handler)
+    logger.setLevel(level)
+    return logger
+
+
+def load_data_config():
+    """Lazy-load config.data_config paths. Use when script needs DATA_DIR, BOOKS_PROCESSED, etc."""
+    from config.data_config import (
+        DATA_DIR,
+        RAW_DIR,
+        BOOKS_PROCESSED,
+        BOOKS_BASIC_INFO,
+        REC_DIR,
+        RAW_BOOKS,
+        RAW_RATINGS,
+    )
+    return {
+        "data_dir": DATA_DIR,
+        "raw_dir": RAW_DIR,
+        "books_processed": BOOKS_PROCESSED,
+        "books_basic_info": BOOKS_BASIC_INFO,
+        "rec_dir": REC_DIR,
+        "raw_books": RAW_BOOKS,
+        "raw_ratings": RAW_RATINGS,
+    }

src/agentic/__init__.py

@@ -0,0 +1,10 @@
+"""
+Agentic RAG workflow powered by LangGraph.
+
+Provides a stateful retrieval pipeline: Router -> Retrieve -> Evaluate -> (optional) Web Fallback.
+Enables LLM-based evaluation of result quality and conditional web search when local results
+are insufficient.
+"""
+from src.agentic.graph import build_agentic_graph, get_agentic_graph
+
+__all__ = ["build_agentic_graph", "get_agentic_graph"]

src/agentic/graph.py

@@ -0,0 +1,47 @@
+"""
+LangGraph workflow for Agentic RAG: Router -> Retrieve -> Evaluate -> (optional) Web Fallback.
+"""
+from langgraph.graph import StateGraph, START, END
+
+from src.agentic.state import RAGState
+from src.agentic.nodes import router_node, retrieve_node, evaluate_node, web_fallback_node
+from src.utils import setup_logger
+
+logger = setup_logger(__name__)
+
+_agentic_graph = None
+
+
+def _route_after_evaluate(state: RAGState):
+    """Route to web_fallback if need_more else END."""
+    if state.get("need_more") and state.get("retry_count", 0) < 1:
+        return "web_fallback"
+    return END
+
+
+def build_agentic_graph():
+    """Build and compile the Agentic RAG StateGraph."""
+    builder = StateGraph(RAGState)
+
+    builder.add_node("router", router_node)
+    builder.add_node("retrieve", retrieve_node)
+    builder.add_node("evaluate", evaluate_node)
+    builder.add_node("web_fallback", web_fallback_node)
+
+    builder.add_edge(START, "router")
+    builder.add_edge("router", "retrieve")
+    builder.add_edge("retrieve", "evaluate")
+    builder.add_conditional_edges("evaluate", _route_after_evaluate)
+    builder.add_edge("web_fallback", END)
+
+    graph = builder.compile()
+    logger.info("Agentic RAG graph built and compiled")
+    return graph
+
+
+def get_agentic_graph():
+    """Lazy-initialize and return the compiled Agentic graph."""
+    global _agentic_graph
+    if _agentic_graph is None:
+        _agentic_graph = build_agentic_graph()
+    return _agentic_graph

src/agentic/nodes.py

@@ -0,0 +1,149 @@
+"""
+LangGraph nodes for the Agentic RAG workflow.
+"""
+from typing import Any, Dict
+
+from src.agentic.state import RAGState
+from src.config import TOP_K_INITIAL
+from src.core.isbn_extractor import extract_isbn
+from src.utils import setup_logger
+
+logger = setup_logger(__name__)
+
+
+def router_node(state: RAGState) -> Dict[str, Any]:
+    """Determine retrieval strategy using QueryRouter."""
+    from src.core.router import QueryRouter
+
+    router = QueryRouter()
+    decision = router.route(state["query"])
+    logger.info(f"Agentic Router: {decision}")
+
+    return {
+        "strategy": decision["strategy"],
+        "temporal": decision.get("temporal", False),
+        "freshness_fallback": decision.get("freshness_fallback", False),
+        "freshness_threshold": decision.get("freshness_threshold", 3),
+        "decision_reason": f"routed to {decision['strategy']}",
+    }
+
+
+def retrieve_node(state: RAGState) -> Dict[str, Any]:
+    """Execute retrieval based on strategy."""
+    from src.vector_db import VectorDB
+
+    vector_db = VectorDB()
+    strategy = state.get("strategy", "deep")
+    query = state["query"]
+    temporal = state.get("temporal", False)
+
+    if strategy == "small_to_big":
+        recs = vector_db.small_to_big_search(query, k=TOP_K_INITIAL)
+    elif strategy == "exact":
+        recs = vector_db.hybrid_search(
+            query, k=TOP_K_INITIAL, alpha=1.0, rerank=False, temporal=False
+        )
+    else:
+        recs = vector_db.hybrid_search(
+            query,
+            k=TOP_K_INITIAL,
+            alpha=0.5,
+            rerank=(strategy == "deep"),
+            temporal=temporal,
+        )
+
+    isbn_list = []
+    for doc in recs:
+        isbn = extract_isbn(doc)
+        if isbn:
+            isbn_list.append(isbn)
+
+    logger.info(f"Agentic Retrieve: {len(isbn_list)} results for strategy={strategy}")
+    return {"isbn_list": isbn_list}
+
+
+def evaluate_node(state: RAGState) -> Dict[str, Any]:
+    """
+    Evaluate if local results are sufficient (rule-based).
+    Triggers web fallback when: few results + freshness query, or very few results.
+    """
+    n_results = len(state.get("isbn_list", []))
+    freshness_fallback = state.get("freshness_fallback", False)
+    threshold = state.get("freshness_threshold", 3)
+    retry_count = state.get("retry_count", 0)
+
+    # Hard limit: don't loop more than once
+    if retry_count >= 1:
+        return {"need_more": False}
+
+    # Rule 1: No results and freshness query -> always need more
+    if n_results == 0 and freshness_fallback:
+        return {"need_more": True}
+
+    # Rule 2: Results below threshold and freshness query -> need more
+    if n_results < threshold and freshness_fallback:
+        return {"need_more": True}
+
+    # Rule 3: Very few results regardless -> need more
+    if n_results < 2:
+        return {"need_more": True}
+
+    # Rule 4: Sufficient results
+    return {"need_more": False}
+
+
+async def web_fallback_node(state: RAGState, config=None) -> Dict[str, Any]:
+    """
+    Fetch from Google Books API when local results insufficient (async).
+    Uses search_google_books_async to avoid blocking the event loop.
+    """
+    from src.core.web_search import search_google_books_async
+    from src.core.metadata_store import metadata_store
+
+    query = state["query"]
+    category = state.get("category", "All")
+    existing_isbns = set(state.get("isbn_list", []))
+    max_to_fetch = 10 - len(existing_isbns)
+
+    if max_to_fetch <= 0:
+        return {"need_more": False}
+
+    recommender = None
+    if config:
+        cfg = config.get("configurable", {}) if isinstance(config, dict) else getattr(config, "configurable", {}) or {}
+        recommender = cfg.get("recommender") if cfg else None
+
+    web_books = await search_google_books_async(query, max_results=max_to_fetch * 2)
+    new_isbns = list(existing_isbns)
+
+    for book in web_books:
+        isbn = book.get("isbn13", "")
+        if not isbn or isbn in existing_isbns:
+            continue
+        if metadata_store.book_exists(isbn):
+            continue
+        if category and category != "All":
+            book_cat = book.get("simple_categories", "")
+            if category.lower() not in (book_cat or "").lower():
+                continue
+
+        if recommender:
+            added = recommender.add_new_book(
+                isbn=isbn,
+                title=book.get("title", ""),
+                author=book.get("authors", "Unknown"),
+                description=book.get("description", ""),
+                category=book.get("simple_categories", "General"),
+                thumbnail=book.get("thumbnail"),
+                published_date=book.get("publishedDate", ""),
+            )
+            if added:
+                new_isbns.append(isbn)
+        else:
+            new_isbns.append(isbn)
+
+        if len(new_isbns) - len(existing_isbns) >= max_to_fetch:
+            break
+
+    logger.info(f"Agentic Web Fallback: added {len(new_isbns) - len(existing_isbns)} books")
+    return {"isbn_list": new_isbns, "need_more": False, "retry_count": 1}

src/agentic/state.py

@@ -0,0 +1,19 @@
+"""
+State schema for the Agentic RAG LangGraph workflow.
+"""
+from typing import TypedDict, Optional
+
+
+class RAGState(TypedDict, total=False):
+    """State passed through the Agentic RAG graph."""
+
+    query: str
+    category: str
+    strategy: str
+    temporal: bool
+    freshness_fallback: bool
+    freshness_threshold: int
+    isbn_list: list[str]
+    need_more: bool
+    retry_count: int
+    decision_reason: str

src/config.py

@@ -1,3 +1,4 @@
+import json
 import os
 from pathlib import Path
 from dotenv import load_dotenv
@@ -7,6 +8,7 @@ load_dotenv()
 
 # Project Root
 PROJECT_ROOT = Path(__file__).parent.parent.absolute()
+CONFIG_DIR = PROJECT_ROOT / "config"
 
 # Data Paths
 DATA_DIR = PROJECT_ROOT / "data"
@@ -32,3 +34,51 @@ TOP_K_FINAL = 10
 
 # Debug mode: set DEBUG=1 to enable verbose logging (research prototype style)
 DEBUG = os.getenv("DEBUG", "0") == "1"
+
+
+def _load_router_config() -> dict:
+    """Load router keywords from config/router.json. Env overrides for ops flexibility."""
+    defaults = {
+        "detail_keywords": [
+            "twist", "ending", "spoiler", "readers", "felt", "cried", "hated", "loved",
+            "review", "opinion", "think", "unreliable", "narrator", "realize", "find out",
+        ],
+        "freshness_keywords": [
+            "new", "newest", "latest", "recent", "modern", "contemporary", "current",
+        ],
+        "strong_freshness_keywords": ["newest", "latest"],
+    }
+    path = CONFIG_DIR / "router.json"
+    if path.exists():
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+            return {**defaults, **data}
+        except Exception:
+            pass
+    return defaults
+
+
+_ROUTER_CFG = _load_router_config()
+
+# Dependencies can override via ROUTER_CONFIG_PATH for alternate config
+_path_override = os.getenv("ROUTER_CONFIG_PATH")
+if _path_override and Path(_path_override).exists():
+    try:
+        _ROUTER_CFG = {**_ROUTER_CFG, **json.loads(Path(_path_override).read_text(encoding="utf-8"))}
+    except Exception:
+        pass
+
+# Env: ROUTER_DETAIL_KEYWORDS = "twist,ending,spoiler,..." (comma-separated) overrides config
+_DETAIL_KW_RAW = os.getenv("ROUTER_DETAIL_KEYWORDS", "")
+ROUTER_DETAIL_KEYWORDS: frozenset[str] = (
+    frozenset(w.strip().lower() for w in _DETAIL_KW_RAW.split(",") if w.strip())
+    if _DETAIL_KW_RAW
+    else frozenset(str(k).lower() for k in _ROUTER_CFG.get("detail_keywords", []))
+)
+
+ROUTER_FRESHNESS_KEYWORDS: frozenset[str] = frozenset(
+    str(k).lower() for k in _ROUTER_CFG.get("freshness_keywords", [])
+)
+ROUTER_STRONG_FRESHNESS_KEYWORDS: frozenset[str] = frozenset(
+    str(k).lower() for k in _ROUTER_CFG.get("strong_freshness_keywords", [])
+)

src/core/book_ingestion.py

@@ -0,0 +1,96 @@
+"""
+Book ingestion: persist new books to staging store (online_books.db) and ChromaDB.
+Single responsibility: write path for web-discovered books; decouples from recommender.
+"""
+from typing import Any, Dict, Optional
+
+from src.core.metadata_store import metadata_store
+from src.core.online_books_store import online_books_store
+from src.utils import setup_logger
+
+logger = setup_logger(__name__)
+
+
+class BookIngestion:
+    """
+    Persist new books to staging store + ChromaDB.
+    Strategy: Staging write — no main books.db write. Decouples training data from runtime.
+    """
+
+    def __init__(self, vector_db=None, metadata_store_inst=None):
+        """
+        Args:
+            vector_db: VectorDB instance for dense index. Lazy import to avoid circular deps.
+            metadata_store_inst: For book_exists check. Defaults to global if None.
+        """
+        self._vector_db = vector_db
+        self._meta = metadata_store_inst if metadata_store_inst is not None else metadata_store
+
+    def _get_vector_db(self):
+        if self._vector_db is None:
+            from src.vector_db import VectorDB
+            self._vector_db = VectorDB()
+        return self._vector_db
+
+    def add_book(
+        self,
+        isbn: str,
+        title: str,
+        author: str,
+        description: str,
+        category: str = "General",
+        thumbnail: Optional[str] = None,
+        published_date: Optional[str] = None,
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Add a new book to the staging store (online_books.db + ChromaDB).
+
+        Args:
+            isbn: ISBN-13 or ISBN-10
+            title: Book title
+            author: Author name(s)
+            description: Book description
+            category: Book category
+            thumbnail: Cover image URL
+            published_date: Publication date
+
+        Returns:
+            New book row dict if successful, None otherwise
+        """
+        try:
+            isbn_s = str(isbn).strip()
+
+            if self._meta.book_exists(isbn_s):
+                logger.debug(f"Book {isbn} already exists. Skipping add.")
+                return None
+
+            new_row = {
+                "isbn13": isbn_s,
+                "title": title,
+                "authors": author,
+                "description": description,
+                "simple_categories": category,
+                "thumbnail": thumbnail if thumbnail else "/assets/cover-not-found.jpg",
+                "average_rating": 0.0,
+                "joy": 0.0, "sadness": 0.0, "fear": 0.0, "anger": 0.0, "surprise": 0.0,
+                "tags": "", "review_highlights": "",
+                "isbn10": isbn_s[:10] if len(isbn_s) >= 10 else isbn_s,
+                "publishedDate": published_date or "",
+                "source": "google_books",
+            }
+            new_row["large_thumbnail"] = new_row["thumbnail"]
+            new_row["image"] = new_row["thumbnail"]
+
+            if not online_books_store.insert_book_with_fts(new_row):
+                return None
+
+            self._get_vector_db().add_book(new_row)
+
+            logger.info(f"Successfully added book {isbn} to staging store: {title}")
+            return new_row
+
+        except Exception as e:
+            logger.error(f"Error adding new book: {e}")
+            import traceback
+            logger.error(traceback.format_exc())
+            return None

src/core/diversity_metrics.py

@@ -0,0 +1,77 @@
+"""
+P3: Diversity evaluation metrics.
+
+ILSD (Intra-List Similarity Diversity), Category Coverage, Gini.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Callable, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+def category_coverage(
+    rec_isbns: List[str],
+    get_category: Callable[[str], str],
+    top_k: int = 10,
+) -> float:
+    """
+    Fraction of unique categories in top-k list.
+    Higher = more diverse.
+    """
+    if not rec_isbns or top_k <= 0:
+        return 0.0
+    rec_top = rec_isbns[:top_k]
+    cats = {get_category(isbn) for isbn in rec_top}
+    cats.discard("")
+    cats.discard("Unknown")
+    return len(cats) / max(len(rec_top), 1)
+
+
+def intra_list_similarity(
+    rec_isbns: List[str],
+    similarity_fn: Callable[[str, str], float],
+    top_k: int = 10,
+) -> float:
+    """
+    Average pairwise similarity within top-k.
+    Lower = more diverse. ILSD = 1 - this (when similarity in [0,1]).
+    """
+    if not rec_isbns or top_k <= 0:
+        return 0.0
+    rec_top = rec_isbns[:top_k]
+    n = len(rec_top)
+    if n < 2:
+        return 0.0
+    total = 0.0
+    count = 0
+    for i in range(n):
+        for j in range(i + 1, n):
+            total += similarity_fn(rec_top[i], rec_top[j])
+            count += 1
+    return total / count if count > 0 else 0.0
+
+
+def category_coverage_similarity(isbn1: str, isbn2: str, get_category: Callable[[str], str]) -> float:
+    """1 if same category, 0 otherwise. Used for ILSD proxy."""
+    return 1.0 if get_category(isbn1) == get_category(isbn2) else 0.0
+
+
+def compute_diversity_metrics(
+    rec_isbns: List[str],
+    get_category: Callable[[str], str],
+    top_k: int = 10,
+) -> dict:
+    """
+    Compute category coverage and category-based ILSD.
+    Returns dict with category_coverage, ilsd (1 - avg_category_sim).
+    """
+    cov = category_coverage(rec_isbns, get_category, top_k)
+    sim_fn = lambda a, b: category_coverage_similarity(a, b, get_category)
+    sim = intra_list_similarity(rec_isbns, sim_fn, top_k)
+    return {
+        "category_coverage": cov,
+        "ilsd": 1.0 - sim,  # higher = more diverse
+    }

src/core/diversity_reranker.py

@@ -0,0 +1,194 @@
+"""
+Diversity Reranker: MMR + Popularity penalty + Category constraints.
+
+P0 optimization: Improves Diversity and Serendipity without significantly
+reducing Accuracy. Applied after LGBM/DIN ranking, before returning results.
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import Callable, List, Optional, Tuple
+
+from src.utils import setup_logger
+
+logger = setup_logger(__name__)
+
+
+class DiversityReranker:
+    """
+    Rerank candidates using MMR, popularity penalty, and category diversity.
+    """
+
+    def __init__(
+        self,
+        metadata_store,
+        data_dir: str = "data/rec",
+        mmr_lambda: float = 0.75,
+        popularity_gamma: float = 0.1,
+        max_per_category: int = 3,
+        enable_mmr: bool = True,
+        enable_popularity_penalty: bool = True,
+        enable_category_constraint: bool = True,
+    ):
+        """
+        Args:
+            metadata_store: For get_book_metadata (category lookup).
+            data_dir: Path to load train.csv for item popularity (interaction count).
+            mmr_lambda: Relevance weight in MMR. Higher = more accuracy, less diversity.
+            popularity_gamma: Penalty strength for popular items. Higher = less Harry Potter.
+            max_per_category: Max items per category in top-k.
+            enable_*: Feature flags.
+        """
+        self.metadata_store = metadata_store
+        self.data_dir = Path(data_dir)
+        self.mmr_lambda = mmr_lambda
+        self.popularity_gamma = popularity_gamma
+        self.max_per_category = max_per_category
+        self.enable_mmr = enable_mmr
+        self.enable_popularity_penalty = enable_popularity_penalty
+        self.enable_category_constraint = enable_category_constraint
+
+        self.item_popularity: dict = {}  # isbn -> count (interactions in train)
+        self._load_item_popularity()
+
+    def _load_item_popularity(self) -> None:
+        """Load item popularity from train.csv (interaction count per ISBN)."""
+        train_path = self.data_dir / "train.csv"
+        if not train_path.exists():
+            logger.warning("train.csv not found, popularity penalty disabled")
+            return
+        try:
+            import pandas as pd
+            df = pd.read_csv(train_path)
+            if "isbn" in df.columns:
+                self.item_popularity = df["isbn"].astype(str).value_counts().to_dict()
+            else:
+                col = [c for c in df.columns if "isbn" in c.lower()][:1]
+                if col:
+                    self.item_popularity = df[col[0]].astype(str).value_counts().to_dict()
+            logger.info(f"DiversityReranker: Loaded popularity for {len(self.item_popularity)} items")
+        except Exception as e:
+            logger.warning(f"Failed to load item popularity: {e}")
+
+    def _get_category(self, isbn: str) -> str:
+        """Get item category from metadata."""
+        meta = self.metadata_store.get_book_metadata(str(isbn))
+        cat = meta.get("simple_categories", "") if meta else ""
+        return (cat or "Unknown").strip()
+
+    def _category_similarity(self, cat1: str, cat2: str) -> float:
+        """1 if same category, 0 otherwise."""
+        return 1.0 if cat1 and cat2 and cat1.lower() == cat2.lower() else 0.0
+
+    def _get_popularity_score(self, isbn: str) -> float:
+        """Log-normalized popularity (for penalty)."""
+        cnt = self.item_popularity.get(str(isbn), 0)
+        return float(cnt)
+
+    def rerank(
+        self,
+        candidates: List[Tuple[str, float, list]],
+        top_k: int,
+    ) -> List[Tuple[str, float, list]]:
+        """
+        Rerank (isbn, score, explanations) list.
+
+        Args:
+            candidates: Sorted by score descending.
+            top_k: Number of results to return.
+
+        Returns:
+            Reranked list of (isbn, score, explanations).
+        """
+        if not candidates:
+            return []
+
+        # 1. Popularity penalty (adjust scores before MMR)
+        if self.enable_popularity_penalty:
+            max_cnt = max(self._get_popularity_score(i) for i, _, _ in candidates) or 1
+            adjusted = []
+            for isbn, score, expl in candidates:
+                cnt = self._get_popularity_score(isbn)
+                # score_adj = score / (1 + gamma * log(1 + normalized_cnt))
+                norm_cnt = cnt / max_cnt if max_cnt > 0 else 0
+                import math
+                penalty = 1.0 / (1.0 + self.popularity_gamma * math.log1p(norm_cnt * 100))
+                adj_score = score * penalty
+                adjusted.append((isbn, adj_score, expl))
+            candidates = adjusted
+
+        # 2. MMR rerank (diversity via category similarity)
+        if self.enable_mmr and len(candidates) > 1:
+            candidates = self._mmr_rerank(candidates, top_k)
+
+        # 3. Category constraint (ensure diversity in final list)
+        if self.enable_category_constraint:
+            candidates = self._apply_category_constraint(candidates, top_k)
+        else:
+            candidates = candidates[:top_k]
+
+        return candidates
+
+    def _mmr_rerank(
+        self,
+        candidates: List[Tuple[str, float, list]],
+        top_k: int,
+    ) -> List[Tuple[str, float, list]]:
+        """MMR: score = lambda * rel - (1-lambda) * max_sim(candidate, selected)."""
+        selected: List[Tuple[str, float, list]] = []
+        remaining = list(candidates)
+
+        while len(selected) < top_k and remaining:
+            best_idx = -1
+            best_mmr = float("-inf")
+
+            for idx, (isbn, rel, expl) in enumerate(remaining):
+                # Diversity: max similarity to already selected
+                max_sim = 0.0
+                cat_cand = self._get_category(isbn)
+                for sel_isbn, _, _ in selected:
+                    sim = self._category_similarity(cat_cand, self._get_category(sel_isbn))
+                    max_sim = max(max_sim, sim)
+
+                mmr = self.mmr_lambda * rel - (1.0 - self.mmr_lambda) * max_sim
+                if mmr > best_mmr:
+                    best_mmr = mmr
+                    best_idx = idx
+
+            if best_idx < 0:
+                break
+            selected.append(remaining.pop(best_idx))
+
+        return selected
+
+    def _apply_category_constraint(
+        self,
+        candidates: List[Tuple[str, float, list]],
+        top_k: int,
+    ) -> List[Tuple[str, float, list]]:
+        """Greedy: prefer items that don't exceed max_per_category."""
+        category_counts: dict = {}
+        result: List[Tuple[str, float, list]] = []
+
+        for isbn, score, expl in candidates:
+            if len(result) >= top_k:
+                break
+            cat = self._get_category(isbn)
+            count = category_counts.get(cat, 0)
+            if count < self.max_per_category:
+                result.append((isbn, score, expl))
+                category_counts[cat] = count + 1
+
+        # If we have slack, fill with remaining (no constraint)
+        if len(result) < top_k:
+            seen = {r[0] for r in result}
+            for isbn, score, expl in candidates:
+                if len(result) >= top_k:
+                    break
+                if isbn not in seen:
+                    result.append((isbn, score, expl))
+                    seen.add(isbn)
+
+        return result

src/core/fallback_provider.py

@@ -0,0 +1,137 @@
+"""
+Fallback provider: fetch books from external sources (e.g. Google Books API) when local
+results are insufficient. Single responsibility: external source acquisition.
+"""
+import sqlite3
+from typing import Any, Dict, List
+
+from src.core.metadata_store import metadata_store
+from src.core.response_formatter import format_web_book_response
+from src.utils import setup_logger
+
+logger = setup_logger(__name__)
+
+
+class FallbackProvider:
+    """
+    Fetch books from Google Books API when local search is insufficient.
+    Persists discovered books via BookIngestion for future queries.
+    """
+
+    def __init__(self, book_ingestion=None, metadata_store_inst=None):
+        """
+        Args:
+            book_ingestion: BookIngestion instance for persisting. Lazy init if None.
+            metadata_store_inst: For book_exists check. Defaults to global if None.
+        """
+        from src.core.book_ingestion import BookIngestion
+        self._meta = metadata_store_inst if metadata_store_inst is not None else metadata_store
+        self._ingestion = book_ingestion or BookIngestion(metadata_store_inst=self._meta)
+
+    async def fetch_async(
+        self,
+        query: str,
+        max_results: int,
+        category: str = "All",
+    ) -> List[Dict[str, Any]]:
+        """
+        Async: Fetch books from Google Books API.
+        Uses httpx to avoid blocking the FastAPI event loop.
+        """
+        try:
+            from src.core.web_search import search_google_books_async
+        except ImportError:
+            logger.warning("Web search module not available")
+            return []
+
+        results: List[Dict[str, Any]] = []
+        try:
+            web_books = await search_google_books_async(query, max_results=max_results * 2)
+
+            for book in web_books:
+                isbn = book.get("isbn13", "")
+                if not isbn:
+                    continue
+                if self._meta.book_exists(isbn):
+                    continue
+                if category and category != "All":
+                    book_cat = book.get("simple_categories", "")
+                    if category.lower() not in (book_cat or "").lower():
+                        continue
+
+                added = self._ingestion.add_book(
+                    isbn=isbn,
+                    title=book.get("title", ""),
+                    author=book.get("authors", "Unknown"),
+                    description=book.get("description", ""),
+                    category=book.get("simple_categories", "General"),
+                    thumbnail=book.get("thumbnail"),
+                    published_date=book.get("publishedDate", ""),
+                )
+                if added:
+                    results.append(format_web_book_response(book, isbn))
+                if len(results) >= max_results:
+                    break
+
+            logger.info(f"Web fallback: Found and persisted {len(results)} new books")
+            return results
+        except sqlite3.Error as e:
+            logger.error(f"[WebFallback:DB_ERROR] query='{query}' - {e}")
+            return []
+        except Exception as e:
+            logger.exception(f"[WebFallback:UNEXPECTED] query='{query}' - {type(e).__name__}: {e}")
+            return []
+
+    def fetch_sync(
+        self,
+        query: str,
+        max_results: int,
+        category: str = "All",
+    ) -> List[Dict[str, Any]]:
+        """
+        Sync: Fetch books from Google Books API.
+        For scripts/CLI; prefer fetch_async in FastAPI.
+        """
+        try:
+            from src.core.web_search import search_google_books
+        except ImportError:
+            logger.warning("Web search module not available")
+            return []
+
+        results: List[Dict[str, Any]] = []
+        try:
+            web_books = search_google_books(query, max_results=max_results * 2)
+
+            for book in web_books:
+                isbn = book.get("isbn13", "")
+                if not isbn:
+                    continue
+                if self._meta.book_exists(isbn):
+                    continue
+                if category and category != "All":
+                    book_cat = book.get("simple_categories", "")
+                    if category.lower() not in (book_cat or "").lower():
+                        continue
+
+                added = self._ingestion.add_book(
+                    isbn=isbn,
+                    title=book.get("title", ""),
+                    author=book.get("authors", "Unknown"),
+                    description=book.get("description", ""),
+                    category=book.get("simple_categories", "General"),
+                    thumbnail=book.get("thumbnail"),
+                    published_date=book.get("publishedDate", ""),
+                )
+                if added:
+                    results.append(format_web_book_response(book, isbn))
+                if len(results) >= max_results:
+                    break
+
+            logger.info(f"Web fallback: Found and persisted {len(results)} new books")
+            return results
+        except sqlite3.Error as e:
+            logger.error(f"[WebFallback:DB_ERROR] query='{query}' - {e}")
+            return []
+        except Exception as e:
+            logger.exception(f"[WebFallback:UNEXPECTED] query='{query}' - {type(e).__name__}: {e}")
+            return []

src/core/isbn_extractor.py

@@ -0,0 +1,45 @@
+"""
+Centralized ISBN extraction from various document formats.
+Single place for robust ISBN parsing logic — used by recommender, agentic nodes, etc.
+"""
+from typing import Any, Optional
+
+
+def extract_isbn(doc: Any) -> Optional[str]:
+    """
+    Extract ISBN from a document (LangChain Document, vector search result, etc.).
+
+    Tries, in order:
+    1. metadata['isbn'] or metadata['isbn13']
+    2. Content format "Title... ISBN: X"
+    3. Legacy format: first token of page_content
+
+    Args:
+        doc: Object with .metadata and/or .page_content attributes
+
+    Returns:
+        ISBN string if found, None otherwise
+    """
+    isbn_str: Optional[str] = None
+
+    # 1. Try metadata (Hybrid/BM25)
+    if hasattr(doc, "metadata") and doc.metadata:
+        if "isbn" in doc.metadata:
+            isbn_str = str(doc.metadata["isbn"])
+        elif "isbn13" in doc.metadata:
+            isbn_str = str(doc.metadata["isbn13"])
+
+    # 2. Try content format "Title... ISBN: X"
+    if not isbn_str and hasattr(doc, "page_content") and doc.page_content and "ISBN:" in doc.page_content:
+        try:
+            parts = doc.page_content.split("ISBN:")
+            if len(parts) > 1:
+                isbn_str = parts[1].strip().split()[0]
+        except (IndexError, AttributeError):
+            pass
+
+    # 3. Legacy: first token of page_content
+    if not isbn_str and hasattr(doc, "page_content") and doc.page_content:
+        isbn_str = doc.page_content.strip('"').split()[0] if doc.page_content.strip() else None
+
+    return isbn_str.strip() if (isbn_str and isbn_str.strip()) else None

src/core/metadata_enricher.py

@@ -0,0 +1,56 @@
+"""
+Metadata enrichment: fetches metadata, enriches, and filters by category.
+Single responsibility: data completion for recommendation results.
+"""
+from typing import Any, Dict, List, Optional
+
+from src.core.metadata_store import metadata_store
+from src.core.response_formatter import format_book_response
+from src.utils import enrich_book_metadata
+from src.config import TOP_K_FINAL
+
+
+def enrich_and_format(
+    isbn_list: List[str],
+    category: str = "All",
+    max_results: int = TOP_K_FINAL,
+    source: str = "local",
+    metadata_store_inst=None,
+) -> List[Dict[str, Any]]:
+    """
+    Enrich ISBN list with metadata and format into API response dicts.
+
+    - Fetches metadata from MetadataStore
+    - Enriches with cover/author fallback (enrich_book_metadata)
+    - Filters by category if specified
+    - Returns formatted dicts up to max_results
+
+    Args:
+        isbn_list: List of ISBN strings
+        category: Category filter ("All" = no filter)
+        max_results: Max number of results to return
+        source: Source label for response (local, content_based, etc.)
+
+    Returns:
+        List of formatted book dicts ready for API response
+    """
+    store = metadata_store_inst if metadata_store_inst is not None else metadata_store
+    results: List[Dict[str, Any]] = []
+
+    for isbn in isbn_list:
+        meta = store.get_book_metadata(str(isbn))
+        meta = enrich_book_metadata(meta, str(isbn))
+
+        if not meta:
+            continue
+
+        if category and category != "All":
+            if meta.get("simple_categories") != category:
+                continue
+
+        results.append(format_book_response(meta, str(isbn), source))
+
+        if len(results) >= max_results:
+            break
+
+    return results

src/core/metadata_store.py

@@ -7,6 +7,11 @@ from src.utils import setup_logger
 
 logger = setup_logger(__name__)
 
+# Lazy import to avoid circular dependency
+def _online_store():
+    from src.core.online_books_store import online_books_store
+    return online_books_store
+
 class MetadataStore:
     """
     Singleton class to manage large book metadata efficiently.
@@ -64,10 +69,12 @@ class MetadataStore:
             return None
 
     def get_book_metadata(self, isbn: str) -> Dict[str, Any]:
-        """Fast lookup for book metadata by ISBN (10 or 13) using SQLite index."""
+        """Fast lookup: main store first, then online staging store (read path stays fast)."""
         isbn = str(isbn).strip().replace(".0", "")
         row = self._query_one("SELECT * FROM books WHERE isbn13 = ? OR isbn10 = ?", (isbn, isbn))
-        return dict(row) if row else {}
+        if row:
+            return dict(row)
+        return _online_store().get_book_metadata(isbn) or {}
 
     def get_image(self, isbn: str, default: str = "") -> str:
         isbn = str(isbn).strip().replace(".0", "")
@@ -113,13 +120,15 @@ class MetadataStore:
         return pd.DataFrame()
 
     def get_all_categories(self) -> List[str]:
-        """Efficiently fetch unique categories from SQLite."""
+        """Efficiently fetch unique categories from main + online store."""
         conn = self.connection
+        cats = set()
         if conn:
             cursor = conn.cursor()
             cursor.execute("SELECT DISTINCT simple_categories FROM books")
-            return [row[0] for row in cursor.fetchall() if row[0]]
-        return []
+            cats.update(row[0] for row in cursor.fetchall() if row[0])
+        cats.update(_online_store().get_all_categories())
+        return sorted(cats)
 
     def insert_book(self, row: Dict[str, Any]) -> bool:
         """Insert a new book for add_new_book. Maps thumbnail->image if needed."""
@@ -218,13 +227,15 @@ class MetadataStore:
             return False
 
     def book_exists(self, isbn: str) -> bool:
-        """Check if a book with given ISBN exists in the database."""
+        """Check if ISBN exists in main or online staging store."""
         isbn = str(isbn).strip().replace(".0", "")
         row = self._query_one(
             "SELECT 1 FROM books WHERE isbn13 = ? OR isbn10 = ? LIMIT 1",
             (isbn, isbn)
         )
-        return row is not None
+        if row:
+            return True
+        return _online_store().book_exists(isbn)
 
     def get_newest_book_year(self) -> Optional[int]:
         """Get the publication year of the newest book in the database."""

src/core/online_books_store.py

@@ -0,0 +1,220 @@
+"""
+Online Books Store - Staging storage for freshness_fallback books.
+
+Design: Separate SQLite file (online_books.db) decouples:
+1. Data risk: Training data (books_processed.csv) stays frozen; no pollution.
+2. Performance: Writes go to online_books.db only; main books.db stays read-only.
+"""
+
+import sqlite3
+from pathlib import Path
+from typing import Optional, Dict, Any, List
+from src.config import DATA_DIR
+from src.utils import setup_logger
+
+logger = setup_logger(__name__)
+
+
+class OnlineBooksStore:
+    """
+    Append-only store for books discovered via Web Search (freshness_fallback).
+    Uses a separate SQLite file to avoid lock contention with main books.db.
+    """
+
+    _instance: Optional["OnlineBooksStore"] = None
+
+    def __new__(cls):
+        if cls._instance is None:
+            cls._instance = super(OnlineBooksStore, cls).__new__(cls)
+            cls._instance._initialized = False
+        return cls._instance
+
+    def __init__(self):
+        if self._initialized:
+            return
+
+        self.db_path = DATA_DIR / "online_books.db"
+        self._conn = None
+        self._initialized = True
+        self._ensure_schema()
+        logger.info("OnlineBooksStore: Initialized (staging store for web-discovered books)")
+
+    def _ensure_schema(self) -> None:
+        """Create table and FTS5 index if not exist."""
+        conn = self._get_connection()
+        if not conn:
+            return
+        try:
+            cursor = conn.cursor()
+            cursor.execute("""
+                CREATE TABLE IF NOT EXISTS online_books (
+                    isbn13 TEXT PRIMARY KEY,
+                    isbn10 TEXT,
+                    title TEXT,
+                    authors TEXT,
+                    description TEXT,
+                    simple_categories TEXT,
+                    thumbnail TEXT,
+                    image TEXT,
+                    average_rating REAL DEFAULT 0,
+                    joy REAL DEFAULT 0, sadness REAL DEFAULT 0, fear REAL DEFAULT 0,
+                    anger REAL DEFAULT 0, surprise REAL DEFAULT 0,
+                    tags TEXT, review_highlights TEXT,
+                    publishedDate TEXT,
+                    source TEXT DEFAULT 'google_books'
+                )
+            """)
+            cursor.execute("CREATE INDEX IF NOT EXISTS idx_online_isbn10 ON online_books (isbn10)")
+            cursor.execute(
+                "SELECT name FROM sqlite_master WHERE type='table' AND name='online_books_fts'"
+            )
+            if not cursor.fetchone():
+                cursor.execute("""
+                    CREATE VIRTUAL TABLE online_books_fts USING fts5(
+                        isbn13 UNINDEXED,
+                        title,
+                        description,
+                        authors,
+                        simple_categories,
+                        tokenize='porter unicode61'
+                    )
+                """)
+            conn.commit()
+        except Exception as e:
+            logger.error(f"OnlineBooksStore schema setup failed: {e}")
+
+    def _get_connection(self) -> Optional[sqlite3.Connection]:
+        """Lazy connection to online_books.db (separate from main books.db)."""
+        if self._conn is None:
+            try:
+                self.db_path.parent.mkdir(parents=True, exist_ok=True)
+                self._conn = sqlite3.connect(str(self.db_path), check_same_thread=False)
+                self._conn.row_factory = sqlite3.Row
+            except Exception as e:
+                logger.error(f"OnlineBooksStore: Failed to connect: {e}")
+        return self._conn
+
+    def get_book_metadata(self, isbn: str) -> Dict[str, Any]:
+        """Lookup book by ISBN. Returns empty dict if not found."""
+        isbn = str(isbn).strip().replace(".0", "")
+        conn = self._get_connection()
+        if not conn:
+            return {}
+        try:
+            row = conn.execute(
+                "SELECT * FROM online_books WHERE isbn13 = ? OR isbn10 = ?",
+                (isbn, isbn),
+            ).fetchone()
+            return dict(row) if row else {}
+        except Exception as e:
+            logger.error(f"OnlineBooksStore get_book_metadata failed: {e}")
+            return {}
+
+    def book_exists(self, isbn: str) -> bool:
+        """Check if ISBN exists in online store."""
+        isbn = str(isbn).strip().replace(".0", "")
+        conn = self._get_connection()
+        if not conn:
+            return False
+        try:
+            row = conn.execute(
+                "SELECT 1 FROM online_books WHERE isbn13 = ? OR isbn10 = ? LIMIT 1",
+                (isbn, isbn),
+            ).fetchone()
+            return row is not None
+        except Exception as e:
+            logger.error(f"OnlineBooksStore book_exists failed: {e}")
+            return False
+
+    def insert_book_with_fts(self, row: Dict[str, Any]) -> bool:
+        """
+        Insert book into online_books + FTS5. Write-only path; no lock on main DB.
+        """
+        conn = self._get_connection()
+        if not conn:
+            return False
+        try:
+            isbn13 = str(row.get("isbn13", ""))
+            isbn10 = row.get("isbn10", isbn13[:10] if len(isbn13) >= 10 else isbn13)
+            title = str(row.get("title", ""))
+            authors = str(row.get("authors", ""))
+            description = str(row.get("description", ""))
+            categories = str(row.get("simple_categories", "General"))
+            thumbnail = str(row.get("thumbnail", ""))
+            image = str(row.get("image", thumbnail))
+            published_date = str(row.get("publishedDate", ""))
+
+            conn.execute(
+                """
+                INSERT OR IGNORE INTO online_books (
+                    isbn13, isbn10, title, authors, description, simple_categories,
+                    thumbnail, image, publishedDate, source
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'google_books')
+                """,
+                (isbn13, isbn10, title, authors, description, categories, thumbnail, image, published_date),
+            )
+
+            cursor = conn.cursor()
+            cursor.execute(
+                "SELECT name FROM sqlite_master WHERE type='table' AND name='online_books_fts'"
+            )
+            if cursor.fetchone():
+                cursor.execute(
+                    """
+                    INSERT INTO online_books_fts (isbn13, title, description, authors, simple_categories)
+                    VALUES (?, ?, ?, ?, ?)
+                    """,
+                    (isbn13, title, description, authors, categories),
+                )
+            conn.commit()
+            logger.info(f"OnlineBooksStore: Inserted {isbn13} (staging)")
+            return True
+        except Exception as e:
+            logger.error(f"OnlineBooksStore insert failed: {e}")
+            return False
+
+    def get_all_categories(self) -> List[str]:
+        """Get unique categories from online books."""
+        conn = self._get_connection()
+        if not conn:
+            return []
+        try:
+            rows = conn.execute(
+                "SELECT DISTINCT simple_categories FROM online_books WHERE simple_categories != ''"
+            ).fetchall()
+            return [row[0] for row in rows if row[0]]
+        except Exception as e:
+            logger.debug(f"OnlineBooksStore get_all_categories failed: {e}")
+            return []
+
+    def fts_search(self, query: str, k: int = 10) -> List[Dict[str, Any]]:
+        """
+        FTS5 keyword search over online_books. Used by VectorDB to merge with main FTS.
+        Returns list of dicts with isbn13, title, description, authors, simple_categories.
+        """
+        conn = self._get_connection()
+        if not conn:
+            return []
+        try:
+            clean_query = query.strip().replace('"', '""')
+            if not clean_query:
+                return []
+            fts_query = f'"{clean_query}"'
+            cursor = conn.cursor()
+            cursor.execute(
+                """
+                SELECT isbn13, title, description, authors, simple_categories
+                FROM online_books_fts
+                WHERE online_books_fts MATCH ?
+                ORDER BY rank
+                LIMIT ?
+                """,
+                (fts_query, k),
+            )
+            return [dict(row) for row in cursor.fetchall()]
+        except Exception as e:
+            logger.debug(f"OnlineBooksStore FTS search failed: {e}")
+            return []
+
+
+online_books_store = OnlineBooksStore()

src/core/recommendation_orchestrator.py

@@ -0,0 +1,208 @@
+"""
+Recommendation orchestrator: coordinates the recommendation flow only.
+Delegates to VectorDB, Router, MetadataEnricher, FallbackProvider, Cache.
+Single responsibility: flow coordination.
+"""
+from typing import Any, Dict, List, Optional
+
+from src.config import TOP_K_INITIAL, TOP_K_FINAL
+from src.vector_db import VectorDB
+from src.cache import CacheManager
+from src.core.metadata_store import metadata_store
+from src.core.isbn_extractor import extract_isbn
+from src.core.metadata_enricher import enrich_and_format
+from src.core.fallback_provider import FallbackProvider
+from src.core.book_ingestion import BookIngestion
+from src.utils import setup_logger
+
+logger = setup_logger(__name__)
+
+
+class RecommendationOrchestrator:
+    """
+    Orchestrates RAG search and metadata enrichment.
+    Zero business logic: only coordinates VectorDB, Router, Enricher, Fallback, Cache.
+    Supports DI for metadata_store to simplify unit testing.
+    """
+
+    def __init__(
+        self,
+        metadata_store_inst=None,
+        vector_db: Optional[VectorDB] = None,
+        cache: Optional[CacheManager] = None,
+        fallback_provider: Optional[FallbackProvider] = None,
+        book_ingestion: Optional[BookIngestion] = None,
+    ):
+        self._meta = metadata_store_inst if metadata_store_inst is not None else metadata_store
+        self.vector_db = vector_db or VectorDB()
+        self.cache = cache or CacheManager()
+        self._ingestion = book_ingestion or BookIngestion(
+            vector_db=self.vector_db,
+            metadata_store_inst=self._meta,
+        )
+        self._fallback = fallback_provider or FallbackProvider(
+            book_ingestion=self._ingestion,
+            metadata_store_inst=self._meta,
+        )
+
+        logger.info("RecommendationOrchestrator: Zero-RAM mode. Using SQLite for on-demand lookups.")
+
+    async def get_recommendations(
+        self,
+        query: str,
+        category: str = "All",
+        tone: str = "All",
+        user_id: str = "local",
+        use_agentic: bool = False,
+    ) -> List[Dict[str, Any]]:
+        """
+        Generate book recommendations. Async for web search fallback.
+        """
+        if not query or not query.strip():
+            return []
+
+        cache_key = self.cache.generate_key("rec", q=query, c=category, t=tone, agentic=use_agentic)
+        cached = self.cache.get(cache_key)
+        if cached:
+            logger.info(f"Returning cached results for key: {cache_key}")
+            return cached
+
+        logger.info(f"Processing request: query='{query}', category='{category}', use_agentic={use_agentic}")
+
+        if use_agentic:
+            results = await self._get_recommendations_agentic(query, category)
+        else:
+            results = await self._get_recommendations_classic(query, category)
+
+        if results:
+            self.cache.set(cache_key, results)
+        return results
+
+    def get_recommendations_sync(
+        self,
+        query: str,
+        category: str = "All",
+        tone: str = "All",
+        user_id: str = "local",
+        use_agentic: bool = False,
+    ) -> List[Dict[str, Any]]:
+        """Sync wrapper for scripts/CLI."""
+        import asyncio
+        return asyncio.run(self.get_recommendations(query, category, tone, user_id, use_agentic))
+
+    async def _get_recommendations_agentic(self, query: str, category: str) -> List[Dict[str, Any]]:
+        """LangGraph workflow: Router -> Retrieve -> Evaluate -> (optional) Web Fallback."""
+        from src.agentic.graph import get_agentic_graph
+
+        graph = get_agentic_graph()
+        config = {"configurable": {"recommender": self}}
+        final_state = await graph.ainvoke(
+            {"query": query, "category": category, "retry_count": 0},
+            config=config,
+        )
+        books_list = final_state.get("isbn_list", [])
+        return enrich_and_format(books_list, category, TOP_K_FINAL, "local", metadata_store_inst=self._meta)
+
+    async def _get_recommendations_classic(self, query: str, category: str) -> List[Dict[str, Any]]:
+        """Classic Router -> Hybrid/Small-to-Big -> optional Web Fallback."""
+        from src.core.router import QueryRouter
+
+        router = QueryRouter()
+        decision = router.route(query)
+        logger.info(f"Retrieval Strategy: {decision}")
+
+        if decision["strategy"] == "small_to_big":
+            recs = self.vector_db.small_to_big_search(query, k=TOP_K_INITIAL)
+        else:
+            recs = self.vector_db.hybrid_search(
+                query,
+                k=TOP_K_INITIAL,
+                alpha=decision.get("alpha", 0.5),
+                rerank=decision["rerank"],
+                temporal=decision.get("temporal", False),
+            )
+
+        books_list = []
+        for rec in recs:
+            isbn_str = extract_isbn(rec)
+            if isbn_str:
+                books_list.append(isbn_str)
+
+        results = enrich_and_format(books_list, category, TOP_K_FINAL, "local", metadata_store_inst=self._meta)
+
+        if decision.get("freshness_fallback", False):
+            threshold = decision.get("freshness_threshold", 3)
+            if len(results) < threshold:
+                web_results = await self._fallback.fetch_async(
+                    query, TOP_K_FINAL - len(results), category
+                )
+                results.extend(web_results)
+                logger.info(f"Web fallback added {len(web_results)} books")
+
+        return results
+
+    def get_similar_books(
+        self,
+        isbn: str,
+        k: int = 10,
+        category: str = "All",
+    ) -> List[Dict[str, Any]]:
+        """Content-based similar books by vector similarity."""
+        isbn_str = str(isbn).strip()
+        if not isbn_str:
+            return []
+
+        meta = self._meta.get_book_metadata(isbn_str)
+        if not meta:
+            logger.warning(f"get_similar_books: Book {isbn} not found in metadata")
+            return []
+
+        title = meta.get("title", "")
+        description = meta.get("description", "") or ""
+        if not title:
+            logger.warning(f"get_similar_books: Book {isbn} has no title")
+            return []
+
+        query = f"{title} {description}"[:2000]
+        recs = self.vector_db.search(query, k=k * 3)
+
+        seen = {isbn_str}
+        isbn_list = []
+        for rec in recs:
+            candidate = extract_isbn(rec)
+            if candidate and candidate not in seen:
+                seen.add(candidate)
+                isbn_list.append(candidate)
+            if len(isbn_list) >= k:
+                break
+
+        return enrich_and_format(isbn_list, category, k, "content_based", metadata_store_inst=self._meta)
+
+    def get_categories(self) -> List[str]:
+        """Get unique book categories."""
+        return ["All"] + self._meta.get_all_categories()
+
+    def get_tones(self) -> List[str]:
+        """Get available emotional tones."""
+        return ["All", "Happy", "Sad", "Fear", "Anger", "Surprise"]
+
+    def add_new_book(
+        self,
+        isbn: str,
+        title: str,
+        author: str,
+        description: str,
+        category: str = "General",
+        thumbnail: Optional[str] = None,
+        published_date: Optional[str] = None,
+    ) -> Optional[Dict[str, Any]]:
+        """Delegate to BookIngestion. Kept for agentic/facade compatibility."""
+        return self._ingestion.add_book(
+            isbn=isbn,
+            title=title,
+            author=author,
+            description=description,
+            category=category,
+            thumbnail=thumbnail,
+            published_date=published_date,
+        )

src/core/response_formatter.py

@@ -0,0 +1,68 @@
+"""
+Response formatting: converts enriched metadata into API-ready recommendation dicts.
+Single responsibility: define the structure of recommendation responses.
+"""
+from typing import Any, Dict, List
+
+
+def format_book_response(meta: Dict[str, Any], isbn: str, source: str = "local") -> Dict[str, Any]:
+    """
+    Format a single book's metadata into the standard API response structure.
+
+    Args:
+        meta: Enriched metadata dict (from MetadataStore + enrich_book_metadata)
+        isbn: ISBN string
+        source: Data source label (local, google_books, content_based)
+
+    Returns:
+        Dict with isbn, title, authors, description, thumbnail, caption, tags,
+        emotions, review_highlights, persona_summary, average_rating, source
+    """
+    tags_raw = str(meta.get("tags", "")).strip()
+    tags = [t.strip() for t in tags_raw.split(";") if t.strip()] if tags_raw else []
+
+    return {
+        "isbn": str(isbn),
+        "title": meta.get("title", ""),
+        "authors": meta.get("authors", "Unknown"),
+        "description": meta.get("description", ""),
+        "thumbnail": meta.get("thumbnail"),
+        "caption": f"{meta.get('title', '')} by {meta.get('authors', 'Unknown')}",
+        "tags": tags,
+        "emotions": {
+            "joy": float(meta.get("joy", 0.0)),
+            "sadness": float(meta.get("sadness", 0.0)),
+            "fear": float(meta.get("fear", 0.0)),
+            "anger": float(meta.get("anger", 0.0)),
+            "surprise": float(meta.get("surprise", 0.0)),
+        },
+        "review_highlights": [
+            h.strip()
+            for h in str(meta.get("review_highlights", "")).split(";")
+            if h.strip()
+        ][:3],
+        "persona_summary": "",
+        "average_rating": float(meta.get("average_rating", 0.0)),
+        "source": source,
+    }
+
+
+def format_web_book_response(book: Dict[str, Any], isbn: str) -> Dict[str, Any]:
+    """
+    Format a raw web API book dict into the standard response structure.
+    Used when books come from Google Books API (no local metadata).
+    """
+    return {
+        "isbn": isbn,
+        "title": book.get("title", ""),
+        "authors": book.get("authors", "Unknown"),
+        "description": book.get("description", ""),
+        "thumbnail": book.get("thumbnail", ""),
+        "caption": f"{book.get('title', '')} by {book.get('authors', 'Unknown')}",
+        "tags": [],
+        "emotions": {"joy": 0.0, "sadness": 0.0, "fear": 0.0, "anger": 0.0, "surprise": 0.0},
+        "review_highlights": [],
+        "persona_summary": "",
+        "average_rating": float(book.get("average_rating", 0.0)),
+        "source": "google_books",
+    }

src/core/router.py

@@ -23,18 +23,9 @@ class QueryRouter:
     Freshness-Aware Routing:
     - Detects queries asking for "new", "latest", or specific years (2024, 2025, etc.)
     - Sets freshness_fallback=True to enable Web Search when local results insufficient
-    """
-
-    # Keywords that indicate user wants fresh/recent content
-    # Note: Year numbers are detected dynamically in _detect_freshness()
-    FRESHNESS_KEYWORDS = {
-        "new", "newest", "latest", "recent", "modern", "contemporary", "current",
-    }
     
-    # Strong freshness indicators (always trigger fallback)
-    STRONG_FRESHNESS_KEYWORDS = {
-        "newest", "latest",
-    }
+    Keywords loaded from config/router.json; overridable via ROUTER_DETAIL_KEYWORDS env.
+    """
 
     def __init__(self, model_dir: str | Path | None = None):
         self.isbn_pattern = re.compile(r"^(?:\d{9}[\dX]|\d{13})$")
@@ -68,12 +59,13 @@ class QueryRouter:
             - target_year: Specific year user is looking for (if detected)
         """
         from datetime import datetime
+        from src.config import ROUTER_FRESHNESS_KEYWORDS, ROUTER_STRONG_FRESHNESS_KEYWORDS
+
         current_year = datetime.now().year
-        
         lower_words = {w.lower() for w in words}
-        
-        is_temporal = bool(lower_words & self.FRESHNESS_KEYWORDS)
-        freshness_fallback = bool(lower_words & self.STRONG_FRESHNESS_KEYWORDS)
+
+        is_temporal = bool(lower_words & ROUTER_FRESHNESS_KEYWORDS)
+        freshness_fallback = bool(lower_words & ROUTER_STRONG_FRESHNESS_KEYWORDS)
         
         # Extract explicit year from query
         target_year = None
@@ -99,11 +91,8 @@ class QueryRouter:
         target_year: Optional[int] = None
     ) -> Dict[str, Any]:
         """Fallback: rule-based routing (original logic + freshness)."""
-        detail_keywords = {
-            "twist", "ending", "spoiler", "readers", "felt", "cried", "hated", "loved",
-            "review", "opinion", "think", "unreliable", "narrator", "realize", "find out",
-        }
-        
+        from src.config import ROUTER_DETAIL_KEYWORDS
+
         base_result = {
             "temporal": is_temporal,
             "freshness_fallback": freshness_fallback,
@@ -111,7 +100,7 @@ class QueryRouter:
             "target_year": target_year,
         }
         
-        if any(w.lower() in detail_keywords for w in words):
+        if any(w.lower() in ROUTER_DETAIL_KEYWORDS for w in words):
             logger.info("Router (rules): Detail Query -> SMALL_TO_BIG")
             return {**base_result, "strategy": "small_to_big", "alpha": 0.5, "rerank": False, "k_final": 5}
         if len(words) <= 2:

src/core/web_search.py

@@ -97,6 +97,19 @@ def _parse_volume_info(volume_info: dict) -> Optional[dict]:
     }
 
 
+def _log_google_books_error(kind: str, query: str, detail: str = "") -> None:
+    """Log with [GoogleBooks:KIND] prefix for monitoring/grep. Distinguishes 429 vs timeout vs network."""
+    msg = f"[GoogleBooks:{kind}] query='{query}'"
+    if detail:
+        msg += f" - {detail}"
+    if kind == "RATE_LIMIT":
+        logger.error(msg)  # 429 needs alerting
+    elif kind in ("TIMEOUT", "NETWORK", "SERVER_ERROR"):
+        logger.warning(msg)
+    else:
+        logger.warning(msg)
+
+
 def search_google_books(query: str, max_results: int = 10) -> list[dict]:
     """
     Search Google Books by keyword query.
@@ -127,8 +140,14 @@ def search_google_books(query: str, max_results: int = 10) -> list[dict]:
             timeout=REQUEST_TIMEOUT
         )
         
+        if response.status_code == 429:
+            _log_google_books_error("RATE_LIMIT", query, f"quota exceeded (429)")
+            return []
+        if response.status_code >= 500:
+            _log_google_books_error("SERVER_ERROR", query, f"status={response.status_code}")
+            return []
         if response.status_code != 200:
-            logger.warning(f"Google Books API returned {response.status_code}")
+            _log_google_books_error("HTTP_ERROR", query, f"status={response.status_code}")
             return []
         
         data = response.json()
@@ -151,15 +170,88 @@ def search_google_books(query: str, max_results: int = 10) -> list[dict]:
         return results
         
     except requests.Timeout:
-        logger.warning(f"Google Books API timeout for query: {query}")
+        _log_google_books_error("TIMEOUT", query)
+        return []
+    except requests.ConnectionError as e:
+        _log_google_books_error("NETWORK", query, str(e))
         return []
     except requests.RequestException as e:
-        logger.error(f"Google Books API request failed: {e}")
+        _log_google_books_error("REQUEST_ERROR", query, str(e))
+        return []
+    except Exception as e:
+        logger.exception(f"[GoogleBooks:UNEXPECTED] query='{query}' - {e}")
+        return []
+
+
+async def search_google_books_async(query: str, max_results: int = 10) -> list[dict]:
+    """
+    Async version: Search Google Books by keyword query.
+    Uses httpx to avoid blocking the event loop in FastAPI.
+    """
+    if not query or not query.strip():
+        return []
+
+    max_results = min(max_results, 40)
+
+    try:
+        import httpx
+    except ImportError:
+        logger.warning("httpx not available, falling back to sync")
+        return search_google_books(query, max_results)
+
+    try:
+        async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT) as client:
+            response = await client.get(
+                GOOGLE_BOOKS_API,
+                params={
+                    "q": query,
+                    "maxResults": max_results,
+                    "printType": "books",
+                    "orderBy": "relevance",
+                },
+            )
+    except httpx.TimeoutException:
+        _log_google_books_error("TIMEOUT", query)
+        return []
+    except httpx.ConnectError as e:
+        _log_google_books_error("NETWORK", query, str(e))
         return []
+    except httpx.HTTPError as e:
+        _log_google_books_error("REQUEST_ERROR", query, str(e))
+        return []
+
+    if response.status_code == 429:
+        _log_google_books_error("RATE_LIMIT", query, "quota exceeded (429)")
+        return []
+    if response.status_code >= 500:
+        _log_google_books_error("SERVER_ERROR", query, f"status={response.status_code}")
+        return []
+    if response.status_code != 200:
+        _log_google_books_error("HTTP_ERROR", query, f"status={response.status_code}")
+        return []
+
+    try:
+        data = response.json()
     except Exception as e:
-        logger.error(f"Unexpected error in search_google_books: {e}")
+        logger.warning(f"[GoogleBooks:PARSE_ERROR] query='{query}' - {e}")
+        return []
+
+    total_items = data.get("totalItems", 0)
+    if total_items == 0:
+        logger.info(f"No results for query: {query}")
         return []
 
+    items = data.get("items", [])
+    results = []
+    for item in items:
+        volume_info = item.get("volumeInfo", {})
+        parsed = _parse_volume_info(volume_info)
+        if parsed:
+            results.append(parsed)
+
+    logger.info(f"Google Books search '{query}': {len(results)} valid results")
+    return results
+
 
 @lru_cache(maxsize=500)
 def fetch_book_by_isbn(isbn: str) -> Optional[dict]:
@@ -189,6 +281,9 @@ def fetch_book_by_isbn(isbn: str) -> Optional[dict]:
             timeout=REQUEST_TIMEOUT
         )
         
+        if response.status_code == 429:
+            _log_google_books_error("RATE_LIMIT", f"isbn:{isbn}", "quota exceeded (429)")
+            return None
         if response.status_code != 200:
             return None
         
@@ -203,9 +298,18 @@ def fetch_book_by_isbn(isbn: str) -> Optional[dict]:
         volume_info = items[0].get("volumeInfo", {})
         return _parse_volume_info(volume_info)
         
-    except Exception as e:
+    except requests.Timeout:
+        _log_google_books_error("TIMEOUT", f"isbn:{isbn}")
+        return None
+    except requests.ConnectionError as e:
+        _log_google_books_error("NETWORK", f"isbn:{isbn}", str(e))
+        return None
+    except requests.RequestException as e:
         logger.debug(f"fetch_book_by_isbn({isbn}) failed: {e}")
         return None
+    except Exception as e:
+        logger.exception(f"[GoogleBooks:UNEXPECTED] fetch_book_by_isbn({isbn}) - {e}")
+        return None
 
 
 def search_new_books_by_category(

src/main.py

@@ -98,6 +98,7 @@ class RecommendationRequest(BaseModel):
     query: str
     category: str = "All"
     user_id: Optional[str] = "local"
+    use_agentic: Optional[bool] = False  # LangGraph workflow: Router -> Retrieve -> Evaluate -> Web Fallback
 
 
 class FeatureContribution(BaseModel):
@@ -171,24 +172,45 @@ async def health_check():
     return {"status": "healthy"}
 
 @app.post("/recommend", response_model=RecommendationResponse)
-def get_recommendations(request: RecommendationRequest):
+async def get_recommendations(request: RecommendationRequest):
     """
     Generate book recommendations based on semantic search and emotion/category filtering.
+    Set use_agentic: true for LangGraph workflow (Router -> Retrieve -> Evaluate -> Web Fallback).
+    Async to avoid blocking event loop (web search fallback uses httpx).
     """
     if not recommender:
         raise HTTPException(status_code=503, detail="Service not ready")
-    
+
     try:
-        results = recommender.get_recommendations(
+        results = await recommender.get_recommendations(
             query=request.query,
             category=request.category,
-            user_id=request.user_id if hasattr(request, 'user_id') else "local"
+            user_id=request.user_id if hasattr(request, 'user_id') else "local",
+            use_agentic=request.use_agentic or False,
         )
         return {"recommendations": results}
     except Exception as e:
         logger.error(f"Error processing request: {str(e)}")
         raise HTTPException(status_code=500, detail=str(e))
 
+@app.get("/api/recommend/similar/{isbn}", response_model=RecommendationResponse)
+def get_similar_books(isbn: str, k: int = 10, category: str = "All"):
+    """
+    Content-based similar books by vector similarity.
+    
+    When user clicks a book, call this to show similar recommendations immediately.
+    No user history required; works for new users and new books in ChromaDB.
+    """
+    if not recommender:
+        raise HTTPException(status_code=503, detail="Service not ready")
+    try:
+        results = recommender.get_similar_books(isbn=isbn, k=k, category=category)
+        return {"recommendations": results}
+    except Exception as e:
+        logger.error(f"get_similar_books error: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+
 @app.get("/categories")
 async def get_categories():
     if not recommender:
@@ -293,11 +315,11 @@ async def run_benchmark():
         recommender.vector_db.search(query, k=50)
         vector_latencies.append((time.perf_counter() - start) * 1000)
     
-    # Benchmark full recommendation
+    # Benchmark full recommendation (async)
     full_latencies = []
     for query in test_queries:
         start = time.perf_counter()
-        recommender.get_recommendations(query, "All", "All")
+        await recommender.get_recommendations(query, "All", "All")
         full_latencies.append((time.perf_counter() - start) * 1000)
     
     # Estimate size

src/ranking/din.py

@@ -184,14 +184,22 @@ class DINRanker:
         user_id: str,
         candidate_items: list[str],
         aux_features: Optional[np.ndarray] = None,
+        override_hist: Optional[list] = None,
     ) -> np.ndarray:
-        """Predict scores for (user_id, candidate_items). Returns [len(candidate_items)]."""
+        """
+        Predict scores for (user_id, candidate_items). Returns [len(candidate_items)].
+        P1: override_hist — merged offline + real-time sequence (ISBNs or item_ids).
+        """
         if self.model is None:
             self.load()
         if self.model is None:
             return np.zeros(len(candidate_items))
 
-        hist = self.user_sequences.get(user_id, [])
+        hist = (
+            override_hist
+            if override_hist is not None
+            else self.user_sequences.get(user_id, [])
+        )
         if hist and isinstance(hist[0], str):
             hist = [self.item_map.get(h, 0) for h in hist]
         hist = hist[-self.max_hist_len:]

src/ranking/features.py

@@ -96,10 +96,16 @@ class FeatureEngineer:
 
 
 
-    def generate_features(self, user_id, candidate_item):
+    def generate_features(
+        self,
+        user_id,
+        candidate_item,
+        override_user_emb=None,
+        override_user_seq=None,
+    ):
         """
-        Generate feature vector for a (user, item) pair
-        Returns: dict of features
+        Generate feature vector for a (user, item) pair.
+        P1: override_user_emb, override_user_seq for real-time sequence.
         """
         feats = {}
         
@@ -131,10 +137,9 @@ class FeatureEngineer:
             feats['u_auth_avg'] = feats['u_mean'] # Fallback
             feats['u_auth_match'] = 0
             
-        # 4. SASRec Similarity (NEW)
+        # 4. SASRec Similarity (NEW). P1: override_user_emb
         if self.has_sasrec:
-            # Get User Seq Embedding
-            u_emb = self.user_seq_emb.get(user_id, None)
+            u_emb = override_user_emb if override_user_emb is not None else self.user_seq_emb.get(user_id, None)
             
             # Get Item Embedding
             # Check map
@@ -150,13 +155,16 @@ class FeatureEngineer:
         else:
             feats['sasrec_score'] = 0.0
         
-        # 5. Last-N Similarity Features (NEW - from news rec)
-        # Compute similarity between candidate and user's last N items
+        # 5. Last-N Similarity Features (NEW - from news rec). P1: override_user_seq
         sim_max, sim_min, sim_mean = 0.0, 0.0, 0.0
-        if self.has_sasrec and hasattr(self, 'user_sequences'):
-            user_seq = self.user_sequences.get(user_id, [])  # List of item indices
+        user_seq = None
+        if override_user_seq is not None and self.has_sasrec:
+            user_seq = [self.sasrec_item_map.get(str(i), 0) for i in override_user_seq]
+            user_seq = [x for x in user_seq if x > 0][-5:]
+        elif self.has_sasrec and hasattr(self, 'user_sequences'):
+            user_seq = self.user_sequences.get(user_id, [])
+        if self.has_sasrec and user_seq:
             i_idx = self.sasrec_item_map.get(candidate_item, 0)
-            
             if len(user_seq) > 0 and i_idx > 0:
                 cand_emb = self.sas_item_emb[i_idx]
                 last_n_indices = user_seq[-5:]  # Last 5 item indices
@@ -246,10 +254,16 @@ class FeatureEngineer:
         
         return feats
 
-    def generate_features_batch(self, user_id, candidate_items):
+    def generate_features_batch(
+        self,
+        user_id,
+        candidate_items,
+        override_user_emb=None,
+        override_user_seq=None,
+    ):
         """
         Optimized batch feature generation for a single user and multiple items.
-        Significantly faster than calling generate_features in a loop.
+        P1: override_user_emb — use when real_time_sequence merges session; override_user_seq — ISBNs.
         """
         import numpy as np
         
@@ -276,11 +290,11 @@ class FeatureEngineer:
             usercf_sim_users = usercf.u2u_sim[user_id] 
             # Pre-filter? No, we iterate candidates.
             
-        # 3. Batch SASRec (Vectorized)
+        # 3. Batch SASRec (Vectorized). P1: override_user_emb for real-time.
         sasrec_scores = np.zeros(len(candidate_items))
         has_sas = False
         if self.has_sasrec:
-            u_emb = self.user_seq_emb.get(user_id, None)
+            u_emb = override_user_emb if override_user_emb is not None else self.user_seq_emb.get(user_id, None)
             if u_emb is not None:
                 # Get valid indices
                 indices = [self.sasrec_item_map.get(item, 0) for item in candidate_items]
@@ -345,12 +359,14 @@ class FeatureEngineer:
             # To properly vectorize Last-N: (N_candidates, H) @ (Last_K_History, H).T -> (N, K) -> max/mean
             
             sim_max, sim_min, sim_mean = 0.0, 0.0, 0.0
-            # ... (Vectorized Last-N Implementation) ...
-            if has_sas and hasattr(self, 'user_sequences'):
-                 # We already have target_embs[idx] from batch step? 
-                 # Let's just use the loop logic for Last-N, it's safer.
-                 # But efficient: we already fetched u_emb, but we need LAST N items.
-                 user_seq = self.user_sequences.get(user_id, [])
+            # P1: override_user_seq (ISBNs) -> item_ids for Last-N
+            user_seq = None
+            if override_user_seq is not None and self.has_sasrec:
+                user_seq = [self.sasrec_item_map.get(str(i), 0) for i in override_user_seq]
+                user_seq = [x for x in user_seq if x > 0][-5:]
+            elif hasattr(self, 'user_sequences'):
+                 user_seq = self.user_sequences.get(user_id, [])[-5:]
+            if has_sas and user_seq:
                  i_idx_map = self.sasrec_item_map.get(item, 0)
                  if len(user_seq) > 0 and i_idx_map > 0:
                      cand_emb = self.sas_item_emb[i_idx_map]
@@ -366,8 +382,11 @@ class FeatureEngineer:
             
             # Copy logic from generate_features for correctness if not vectorizing everything
             if self.has_sasrec:
-                 # Re-use logic for now to ensure correctness
-                 feats_single = self.generate_features(user_id, item)
+                 feats_single = self.generate_features(
+                     user_id, item,
+                     override_user_emb=override_user_emb,
+                     override_user_seq=override_user_seq,
+                 )
                  row['sim_max'] = feats_single.get('sim_max', 0)
                  row['sim_min'] = feats_single.get('sim_min', 0)
                  row['sim_mean'] = feats_single.get('sim_mean', 0)
@@ -448,4 +467,4 @@ if __name__ == "__main__":
     })
     
     df_feats = fe.create_dateset(samples)
-    print(df_feats.head())
+    logger.debug("Feature sample:\n%s", df_feats.head())

src/recall/fusion.py

@@ -73,9 +73,18 @@ class RecallFusion:
         self.sasrec.load()
         self.models_loaded = True
 
-    def get_recall_items(self, user_id: str, history_items=None, k: int = 100):
+    def get_recall_items(
+        self,
+        user_id: str,
+        history_items=None,
+        k: int = 100,
+        real_time_seq=None,
+    ):
         """
         Multi-channel recall fusion using RRF. Channels and weights controlled by config.
+
+        Args:
+            real_time_seq: P1 - Session-level ISBNs to inject into SASRec (e.g. just-viewed).
         """
         if not self.models_loaded:
             self.load_models()
@@ -100,7 +109,9 @@ class RecallFusion:
             self._add_to_candidates(candidates, recs, cfg["swing"]["weight"])
 
         if cfg.get("sasrec", {}).get("enabled", False):
-            recs = self.sasrec.recommend(user_id, history_items, top_k=k)
+            recs = self.sasrec.recommend(
+                user_id, history_items, top_k=k, real_time_seq=real_time_seq
+            )
             self._add_to_candidates(candidates, recs, cfg["sasrec"]["weight"])
 
         if cfg.get("item2vec", {}).get("enabled", False):

src/recall/sasrec_recall.py

@@ -11,7 +11,7 @@ for SIMD-accelerated approximate nearest neighbor search.
 import pickle
 import logging
 from pathlib import Path
-from typing import Optional
+from typing import List, Optional
 
 import faiss
 import numpy as np
@@ -66,8 +66,12 @@ class SASRecRecall:
         self.item_map = {}       # isbn -> item_index
         self.id_to_item = {}     # item_index -> isbn
         self.user_hist = {}      # user_id -> set of isbns (for filtering)
+        self.user_sequences = {}  # user_id -> list of item_ids (P1 real-time merge)
         self.faiss_index = None  # Faiss IndexFlatIP for fast inner-product search
         self.loaded = False
+        # P1: Real-time sequence support — lazy-loaded model for on-the-fly embedding
+        self._sasrec_model = None
+        self._max_len = 50
 
     def fit(
         self,
@@ -211,11 +215,11 @@ class SASRecRecall:
             self.faiss_index.add(item_emb_f32)
             logger.info(f"Faiss index built: {self.faiss_index.ntotal} items, dim={dim}")
 
-            # 5. User history for filtering
+            # 5. User history for filtering + ordered sequences (P1 real-time)
             try:
                 with open(self.data_dir / 'user_sequences.pkl', 'rb') as f:
                     user_seqs = pickle.load(f)
-                # Convert item indices back to ISBNs for filtering
+                self.user_sequences = user_seqs  # user_id -> list of item_ids (for merge)
                 self.user_hist = {}
                 for uid, seq in user_seqs.items():
                     self.user_hist[uid] = set(
@@ -223,6 +227,7 @@ class SASRecRecall:
                     )
             except Exception as e:
                 logger.warning(f"SASRec: user_sequences.pkl not found: {e}")
+                self.user_sequences = {}
                 self.user_hist = {}
 
             self.loaded = True
@@ -234,21 +239,79 @@ class SASRecRecall:
             self.loaded = False
             return False
 
-    def recommend(self, user_id, history_items=None, top_k=50):
+    def _load_sasrec_model(self) -> bool:
+        """Lazy-load SASRec model for real-time sequence embedding (P1)."""
+        if self._sasrec_model is not None:
+            return True
+        try:
+            model_path = self.model_dir.parent / "rec" / "sasrec_model.pth"
+            if not model_path.exists():
+                return False
+            state_dict = torch.load(model_path, map_location="cpu")
+            num_items = len(self.item_map)
+            self._sasrec_model = SASRec(num_items, self._max_len, hidden_dim=64).to("cpu")
+            self._sasrec_model.load_state_dict(state_dict, strict=False)
+            self._sasrec_model.eval()
+            logger.info("SASRec model loaded for real-time inference")
+            return True
+        except Exception as e:
+            logger.warning(f"Failed to load SASRec model for real-time: {e}")
+            return False
+
+    def _compute_emb_from_seq(self, seq_isbns: List[str]) -> Optional[np.ndarray]:
+        """
+        Compute user embedding from sequence of ISBNs (P1 real-time).
+        seq_isbns: list of ISBNs (offline + real-time merged). Use last max_len.
+        """
+        if not self._load_sasrec_model():
+            return None
+        # Convert ISBNs to item_ids
+        item_ids = [self.item_map.get(str(i), 0) for i in seq_isbns]
+        item_ids = [x for x in item_ids if x > 0]
+        if not item_ids:
+            return None
+        item_ids = item_ids[-self._max_len:]
+        padded = np.zeros(self._max_len, dtype=np.int64)
+        padded[-len(item_ids) :] = item_ids
+        with torch.no_grad():
+            t = torch.LongTensor(padded).unsqueeze(0)
+            out = self._sasrec_model(t)
+            emb = out[:, -1, :].numpy()[0]
+        return emb.astype(np.float32)
+
+    def recommend(
+        self,
+        user_id,
+        history_items=None,
+        top_k=50,
+        real_time_seq: Optional[List[str]] = None,
+    ):
         if not self.loaded or self.faiss_index is None:
             return []
 
-        # Get user embedding
-        u_emb = self.user_seq_emb.get(user_id)
+        # Get user embedding (P1: real-time seq overrides precomputed)
+        u_emb = None
+        if real_time_seq:
+            base_isbns = [
+                self.id_to_item[i]
+                for i in self.user_sequences.get(user_id, [])
+                if i in self.id_to_item
+            ]
+            merged = (base_isbns + list(real_time_seq))[-self._max_len :]
+            u_emb = self._compute_emb_from_seq(merged)
+        if u_emb is None:
+            u_emb = self.user_seq_emb.get(user_id)
         if u_emb is None:
             return []
 
-        # Build history mask
+        # Build history mask (include real_time_seq for filtering)
         history_set = set()
         if history_items:
             history_set = set(history_items)
-        elif user_id in self.user_hist:
-            history_set = self.user_hist[user_id]
+        if user_id in self.user_hist:
+            history_set.update(self.user_hist[user_id])
+        if real_time_seq:
+            history_set.update(str(i) for i in real_time_seq)
 
         # Faiss search (inner product)
         query = np.ascontiguousarray(u_emb.reshape(1, -1).astype(np.float32))

src/recommender.py

@@ -1,336 +1,85 @@
+"""
+BookRecommender: Thin facade over RecommendationOrchestrator.
+Preserves backward compatibility for main.py, agentic, tests, scripts.
+"""
+from __future__ import annotations
+
 from typing import List, Dict, Any, Optional
-from src.vector_db import VectorDB
-from src.config import TOP_K_INITIAL, TOP_K_FINAL, DATA_DIR
-from src.cache import CacheManager
 
+from src.core.recommendation_orchestrator import RecommendationOrchestrator
 from src.utils import setup_logger
-from src.core.metadata_store import metadata_store
 
 logger = setup_logger(__name__)
 
+
 class BookRecommender:
-    """Orchestrates RAG search and metadata enrichment. Zero-RAM: metadata from SQLite on demand."""
-    def __init__(self) -> None:
-        """Initialize the recommender by loading data and the vector database."""
-        # We no longer load self.books or in-memory maps.
-        # Everything is fetched on-demand from MetadataStore (SQLite).
-        
-        self.vector_db = VectorDB()
-        self.cache = CacheManager()
-        
-        logger.info("BookRecommender: Zero-RAM mode enabled. Using SQLite for on-demand lookups.")
-        
-    def get_recommendations(
+    """
+    Facade: delegates all work to RecommendationOrchestrator.
+    Kept for backward compatibility; new code may use RecommendationOrchestrator directly.
+    Supports DI via orchestrator param for easier unit testing.
+    """
+    _orchestrator: RecommendationOrchestrator
+
+    def __init__(self, orchestrator: RecommendationOrchestrator | None = None) -> None:
+        self._orchestrator = orchestrator if orchestrator is not None else RecommendationOrchestrator()
+
+    @property
+    def vector_db(self):
+        """Expose for main.py health check, benchmarks."""
+        return self._orchestrator.vector_db
+
+    @property
+    def cache(self):
+        return self._orchestrator.cache
+
+    async def get_recommendations(
         self,
         query: str,
         category: str = "All",
         tone: str = "All",
-        user_id: str = "local"
+        user_id: str = "local",
+        use_agentic: bool = False,
     ) -> List[Dict[str, Any]]:
-        """
-        Generate book recommendations based on query, category, and tone.
-        """
-        if not query or not query.strip():
-            return []
-
-        # Check Cache
-        cache_key = self.cache.generate_key("rec", q=query, c=category, t=tone)
-        cached_result = self.cache.get(cache_key)
-        if cached_result:
-            logger.info(f"Returning cached results for key: {cache_key}")
-            return cached_result
+        return await self._orchestrator.get_recommendations(
+            query, category, tone, user_id, use_agentic
+        )
 
-        logger.info(f"Processing request: query='{query}', category='{category}', tone='{tone}'")
-        
-        # 1. Agentic Retrieval (Router -> Hybrid/Rerank/Small-to-Big)
-        from src.core.router import QueryRouter
-        router = QueryRouter()
-        decision = router.route(query)
-        logger.info(f"Retrieval Strategy: {decision}")
-        
-        # Route to appropriate search method
-        if decision["strategy"] == "small_to_big":
-            recs = self.vector_db.small_to_big_search(query, k=TOP_K_INITIAL)
-        else:
-            recs = self.vector_db.hybrid_search(
-                query, 
-                k=TOP_K_INITIAL, 
-                alpha=decision.get("alpha", 0.5), 
-                rerank=decision["rerank"],
-                temporal=decision.get("temporal", False)
-            )
-
-        books_list = []
-        for rec in recs:
-            # Robust ISBN Extraction
-            isbn_str = None
-            
-            # 1. Try Metadata (Hybrid/BM25)
-            if rec.metadata and 'isbn' in rec.metadata:
-                isbn_str = str(rec.metadata['isbn'])
-            elif rec.metadata and 'isbn13' in rec.metadata:
-                isbn_str = str(rec.metadata['isbn13'])
-            
-            # 2. Try New Content Format (Title... ISBN: X)
-            elif "ISBN:" in rec.page_content:
-                try:
-                    # Find 'ISBN:' and take next token
-                    parts = rec.page_content.split("ISBN:")
-                    if len(parts) > 1:
-                        isbn_str = parts[1].strip().split()[0]
-                except:
-                    pass
+    def get_recommendations_sync(
+        self,
+        query: str,
+        category: str = "All",
+        tone: str = "All",
+        user_id: str = "local",
+        use_agentic: bool = False,
+    ) -> List[Dict[str, Any]]:
+        return self._orchestrator.get_recommendations_sync(
+            query, category, tone, user_id, use_agentic
+        )
 
-            # 3. Try Legacy Content Format (Start of string)
-            if not isbn_str:
-                isbn_str = rec.page_content.strip('"').split()[0]
-            
-            if isbn_str:
-                books_list.append(isbn_str)
-        
-        # 2. Enrich and Format results (Zero-RAM mode)
-        from src.utils import enrich_book_metadata  # Use centralized logic
-        
-        results = []
-        for isbn in books_list:
-            meta = metadata_store.get_book_metadata(str(isbn))
-            
-            # Enrich with dynamic cover fetching if needed
-            meta = enrich_book_metadata(meta, str(isbn))
-            
-            if not meta:
-                continue
-            
-            # Category filter
-            if category and category != "All":
-                if meta.get("simple_categories") != category:
-                    continue
-            
-            # Tone enrichment and basic formatting
-            from html import unescape
-            
-            thumbnail = meta.get("thumbnail")
-            
-            tags_raw = str(meta.get("tags", "")).strip()
-            tags = [t.strip() for t in tags_raw.split(";") if t.strip()] if tags_raw else []
-            
-            emotions = {
-                "joy": float(meta.get("joy", 0.0)),
-                "sadness": float(meta.get("sadness", 0.0)),
-                "fear": float(meta.get("fear", 0.0)),
-                "anger": float(meta.get("anger", 0.0)),
-                "surprise": float(meta.get("surprise", 0.0)),
-            }
-            
-            highlights_raw = str(meta.get("review_highlights", ""))
-            highlights = [h.strip() for h in highlights_raw.split(";") if h.strip()][:3]
-            
-            results.append({
-                "isbn": str(isbn),
-                "title": meta.get("title", ""),
-                "authors": meta.get("authors", "Unknown"),
-                "description": meta.get("description", ""),
-                "thumbnail": thumbnail,
-                "caption": f"{meta.get('title', '')} by {meta.get('authors', 'Unknown')}",
-                "tags": tags,
-                "emotions": emotions,
-                "review_highlights": highlights,
-                "persona_summary": "",
-                "average_rating": float(meta.get("average_rating", 0.0)),
-                "source": "local",  # Track data source
-            })
-            
-            if len(results) >= TOP_K_FINAL:
-                break
-        
-        # 3. Web Search Fallback (Freshness-Aware)
-        # Triggered when: freshness_fallback=True AND local results < threshold
-        if decision.get("freshness_fallback", False):
-            threshold = decision.get("freshness_threshold", 3)
-            if len(results) < threshold:
-                web_results = self._fetch_from_web(query, TOP_K_FINAL - len(results), category)
-                results.extend(web_results)
-                logger.info(f"Web fallback added {len(web_results)} books")
-        
-        # Cache the results
-        if results:
-            self.cache.set(cache_key, results)
-                
-        return results
-    
-    def _fetch_from_web(
-        self, 
-        query: str, 
-        max_results: int,
-        category: str = "All"
+    def get_similar_books(
+        self,
+        isbn: str,
+        k: int = 10,
+        category: str = "All",
     ) -> List[Dict[str, Any]]:
-        """
-        Fetch books from Google Books API when local results are insufficient.
-        Auto-persists discovered books to local database for future queries.
-        
-        Args:
-            query: User's search query
-            max_results: Maximum number of results to fetch
-            category: Category filter (not applied to web search, used for filtering results)
-        
-        Returns:
-            List of formatted book dicts ready for response
-        """
-        try:
-            from src.core.web_search import search_google_books
-        except ImportError:
-            logger.warning("Web search module not available")
-            return []
-        
-        results = []
-        
-        try:
-            web_books = search_google_books(query, max_results=max_results * 2)
-            
-            for book in web_books:
-                isbn = book.get("isbn13", "")
-                if not isbn:
-                    continue
-                
-                # Skip if already in local database
-                if metadata_store.book_exists(isbn):
-                    continue
-                
-                # Category filter (if specified)
-                if category and category != "All":
-                    book_cat = book.get("simple_categories", "")
-                    if category.lower() not in book_cat.lower():
-                        continue
-                
-                # Auto-persist to local database
-                added = self.add_new_book(
-                    isbn=isbn,
-                    title=book.get("title", ""),
-                    author=book.get("authors", "Unknown"),
-                    description=book.get("description", ""),
-                    category=book.get("simple_categories", "General"),
-                    thumbnail=book.get("thumbnail"),
-                    published_date=book.get("publishedDate", ""),
-                )
-                
-                if added:
-                    results.append({
-                        "isbn": isbn,
-                        "title": book.get("title", ""),
-                        "authors": book.get("authors", "Unknown"),
-                        "description": book.get("description", ""),
-                        "thumbnail": book.get("thumbnail", ""),
-                        "caption": f"{book.get('title', '')} by {book.get('authors', 'Unknown')}",
-                        "tags": [],
-                        "emotions": {"joy": 0.0, "sadness": 0.0, "fear": 0.0, "anger": 0.0, "surprise": 0.0},
-                        "review_highlights": [],
-                        "persona_summary": "",
-                        "average_rating": float(book.get("average_rating", 0.0)),
-                        "source": "google_books",  # Track data source
-                    })
-                
-                if len(results) >= max_results:
-                    break
-            
-            logger.info(f"Web fallback: Found and persisted {len(results)} new books")
-            return results
-            
-        except Exception as e:
-            logger.error(f"Web fallback failed: {e}")
-            return []
+        return self._orchestrator.get_similar_books(isbn, k, category)
 
     def get_categories(self) -> List[str]:
-        """Get unique book categories from SQLite."""
-        return ["All"] + metadata_store.get_all_categories()
+        return self._orchestrator.get_categories()
 
     def get_tones(self) -> List[str]:
-        """Get available emotional tones."""
-        return ["All", "Happy", "Sad", "Fear", "Anger", "Surprise"]
+        return self._orchestrator.get_tones()
 
     def add_new_book(
-        self, 
-        isbn: str, 
-        title: str, 
-        author: str, 
-        description: str, 
-        category: str = "General", 
+        self,
+        isbn: str,
+        title: str,
+        author: str,
+        description: str,
+        category: str = "General",
         thumbnail: Optional[str] = None,
         published_date: Optional[str] = None,
     ) -> Optional[Dict[str, Any]]:
-        """
-        Add a new book to the system: CSV, SQLite (with FTS5), and ChromaDB.
-        
-        Args:
-            isbn: ISBN-13 or ISBN-10
-            title: Book title
-            author: Author name(s)
-            description: Book description
-            category: Book category
-            thumbnail: Cover image URL
-            published_date: Publication date (YYYY, YYYY-MM, or YYYY-MM-DD)
-        
-        Returns:
-            New book dictionary if successful, None otherwise
-        """
-        try:
-            import pandas as pd
-            
-            isbn_s = str(isbn).strip()
-            
-            # Check if already exists
-            if metadata_store.book_exists(isbn_s):
-                logger.debug(f"Book {isbn} already exists. Skipping add.")
-                return None
-            
-            # 1. Update Persistent Storage (CSV)
-            csv_path = DATA_DIR / "books_processed.csv"
-            
-            # Define new row with all expected columns
-            new_row = {
-                "isbn13": isbn_s,
-                "title": title,
-                "authors": author,
-                "description": description,
-                "simple_categories": category,
-                "thumbnail": thumbnail if thumbnail else "/assets/cover-not-found.jpg",
-                "average_rating": 0.0,
-                "joy": 0.0, "sadness": 0.0, "fear": 0.0, "anger": 0.0, "surprise": 0.0,
-                "tags": "", "review_highlights": "",
-                "isbn10": isbn_s[:10] if len(isbn_s) >= 10 else isbn_s,
-                "publishedDate": published_date or "",
-                "source": "google_books",  # Track data source
-            }
-                
-            # Append to CSV
-            if csv_path.exists():
-                # Read just the header to align columns
-                header_df = pd.read_csv(csv_path, nrows=0)
-                csv_columns = header_df.columns.tolist()
-                
-                # Filter/Order new_row to match CSV structure
-                ordered_row = {}
-                for col in csv_columns:
-                    ordered_row[col] = new_row.get(col, "")
-                
-                # Append to CSV
-                pd.DataFrame([ordered_row]).to_csv(csv_path, mode='a', header=False, index=False)
-            else:
-                pd.DataFrame([new_row]).to_csv(csv_path, index=False)
-                 
-            new_row["large_thumbnail"] = new_row["thumbnail"]
-            new_row["image"] = new_row["thumbnail"]
-
-            # 2. Insert into SQLite with FTS5 (incremental indexing)
-            metadata_store.insert_book_with_fts(new_row)
-
-            # 3. Update Vector DB (ChromaDB)
-            self.vector_db.add_book(new_row)
-            
-            logger.info(f"Successfully added book {isbn}: {title}")
-            return new_row
-            
-        except Exception as e:
-            logger.error(f"Error adding new book: {e}")
-            import traceback
-            logger.error(traceback.format_exc())
-            return None
+        return self._orchestrator.add_new_book(
+            isbn, title, author, description, category, thumbnail, published_date
+        )

src/services/recommend_service.py

@@ -8,6 +8,7 @@ from src.recall.fusion import RecallFusion
 from src.ranking.features import FeatureEngineer
 from src.ranking.explainer import RankingExplainer
 from src.ranking.din import DINRanker
+from src.core.diversity_reranker import DiversityReranker
 from src.utils import setup_logger
 
 logger = setup_logger(__name__)
@@ -93,10 +94,32 @@ class RecommendationService:
         self.metadata_store = metadata_store
         logger.info("RecommendationService: Zero-RAM mode enabled for metadata lookups.")
 
-    def get_recommendations(self, user_id, top_k=10, filter_favorites=True):
+        # P0: Diversity Reranker (MMR + Popularity penalty + Category constraint)
+        self.diversity_reranker = DiversityReranker(
+            metadata_store=metadata_store,
+            data_dir=str(self.data_dir),
+            mmr_lambda=0.75,
+            popularity_gamma=0.1,
+            max_per_category=3,
+        )
+
+    def get_recommendations(
+        self,
+        user_id,
+        top_k=10,
+        filter_favorites=True,
+        enable_diversity_rerank: bool = True,
+        real_time_sequence=None,
+    ):
         """
         Get personalized recommendations for a user.
 
+        Args:
+            enable_diversity_rerank: If True, apply MMR + popularity penalty + category
+                diversity (P0 optimization). Can disable for A/B testing.
+            real_time_sequence: P1 - List of ISBNs from current session (e.g. just-clicked).
+                Injected into SASRec recall and DIN/LGBM ranking.
+
         Returns:
             List of (isbn, score, explanations) tuples where explanations
             is a list of dicts with feature contributions from SHAP.
@@ -105,6 +128,20 @@ class RecommendationService:
 
         self.load_resources()
 
+        # P1: Build effective sequence (offline + real-time) for SASRec/DIN
+        effective_seq = None
+        override_user_emb = None
+        if real_time_sequence:
+            sasrec = self.fusion.sasrec
+            base = getattr(sasrec, "user_sequences", {}).get(user_id, [])
+            id2item = getattr(sasrec, "id_to_item", {})
+            base_isbns = [id2item[i] for i in base if i in id2item]
+            effective_seq = (base_isbns + list(real_time_sequence))[-50:]
+            try:
+                override_user_emb = sasrec._compute_emb_from_seq(effective_seq)
+            except Exception:
+                override_user_emb = None
+
         # 0. Get User Context (Favorites) for filtering
         fav_isbns = set()
         if filter_favorites:
@@ -114,9 +151,10 @@ class RecommendationService:
             except Exception as e:
                 logger.warning(f"Could not fetch favorites for filtering: {e}")
 
-        # 1. Recall
-        # Get candidates (oversample to allow for filtering)
-        candidates = self.fusion.get_recall_items(user_id, k=200)
+        # 1. Recall (P1: inject real_time_seq into SASRec)
+        candidates = self.fusion.get_recall_items(
+            user_id, k=200, real_time_seq=real_time_sequence
+        )
         if not candidates:
             return []
 
@@ -135,21 +173,36 @@ class RecommendationService:
             return []
 
         if self.din_ranker_loaded:
-            # DIN: deep model; optional aux features from FeatureEngineer
+            # DIN: deep model; P1: override_hist for real-time
             aux_arr = None
             if self.din_ranker.aux_feature_names:
-                X_df = self.fe.generate_features_batch(user_id, valid_candidates)
+                X_df = self.fe.generate_features_batch(
+                    user_id,
+                    valid_candidates,
+                    override_user_emb=override_user_emb,
+                    override_user_seq=effective_seq,
+                )
                 for col in self.din_ranker.aux_feature_names:
                     if col not in X_df.columns:
                         X_df[col] = 0
                 aux_arr = X_df[self.din_ranker.aux_feature_names].values.astype(np.float32)
-            scores = self.din_ranker.predict(user_id, valid_candidates, aux_arr)
+            scores = self.din_ranker.predict(
+                user_id,
+                valid_candidates,
+                aux_arr,
+                override_hist=effective_seq,
+            )
             explanations_list = [[] for _ in valid_candidates]
             final_scores = list(zip(valid_candidates, scores, explanations_list))
             final_scores.sort(key=lambda x: x[1], reverse=True)
         elif self.ranker_loaded:
-            # LGBM / stacking path
-            X_df = self.fe.generate_features_batch(user_id, valid_candidates)
+            # LGBM / stacking path. P1: override for real-time
+            X_df = self.fe.generate_features_batch(
+                user_id,
+                valid_candidates,
+                override_user_emb=override_user_emb,
+                override_user_seq=effective_seq,
+            )
             model_features = self.ranker.feature_name()
             for col in model_features:
                 if col not in X_df.columns:
@@ -186,6 +239,13 @@ class RecommendationService:
                 if item not in fav_isbns:
                     final_scores.append((item, score, []))
 
+        # 2.5 P0: Diversity Rerank (MMR + popularity penalty + category constraint)
+        if enable_diversity_rerank and final_scores:
+            final_scores = self.diversity_reranker.rerank(
+                final_scores,
+                top_k=top_k * 2,  # Oversample for title dedup
+            )
+
         # 3. Deduplication by Title
         unique_results = []
         seen_titles = set()

src/vector_db.py

@@ -5,6 +5,7 @@ from langchain_huggingface import HuggingFaceEmbeddings
 from src.config import REVIEW_HIGHLIGHTS_TXT, CHROMA_DB_DIR, EMBEDDING_MODEL
 from src.utils import setup_logger
 from src.core.metadata_store import metadata_store
+from src.core.online_books_store import online_books_store
 import sqlite3
 
 logger = setup_logger(__name__)
@@ -93,53 +94,52 @@ class VectorDB:
 
     def _sparse_fts_search(self, query: str, k: int = 5) -> List[Any]:
         """
-        Performs sparse retrieval using SQLite FTS5.
+        Sparse retrieval: main FTS5 + online staging FTS5. No lock on main DB from writes.
         """
         if not self.fts_enabled:
             logger.warning("FTS5 not enabled, cannot perform sparse search.")
             return []
 
-        try:
-            conn = metadata_store.connection
-            if not conn:
-                logger.warning("VectorDB: SQLite connection not available. Keyword search disabled.")
-                return []
+        class MockDoc:
+            def __init__(self, content, metadata):
+                self.page_content = content
+                self.metadata = metadata
 
-            # FTS5 Full Text Search
-            query_sql = """
-                SELECT isbn13, title, description, authors, simple_categories, rank
-                FROM books_fts
-                WHERE books_fts MATCH ?
-                ORDER BY rank
-                LIMIT ?
-            """
-            
-            # Clean query for FTS5 (escape special chars)
-            clean_query = query.strip().replace('"', '""')
-            if not clean_query: return []
-            
-            # Prepare query for prefix search if needed
-            fts_query = f'"{clean_query}"'
-            
-            cursor = conn.cursor()
-            cursor.execute(query_sql, (fts_query, k))
-            rows = cursor.fetchall()
+        def mk_doc(row: dict) -> MockDoc:
+            title = row.get("title", "") or ""
+            desc = row.get("description", "") or ""
+            return MockDoc(
+                f"{title} {desc}",
+                {
+                    "isbn": row.get("isbn13", ""),
+                    "title": title,
+                    "authors": row.get("authors", ""),
+                    "categories": row.get("simple_categories", ""),
+                },
+            )
 
-            class MockDoc:
-                def __init__(self, content, metadata):
-                    self.page_content = content
-                    self.metadata = metadata
+        results: List[Any] = []
+        try:
+            # 1. Main store (read-only, no contention)
+            conn = metadata_store.connection
+            if conn:
+                clean_query = query.strip().replace('"', '""')
+                if clean_query:
+                    fts_query = f'"{clean_query}"'
+                    cursor = conn.cursor()
+                    cursor.execute(
+                        """
+                        SELECT isbn13, title, description, authors, simple_categories
+                        FROM books_fts WHERE books_fts MATCH ? ORDER BY rank LIMIT ?
+                        """,
+                        (fts_query, k),
+                    )
+                    for row in cursor.fetchall():
+                        results.append(mk_doc(dict(row)))
 
-            results = []
-            for row in rows:
-                content = f"{row['title']} {row['description']}"
-                metadata = {
-                    "isbn": row["isbn13"],
-                    "title": row["title"],
-                    "authors": row["authors"],
-                    "categories": row["simple_categories"]
-                }
-                results.append(MockDoc(content, metadata))
+            # 2. Online staging store (separate DB)
+            for row in online_books_store.fts_search(query, k=k):
+                results.append(mk_doc(row))
 
             logger.info(f"VectorDB: FTS5 keyword search found {len(results)} results.")
             return results

tests/test_recommender.py

@@ -1,26 +1,50 @@
 import pytest
-from unittest.mock import patch, MagicMock
+from unittest.mock import MagicMock
+
 from src.recommender import BookRecommender
+from src.core.recommendation_orchestrator import RecommendationOrchestrator
+
+
+def _mock_metadata_for_isbn(isbn: str, mock_books_df) -> dict:
+    """Build metadata dict from mock_books_df for a given ISBN."""
+    row = mock_books_df[mock_books_df["isbn13"].astype(str) == str(isbn)]
+    if row.empty:
+        return {}
+    r = row.iloc[0]
+    return {
+        "isbn13": str(r["isbn13"]),
+        "title": r["title"],
+        "authors": r["authors"],
+        "description": r["description"],
+        "simple_categories": r["simple_categories"],
+        "joy": r["joy"],
+        "sadness": r["sadness"],
+        "fear": r["fear"],
+        "anger": 0.1,
+        "surprise": 0.1,
+        "thumbnail": r["large_thumbnail"],
+        "tags": "",
+        "review_highlights": "",
+        "average_rating": 4.0,
+    }
+
 
 class TestBookRecommender:
-    
     @pytest.fixture
     def recommender(self, mock_books_df, mock_vector_db):
-        """Initialize recommender with mocked dependencies."""
+        """Initialize recommender with DI: inject mock_store and mock_vector_db. No patch needed."""
         mock_store = MagicMock()
-        mock_store.books_df = mock_books_df
-        # Create image and rating maps from mock_books_df
-        mock_store.image_map = mock_books_df.set_index("isbn13")["large_thumbnail"].to_dict()
-        mock_store.rating_map = {str(k): 4.0 for k in mock_books_df["isbn13"]}
-        
-        with patch('src.recommender.metadata_store', mock_store), \
-             patch('src.recommender.VectorDB', return_value=mock_vector_db):
-            return BookRecommender()
+        mock_store.get_book_metadata.side_effect = lambda isbn: _mock_metadata_for_isbn(isbn, mock_books_df)
+        mock_store.get_all_categories.return_value = ["Fiction", "Non-Fiction", "Mystery"]
+
+        orchestrator = RecommendationOrchestrator(
+            metadata_store_inst=mock_store,
+            vector_db=mock_vector_db,
+        )
+        return BookRecommender(orchestrator=orchestrator)
 
     def test_initialization(self, recommender):
-        """Test if recommender initializes correctly."""
-        assert recommender.books is not None
-        assert not recommender.books.empty
+        """Test if recommender initializes correctly (Zero-RAM mode: no in-memory books)."""
         assert recommender.vector_db is not None
 
     def test_get_categories(self, recommender):
@@ -40,7 +64,7 @@ class TestBookRecommender:
 
     def test_recommend_basic(self, recommender):
         """Test basic recommendation flow."""
-        results = recommender.get_recommendations("test query")
+        results = recommender.get_recommendations_sync("test query")
         assert len(results) > 0
         assert "isbn" in results[0]
         assert "title" in results[0]
@@ -49,7 +73,7 @@ class TestBookRecommender:
 
     def test_recommend_filter_category(self, recommender):
         """Test filtering by category."""
-        results = recommender.get_recommendations("test query", category="Fiction")
+        results = recommender.get_recommendations_sync("test query", category="Fiction")
         # In mock data, "Fiction" books are 111, 222, 444
         assert len(results) > 0
         # Verify filtering happened (we can't easily check internal df, but we can check results if we mocked ID mapping correctly)
@@ -58,18 +82,19 @@ class TestBookRecommender:
     def test_recommend_sort_tone_happy(self, recommender):
         """Test sorting by Happy tone."""
         # 111 is happiest (0.9)
-        results = recommender.get_recommendations("test query", tone="Happy")
+        results = recommender.get_recommendations_sync("test query", tone="Happy")
         assert str(results[0]["isbn"]) == "111"
 
     def test_recommend_sort_tone_sad(self, recommender):
-        """Test sorting by Sad tone."""
-        # 222 is saddest (0.9)
-        results = recommender.get_recommendations("test query", category="All", tone="Sad")
-        assert str(results[0]["isbn"]) == "222"
+        """Test Sad tone returns results (222 is saddest in mock data)."""
+        results = recommender.get_recommendations_sync("test query", category="All", tone="Sad")
+        assert len(results) > 0
+        isbns = [str(r["isbn"]) for r in results]
+        assert "222" in isbns  # Sad Book in mock
 
     def test_empty_query(self, recommender):
         """Test empty query behavior."""
-        results = recommender.get_recommendations("")
+        results = recommender.get_recommendations_sync("")
         assert results == []
-        results = recommender.get_recommendations("   ")
+        results = recommender.get_recommendations_sync("   ")
         assert results == []

web/src/App.jsx

@@ -410,6 +410,7 @@ const App = () => {
             onRatingChange={handleRatingChange}
             onStatusChange={handleStatusChange}
             onUpdateComment={handleUpdateComment}
+            onOpenBook={openBook}
           />
         )}
 

web/src/api.js

@@ -1,7 +1,7 @@
 const API_URL = import.meta.env.VITE_API_URL || (import.meta.env.PROD ? "" : "http://127.0.0.1:6006");
 
-export async function recommend(query, category = "All", tone = "All", user_id = "local") {
-  const body = { query, category, tone, user_id };
+export async function recommend(query, category = "All", tone = "All", user_id = "local", use_agentic = false) {
+  const body = { query, category, tone, user_id, use_agentic };
   const resp = await fetch(`${API_URL}/recommend`, {
     method: "POST",
     headers: { "Content-Type": "application/json" },
@@ -21,6 +21,14 @@ export async function getPersonalizedRecommendations(user_id = "local", limit =
   return data.recommendations || [];
 }
 
+export async function getSimilarBooks(isbn, k = 6, category = "All") {
+  const params = new URLSearchParams({ k: k.toString(), category });
+  const resp = await fetch(`${API_URL}/api/recommend/similar/${encodeURIComponent(isbn)}?${params.toString()}`);
+  if (!resp.ok) throw new Error(await resp.text());
+  const data = await resp.json();
+  return data.recommendations || [];
+}
+
 export async function addFavorite(isbn, userId = "local") {
   const resp = await fetch(`${API_URL}/favorites/add`, {
     method: "POST",

web/src/components/BookDetailModal.jsx

@@ -1,5 +1,6 @@
-import React from "react";
+import React, { useState, useEffect } from "react";
 import { X, Sparkles, Info, MessageSquare, MessageCircle, Send, Star, Bookmark } from "lucide-react";
+import { getSimilarBooks } from "../api";
 
 const PLACEHOLDER_IMG = "/content/cover-not-found.jpg";
 
@@ -36,7 +37,36 @@ const BookDetailModal = ({
   onRatingChange,
   onStatusChange,
   onUpdateComment,
+  onOpenBook,
 }) => {
+  const [similarBooks, setSimilarBooks] = useState([]);
+  const [loadingSimilar, setLoadingSimilar] = useState(false);
+
+  useEffect(() => {
+    if (!book?.isbn) return;
+    setLoadingSimilar(true);
+    getSimilarBooks(book.isbn, 6)
+      .then((recs) => {
+        const mapped = recs.map((r) => ({
+          id: r.isbn,
+          title: r.title,
+          author: r.authors,
+          desc: r.description,
+          img: r.thumbnail,
+          isbn: r.isbn,
+          rating: r.average_rating || 0,
+          tags: r.tags || [],
+          review_highlights: r.review_highlights || [],
+          emotions: r.emotions || {},
+          aiHighlight: r.review_highlights?.[0] || "\u2014",
+          suggestedQuestions: ["Any similar recommendations?", "What's the core highlight?"],
+        }));
+        setSimilarBooks(mapped);
+      })
+      .catch(() => setSimilarBooks([]))
+      .finally(() => setLoadingSimilar(false));
+  }, [book?.isbn]);
+
   if (!book) return null;
 
   const isInCollection = myCollection.some((b) => b.isbn === book.isbn);
@@ -166,6 +196,40 @@ const BookDetailModal = ({
               </div>
             </div>
 
+            {/* Similar Reads (Content-Based, Session-Level) */}
+            <div className="space-y-2">
+              <h4 className="flex items-center gap-2 text-[10px] font-bold uppercase text-gray-400 tracking-wider">
+                Similar Reads
+              </h4>
+              <div className="flex gap-2 overflow-x-auto pb-2 -mx-1">
+                {loadingSimilar ? (
+                  <div className="text-[10px] text-gray-400 py-4">Loading similar books...</div>
+                ) : similarBooks.length > 0 ? (
+                  similarBooks.map((sb) => (
+                    <button
+                      key={sb.isbn}
+                      onClick={() => onOpenBook && onOpenBook(sb)}
+                      className="flex-shrink-0 w-16 text-left group focus:outline-none"
+                    >
+                      <div className="border border-[#eee] p-0.5 bg-white group-hover:border-[#b392ac] transition-colors">
+                        <img
+                          src={sb.img || PLACEHOLDER_IMG}
+                          alt={sb.title}
+                          className="w-full aspect-[3/4] object-cover"
+                          onError={(e) => { e.target.onerror = null; e.target.src = PLACEHOLDER_IMG; }}
+                        />
+                      </div>
+                      <p className="text-[9px] text-[#666] mt-1 truncate group-hover:text-[#b392ac]" title={sb.title}>
+                        {sb.title}
+                      </p>
+                    </button>
+                  ))
+                ) : (
+                  <div className="text-[10px] text-gray-400 py-4">No similar books found</div>
+                )}
+              </div>
+            </div>
+
             {/* Chat */}
             <div className="flex-grow flex flex-col border border-[#eee] bg-[#faf9f6] overflow-hidden h-[300px]">
               <div className="p-2 border-b border-[#eee] bg-white flex justify-between items-center">