"""Knowledge plugin — LLM-based entity/relationship extraction, ask-knowledge, review queue.""" from __future__ import annotations import logging import uuid from typing import Any from app.plugins.base import BasePlugin from app.plugins.manifest import PluginManifest, PluginRouteDef logger = logging.getLogger(__name__) class KnowledgePlugin(BasePlugin): manifest = PluginManifest( name="knowledge", version="1.0.0", display_name="Knowledge", description="LLM-based knowledge extraction, ask-knowledge, review queue. Builds on graph_rag + unified_search.", dependencies=["permissions", "graph_rag", "unified_search"], routes=[ PluginRouteDef(path="/api/v1/knowledge", module="app.plugins.builtins.knowledge.routes", router_attr="router"), ], permissions=["knowledge:read", "knowledge:write", "knowledge:admin"], ) async def on_activate(self, db, service_container, event_bus) -> None: """Register event-driven extraction hooks on activation.""" await super().on_activate(db, service_container, event_bus) try: from app.core.hooks import register_action from app.plugins.builtins.knowledge.services import extract_knowledge async def on_wiki_create(*args, **kwargs): article_id = kwargs.get("article_id") or kwargs.get("entity_id") tenant_id = kwargs.get("tenant_id") title = kwargs.get("title", "") content = kwargs.get("content", "") if article_id and tenant_id and content: from app.core.db import get_worker_session_factory factory = get_worker_session_factory() async with factory() as session: await extract_knowledge( db=session, tenant_id=uuid.UUID(str(tenant_id)), source_type="wiki_article", source_id=uuid.UUID(str(article_id)), source_title=title, source_text=content, ) register_action("wiki.article.created", on_wiki_create, priority=20, owner_tag="knowledge") # H-DATA-LIFE: Re-extract when wiki article is updated async def on_wiki_update(*args, **kwargs): article_id = kwargs.get("article_id") or kwargs.get("entity_id") tenant_id = kwargs.get("tenant_id") title = kwargs.get("title", "") content = kwargs.get("content", "") if article_id and tenant_id and content: from app.core.db import get_worker_session_factory factory = get_worker_session_factory() async with factory() as session: await extract_knowledge( db=session, tenant_id=uuid.UUID(str(tenant_id)), source_type="wiki_article", source_id=uuid.UUID(str(article_id)), source_title=title, source_text=content, ) register_action("wiki.article.updated", on_wiki_update, priority=20, owner_tag="knowledge") logger.info("Registered knowledge extraction hooks") except Exception: logger.exception("Failed to register knowledge hooks") # Register knowledge agent tools (I-AK: Agent→Knowledge) try: self._register_knowledge_agent_tools() except Exception: logger.exception("Failed to register knowledge agent tools") def _register_knowledge_agent_tools(self) -> None: """Register I-AK agent tools for asking and searching knowledge.""" from app.ai.tool_registry import get_tool_registry registry = get_tool_registry() async def _ask_knowledge_handler(arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]: """Ask a knowledge question.""" from app.plugins.builtins.knowledge.services import ask_knowledge from app.core.db import get_worker_session_factory question = arguments.get("question", "") tenant_id = context.get("tenant_id") if not question or not tenant_id: return {"error": "question and tenant_id required"} factory = get_worker_session_factory() async with factory() as db: result = await ask_knowledge(db=db, tenant_id=uuid.UUID(str(tenant_id)), question=question) return {"answer": result.get("answer", ""), "evidence_count": len(result.get("evidence", []))} registry.register( name="ask_knowledge", description="Ask a knowledge question. Searches wiki articles and graph relationships for context.", parameters={ "type": "object", "properties": { "question": {"type": "string", "description": "The question to ask"}, }, "required": ["question"], }, handler=_ask_knowledge_handler, plugin_name=self.manifest.name, required_permission="wiki:read", category="knowledge", ) async def _search_knowledge_handler(arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]: """Search wiki articles via unified search.""" from app.plugins.builtins.unified_search.provider_registry import get_search_registry from app.core.db import get_worker_session_factory query = arguments.get("query", "") tenant_id = context.get("tenant_id") if not query or not tenant_id: return {"error": "query and tenant_id required"} factory = get_worker_session_factory() async with factory() as db: search_registry = get_search_registry() wiki_provider = search_registry.get("wiki_article") if not wiki_provider: return {"error": "Wiki search provider not available"} results = await wiki_provider._search_fts_filtered( db=db, tsquery=query, tenant_id=uuid.UUID(str(tenant_id)), limit=5, visible_ids=None ) return { "results": [ {"title": r.get("title", ""), "summary": (r.get("summary") or r.get("content", "")[:200] or "")} for r in results ], "count": len(results), } registry.register( name="search_knowledge", description="Search wiki articles by keyword. Returns matching articles with title and summary.", parameters={ "type": "object", "properties": { "query": {"type": "string", "description": "Search query"}, }, "required": ["query"], }, handler=_search_knowledge_handler, plugin_name=self.manifest.name, required_permission="wiki:read", category="knowledge", ) logger.info("Registered knowledge agent tools: ask_knowledge, search_knowledge") def get_job_modules(self) -> list[str]: """ARQ job modules — the retention cleanup job lives with this plugin.""" return ["app.plugins.builtins.knowledge.jobs"] async def on_deactivate(self, db, service_container, event_bus) -> None: """Clean up on deactivation.""" from app.core.hooks import unregister_actions_by_owner unregister_actions_by_owner("knowledge") # Unregister knowledge agent tools from the core AI tool registry try: from app.ai.tool_registry import get_tool_registry get_tool_registry().unregister_plugin(self.manifest.name) logger.info("Unregistered AI agent tools for plugin '%s'", self.manifest.name) except Exception: logger.exception("Failed to unregister AI agent tools") await super().on_deactivate(db, service_container, event_bus)