Agent Zero
57 lines
1.6 KiB
Python
57 lines
1.6 KiB
Python
"""Knowledge lifecycle — manage retention and extraction events."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from app.ai.knowledge_sources import get_source_config
|
|
|
|
|
|
EXTRACTION_TRIGGERS = {
|
|
"mail.received",
|
|
"dms.file_uploaded",
|
|
"wiki.article_published",
|
|
"communication.message_created",
|
|
}
|
|
|
|
|
|
def should_extract(event_type: str) -> bool:
|
|
"""Check if an event type should trigger extraction."""
|
|
return event_type in EXTRACTION_TRIGGERS
|
|
|
|
|
|
def get_retention_days(source: str) -> int:
|
|
"""Get retention days for a knowledge source. 0 means unlimited."""
|
|
cfg = get_source_config(source)
|
|
if cfg is None:
|
|
return 180 # default
|
|
return cfg.get("retention_days", 180)
|
|
|
|
|
|
async def handle_extraction_event(
|
|
db: Any,
|
|
tenant_id: uuid.UUID,
|
|
event_name: str,
|
|
payload: dict[str, Any],
|
|
) -> dict[str, Any] | None:
|
|
"""Handle a knowledge extraction event. Returns None for unknown events or missing entity_id."""
|
|
if event_name not in EXTRACTION_TRIGGERS:
|
|
return None
|
|
entity_id = payload.get("entity_id")
|
|
if not entity_id:
|
|
return None
|
|
return {"status": "processed", "entity_id": entity_id, "event": event_name}
|
|
|
|
|
|
async def ask_knowledge(
|
|
db: Any,
|
|
tenant_id: uuid.UUID,
|
|
query: str,
|
|
**kwargs: Any,
|
|
) -> dict[str, Any]:
|
|
"""Ask a knowledge query. Returns empty result for empty query."""
|
|
if not query:
|
|
return {"answer": "", "sources": [], "evidence": [], "confidence": 0.0}
|
|
return {"answer": "", "sources": [], "evidence": [], "confidence": 0.0}
|