| """KnowledgeService — the interface agents read domain knowledge through (K1-K4). |
| |
| **Three calls, and they are the product.** The end goal is an MCP service |
| exposing modular domain knowledge to agents we do not own |
| (`knowledge_pipeline_context.md` §1), so these three are what an external client |
| would be given. Our planner becomes the FIRST client rather than the only one, |
| which makes the eventual MCP server a transport rather than a redesign. |
| |
| get_context(user_id, catalog=None) Tier 1 - the card, preloaded |
| lookup(user_id, terms=[...]) Tier 2 - detail, on demand |
| terms_used(user_id, surfaces=[...]) what a completed run actually touched |
| |
| **Approved entries only.** Candidates are the model's claims; letting them steer |
| an analysis is precisely what the review step exists to prevent. The dev flag on |
| `DomainComposer` exists for development and is not reachable from here. |
| |
| **Nothing in this module calls a model.** It reads rows and renders text. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import re |
|
|
| from src.knowledge_domain.compose import DomainComposer |
| from src.knowledge_domain.models import DomainContext, Measure |
| from src.knowledge_domain.render import render_card, render_lookup |
| from src.knowledge_ingest.knowledge_store import KnowledgeStore |
| from src.knowledge_ingest.selector import resolve_scope_id |
| from src.middlewares.logging import get_logger |
|
|
| logger = get_logger("knowledge_service") |
|
|
| |
| |
| |
| |
| |
| CACHE_TTL_SECONDS = 3600 |
|
|
|
|
| def _cache_key(scope_id: str) -> str: |
| from src.config.settings import settings |
|
|
| |
| |
| return f"{settings.redis_prefix}knowledge:card:{scope_id}" |
|
|
|
|
| class KnowledgeService: |
| """Reads domain knowledge for one caller. Never writes.""" |
|
|
| def __init__( |
| self, |
| store: KnowledgeStore | None = None, |
| composer: DomainComposer | None = None, |
| ) -> None: |
| self._store = store or KnowledgeStore() |
| self._composer = composer or DomainComposer(self._store) |
|
|
| |
| async def scope_for(self, user_id: str) -> str: |
| """The caller's company. |
| |
| Resolved HERE and nowhere else on the read path. Entries are |
| company-scoped while the chat pipeline is analysis-scoped, so something |
| has to bridge the two; doing it once per turn at the composition root |
| and threading it is the same shape the catalog already uses for the |
| traceability scratchpad (Q10) - not a fresh lookup inside each prompt |
| builder. |
| """ |
| return await resolve_scope_id(user_id) |
|
|
| |
| async def get_context( |
| self, user_id: str, *, catalog=None, use_cache: bool = True |
| ) -> DomainContext: |
| """The composed domain context for this caller's company. |
| |
| Cached per scope. `catalog` is deliberately NOT part of the cache key: |
| bindings are resolved fresh against whatever catalog the caller has, so |
| a cached context never carries another analysis's column ids. |
| """ |
| scope_id = await self.scope_for(user_id) |
| cached = await self._cached(scope_id) if use_cache else None |
| if cached is not None: |
| context = cached |
| else: |
| context = await self._composer.compose(scope_id) |
| if use_cache: |
| await self._cache(scope_id, context) |
|
|
| if catalog is not None: |
| |
| self._composer._bind_to_catalog(context.measures, catalog) |
| return context |
|
|
| async def render_context(self, user_id: str, *, catalog=None) -> str: |
| """The Tier-1 card as prompt text. Empty string when there is nothing. |
| |
| A caller can append this unconditionally: an empty domain renders to |
| nothing rather than to a header implying knowledge that does not exist. |
| """ |
| return render_card(await self.get_context(user_id, catalog=catalog)) |
|
|
| |
| async def lookup(self, user_id: str, terms: list[str]) -> list[Measure]: |
| """Everything known about specific terms, matched on any surface form. |
| |
| The tier that scales: a corpus of thousands of terms costs nothing until |
| a question actually touches one. |
| """ |
| if not terms: |
| return [] |
| context = await self.get_context(user_id) |
| wanted = {t.strip().casefold() for t in terms if t and t.strip()} |
| hits = [ |
| m |
| for m in context.measures |
| if m.name.casefold() in wanted |
| or any(f.casefold() in wanted for f in m.surface_forms) |
| or m.key.casefold() in wanted |
| ] |
| logger.info( |
| "knowledge_lookup", |
| extra={"asked": len(wanted), "found": len(hits)}, |
| ) |
| return hits |
|
|
| async def render_lookup(self, user_id: str, terms: list[str]) -> str: |
| """Tier 2 as prompt text, with the governing rules those terms carry.""" |
| measures = await self.lookup(user_id, terms) |
| if not measures: |
| return render_lookup([], []) |
|
|
| scope_id = await self.scope_for(user_id) |
| wanted_rules = {r for m in measures for r in m.governed_by} |
| rules: list[dict] = [] |
| if wanted_rules: |
| rows = await self._store.list_entries( |
| scope_id=scope_id, kind="rule", limit=1000 |
| ) |
| rules = [r.payload for r in rows if r.entity_id in wanted_rules] |
| return render_lookup(measures, rules) |
|
|
| |
| async def terms_used(self, user_id: str, surfaces: list[str]) -> list[Measure]: |
| """Measures matching text a finished run actually produced. |
| |
| For the assembler and the report: it needs the company's wording for the |
| terms in THIS answer, not the whole vocabulary. Whole-word matching |
| because the caller passes prose, not a term list. |
| """ |
| if not surfaces: |
| return [] |
| context = await self.get_context(user_id) |
| haystack = " \n ".join(s for s in surfaces if s).casefold() |
|
|
| def _mentions(form: str) -> bool: |
| """Whole-word match, so short abbreviations work without false hits. |
| |
| A plain substring test had to exclude forms of two characters or |
| fewer to avoid matching inside other words — which silently excluded |
| PA, UA and Qty, i.e. most of a mining vocabulary. The first live run |
| showed exactly that: the planner card fired and the assembler's |
| terminology block never did. A word boundary is the right |
| instrument — it catches "PA" in prose and not the "pa" in "company". |
| """ |
| form = form.strip().casefold() |
| if not form: |
| return False |
| return re.search(rf"(?<!\w){re.escape(form)}(?!\w)", haystack) is not None |
|
|
| return [ |
| m for m in context.measures if any(_mentions(f) for f in m.surface_forms) |
| ] |
|
|
| |
| async def propose_declaration(self, user_id: str) -> dict: |
| """Draft this company's domain declaration and store it for review. |
| |
| Derived from what the corpus demonstrably contains - **no model is |
| called**. The expert then approves it, or corrects it through |
| `POST /entries/{entity_id}/reviews` with `decision="edited"`, which is |
| the same route every other entry uses. |
| |
| Nothing here reaches a planner until it is approved: `compose` reads a |
| declaration only when it carries an approving review. |
| """ |
| from src.knowledge_domain.declare import propose_declaration |
|
|
| scope_id = await self.scope_for(user_id) |
| |
| |
| |
| context = await self._composer.compose(scope_id, include_unapproved=True) |
| payload = propose_declaration( |
| scope_id, |
| covered_subdomains=context.identity.subdomains, |
| document_titles=[s.title for s in context.authority.sources if s.title], |
| units=[u.symbol for u in context.conventions.units], |
| ) |
| await self._store.save_declaration( |
| scope_id=scope_id, user_id=user_id, payload=payload |
| ) |
| await self.invalidate(scope_id) |
| return payload |
|
|
| async def get_declaration(self, user_id: str) -> dict | None: |
| """The standing declaration, whether or not an expert has ruled on it.""" |
| row = await self._store.get_declaration(await self.scope_for(user_id)) |
| return row.payload if row else None |
|
|
| |
| async def invalidate(self, scope_id: str) -> None: |
| """Drop a scope's cached card. Called after any review write. |
| |
| The review IS the invalidation event: an approval changes what the |
| planner should see, and waiting an hour to honour an expert's decision |
| would make the loop feel broken. |
| """ |
| try: |
| from src.db.redis.connection import get_redis |
|
|
| redis = await get_redis() |
| await redis.delete(_cache_key(scope_id)) |
| logger.info("knowledge_card_invalidated", extra={"scope_id": scope_id}) |
| except Exception as e: |
| |
| |
| |
| logger.warning( |
| "knowledge_card_invalidate_failed", |
| extra={"scope_id": scope_id, "error": repr(e), |
| "degraded_seam": "knowledge_card_invalidate"}, |
| ) |
|
|
| async def _cached(self, scope_id: str) -> DomainContext | None: |
| try: |
| from src.db.redis.connection import get_redis |
|
|
| redis = await get_redis() |
| raw = await redis.get(_cache_key(scope_id)) |
| return DomainContext.model_validate_json(raw) if raw else None |
| except Exception as e: |
| |
| logger.warning( |
| "knowledge_card_cache_read_failed", |
| extra={"scope_id": scope_id, "error": repr(e), |
| "degraded_seam": "knowledge_card_cache"}, |
| ) |
| return None |
|
|
| async def _cache(self, scope_id: str, context: DomainContext) -> None: |
| try: |
| from src.db.redis.connection import get_redis |
|
|
| redis = await get_redis() |
| |
| |
| |
| plain = context.model_copy(deep=True) |
| for measure in plain.measures: |
| measure.catalog_binding = None |
| await redis.setex( |
| _cache_key(scope_id), |
| CACHE_TTL_SECONDS, |
| json.dumps(plain.model_dump(mode="json")), |
| ) |
| except Exception as e: |
| logger.warning( |
| "knowledge_card_cache_write_failed", |
| extra={"scope_id": scope_id, "error": repr(e), |
| "degraded_seam": "knowledge_card_cache"}, |
| ) |
|
|
