diff --git a/coach/coach_service.py b/coach/coach_service.py index 375d949..685fdb0 100644 --- a/coach/coach_service.py +++ b/coach/coach_service.py @@ -5,7 +5,6 @@ Main orchestrator: loads deck reports, identifies underperformers, finds replacement candidates via embeddings, builds prompts, calls the LLM, parses responses, and persists coaching sessions. """ - import json import logging import uuid @@ -65,8 +64,7 @@ class CoachService: self.embeddings = embeddings ensure_dirs() - # ── Deck Report I/O ──────────────────────────────────── - + # ── Deck Report I/O ────────────────────── def load_deck_report(self, deck_id: str) -> Optional[DeckReport]: """Load a deck report from disk.""" report_path = DECK_REPORTS_DIR / f"{deck_id}.json" @@ -92,8 +90,7 @@ class CoachService: return [] return [f.stem for f in DECK_REPORTS_DIR.glob("*.json")] - # ── Coach Session I/O ────────────────────────────── - + # ── Coach Session I/O ────────────────── def save_session(self, session: CoachSession) -> Path: """Persist a coaching session to disk and SQLite.""" COACH_SESSIONS_DIR.mkdir(parents=True, exist_ok=True) @@ -187,8 +184,7 @@ class CoachService: continue return sessions - # ── Main Coaching Pipeline ─────────────────────────────── - + # ── Main Coaching Pipeline ─────────────────────── async def run_coaching_session( self, deck_id: str, @@ -224,7 +220,6 @@ class CoachService: raise ValueError(f"Deck report not found for: {deck_id}") deck_card_names = [c.name for c in report.cards] - candidates: Dict[str, List[dict]] = {} underperformers = report.underperformers[:MAX_UNDERPERFORMERS] if not underperformers: @@ -252,22 +247,43 @@ class CoachService: candidates[card_name] = [m.to_dict() for m in matches] else: logger.warning("Embeddings not loaded — skipping candidate search") - # Phase 3: RAG vector search for additional card context + + # Phase 4a: behavior-aware RAG — primary + replacement-effect secondary query rag_cards: List[dict] = [] try: from services.rag_store import query_cards - query_text = f"{report.commander} {' '.join(report.colorIdentity)} commander deck" - rag_cards = query_cards(query_text, n_results=15, color_identity=report.colorIdentity) + # Primary: commander synergy context + base_query = f"{report.commander} {' '.join(report.colorIdentity)} commander deck" + rag_cards = query_cards(base_query, n_results=10, color_identity=report.colorIdentity) + # Secondary: replacement effects / graveyard hate / protection + if len(rag_cards) < 15: + repl_cards = query_cards( + f"replacement effect graveyard {' '.join(report.colorIdentity)}", + n_results=5, + color_identity=report.colorIdentity, + require_replacement=True, + ) + # Deduplicate by name before merging + existing_names = {c["name"] for c in rag_cards} + rag_cards += [c for c in repl_cards if c["name"] not in existing_names] logger.info("RAG returned %d cards for '%s'", len(rag_cards), deck_id) except Exception as e: logger.warning("RAG query failed (non-fatal): %s", e) + # Phase 4b: rules context for grounding the coach prompt + rules_context: List[dict] = [] + try: + from services.rag_store import query_rules + rules_context = query_rules( + f"Commander rules {report.commander} {' '.join(report.colorIdentity)}", + n_results=3, + ) + except Exception: + pass system_prompt = build_system_prompt(report, goals) - user_prompt = build_user_prompt(report, candidates, rag_cards=rag_cards) - + user_prompt = build_user_prompt(report, candidates, rag_cards=rag_cards, rules_context=rules_context) logger.info("Calling LLM for deck: %s (provider=%s)", deck_id, COACH_PROVIDER) - if COACH_PROVIDER == "anthropic": import anthropic as _anthropic anthropic_key = ANTHROPIC_API_KEY @@ -283,7 +299,7 @@ class CoachService: ) as stream: content = await stream.get_final_text() final_msg = await stream.get_final_message() - usage = getattr(final_msg, "usage", None) + usage = getattr(final_msg, "usage", None) from types import SimpleNamespace llm_response = SimpleNamespace( content=content, @@ -316,19 +332,16 @@ class CoachService: completionTokens=llm_response.completion_tokens, goals=goals, ) - if llm_response.parsed_json: self._populate_session_from_json(session, llm_response.parsed_json) else: session.rawTextExplanation = llm_response.content session.summary = "LLM returned non-JSON response. See raw explanation." - self.save_session(session) logger.info("Coaching session saved: %s", session_id) return session - # ── Quick Digest ────────────────────────────────────────── - + # ── Quick Digest ────────────────────────────── async def run_quick_digest( self, deck_id: str, @@ -342,7 +355,7 @@ class CoachService: if report is None: raise ValueError(f"Deck report not found for: {deck_id}") - # Phase 3: light RAG query for quick digest too + # Phase 4a: light RAG query for quick digest rag_cards: List[dict] = [] try: from services.rag_store import query_cards @@ -350,10 +363,10 @@ class CoachService: rag_cards = query_cards(query_text, n_results=8, color_identity=report.colorIdentity) except Exception: pass + from .prompt_template import build_quick_system_prompt, build_user_prompt system_prompt = build_quick_system_prompt(report, goals) - user_prompt = build_user_prompt(report, {}, rag_cards=rag_cards) # RAG-enriched quick digest # no candidates for speed - + user_prompt = build_user_prompt(report, {}, rag_cards=rag_cards) # no candidates for speed logger.info("Quick digest for deck: %s (provider=%s)", deck_id, COACH_PROVIDER) if COACH_PROVIDER == "anthropic": @@ -410,18 +423,15 @@ class CoachService: else: session.rawTextExplanation = llm_response.content session.summary = "Quick digest returned non-JSON. See raw explanation." - self.save_session(session) return session - # ── JSON Parser ─────────────────────────────────────────── - + # ── JSON Parser ─────────────────────────────── def _populate_session_from_json(self, session: CoachSession, data: dict): """Parse LLM JSON response into CoachSession fields.""" session.summary = data.get("summary", "") session.rawTextExplanation = data.get("rawTextExplanation", "") session.manaBaseAdvice = data.get("manaBaseAdvice") - hints = data.get("heuristicHints", []) session.heuristicHints = hints if isinstance(hints, list) else [str(hints)] @@ -432,7 +442,6 @@ class CoachService: reason=cut_data.get("reason", ""), replacementOptions=cut_data.get("replacementOptions", []), )) - for add_data in data.get("suggestedAdds", []): if isinstance(add_data, dict): session.suggestedAdds.append(SuggestedAdd( @@ -441,8 +450,6 @@ class CoachService: reason=add_data.get("reason", ""), synergyWith=add_data.get("synergyWith", []), )) - - # Phase 2 for item in data.get("upgradePriority", []): if isinstance(item, dict): session.upgradePriority.append(UpgradePriorityItem( @@ -452,7 +459,6 @@ class CoachService: reasoning=item.get("reasoning", ""), expectedImpact=item.get("expectedImpact", ""), )) - cd = data.get("commanderDependency") if isinstance(cd, dict): session.commanderDependency = CommanderDependency( @@ -460,7 +466,6 @@ class CoachService: dependentCards=cd.get("dependentCards", []), recoveryPlan=cd.get("recoveryPlan", ""), ) - ma = data.get("mulliganAnalysis") if isinstance(ma, dict): session.mulliganAnalysis = MulliganAnalysis( @@ -468,23 +473,17 @@ class CoachService: worstOffenders=ma.get("worstOffenders", []), recommendation=ma.get("recommendation", ""), ) - - # Phase 3: Removal coverage rc = data.get("removalCoverage") if isinstance(rc, dict): session.removalCoverage = RemovalCoverage(**{ k: rc.get(k, "" if k != "gaps" else []) for k in ["creatures", "artifacts", "enchantments", "planeswalkers", "lands", "massRemoval", "gaps"] }) - - # Phase 3: Ramp quality rq = data.get("rampQuality") if isinstance(rq, dict): session.rampQuality = RampQuality(**{ k: rq.get(k, "") for k in ["manaRocks", "landFetch", "manaDorks", "costReducers", "fragility", "canReachFourByTurnThree"] }) - - # Phase 3: Draw engine profile dep = data.get("drawEngineProfile") if isinstance(dep, dict): session.drawEngineProfile = DrawEngineProfile( @@ -493,8 +492,6 @@ class CoachService: assessment=dep.get("assessment", ""), sustainability=dep.get("sustainability", ""), ) - - # Phase 3: Win conditions for wc in data.get("winConditions", []): if isinstance(wc, dict): session.winConditions.append(WinCondition( @@ -503,8 +500,6 @@ class CoachService: independence=wc.get("independence", ""), description=wc.get("description", ""), )) - - # Phase 3: Anti-synergy flags for af in data.get("antiSynergyFlags", []): if isinstance(af, dict): session.antiSynergyFlags.append(AntiSynergyFlag( @@ -512,8 +507,6 @@ class CoachService: conflict=af.get("conflict", ""), severity=af.get("severity", ""), )) - - # Phase 5: Pod presence pp = data.get("podPresence") if isinstance(pp, dict): session.podPresence = PodPresence( @@ -522,8 +515,6 @@ class CoachService: adversarialRating=pp.get("adversarialRating", ""), recommendation=pp.get("recommendation", ""), ) - - # Phase 5: Tempo assessment ta = data.get("tempoAssessment") if isinstance(ta, dict): session.tempoAssessment = TempoAssessment( @@ -532,8 +523,6 @@ class CoachService: survivalWindow=ta.get("survivalWindow", ""), assessment=ta.get("assessment", ""), ) - - # Phase 5: Meta matchups for mm in data.get("metaMatchups", []): if isinstance(mm, dict): session.metaMatchups.append(MetaMatchup( @@ -544,12 +533,10 @@ class CoachService: overallRating=mm.get("overallRating", ""), )) - # ── Status Check ───────────────────────────────────────── - + # ── Status Check ────────────────────────────── def get_status(self) -> CoachStatus: """Check the health of all coach subsystems.""" status = CoachStatus() - if COACH_PROVIDER == "anthropic": anthropic_key = ANTHROPIC_API_KEY if anthropic_key: @@ -568,7 +555,6 @@ class CoachService: status.llmModels = llm_status.get("models", []) if not status.llmConnected: status.error = llm_status.get("error", "Local LLM not reachable") - status.embeddingsLoaded = self.embeddings.loaded status.embeddingCards = self.embeddings.card_count status.deckReportsAvailable = len(self.list_deck_reports()) diff --git a/routes/rag.py b/routes/rag.py index b2a37e3..3b044eb 100644 --- a/routes/rag.py +++ b/routes/rag.py @@ -119,3 +119,53 @@ async def rag_status_detail(): except Exception as exc: log.exception("RAG status check failed") return {"status": "error", "error": str(exc)} + + +# ── Phase 5: Rules RAG endpoints ─────────────────────────────────────────── +class RulesBuildRequest(BaseModel): + rules_path: str = Field( + ..., + description="Absolute or repo-relative path to MagicCompRules.txt", + ) + chunk_size: int = Field(400, ge=100, le=2000, description="Approx characters per chunk") + + +@router.post("/api/rag/build-rules") +async def rag_build_rules(body: RulesBuildRequest): + """ + Build the mtgrules ChromaDB collection from the MTG Comprehensive Rules text file. + + Download MagicCompRules.txt from https://magic.wizards.com/en/rules and pass the + local path as `rules_path`. This is a one-time or yearly operation. + """ + try: + from services.rag_store import build_rules_index + stats = build_rules_index(body.rules_path, chunk_size=body.chunk_size) + return {"status": "ok", **stats} + except FileNotFoundError as exc: + raise HTTPException(status_code=404, detail=str(exc)) + except Exception as exc: + log.exception("Rules RAG build failed") + raise HTTPException(status_code=500, detail=str(exc)) + + +@router.get("/api/rag/query-rules") +async def rag_query_rules( + q: str, + n: int = 5, +): + """ + Semantic search over the MTG Comprehensive Rules (mtgrules collection). + + Args: + q: Natural-language rules question, e.g. "Does Teferi stop split second?" + n: Number of rule chunks to return (default 5, max 20). + + Returns list of {rule_id, text, distance}. + """ + try: + from services.rag_store import query_rules + return {"results": query_rules(q, n_results=min(n, 20))} + except Exception as exc: + log.exception("Rules RAG query failed") + raise HTTPException(status_code=500, detail=str(exc)) diff --git a/services/card_text.py b/services/card_text.py index c9b918a..5ee37cc 100644 --- a/services/card_text.py +++ b/services/card_text.py @@ -3,35 +3,43 @@ services/card_text.py ===================== Shared utility for building card text representations. -Both services/rag_store.py and coach/embeddings.py need to produce -identical text strings for embedding so that vector similarity is -consistent across both systems. This module provides a single -canonical implementation. + Both services/rag_store.py and coach/embeddings.py need to produce + identical text strings for embedding so that vector similarity is + consistent across both systems. This module provides a single + canonical implementation. -Format: Name | Mana Cost | Type | Oracle Text -(fields that are empty/missing are omitted) + Format: Natural-language sentences. + (fields that are empty/missing are omitted) """ + from __future__ import annotations def build_card_text(card: dict) -> str: - """Build the canonical text representation of a card for embedding. + """Build canonical natural-language card text for embedding. + + Produces '{Name} is a {Type}. It costs {ManaCost}. {OracleText}' + Fields that are empty/None are omitted. Args: card: Dict with keys 'name', 'mana_cost', 'type_line', 'oracle_text'. Missing keys are treated as empty strings. Returns: - Pipe-separated string like 'Sol Ring | {1} | Artifact | {T}: Add {C}{C}.' + Natural language string suitable for nomic-embed-text or GTE models. """ - parts = [card.get("name", "") or ""] - mana_cost = card.get("mana_cost", "") or "" - if mana_cost: - parts.append(mana_cost) - type_line = card.get("type_line", "") or "" - if type_line: - parts.append(type_line) - oracle = card.get("oracle_text", "") or "" + name = (card.get("name", "") or "").strip() + mana = (card.get("mana_cost", "") or "").strip() + types = (card.get("type_line", "") or "").strip() + oracle = (card.get("oracle_text", "") or "").strip() + + parts: list[str] = [] + if name: + base = f"{name} is a {types}." if types else f"{name}." + parts.append(base) + if mana: + parts.append(f"It costs {mana}.") if oracle: parts.append(oracle) - return " | ".join(parts) + + return " ".join(parts) if parts else name diff --git a/services/rag_store.py b/services/rag_store.py index 2cff406..9fe0f32 100644 --- a/services/rag_store.py +++ b/services/rag_store.py @@ -11,13 +11,13 @@ This is RAG Phase 2. Phase 3 will query this store to inject relevant card oracle text into the LLM prompt before calling the coach. Public API: - ensure_rag_store() -> None (build if missing / stale) - build_index(force, batch_size) -> dict (explicit build, returns stats) - query_cards(text, n, colors) -> list[dict] - get_rag_stats() -> dict + ensure_rag_store() -> None (build if missing / stale) + build_index(force, batch_size) -> dict (explicit build, returns stats) + query_cards(text, n, colors) -> list[dict] + get_rag_stats() -> dict -Data path: ./data/rag_chroma/ -Embedding: nomic-embed-text via Ollama (localhost:11434) +Data path: ./data/rag_chroma/ +Embedding: nomic-embed-text via Ollama (localhost:11434) """ from __future__ import annotations @@ -46,16 +46,45 @@ _rag_ready = False _chroma_client = None _chroma_collection = None +# ── Phase 2a: Forge behavioral enrichment (optional — graceful if Forge not configured) ── +def _try_import_forge_lookup(): + try: + from src.commanderailab.sim.forge_card_loader import lookup_forge_card + return lookup_forge_card + except ImportError: + return None + +_forge_lookup = _try_import_forge_lookup() # ── Helpers ──────────────────────────────────────────────────────────────── - def _build_document(card: dict) -> str: - """Delegate to shared build_card_text() -- see services/card_text.py.""" - return build_card_text(card) - + """Build embedding document: Scryfall oracle text + optional Forge behavioral tags.""" + base = build_card_text(card) # canonical NL text from card_text.py + if _forge_lookup: + try: + forge = _forge_lookup(card.get("name", "")) + if forge: + tags: list[str] = [] + if forge.trigger_modes: + tags.append("Triggers: " + ", ".join(forge.trigger_modes[:6])) + if forge.has_replacement_effect: + tags.append("Has replacement effect") + if forge.has_static_ability: + tags.append("Has static ability") + if forge.keywords: + tags.append("Keywords: " + ", ".join(forge.keywords[:6])) + if tags: + return base + " " + " ".join(tags) + except Exception as exc: + logger.debug("Forge enrichment skipped for %s: %s", card.get("name"), exc) + return base def _build_metadata(card: dict) -> dict: - """Build ChromaDB metadata dict for filtering.""" + """Build ChromaDB metadata dict for filtering. + + Core fields always present: name, type_line, cmc, color_identity, rarity, mana_cost + Forge fields when Forge is configured: has_trigger, has_replacement, has_static, trigger_modes + """ ci_raw = card.get("color_identity", "[]") if isinstance(ci_raw, str): try: @@ -65,15 +94,30 @@ def _build_metadata(card: dict) -> dict: else: ci_list = ci_raw or [] - return { + meta = { "name": card.get("name", ""), "type_line": card.get("type_line", ""), "cmc": float(card.get("cmc") or 0), "color_identity": ",".join(sorted(ci_list)), "rarity": card.get("rarity", ""), "mana_cost": card.get("mana_cost", ""), + # Forge behavioral fields — default to 0/empty so filtering is always safe + "has_trigger": 0, + "has_replacement": 0, + "has_static": 0, + "trigger_modes": "", } - + if _forge_lookup: + try: + forge = _forge_lookup(card.get("name", "")) + if forge: + meta["has_trigger"] = int(bool(forge.trigger_modes)) + meta["has_replacement"] = int(forge.has_replacement_effect) + meta["has_static"] = int(forge.has_static_ability) + meta["trigger_modes"] = ",".join(forge.trigger_modes[:8]) + except Exception as exc: + logger.debug("Forge metadata enrichment skipped for %s: %s", card.get("name"), exc) + return meta def _get_chromadb_client(): """Return the module-level persistent ChromaDB client (singleton).""" @@ -85,7 +129,7 @@ def _get_chromadb_client(): except ImportError: raise ImportError( "chromadb is required for RAG Phase 2. " - "Install with: pip install 'commander-ai-lab[rag]'" + "Install with: pip install 'commander-ai-lab[rag]'" ) RAG_CHROMA_DIR.mkdir(parents=True, exist_ok=True) _chroma_client = chromadb.PersistentClient(path=str(RAG_CHROMA_DIR)) @@ -99,7 +143,7 @@ def _get_ollama_embedding_fn(): except ImportError: raise ImportError( "chromadb >= 0.4.0 with Ollama support is required. " - "Install with: pip install 'commander-ai-lab[rag]'" + "Install with: pip install 'commander-ai-lab[rag]'" ) return OllamaEmbeddingFunction( url=f"{OLLAMA_BASE_URL}/api/embeddings", @@ -138,12 +182,16 @@ def _read_build_meta() -> dict: return {} -def _write_build_meta(card_count: int) -> None: - """Persist build timestamp and card count to sidecar JSON file.""" +def _write_build_meta(card_count: int, forge_coverage: int = 0) -> None: + """Persist build timestamp, card count, and Forge coverage to sidecar JSON.""" try: RAG_CHROMA_DIR.mkdir(parents=True, exist_ok=True) BUILD_META_PATH.write_text( - json.dumps({"built_at": time.time(), "card_count": card_count}), + json.dumps({ + "built_at": time.time(), + "card_count": card_count, + "forge_coverage": forge_coverage, + }), encoding="utf-8", ) except Exception as exc: @@ -151,16 +199,10 @@ def _write_build_meta(card_count: int) -> None: def _collection_is_current(collection, bulk_card_count: int) -> bool: - """Check if the ChromaDB collection is reasonably up-to-date. - - Uses sidecar build_meta.json for the age check (survives restarts), - and compares card count within a 5% tolerance. - """ + """Check if the ChromaDB collection is reasonably up-to-date.""" chroma_count = collection.count() if chroma_count == 0: return False - - # Age check via sidecar file (not collection.metadata which doesn't persist) meta = _read_build_meta() built_at = meta.get("built_at", 0) if built_at: @@ -168,7 +210,6 @@ def _collection_is_current(collection, bulk_card_count: int) -> bool: if age_days > 14: logger.info("RAG collection older than 14 days (%0.1f) — rebuilding.", age_days) return False - ratio = chroma_count / max(bulk_card_count, 1) if ratio < 0.95 or ratio > 1.05: logger.info( @@ -178,23 +219,10 @@ def _collection_is_current(collection, bulk_card_count: int) -> bool: return False return True - # ── Build pipeline ──────────────────────────────────────────────────────────────── - def _build_collection_from_bulk(batch_size: int = DEFAULT_BATCH_SIZE) -> dict: - """Read all cards from the bulk SQLite DB and insert into ChromaDB. - - Args: - batch_size: Number of cards to embed and insert per ChromaDB call. - - Returns: - Dict with keys: indexed (int), failed (int). - - Raises: - RuntimeError: If more than 1% of cards failed to insert. - """ + """Read all cards from the bulk SQLite DB and insert into ChromaDB.""" from services.scryfall_bulk import get_bulk_db - conn = get_bulk_db() cursor = conn.execute( "SELECT name, oracle_text, type_line, mana_cost, cmc, " @@ -204,44 +232,40 @@ def _build_collection_from_bulk(batch_size: int = DEFAULT_BATCH_SIZE) -> dict: ) all_rows = cursor.fetchall() total_rows = len(all_rows) - client = _get_chromadb_client() embed_fn = _get_ollama_embedding_fn() - - # Drop and recreate collection to ensure a clean build try: client.delete_collection(COLLECTION_NAME) except Exception: pass _invalidate_collection_singleton() - collection = client.create_collection( name=COLLECTION_NAME, embedding_function=embed_fn, metadata={"hnsw:space": "cosine"}, ) - inserted = 0 failed_cards = 0 failed_batches = 0 + forge_enriched_count = 0 # Phase 2e: track Forge coverage batch_ids: list[str] = [] batch_docs: list[str] = [] batch_metas: list[dict] = [] - for row in all_rows: card = dict(row) doc = _build_document(card) if not doc.strip(): continue - card_id = card.get("scryfall_id", "") if not card_id: continue - + meta = _build_metadata(card) + # Phase 2e: count Forge-enriched cards + if meta.get("has_trigger") or meta.get("has_replacement") or meta.get("has_static"): + forge_enriched_count += 1 batch_ids.append(card_id) batch_docs.append(doc) - batch_metas.append(_build_metadata(card)) - + batch_metas.append(meta) if len(batch_ids) >= batch_size: try: collection.add( @@ -251,7 +275,7 @@ def _build_collection_from_bulk(batch_size: int = DEFAULT_BATCH_SIZE) -> dict: ) inserted += len(batch_ids) if inserted % 2000 == 0: - logger.info(" RAG indexed %d / %d cards...", inserted, total_rows) + logger.info(" RAG indexed %d / %d cards...", inserted, total_rows) except Exception as exc: failed_cards += len(batch_ids) failed_batches += 1 @@ -262,8 +286,6 @@ def _build_collection_from_bulk(batch_size: int = DEFAULT_BATCH_SIZE) -> dict: batch_ids.clear() batch_docs.clear() batch_metas.clear() - - # Final partial batch if batch_ids: try: collection.add( @@ -276,13 +298,10 @@ def _build_collection_from_bulk(batch_size: int = DEFAULT_BATCH_SIZE) -> dict: failed_cards += len(batch_ids) failed_batches += 1 logger.warning("RAG final batch insert failed: %s", exc) - logger.info( - "RAG vector store complete: %d indexed, %d failed (%d batches).", - inserted, failed_cards, failed_batches, + "RAG vector store complete: %d indexed, %d failed (%d batches). Forge-enriched: %d", + inserted, failed_cards, failed_batches, forge_enriched_count, ) - - # Abort if failure rate exceeds 1% — delete partial collection if total_rows > 0 and (failed_cards / total_rows) > 0.01: logger.error( "RAG build failure rate %.1f%% exceeds 1%% threshold — deleting partial collection.", @@ -297,29 +316,13 @@ def _build_collection_from_bulk(batch_size: int = DEFAULT_BATCH_SIZE) -> dict: f"RAG build aborted: {failed_cards}/{total_rows} cards failed to insert. " "Check Ollama connectivity and retry." ) - - _write_build_meta(card_count=inserted) - # Invalidate singleton so next _get_collection() picks up the new collection + _write_build_meta(card_count=inserted, forge_coverage=forge_enriched_count) _invalidate_collection_singleton() return {"indexed": inserted, "failed": failed_cards} - # ── Public API ──────────────────────────────────────────────────────────────── - def build_index(force: bool = False, batch_size: int = DEFAULT_BATCH_SIZE) -> dict: - """Explicitly build or rebuild the ChromaDB vector index. - - This is the function called by POST /api/rag/build. - Unlike ensure_rag_store(), this always runs (no _rag_ready short-circuit) - and returns stats so the API can surface them to the caller. - - Args: - force: If True, rebuild even if the index appears current. - batch_size: Cards per ChromaDB insert batch (default 500). - - Returns: - Dict with keys: indexed (int), failed (int). - """ + """Explicitly build or rebuild the ChromaDB vector index.""" global _rag_ready with _rag_lock: if not force: @@ -333,20 +336,13 @@ def build_index(force: bool = False, batch_size: int = DEFAULT_BATCH_SIZE) -> di return {"indexed": collection.count(), "failed": 0, "skipped": True} except Exception: pass - result = _build_collection_from_bulk(batch_size=batch_size) _rag_ready = True return result def ensure_rag_store() -> None: - """Build the ChromaDB vector store if it is missing or stale. - - Safe to call at startup — no-ops quickly if the store is current. - Requires: - - Phase 1 bulk DB to be populated (services/scryfall_bulk) - - Ollama running with nomic-embed-text pulled - """ + """Build the ChromaDB vector store if it is missing or stale.""" global _rag_ready with _rag_lock: if _rag_ready: @@ -361,16 +357,11 @@ def ensure_rag_store() -> None: "Run ensure_bulk_db() first." ) return - collection = _get_collection() if _collection_is_current(collection, bulk_count): - logger.info( - "RAG vector store is current (%d cards).", - collection.count(), - ) + logger.info("RAG vector store is current (%d cards).", collection.count()) _rag_ready = True return - logger.info( "Building RAG vector store from %d bulk cards " "(this may take several minutes on first run)...", @@ -378,51 +369,67 @@ def ensure_rag_store() -> None: ) _build_collection_from_bulk() _rag_ready = True - except ImportError as exc: logger.warning("RAG store unavailable: %s", exc) except Exception as exc: logger.error("RAG store build failed: %s", exc) - def query_cards( query: str, n_results: int = 8, color_identity: Optional[list[str]] = None, type_filter: Optional[str] = None, + require_trigger: Optional[str] = None, + require_replacement: bool = False, + require_static: bool = False, ) -> list[dict]: """Semantic search over card oracle text. Args: - query: Natural language query (e.g. "draw cards when creatures die"). - n_results: Maximum number of results. - color_identity: If provided, only return cards whose color identity - is a subset (e.g. ["W", "U", "B"]). - type_filter: Optional type_line substring filter (e.g. "Creature"). - Applied as a post-query Python filter for ChromaDB compatibility. + query: Natural language query. + n_results: Maximum results returned. + color_identity: Only return cards within these colors. + type_filter: Optional type_line substring filter. + require_trigger: If set, only return cards whose trigger_modes contains this string. + require_replacement: If True, only return cards with replacement effects. + require_static: If True, only return cards with static abilities. Returns: - List of dicts with keys: name, type_line, oracle_text, cmc, - color_identity, mana_cost, distance. + List of dicts: name, type_line, oracle_text, cmc, color_identity, mana_cost, + distance, trigger_modes (new), has_replacement (new), has_static (new). """ if not _rag_ready: logger.warning("RAG store not ready — returning empty results.") return [] - try: collection = _get_collection() except Exception as exc: logger.error("Failed to get RAG collection: %s", exc) return [] + # Build native ChromaDB where clause for Forge boolean flags + where_clause: Optional[dict] = None + forge_filters = [] + if require_replacement: + forge_filters.append({"has_replacement": {"$eq": 1}}) + if require_static: + forge_filters.append({"has_static": {"$eq": 1}}) + if len(forge_filters) == 1: + where_clause = forge_filters[0] + elif len(forge_filters) > 1: + where_clause = {"$and": forge_filters} + try: - # Fetch extra results to account for post-query color/type filtering - fetch_n = n_results * 4 if (color_identity or type_filter) else n_results - results = collection.query( + has_filters = bool(color_identity or type_filter or require_trigger) + fetch_n = n_results * 4 if has_filters else n_results + query_kwargs = dict( query_texts=[query], n_results=min(fetch_n, 200), include=["documents", "metadatas", "distances"], ) + if where_clause: + query_kwargs["where"] = where_clause + results = collection.query(**query_kwargs) except Exception as exc: logger.error("RAG query failed: %s", exc) return [] @@ -440,23 +447,27 @@ def query_cards( meta = metas[i] if i < len(metas) else {} doc = docs[i] if i < len(docs) else "" - # Color identity post-filter (Python-side for ChromaDB compat) + # Color identity post-filter (Python-side) if color_identity: card_ci_str = meta.get("color_identity", "") card_ci = set(c for c in card_ci_str.split(",") if c) if card_ci_str else set() - deck_ci = set(color_identity) - if not card_ci.issubset(deck_ci): + if not card_ci.issubset(set(color_identity)): continue # Type post-filter if type_filter: - card_type = meta.get("type_line", "") - if type_filter.lower() not in card_type.lower(): + if type_filter.lower() not in meta.get("type_line", "").lower(): continue - # Document format: Name | Mana Cost | Type | Oracle Text + # Trigger mode substring post-filter (can't do substring in Chroma where clause) + if require_trigger: + tms = meta.get("trigger_modes", "") + if require_trigger.lower() not in tms.lower(): + continue + + # Parse oracle from document — handle both old pipe format and new NL format parts = doc.split(" | ", 3) - oracle = parts[3] if len(parts) > 3 else (parts[-1] if parts else "") + oracle = parts[-1] if len(parts) > 1 else doc cards.append({ "name": meta.get("name", ""), @@ -466,13 +477,16 @@ def query_cards( "color_identity": meta.get("color_identity", ""), "mana_cost": meta.get("mana_cost", ""), "distance": round(dists[i], 4) if i < len(dists) else 0.0, + # New fields — consumers check for presence, not required + "trigger_modes": meta.get("trigger_modes", ""), + "has_replacement": bool(meta.get("has_replacement", 0)), + "has_static": bool(meta.get("has_static", 0)), }) if len(cards) >= n_results: break return cards - def get_rag_stats() -> dict: """Return metadata about the RAG vector store state.""" result = { @@ -483,14 +497,13 @@ def get_rag_stats() -> dict: "card_count": 0, "built_at": None, "age_days": None, + "forge_enriched": _forge_lookup is not None, # Phase 2e } try: collection = _get_collection() result["card_count"] = collection.count() except Exception: pass - - # Read age from sidecar file (persists across restarts) meta = _read_build_meta() built_at = meta.get("built_at") if built_at: @@ -500,7 +513,8 @@ def get_rag_stats() -> dict: result["age_days"] = round(age, 1) except (ValueError, TypeError): pass - + # Phase 2e: report Forge coverage count from sidecar + result["forge_coverage"] = meta.get("forge_coverage", 0) return result @@ -524,13 +538,7 @@ def is_chroma_available() -> bool: def check_staleness() -> dict: - """Return staleness info so a background scheduler can trigger rebuilds. - - Returns dict with keys: - stale (bool): True if the index should be rebuilt. - reason (str): Human-readable explanation. - age_days (float | None): Age of the current build in days. - """ + """Return staleness info so a background scheduler can trigger rebuilds.""" meta = _read_build_meta() built_at = meta.get("built_at") if not built_at: @@ -556,3 +564,148 @@ def check_staleness() -> dict: except Exception as exc: logger.warning("check_staleness card count check failed: %s", exc) return {"stale": False, "reason": "index is current", "age_days": round(age_days, 1)} + + +# ── Phase 3: Rules RAG ── second ChromaDB collection (mtgrules) ────────────────── +RULES_COLLECTION_NAME = "mtgrules" +RULES_META_PATH = RAG_CHROMA_DIR / "rules_meta.json" +_rules_collection = None + + +def _get_rules_collection(): + """Return the mtgrules ChromaDB collection singleton.""" + global _rules_collection + if _rules_collection is not None: + return _rules_collection + client = _get_chromadb_client() + embed_fn = _get_ollama_embedding_fn() + _rules_collection = client.get_or_create_collection( + name=RULES_COLLECTION_NAME, + embedding_function=embed_fn, + metadata={"hnsw:space": "cosine"}, + ) + return _rules_collection + + +def build_rules_index(rules_text_path: str, chunk_size: int = 400) -> dict: + """Build a rules RAG collection from a plain-text Comprehensive Rules document. + + Chunks the document by rule number boundaries (e.g. 100., 903.), + embeds each chunk, and inserts into the mtgrules ChromaDB collection. + + Args: + rules_text_path: Path to MagicCompRules.txt (download from WotC). + chunk_size: Approximate characters per chunk; splits on rule breaks. + + Returns: + Dict with keys: indexed (int), failed (int). + """ + import re + global _rules_collection + + path = Path(rules_text_path) + if not path.exists(): + raise FileNotFoundError(f"Rules file not found: {rules_text_path}") + + text = path.read_text(encoding="utf-8", errors="replace") + + # Split on lines that start with a rule number like 100., 903.4a, etc. + rule_pattern = re.compile(r"(?m)^\d{3,4}\.") + raw_chunks = rule_pattern.split(text) + + # Merge very short fragments into the previous chunk + chunks: list[str] = [] + buffer = "" + for chunk in raw_chunks: + chunk = chunk.strip() + if not chunk: + continue + buffer = buffer + " " + chunk if buffer else chunk + if len(buffer) >= chunk_size: + chunks.append(buffer) + buffer = "" + if buffer: + chunks.append(buffer) + + client = _get_chromadb_client() + embed_fn = _get_ollama_embedding_fn() + + # Drop and recreate for a clean build + try: + client.delete_collection(RULES_COLLECTION_NAME) + except Exception: + pass + _rules_collection = None + collection = client.create_collection( + name=RULES_COLLECTION_NAME, + embedding_function=embed_fn, + metadata={"hnsw:space": "cosine"}, + ) + + inserted = 0 + failed = 0 + batch_ids: list[str] = [] + batch_docs: list[str] = [] + + for idx, chunk in enumerate(chunks): + rule_id_match = re.match(r"[\d\.a-z]+", chunk) + rule_id = rule_id_match.group(0) if rule_id_match else str(idx) + batch_ids.append(f"rule-{rule_id}-{idx}") + batch_docs.append(chunk[:2000]) # hard cap per doc + + if len(batch_ids) >= 100: + try: + collection.add(ids=batch_ids, documents=batch_docs) + inserted += len(batch_ids) + except Exception as exc: + failed += len(batch_ids) + logger.warning("Rules batch insert failed: %s", exc) + batch_ids.clear() + batch_docs.clear() + + if batch_ids: + try: + collection.add(ids=batch_ids, documents=batch_docs) + inserted += len(batch_ids) + except Exception as exc: + failed += len(batch_ids) + logger.warning("Rules final batch failed: %s", exc) + + RULES_META_PATH.write_text( + json.dumps({"built_at": time.time(), "chunk_count": inserted}), + encoding="utf-8", + ) + logger.info("Rules RAG: %d chunks indexed, %d failed", inserted, failed) + return {"indexed": inserted, "failed": failed} + + +def query_rules(query: str, n_results: int = 5) -> list[dict]: + """Semantic search over the MTG Comprehensive Rules. + + Returns: + List of dicts with keys: rule_id, text, distance. + """ + try: + collection = _get_rules_collection() + if collection.count() == 0: + return [] + results = collection.query( + query_texts=[query], + n_results=min(n_results, 20), + include=["documents", "distances"], + ) + except Exception as exc: + logger.error("Rules RAG query failed: %s", exc) + return [] + + if not results or not results.get("ids") or not results["ids"][0]: + return [] + + out = [] + for i, doc_id in enumerate(results["ids"][0]): + out.append({ + "rule_id": doc_id, + "text": results["documents"][0][i] if results.get("documents") else "", + "distance": round(results["distances"][0][i], 4) if results.get("distances") else 0.0, + }) + return out