Async Knowledge Retrieval

Async-safe RAG for agents running under arun or a FastAPI handler

Async-safe RAG for agents running under arun or a FastAPI handler — knowledge search never blocks the event loop.

Quick Start

1. Agent under arun

   An agent that searches knowledge inside arun() no longer blocks the event loop — concurrent requests stay responsive.

   import asyncio
   from praisonaiagents import Agent
   
   agent = Agent(
       name="Research Assistant",
       instructions="Answer questions using the knowledge base.",
       knowledge=["docs/handbook.pdf"],
   )
   
   async def main():
       # Two agents run concurrently; each RAG lookup is offloaded,
       # so the loop stays free instead of serialising the round trips.
       await asyncio.gather(
           agent.arun("Summarise the onboarding policy"),
           agent.arun("What is the refund window?"),
       )
   
   asyncio.run(main())
   
2. Orchestrator (library authors)

   Call aretrieve_knowledge / aadd_knowledge directly when you build on the persistence layer.

   import asyncio
   from praisonai.persistence import PersistenceOrchestrator
   from praisonai.persistence.knowledge.base import KnowledgeDocument
   
   orchestrator = PersistenceOrchestrator.from_env()
   
   async def main():
       # Add documents (embeddings already computed)
       await orchestrator.aadd_knowledge(
           documents=[
               KnowledgeDocument(
                   id="doc-1",
                   content="Refunds are processed within 30 days.",
                   embedding=[0.12, 0.85, 0.03],
               )
           ],
           collection="policies",
       )
   
       # Retrieve without blocking the loop
       docs = await orchestrator.aretrieve_knowledge(
           query_embedding=[0.10, 0.80, 0.05],
           collection="policies",
           limit=3,
       )
       for doc in docs:
           print(doc.content)
   
   asyncio.run(main())
   
3. Streaming + knowledge

   Stream tokens from an agent that also retrieves knowledge — the retrieved context is normalized the same way as the sync path.

   from praisonaiagents import Agent
   
   agent = Agent(
       name="Research Assistant",
       instructions="Answer using the knowledge base.",
       knowledge=["docs/handbook.pdf"],
   )
   
   for chunk in agent.start_stream("Summarise the onboarding policy"):
       print(chunk, end="", flush=True)
   

Fixed in PraisonAI PR #4887. Prior to this release, .achat() / .astart() on an Agent(knowledge=[...]) could silently return None (a TypeError swallowed by a blanket except), and both streaming branches raised TypeError: can only join an iterable — each async and streaming site re-implemented a partial normalizer that only handled the legacy dict shape and tried to "\n".join(...) a SearchResult dataclass. After this release, async chat and both streaming branches route through the same _get_knowledge_context() helper as the sync path and correctly normalize the SearchResult dataclass, dict, and list shapes.

---

How It Works

aretrieve_knowledge and aadd_knowledge inspect the underlying store and pick the right path automatically.

Store type	Path taken
Native async (search / upsert is a coroutine)	Awaited directly
Sync store	Offloaded via asyncio.to_thread(...) so the loop stays free

Every real vector backend issues a network round trip on search. From an async agent, the sync retrieve_knowledge would block the loop for the whole round trip and serialise every other concurrent request. The a* methods align RAG with the existing async conversation hooks (aon_message, aon_agent_start).

Agent RAG lookups are offloaded automatically. Inside Agent._achat_impl, the built-in knowledge.search(...) call runs via asyncio.to_thread(...), so it is safe under concurrent task execution. Multiple async agents — or an asyncio.gather(...) of tasks — that share one knowledge base run in parallel; no lookup blocks the loop, even when the underlying store is synchronous. See Async Safety.

---

When to Use a* vs Sync

Use the a* variants whenever an event loop is running. In a plain sync script, the existing retrieve_knowledge / add_knowledge methods are the right choice.

---

Configuration Options

aretrieve_knowledge(query_embedding, collection="default", limit=5, filters=None)

Option	Type	Default	Description
query_embedding	List[float]	(required)	Query vector to search against
collection	str	"default"	Collection to search
limit	int	5	Max documents to return
filters	Optional[Dict[str, Any]]	None	Metadata filters

aadd_knowledge(documents, collection="default")

Option	Type	Default	Description
documents	List[KnowledgeDocument]	(required)	Documents with embeddings to upsert
collection	str	"default"	Collection to write to

---

Best Practices

Use a* variants inside arun() or a web handler

Any code path that already runs on an event loop — agent.arun(...), a FastAPI route, a background async task — should call aretrieve_knowledge / aadd_knowledge. The sync methods would block the loop for the full network round trip.

Sync stores are still safe

A sync knowledge store plugged into an async agent is not rejected — it is offloaded to a thread via asyncio.to_thread(...). You do not need a native async backend to benefit.

Concurrency is real — verify it

Two concurrent RAG calls against a store with a 0.3 s round trip finish in ~0.3 s, not 0.6 s. Fire two aretrieve_knowledge calls with asyncio.gather and confirm they overlap.

Async, streaming, and sync now use the same normalization

Since PR #4887, .chat(), .achat(), .start_stream(), and .astart(..., stream=True) all route knowledge retrieval through _get_knowledge_context(). The SearchResult dataclass, legacy dict, and plain list shapes are handled in one place — no per-path guardrails needed. Switch an agent between sync, async, and streaming freely; knowledge injection behaves identically.

---

Related

Async Conversation Store

Sibling async hooks (aon_message, aon_agent_start).

Persistence Overview

Where knowledge stores are configured.

RAG Module

The full RAG pipeline.

Search Results

The SearchResult shape and the “check .results” pitfall this fix protects against.

Streaming

Stream tokens while injecting knowledge context.