Agent Zero
78 lines
2.6 KiB
Python
78 lines
2.6 KiB
Python
"""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
|