Cookbook: Async Usage & Streaming

Recipe: Concurrent document ingestion

RAGSystem.aadd_documents chunks documents concurrently via asyncio.gather. Note the different input shape compared to the sync method — see the warning in the async API reference.

import asyncio

async def ingest():
    result = await rag.aadd_documents({
        "doc1": "First document text...",
        "doc2": "Second document text...",
    })
    print(result)

asyncio.run(ingest())

Recipe: Concurrent retrieval for a batch of queries

async def batch_retrieve(queries):
    return await asyncio.gather(*(rag.aretrieve(q) for q in queries))

results = asyncio.run(batch_retrieve(["question 1", "question 2", "question 3"]))

Recipe: Streaming tokens to a client

async def stream_answer(query: str):
    async for token in rag.astream(query):
        yield token   # e.g. forward to a websocket or SSE connection

astream's behavior adapts automatically to what your LLM's .astream() method returns — async generator, coroutine, or sync iterable — see the astream algorithm for exact branching logic.

async def safe_stream(query: str):
    try:
        async for token in rag.astream(query):
            yield token
    except Exception as e:
        yield f"\n[stream error: {e}]"

Recipe: Running sync and async code side by side

Because RAGSystem implements both sync and native async methods, you can freely mix them in the same application — e.g. a background sync job using add_documents() and an async web handler using agenerate() against the same RAGSystem instance. Be mindful of stats mutation race conditions across threads if the sync path runs in a separate thread pool alongside the async event loop (see thread safety notes).

Recipe: Async context manager

async def main():
    async with RAGSystem(vector_db=vector_db, llm=llm, chunker=chunker,
                          context_manager=context_manager) as rag:
        await rag.aadd_documents({"doc1": "text"})
        print(await rag.agenerate("question"))
    # cleanup() called automatically on exit

asyncio.run(main())