feat(kb): support pluggable knowledge base retrieval backends

astrbot/api/__init__.py

@@ -8,6 +8,17 @@
 from astrbot.core.star.register import register_agent as agent
 from astrbot.core.star.register import register_llm_tool as llm_tool
 
+from .knowledge_base import (
+    BaseKnowledgeBaseBackend,
+    KnowledgeBaseBackendError,
+    KnowledgeBaseError,
+    KnowledgeBaseHit,
+    KnowledgeBaseInfo,
+    KnowledgeBaseQuery,
+    KnowledgeBaseRef,
+    KnowledgeBaseResponse,
+)
+
 _fallback_logger = logging.getLogger("astrbot")
 _logger_cache: dict[
     str,
@@ -73,7 +84,15 @@ def __getattr__(self, item: str):
 __all__ = [
     "AstrBotConfig",
     "BaseFunctionToolExecutor",
+    "BaseKnowledgeBaseBackend",
     "FunctionTool",
+    "KnowledgeBaseBackendError",
+    "KnowledgeBaseError",
+    "KnowledgeBaseHit",
+    "KnowledgeBaseInfo",
+    "KnowledgeBaseQuery",
+    "KnowledgeBaseRef",
+    "KnowledgeBaseResponse",
     "ToolSet",
     "agent",
     "html_renderer",

astrbot/api/all.py

@@ -53,3 +53,4 @@
 from astrbot.core.platform.register import register_platform_adapter
 
 from .message_components import *
+from .knowledge_base import *

astrbot/api/knowledge_base.py

@@ -0,0 +1,163 @@
+"""Public contracts for pluggable knowledge base backends."""
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass(frozen=True, slots=True)
+class KnowledgeBaseRef:
+    """Identify a knowledge base exposed by one backend.
+
+    Args:
+        backend_id: Globally unique backend identifier.
+        knowledge_base_id: Backend-local knowledge base identifier.
+    """
+
+    backend_id: str
+    knowledge_base_id: str
+
+
+@dataclass(frozen=True, slots=True)
+class KnowledgeBaseInfo:
+    """Describe one knowledge base exposed by a backend.
+
+    Args:
+        ref: Backend and knowledge base reference.
+        name: Human-readable knowledge base name.
+        description: Optional knowledge base description.
+        metadata: Backend-specific public metadata.
+    """
+
+    ref: KnowledgeBaseRef
+    name: str
+    description: str | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True, slots=True)
+class KnowledgeBaseQuery:
+    """Represent a backend-independent knowledge base query.
+
+    Args:
+        query: User query text.
+        top_k: Maximum number of results requested from the backend.
+        score_threshold: Optional backend-local relevance threshold.
+        filters: Optional backend-specific metadata filters.
+        umo: Optional unified message origin for session-aware retrieval.
+    """
+
+    query: str
+    top_k: int = 5
+    score_threshold: float | None = None
+    filters: dict[str, Any] = field(default_factory=dict)
+    umo: str | None = None
+
+
+@dataclass(slots=True)
+class KnowledgeBaseHit:
+    """Represent one standardized knowledge base result.
+
+    Args:
+        ref: Backend and knowledge base that produced the result.
+        content: Retrieved text content.
+        source: Human-readable result source.
+        rank: Result rank assigned by the backend, starting from one.
+        score: Optional backend-local relevance score.
+        document_id: Optional backend document identifier.
+        chunk_id: Optional backend chunk identifier.
+        source_uri: Optional URI for the original content.
+        metadata: Backend-specific result metadata.
+    """
+
+    ref: KnowledgeBaseRef
+    content: str
+    source: str
+    rank: int
+    score: float | None = None
+    document_id: str | None = None
+    chunk_id: str | None = None
+    source_uri: str | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(slots=True)
+class KnowledgeBaseResponse:
+    """Represent one backend retrieval response.
+
+    Args:
+        hits: Results ordered from most to least relevant.
+        warnings: Non-fatal backend warnings.
+    """
+
+    hits: list[KnowledgeBaseHit]
+    warnings: list[str] = field(default_factory=list)
+
+
+class KnowledgeBaseError(Exception):
+    """Base exception for standardized knowledge base operations."""
+
+
+class KnowledgeBaseBackendError(KnowledgeBaseError):
+    """Raised when a knowledge base backend request fails."""
+
+
+class BaseKnowledgeBaseBackend(ABC):
+    """Define the public contract implemented by knowledge base backends."""
+
+    @property
+    @abstractmethod
+    def backend_id(self) -> str:
+        """Return the globally unique backend identifier."""
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def display_name(self) -> str:
+        """Return the human-readable backend name."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def list_knowledge_bases(
+        self,
+        *,
+        umo: str | None = None,
+    ) -> list[KnowledgeBaseInfo]:
+        """List knowledge bases enabled and accessible for a caller.
+
+        Args:
+            umo: Optional unified message origin for access filtering.
+
+        Returns:
+            Knowledge bases exposed to AstrBot retrieval for the caller.
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    async def retrieve(
+        self,
+        knowledge_base_ids: list[str],
+        request: KnowledgeBaseQuery,
+    ) -> KnowledgeBaseResponse:
+        """Retrieve from selected backend knowledge bases.
+
+        Args:
+            knowledge_base_ids: Backend-local knowledge base identifiers.
+            request: Standardized retrieval request.
+
+        Returns:
+            Standardized retrieval response.
+        """
+        raise NotImplementedError
+
+
+__all__ = [
+    "BaseKnowledgeBaseBackend",
+    "KnowledgeBaseBackendError",
+    "KnowledgeBaseError",
+    "KnowledgeBaseHit",
+    "KnowledgeBaseInfo",
+    "KnowledgeBaseQuery",
+    "KnowledgeBaseRef",
+    "KnowledgeBaseResponse",
+]

astrbot/core/knowledge_base/builtin_backend.py

@@ -0,0 +1,131 @@
+"""Adapter exposing the built-in knowledge base through the public contract."""
+
+from typing import TYPE_CHECKING
+
+from astrbot.api.knowledge_base import (
+    BaseKnowledgeBaseBackend,
+    KnowledgeBaseHit,
+    KnowledgeBaseInfo,
+    KnowledgeBaseQuery,
+    KnowledgeBaseRef,
+    KnowledgeBaseResponse,
+)
+
+if TYPE_CHECKING:
+    from .kb_mgr import KnowledgeBaseManager
+
+
+class BuiltinKnowledgeBaseBackend(BaseKnowledgeBaseBackend):
+    """Adapt the existing AstrBot knowledge base implementation."""
+
+    def __init__(self, manager: "KnowledgeBaseManager") -> None:
+        """Initialize the built-in backend adapter.
+
+        Args:
+            manager: Existing knowledge base manager.
+        """
+        self.manager = manager
+
+    @property
+    def backend_id(self) -> str:
+        """Return the reserved built-in backend identifier."""
+        return "builtin"
+
+    @property
+    def display_name(self) -> str:
+        """Return the built-in backend display name."""
+        return "AstrBot Built-in Knowledge Base"
+
+    async def list_knowledge_bases(
+        self,
+        *,
+        umo: str | None = None,
+    ) -> list[KnowledgeBaseInfo]:
+        """List built-in knowledge bases.
+
+        Args:
+            umo: Optional unified message origin. The built-in backend does not
+                currently apply session-specific access filtering.
+
+        Returns:
+            Built-in knowledge base descriptors.
+        """
+        records = await self.manager.list_kbs()
+        return [
+            KnowledgeBaseInfo(
+                ref=KnowledgeBaseRef(
+                    backend_id=self.backend_id,
+                    knowledge_base_id=record.kb_id,
+                ),
+                name=record.kb_name,
+                description=record.description,
+                metadata={
+                    "emoji": record.emoji,
+                    "doc_count": record.doc_count,
+                    "chunk_count": record.chunk_count,
+                },
+            )
+            for record in records
+        ]
+
+    async def retrieve(
+        self,
+        knowledge_base_ids: list[str],
+        request: KnowledgeBaseQuery,
+    ) -> KnowledgeBaseResponse:
+        """Retrieve from built-in knowledge bases.
+
+        Args:
+            knowledge_base_ids: Built-in knowledge base identifiers.
+            request: Standardized retrieval request.
+
+        Returns:
+            Standardized built-in retrieval results and warnings.
+        """
+        kb_names = []
+        warnings = []
+        for kb_id in knowledge_base_ids:
+            helper = await self.manager.get_kb(kb_id)
+            if helper is None:
+                warnings.append(f"Built-in knowledge base '{kb_id}' was not found.")
+                continue
+            kb_names.append(helper.kb.kb_name)
+
+        if not kb_names:
+            return KnowledgeBaseResponse(hits=[], warnings=warnings)
+
+        result = await self.manager.retrieve(
+            query=request.query,
+            kb_names=kb_names,
+            top_m_final=request.top_k,
+        )
+        if not result:
+            return KnowledgeBaseResponse(hits=[], warnings=warnings)
+
+        hits = []
+        for rank, item in enumerate(result.get("results", []), start=1):
+            hits.append(
+                KnowledgeBaseHit(
+                    ref=KnowledgeBaseRef(
+                        backend_id=self.backend_id,
+                        knowledge_base_id=item["kb_id"],
+                    ),
+                    content=item["content"],
+                    source=item.get("doc_name")
+                    or item.get("kb_name")
+                    or self.display_name,
+                    rank=rank,
+                    score=item.get("score"),
+                    document_id=item.get("doc_id"),
+                    chunk_id=item.get("chunk_id"),
+                    metadata={
+                        "backend_id": self.backend_id,
+                        "knowledge_base_id": item.get("kb_id"),
+                        "knowledge_base_name": item.get("kb_name"),
+                        "chunk_index": item.get("chunk_index", 0),
+                        "char_count": item.get("char_count", 0),
+                    },
+                )
+            )
+
+        return KnowledgeBaseResponse(hits=hits, warnings=warnings)