{"id":"2fcc802e-1647-4595-acac-c1dc7bb708c3","slug":"knowledge-rag-assistant","kind":"mcp","name":"knowledge-rag-assistant","summary":"RAG-powered assistant with MCP server, Agent SDK orchestration, and Langfuse observability. Built with FastAPI, Qdrant, Redis, and PostgreSQL.","owner":"JaimeRam","repoUrl":"https://github.com/JaimeRam/knowledge-rag-assistant","license":"MIT","licenseTier":"permissive","trust":"community","stars":2,"match":null,"fewStars":true,"format":"mcp","createdAt":"2026-04-29T18:59:47.000Z","addedAt":"2026-10-21T21:03:56.910Z","logoUrl":"https://avatars.githubusercontent.com/u/6652595?v=4","linkOnly":false,"collection":null,"skillPath":null,"categories":["databases","search","cloud-infra"],"tags":[],"runtime":"binary","local":true,"cloudCapable":false,"cloudCapability":"unknown","cloudReadyOnPlatform":false,"needsApiKey":false,"reviewed":false,"syncedAt":"2026-10-22T07:18:26.035Z","sample":false,"url":"https://conduits.dev/tools/knowledge-rag-assistant","language":"Python","transports":["stdio"],"install":{},"requirements":{"env":[],"upstreamMayCharge":false},"tools":null,"readme":"# Knowledge RAG Assistant\n\n![CI](https://github.com/JaimeRam/knowledge-rag-assistant/actions/workflows/ci.yml/badge.svg)\n![Python](https://img.shields.io/badge/Python-3.11+-blue.svg)\n![FastAPI](https://img.shields.io/badge/FastAPI-0.115+-green.svg)\n![License](https://img.shields.io/badge/License-MIT-yellow.svg)\n\nA production-ready RAG-powered knowledge assistant demonstrating modern LLM backend development practices: custom RAG pipeline, MCP server, PydanticAI tool-calling agent, LangGraph Corrective RAG, Langfuse observability, streaming, and semantic caching.\n\n*Example domain: Digimon knowledge base (DAPI)*\n\n## Demo\n\n![Streaming demo](docs/demo.gif)\n\n> **Record your own**: install `asciinema` and `agg`, then run `./scripts/record_demo.sh` — it records the full multi-turn conversation and converts to `docs/demo.gif` automatically.\n>\n> ```bash\n> pip install asciinema          # or: brew install asciinema\n> cargo install --git https://github.com/asciinema/agg   # or: brew install agg\n> ./scripts/record_demo.sh\n> ```\n\n```bash\n# Multi-turn conversation via streaming\ncurl -N -X POST http://localhost:8000/api/v1/chat/stream \\\n  -H \"Content-Type: application/json\" \\\n  -d '{\"query\": \"Tell me about Agumon\"}'\n\n# Follow-up — pass the conversation_id from the previous response\ncurl -N -X POST http://localhost:8000/api/v1/chat/stream \\\n  -H \"Content-Type: application/json\" \\\n  -d '{\"query\": \"What are its digivolutions?\", \"conversation_id\": \"<id from above>\"}'\n```\n\n## Tech Stack\n\n| Layer | Technology |\n|---|---|\n| API | FastAPI (Python 3.11+) |\n| LLM | Anthropic `claude-haiku-4-5` |\n| Embeddings | Voyage AI `voyage-3` (1024 dims) |\n| Vector DB | Qdrant |\n| Relational DB | PostgreSQL + SQLAlchemy |\n| Caching | Redis |\n| Observability | Langfuse |\n| Agent Protocol | MCP (Model Context Protocol) |\n| Typed Agent | PydanticAI (tool-calling agent) |\n| Agentic RAG | LangGraph (Corrective RAG workflow) |\n| Guardrails | Custom `InputGuard` (LLM-based, fail-open) |\n| RAG Evaluation | RAGAS (faithfulness, relevancy, precision, recall) |\n| Infrastructure | Docker + docker-compose |\n\n## Architecture\n\n```\n  ┌─────────────────────────────────────────────────────┐\n  │                    Client (HTTP)                    │\n  └─────────────────────────┬───────────────────────────┘\n                            │\n             ┌──────────────┴─────────────┐\n             │                            │\n  ┌──────────▼──────────────┐  ┌──────────▼───────────────────┐\n  │   POST /api/v1/chat     │  │   POST /api/v1/agent         │\n  │      (Custom RAG)       │  │   POST /api/v1/agent/graph   │\n  └──────────┬──────────────┘  └──────────┬───────────────────┘\n             │                            │\n             └──────────────┬─────────────┘\n                            │\n  ┌─────────────────────────▼───────────────────────────┐\n  │                      RAG Core                       │\n  │      Retriever → PromptBuilder → LLMClient          │\n  └───────────────────┬─────────────────┬───────────────┘\n                      │                 │\n           ┌──────────▼──┐     ┌────────▼──────┐\n           │   Qdrant    │     │     DAPI      │\n           │  (vectors)  │     │  (live API)   │\n           └─────────────┘     └───────────────┘\n\n  LangGraph Corrective RAG flow:\n  retrieve → grade → generate\n                └──→ rewrite → retrieve (max 2 loops)\n\n  Observability: Langfuse traces all LLM calls and retrievals\n  Caching:       Redis caches /chat responses by query + filters\n```\n\n## Features\n\n- **Custom RAG Pipeline**: Retriever → Prompt Builder → LLM, built from scratch without LangChain\n- **PydanticAI Agent** (`/api/v1/agent`): type-safe agent with 4 tools (RAG search, by name, by level, skills) and automatic tool selection\n- **LangGraph Corrective RAG** (`/api/v1/agent/graph`): agentic retrieve → grade → generate workflow with query rewriting when docs aren't relevant\n- **MCP Server**: 4 tools exposing Digimon data (by name, level, ID, skills)\n- **Conversation History**: Multi-turn memory stored in Redis — pass `conversation_id` to maintain context across calls (last 5 turns, 1h TTL)\n- **Streaming**: SSE endpoint (`/chat/stream`) returns tokens as they are generated — zero-buffering, real-time UX\n- **Observability**: Full tracing with Langfuse — per-trace numeric scores (latency, tokens, retrieval quality) and async LLM-as-judge evaluation (faithfulness, relevance, context relevance)\n- **Semantic Caching**: Redis-based response caching to reduce API costs\n- **Data Ingestion**: Async pipeline from [DAPI](https://digi-api.com/) → Voyage AI embeddings → Qdrant + PostgreSQL\n\n## Setup\n\n### Prerequisites\n\n- Docker and Docker Compose\n- Python 3.11+\n- Anthropic API key → [console.anthropic.com](https://console.anthropic.com)\n- Voyage AI API key → [dash.voyageai.com](https://dash.voyageai.com)\n\n### Installation\n\n1. Clone the repository:\n\n```bash\ngit clone https://github.com/JaimeRam/knowledge-rag-assistant.git\ncd knowledge-rag-assistant\n```\n\n2. Create and activate a virtual environment:\n\n```bash\npython -m venv venv\nsource venv/bin/activate  # On Windows: venv\\Scripts\\activate\n```\n\n3. Install Python dependencies:\n\n```bash\npip install -r requirements.txt\n```\n\n4. Copy and configure the environment file:\n\n```bash\ncp .env.example .env\n# Edit .env — set ANTHROPIC_API_KEY and VOYAGE_API_KEY at minimum\n```\n\n5. Start infrastructure services (Qdrant, Redis, PostgreSQL, Langfuse):\n\n```bash\nmake services\n```\n\n6. Run data ingestion:\n\n```bash\nmake ingest\n```\n\n> **Voyage AI free tier note**: without a payment method the limit is 3 RPM. The default `INGEST_LIMIT=100` in `.env` keeps ingestion under ~5 minutes. Set `INGEST_LIMIT=0` to ingest all ~1450 Digimon once you add a payment method (200M free tokens still apply).\n\n7. Start the API:\n\n```bash\nmake run\n```\n\nThe API will be available at `http://localhost:8000`. Interactive docs at `http://localhost:8000/docs`.\n\n### Docker (full stack)\n\n```bash\ndocker build -t digimon-rag-assistant .\ndocker run -p 8000:8000 --env-file .env digimon-rag-assistant\n```\n\n## API Endpoints\n\n### Chat with RAG\n\n```bash\nPOST /api/v1/chat\nContent-Type: application/json\n\n{\n  \"query\": \"What is Agumon?\",\n  \"level_filter\": \"Rookie\",\n  \"type_filter\": \"Reptile\"\n}\n```\n\n### PydanticAI Agent (tool-calling)\n\n```bash\nPOST /api/v1/agent\nContent-Type: application/json\n\n{\"query\": \"What are the skills of Agumon?\"}\n# → agent selects get_digimon_by_name + get_digimon_skills automatically\n\n{\"query\": \"Which Rookie Digimon are best for beginners?\"}\n# → agent selects rag_search + get_digimon_by_level automatically\n```\n\nResponse includes `tool_calls` (list of tools used) and token `usage`.\n\n### LangGraph Corrective RAG\n\n```bash\nPOST /api/v1/agent/graph\nContent-Type: application/json\n\n{\"query\": \"Tell me about fire-type Rookie Digimon\"}\n```\n\nResponse includes `iterations` (0 = direct, 1+ = query was rewritten) and `documents_used`.\n\n### Get Digimon by ID\n\n```bash\nGET /api/v1/digimon/{id}\n```\n\n### List Digimons\n\n```bash\nGET /api/v1/digimon?limit=10&offset=0\n```\n\n### Health Check\n\n```bash\nGET /api/v1/health\n```\n\n## Data Ingestion\n\nThe ingestion pipeline fetches data from [DAPI](https://digi-api.com/):\n\n1. Fetches Digimon data (respects `INGEST_LIMIT` in `.env`)\n2. Prepares text chunks per Digimon (basic info, description, skills, evolutions)\n3. Generates vector embeddings with Voyage AI (`voyage-3`, 1024 dims) in rate-limited batches\n4. Stores embeddings in Qdrant and metadata in PostgreSQL\n\nKey `.env` variables for ingestion:\n\n| Variable | Default | Description |\n|---|---|---|\n| `INGEST_LIMIT` | `100` | Max Digimon to ingest (0 = all ~1450) |\n| `VOYAGE_EMBED_BATCH_SIZE` | `20` | Chunks per embedding API call |\n| `VOYAGE_RPM` | `3` | Voyage AI requests per minute (free tier = 3) |\n\nTo re-run ingestion:\n\n```bash\nmake ingest\n```\n\n## Project Structure\n\n```\nknowledge-rag-assistant/\n├── app/\n│   ├── api/              # FastAPI routes and app setup\n│   │   ├── routes/\n│   │   │   ├── chat.py       # /chat — Custom RAG + Redis caching\n│   │   │   ├── agent.py      # /agent — PydanticAI + /agent/graph — LangGraph\n│   │   │   ├── digimon.py    # Digimon metadata endpoints\n│   │   │   └── health.py     # Health check\n│   │   └── main.py\n│   ├── core/             # Config (pydantic-settings) and logging\n│   ├── ingestion/        # DAPI client, embedder, ingest pipeline\n│   ├── rag/              # Retriever, prompt builder, LLM client\n│   ├── guardrails/       # InputGuard — LLM-based topic classifier (fail-open)\n│   ├── mcp/              # MCP server with 4 Digimon tools\n│   ├── agents/\n│   │   ├── pydantic_agent.py # PydanticAI agent with 4 typed tools\n│   │   └── graph/            # LangGraph Corrective RAG\n│   │       ├── state.py      # GraphState TypedDict\n│   │       ├── nodes.py      # retrieve / grade / generate / rewrite nodes\n│   │       └── workflow.py   # Compiled StateGraph\n│   ├── observability/    # Langfuse tracing integration\n│   └── db/               # Qdrant, PostgreSQL, Redis managers\n├── docker/\n│   └── docker-compose.yml\n├── tests/\n├── requirements.txt\n├── Dockerfile\n├── CONTRIBUTING.md\n└── README.md\n```\n\n## Technical Highlights\n\nThis project demonstrates key skills for LLM Backend Developer roles:\n\n1. **Custom RAG without LangChain**: Built the full pipeline (retrieval → prompt → generation) from first principles — shows deep understanding of how RAG works, not just framework usage.\n\n2. **Three agent patterns side by side**: Custom RAG (`/chat`), PydanticAI typed agent (`/agent`), and LangGraph Corrective RAG (`/agent/graph`) — each demonstrates a different architectural trade-off.\n\n3. **PydanticAI typed tools**: Type-safe agent with `@agent.tool_plain` decorators, `AnthropicModel`, and automatic tool selection. Zero boilerplate.\n\n4. **LangGraph Corrective RAG**: Stateful `retrieve → grade → generate` loop with query rewriting when retrieved docs aren't relevant — demonstrates iterative reasoning beyond single-shot generation.\n\n5. **MCP (Model Context Protocol)**: Implements the emerging standard for tool integration between LLMs and external systems.\n\n6. **Production Observability**: Full tracing with Langfuse — numeric scores per trace (latency, tokens, retrieval quality) plus async LLM-as-judge evaluation (faithfulness, relevance, context relevance) as a background task so it never adds latency to the API response.\n\n7. **Cost Optimization**: Redis caching avoids redundant LLM calls for identical queries, with proper cache-key design to prevent collisions across different filter combinations.\n\n8. **Input Guardrails**: LLM-based classifier blocks off-topic queries before they reach the RAG pipeline. Fail-open design ensures a guardrail outage never takes down the main service.\n\n9. **RAGAS Evaluation**: Industry-standard RAG evaluation with 4 metrics (faithfulness, answer relevancy, context precision, context recall) against a 10-question golden dataset — measures actual pipeline quality, not just test pass/fail.\n\n## Roadmap\n\n- [x] Custom RAG pipeline (Retriever → PromptBuilder → LLMClient)\n- [x] MCP server with 4 tools\n- [x] Anthropic Claude Haiku + Voyage AI voyage-3 integration\n- [x] Langfuse observability\n- [x] Redis caching with proper cache-key design\n- [x] Docker infrastructure\n- [x] PydanticAI typed agent with automatic tool selection\n- [x] LangGraph Corrective RAG with query rewriting\n- [x] Comprehensive test suite with pytest (mocked, no external services required)\n- [x] RAG evaluation pipeline with RAGAS (faithfulness, relevancy, precision, recall)\n- [x] Input guardrails to block off-topic queries (fail-open design)\n- [x] LLM-as-judge scoring in Langfuse (faithfulness, relevance, context relevance — async background task)\n- [x] Streaming SSE endpoint with real-time token delivery\n- [x] Conversation history with Redis (multi-turn, last 5 turns, 1h TTL)\n- [ ] Rate limiting and authentication\n\n## Running Tests\n\nTests are fully mocked — no external services required:\n\n```bash\n# Activate venv\nsource venv/bin/activate\n\n# Run all unit tests (test_ingestion.py excluded by default — needs live services)\nmake test\n# or: pytest tests/ -v --ignore=tests/test_ingestion.py\n\n# With coverage report\npytest tests/ -v --ignore=tests/test_ingestion.py --cov=app --cov-report=term-missing\n\n# Integration tests (requires make services + valid API keys)\npytest -m integration\n```\n\nTest suite covers: `PromptBuilder`, `LLMClient`, `Retriever`, `/chat` endpoint (including cache hit), and `InputGuard` guardrails.\n\n`tests/test_ingestion.py` is marked `@pytest.mark.integration` and excluded from the default run.\n\n## RAG Evaluation\n\nEvaluates the pipeline against a 10-question golden dataset using [RAGAS](https://docs.ragas.io/) metrics:\n\n| Metric | Measures |\n|---|---|\n| `faithfulness` | Is the answer grounded in the retrieved context? |\n| `answer_relevancy` | Is the answer relevant to the question? |\n| `context_precision` | Are the retrieved chunks precise (not noisy)? |\n| `context_recall` | Do the chunks cover the ground-truth answer? |\n\n```bash\n# Requires: make services + make ingest (Qdrant must have data)\nsource venv/bin/activate\npython scripts/evaluate_rag.py            # build dataset + evaluate (~60 min total)\npython scripts/evaluate_rag.py --eval-only  # reuse cached dataset, skip rebuild (~55 min)\n```\n\nOutput: per-question scores table + overall averages. Faithfulness < 0.7 indicates a prompt issue; context recall < 0.6 indicates an ingestion or embedding issue.\n\n**Baseline scores on default 100-Digimon ingest:**\n\n| Metric | Score | Threshold |\n|---|---|---|\n| `faithfulness` | 1.000 | ≥ 0.70 ✓ |\n| `answer_relevancy` | 0.848 | ≥ 0.70 ✓ |\n| `context_precision` | 0.883 | ≥ 0.70 ✓ |\n| `context_recall` | 1.000 | ≥ 0.60 ✓ |\n\nAll four metrics pass threshold. `answer_relevancy` originally scored 0.558 — two compounding root causes were found and fixed:\n\n1. **Prompt scope discipline**: the system prompt didn't constrain answers to what was actually asked, so narrow questions (\"What level is X?\") got full Digimon profiles back. RAGAS's `ResponseRelevancy` reconstructs questions from the *answer* and compares them to the original — a bloated answer reconstructs into a generic question that doesn't match a narrow original. Fixed by instructing the model to answer only the requested aspect unless the question is an open request (\"tell me about X\").\n2. **Evaluation harness bug**: RAGAS's `ResponseRelevancy` embeds the original question with `embed_query()` and the reconstructed questions with `embed_documents()`. Voyage-3 is an asymmetric retrieval model — `input_type=\"query\"` and `input_type=\"document\"` land in different subspaces — so comparing a question against other questions this way artificially compressed the cosine similarity (verified directly: two near-identical paraphrases scored only 0.59 under the asymmetric call vs. 0.96 with symmetric embeddings). Fixed in `scripts/evaluate_rag.py` by embedding both sides with `input_type=None` (Voyage's general-purpose symmetric mode) for this metric only — production retrieval is untouched and still correctly uses the asymmetric query/document embeddings it was designed for.\n\n`context_precision` (0.883) has two individually low-scoring questions (~0.33–0.5) despite retrieving the correct chunk every time — RAGAS's precision metric is rank-sensitive, and for a couple of narrow single-fact questions the info-bearing chunk is outranked by that same Digimon's skills or evolution chunk. `context_recall` stays perfect because it only checks presence, not rank.\n\n## Contributing\n\nSee [CONTRIBUTING.md](CONTRIBUTING.md) for development setup and guidelines.\n\n## License\n\nMIT — see [LICENSE](LICENSE) for details.\n"}