This project is a small, fully async RAG (Retrieval-Augmented Generation) system that turns a flat JSON file of question/answer pairs into a semantic QA engine. Instead of matching on keywords, it:
Loads and validates a knowledge base concurrently with asyncio + aiofiles + pydantic Embeds every document through an OpenAI-compatible embeddings API, with concurrency capped by an asyncio.Semaphore Embeds the incoming query the same way, and finds the closest document with NumPy cosine similarity Hands the retrieved answer to an LLM as context, and returns a natural-language response
It's built as a learning/reference project for async Python patterns applied to a real RAG pipeline — not a production vector database.
Features Async I/O throughout — file loading, embedding calls, and generation calls all run on asyncio, so nothing blocks the event loop Schema validation on load — every record in the knowledge base is parsed into a pydantic.BaseModel, so malformed JSON fails loudly and early Concurrency-controlled embedding — a semaphore caps in-flight embedding requests (MAX_CONCURRENT_REQUESTS), so you don't blow through API rate limits when embedding large datasets Semantic retrieval — cosine similarity over embedding vectors, computed with NumPy, instead of keyword/substring search Pluggable LLM backend — uses the openai SDK's AsyncOpenAI client pointed at a custom base_url, so it works with OpenAI, Hugging Face's Inference Router, or any OpenAI-compatible endpoint How it works aiofiles.open + json.loads qa_pairs.json Pydantic validation(Document model) Concurrent embedding(asyncio.gather +Semaphore) In-memorydocument store User query Embed query Cosine similarityvs. every document(NumPy) Best-matching answer LLM generation(context + question) Final response
Indexing (offline / startup path)
load_documents() reads qa_pairs.json asynchronously and validates each entry against a Document pydantic model (question, answer). index_documents() fires off one embedding request per document via asyncio.gather, with get_embeddings_with_semaphore() limiting how many run at once.
Query (runtime path)
retrieve() embeds the incoming query, scores it against every stored document with compute_similarity() (cosine similarity), and returns the single best-matching answer. generate_response() passes that answer to the LLM as context alongside the original question, and returns the model's synthesized reply. Project structure . ├── main.py # Pipeline: load → embed → retrieve → generate ├── qa_pairs.json # Knowledge base (question/answer pairs) — add this to the repo ├── requirements.txt # Dependencies ├── .env # HF_TOKEN / API key (not committed) └── README.md Tech stack Purpose Library Async runtime asyncio Async file I/O aiofiles Data validation pydantic HTTP clients httpx, aiohttp Numerical ops (cosine sim.) numpy LLM / embeddings client openai (AsyncOpenAI) Env config python-dotenv Getting started Prerequisites Python 3.10+ An API key for an OpenAI-compatible endpoint (the code defaults to Hugging Face's Inference Router, but works with OpenAI or any compatible provider) Installation bash git clone https://github.com/ivyanalyst/Python-Asynchronous-RAG-Program.git cd Python-Asynchronous-RAG-Program
python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate
pip install -r requirements.txt Configuration
Create a .env file in the project root:
env HF_TOKEN=your_api_key_here
By default the client points at Hugging Face's router:
python client = AsyncOpenAI( http_client=http_client, base_url="https://router.huggingface.co/v1", api_key=os.getenv("HF_TOKEN") )
To use OpenAI directly instead, swap the base_url for https://api.openai.com/v1 and set OPENAI_API_KEY (updating os.getenv(...) to match).
Add your knowledge base
Drop a qa_pairs.json file in the project root — a list of question/answer objects:
json [ { "question": "Who was the first President of the United States?", "answer": "George Washington" }, { "question": "What year did World War II end?", "answer": "1945" } ] Run it bash python main.py
Expected output:
Loading databases from qa_pairs.json... Success Loaded 50 documents. Fist Question: Who was the first President of the United States? First Answer: George Washington Embedding 50 chunk. Max concurrency: 2... All chunks embedded. documents indexing completed.
User Query: Who was England's longest-ruling monarch? Retrieved Answer: Queen Elizabeth II
Generating response... Final Response: Queen Elizabeth II was England's longest-ruling monarch. Configuration knobs Setting Where Default Max concurrent embedding calls MAX_CONCURRENT_REQUESTS 2 Embedding model get_embeddings() text-embedding-3-small Chat/generation model generate_response() gpt-4o Knowledge base path main() → db_path "qa_pairs.json" Query main() → user_query hardcoded — swap for input() or CLI args Known limitations Retrieval is a brute-force cosine similarity scan over every document in memory — fine for tens or hundreds of QA pairs, but it won't scale to a large corpus without a proper vector index (FAISS, pgvector, etc.). Embeddings are recomputed on every run; there's no caching or persistence layer. retrieve() accepts a top_k parameter but currently only returns the single best match — multi-document context assembly isn't wired up yet. The query is hardcoded in main() rather than accepted as user input. Roadmap Persist embeddings (e.g., to disk or a lightweight vector store) instead of recomputing on every run Use top_k to assemble multi-document context for generation Accept queries via CLI args or an interactive loop Add retry/backoff around embedding and generation calls Basic test coverage for compute_similarity and the loading/validation path Contributing
Issues and PRs are welcome — this is very much a learning project in progress.
License
This project is open source and available under the MIT License.