ak-dev-new-knowledgebase-integration

Adding a New Knowledge Base Integration

Use this skill to add a new knowledge base backend under ak-py/src/agentkernel/knowledgebase/.

This applies when you are integrating any durable source not already covered by:

• ChromaManager (semantic vector search)
• Neo4jManager (graph relationships)
• StarburstManager (read-only SQL via Trino)

Prerequisites

• Understand architecture and contribution patterns from .agents/skills/ak-dev-architecture/SKILL.md
• Understand existing knowledge base APIs:
  • ak-py/src/agentkernel/knowledgebase/base.py
  • ak-py/src/agentkernel/knowledgebase/knowledgebuilder.py
• Have provider credentials and a local/dev test instance for the target backend

Step-by-Step

1. Create Backend Module

Create a new file:

ak-py/src/agentkernel/knowledgebase/<backend>.py

Use lowercase file names (for example qdrant.py, milvus.py, elasticsearch.py).

2. Subclass KnowledgeBase

Implement a concrete class that extends KnowledgeBase:

from typing import Any, Iterable, List, Mapping

from .base import KnowledgeBase, Record


class MyBackendManager(KnowledgeBase):
    def __init__(self, name: str = "", description: str | None = None, **kwargs):
        super().__init__()
        self.name = name
        self.description = description or "my backend"
        self._client = None
        self.connect(**kwargs)

    @property
    def backend_name(self) -> str:
        return self.name if self.name else "mybackend"

    def connect(self, **kwargs) -> None:
        # Initialize backend client and verify connectivity.
        self._client = ...

    def write(self, records: Iterable[Record], **kwargs) -> None:
        for record in records:
            text = str(record.get("text", "")).strip()
            metadata = dict(record.get("metadata", {}))
            if not text:
                continue
            # Persist text + metadata using backend-native API.
            self._client.store(text=text, metadata=metadata)

    def read(self, query: str, limit: int = 3, **kwargs) -> List[Mapping[str, Any]]:
        rows = self._client.search(query=query, limit=limit)
        return [{"text": row["text"], "metadata": row.get("metadata", {})} for row in rows]

    def get_description(self) -> str:
        return f"{self.backend_name}: {self.description}"