Async Knowledge Store

Non-blocking vector and RAG persistence with the AsyncKnowledgeStore protocol

Async knowledge stores provide non-blocking vector storage and semantic search so native-async RAG backends never block the event loop.

from praisonaiagents import Agent

agent = Agent(
    name="Researcher",
    instructions="Answer from the indexed knowledge base.",
)
agent.start("Summarise our onboarding docs.")

The agent runs a semantic search; an async knowledge store queries the vector backend without blocking other agents on the same loop.

Quick Start

1. Subclass AsyncKnowledgeStore

   Implement a native-async vector backend by inheriting from AsyncKnowledgeStore:

   from praisonai.persistence import AsyncKnowledgeStore
   from praisonai.persistence.knowledge import KnowledgeDocument
   from typing import Any, Dict, List, Optional
   
   class MyAsyncKnowledgeStore(AsyncKnowledgeStore):
       async def create_collection(self, name, dimension, distance="cosine", metadata=None):
           ...
   
       async def delete_collection(self, name) -> bool:
           ...
   
       async def collection_exists(self, name) -> bool:
           ...
   
       async def list_collections(self) -> List[str]:
           ...
   
       async def insert(self, collection, documents) -> List[str]:
           ...
   
       async def upsert(self, collection, documents) -> List[str]:
           ...
   
       async def search(self, collection, query_embedding, limit=5, filters=None, score_threshold=None):
           ...
   
       async def get(self, collection, ids) -> List[KnowledgeDocument]:
           ...
   
       async def delete(self, collection, ids=None, filters=None) -> int:
           ...
   
       async def count(self, collection) -> int:
           ...
   
       async def close(self) -> None:
           ...

2. Wire It Into an Agent

   Pass the store through the persistence config so the agent dispatches async calls natively:

   from praisonaiagents import Agent
   
   store = MyAsyncKnowledgeStore()
   
   agent = Agent(
       name="Researcher",
       instructions="Answer from the indexed knowledge base.",
       knowledge_store=store,
   )
   agent.start("Summarise our onboarding docs.")

How It Works

A sync caller inside a running loop dispatches to the async store through PraisonAIDB._dispatch_async.

Abstract Methods

The ABC also ships __aenter__ / __aexit__ for async with usage.

Configuration Options

Common Patterns

Custom Async Backend Inheritance

Subclass AsyncKnowledgeStore (not KnowledgeStore) when your vector backend is natively async:

from praisonai.persistence import AsyncKnowledgeStore

class QdrantAsyncKnowledgeStore(AsyncKnowledgeStore):
    def __init__(self, client):
        self._client = client

    async def search(self, collection, query_embedding, limit=5, filters=None, score_threshold=None):
        return await self._client.search(collection, query_embedding, limit=limit)

    async def close(self):
        await self._client.close()
    # ... implement the remaining abstract methods

async with Context Manager

The ABC defines __aenter__ / __aexit__, so async with closes the store automatically:

async with MyAsyncKnowledgeStore() as store:
    await store.create_collection("docs", dimension=1536)
    ids = await store.insert("docs", documents)

Semantic Search

Query by embedding with an optional score threshold and metadata filters:

async with MyAsyncKnowledgeStore() as store:
    results = await store.search(
        "docs",
        query_embedding=embedding,
        limit=5,
        score_threshold=0.75,
    )

Best Practices

# ✅ Good - native async dispatch
class MyStore(AsyncKnowledgeStore):
    async def search(self, collection, query_embedding, limit=5, filters=None, score_threshold=None): ...

# ❌ Bad - sync ABC forces run_sync shims
class MyStore(KnowledgeStore):
    def search(self, collection, query_embedding): ...

# ✅ Good - automatic cleanup
async with MyAsyncKnowledgeStore() as store:
    await store.count("docs")

# ❌ Bad - leaked connections
store = MyAsyncKnowledgeStore()
await store.count("docs")

# ✅ Good - pure async
async with store:
    await store.insert("docs", documents)

# ❌ Bad - calling an async method without await
store.insert("docs", documents)  # returns an un-awaited coroutine

Related