"""Knowledge source registry — manages available evidence sources for AI.""" from __future__ import annotations from dataclasses import dataclass from typing import Any @dataclass class EvidenceReference: """A reference to a piece of evidence from a knowledge source.""" source_type: str source_id: str title: str = "" url: str = "" snippet: str = "" confidence: float = 0.0 def to_dict(self) -> dict[str, Any]: return { "source_type": self.source_type, "source_id": self.source_id, "title": self.title, "url": self.url, "snippet": self.snippet, "confidence": self.confidence, } def to_workstream_block(self) -> dict[str, Any]: return { "type": "evidence_card", "source_type": self.source_type, "source_id": self.source_id, "title": self.title, "url": self.url, "snippet": self.snippet, "confidence": self.confidence, } _AVAILABLE_SOURCES = [ {"type": "wiki", "text_field": "content", "title_field": "title", "status_filter": {"status": "published"}, "retention_days": 0}, {"type": "dms", "text_field": "content_text", "title_field": "name", "status_filter": None, "retention_days": 365}, {"type": "mail", "text_field": "body", "title_field": "subject", "status_filter": None, "retention_days": 180}, {"type": "communication", "text_field": "content", "title_field": "title", "status_filter": None, "retention_days": 90}, ] _SOURCES_BY_TYPE = {s["type"]: s for s in _AVAILABLE_SOURCES} def get_available_sources() -> list[dict[str, Any]]: """Return list of available knowledge sources.""" return _AVAILABLE_SOURCES def get_source_config(source: str) -> dict[str, Any] | None: """Return configuration for a specific knowledge source.""" return _SOURCES_BY_TYPE.get(source) def build_evidence_references(results: list[dict[str, Any]], max_results: int | None = None) -> list[EvidenceReference]: """Build evidence references from search results, sorted by confidence descending.""" refs = [ EvidenceReference( source_type=r.get("source_type", ""), source_id=r.get("source_id", ""), title=r.get("title", ""), url=r.get("url", ""), snippet=r.get("snippet", ""), confidence=r.get("score", 0.0), ) for r in results ] refs.sort(key=lambda x: x.confidence, reverse=True) if max_results is not None: refs = refs[:max_results] return refs