Agent Zero
52 lines
2.6 KiB
Python
52 lines
2.6 KiB
Python
"""Knowledge plugin — LLM-based entity/relationship extraction, ask-knowledge, review queue."""
|
|
from __future__ import annotations
|
|
import logging
|
|
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")
|
|
logger.info("Registered knowledge extraction hooks")
|
|
except Exception:
|
|
logger.exception("Failed to register knowledge hooks")
|
|
|
|
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")
|
|
await super().on_deactivate(db, service_container, event_bus)
|