7dayrag

Production-oriented RAG + AI-agent workflow exposed as a FastAPI service. Built as the reference implementation for a 7-day SaaS AI engagement — grounded Q&A over business data with citations, refusal guardrails, and a tool-using agent that calls internal APIs.

See ARCHITECTURE.md for design rationale and the day-by-day delivery plan.

Quick start (zero API keys required)

The app runs fully offline in stub mode (deterministic pseudo-embeddings + scripted LLM). Add real keys later to switch to OpenAI/Anthropic with automatic failover.

# 1. Postgres + pgvector
docker compose up -d db

# 2. Python deps
pip install -r requirements.txt

# 3. Configure (or skip: defaults match compose)
copy .env.example .env

# 4. Create schema + load the sample knowledge base
python -m scripts.seed_sample_data

# 5. Serve
uvicorn app.main:app --port 8000 --reload

Try it

# Grounded Q&A with citations
curl -X POST localhost:8000/api/v1/query \
  -H "Content-Type: application/json" \
  -d '{"question": "What is the uptime SLA for Business plans?"}'

# Agent that calls tools (ticket lookup)
curl -X POST localhost:8000/api/v1/agent/run \
  -H "Content-Type: application/json" \
  -d '{"task": "Check ticket TICKET-1001 and summarize its status."}'

# Raw hybrid retrieval (debug/tuning)
curl -X POST localhost:8000/api/v1/documents/search \
  -H "Content-Type: application/json" \
  -d '{"query": "refund window annual plan", "top_n": 3}'

Interactive docs: http://localhost:8000/docs

API

Every response carries an x-request-id; errors are structured {error: {code, message}}.

Configuration

All via environment / .env (see .env.example). Key settings:

• LLM_PROVIDER: openai | anthropic | stub | auto (auto walks PROVIDER_ORDER with per-provider retry + backoff and failover; ends at stub if no keys are set)
• OPENAI_BASE_URL: point at any OpenAI-compatible endpoint (Ollama, vLLM, gateways)
• MIN_VECTOR_SCORE: best-hit cosine floor below which the API refuses instead of guessing
• TICKETS_API_BASE_URL / ACCOUNTS_API_BASE_URL: point agent tools at real internal APIs; blank = built-in sandbox data
• REDIS_URL, CACHE_ENABLED, CACHE_TTL_SECONDS, RATE_LIMIT_PER_MINUTE: caching + rate limiting; a missing Redis only costs performance, never availability

Redis (caching + rate limiting)

Grounded answers are cached (keyed by question + config) and /api/v1/* is rate limited per client IP with a fixed 60s window. Responses carry x-ratelimit-remaining; exceeding the limit returns structured 429. /readyz reports Redis health; the API fails open if Redis is down. Only non-refusal answers are cached (refusals can change as docs update).

docker compose up -d redis   # or just: docker compose up -d  (brings up db+redis+api+n8n)

MCP server

Expose the same capabilities to Claude Desktop or any MCP client:

python mcp_server.py        # stdio transport

Tools: search_knowledge_base, answer_question, run_agent, lookup_ticket, lookup_account. Claude Desktop config snippet:

{
  "mcpServers": {
    "7dayrag": {
      "command": "python",
      "args": ["/absolute/path/to/7dayrag/mcp_server.py"]
    }
  }
}

n8n workflow automation

docker compose up -d n8n → open http://localhost:5678 → import from workflows/:

Workflows call http://api:8000 (compose network). If you run n8n outside Compose, change the base URL to http://localhost:8000.

Test the triage webhook after activating:

curl -X POST localhost:5678/webhook/ticket-triage \
  -H "Content-Type: application/json" -d '{"ticket_id": "TICKET-1001"}'

How grounding works

1. Question is embedded (same model as ingestion) and run through hybrid retrieval: pgvector cosine top-K + Postgres full-text top-K, fused with Reciprocal Rank Fusion.
2. If the best hit's vector score is below MIN_VECTOR_SCORE → refusal (no LLM call).
3. Otherwise the numbered context goes to the model with strict rules: cite as [n], answer only from context, reply NOT_ENOUGH_CONTEXT otherwise.
4. Citations in the answer are mapped back to source documents and returned.

Tests

docker compose up -d db      # integration tests need Postgres on :5433
pytest tests -q              # unit + integration; integration skips cleanly without DB
ruff check app tests scripts

21 tests: chunking invariants, RRF fusion, embedding determinism, stub provider behavior, agent loop parsing, plus end-to-end API round-trips against real Postgres/pgvector.

Deploy (staging)

cp .env.example .env   # add OPENAI_API_KEY
docker compose up -d --build
curl localhost:8000/readyz
curl -X POST localhost:8000/api/v1/admin/seed

For AWS: same images → ECS Fargate + RDS Postgres (enable pgvector extension). For DigitalOcean: droplet + managed Postgres. Secrets via environment/secret manager only.

Project layout

app/
  api/        FastAPI routes (documents, query, agent, health/admin)
  agent/      tool registry (KB search, ticket/account lookup) + bounded agent loop
  llm/        provider abstraction: openai, anthropic, stub + retry/failover router
  rag/        chunking, ingestion, hybrid retrieval (RRF), grounded generation
  cache.py    Redis: response cache + fixed-window rate limiting (fail-open)
  config.py   env-driven settings · db.py engine/session · db_init.py schema bootstrap
data/sample_docs/*.md    demo knowledge base
scripts/seed_sample_data.py
workflows/*.json         importable n8n automations (ticket triage, KB sync)
mcp_server.py            MCP tool server (stdio) for Claude Desktop / MCP clients
tests/

Next steps (post-engagement backlog)

Streaming (SSE), feedback capture into an eval set, reranker stage, multi-tenant RLS, scheduled re-indexing, prompt versioning/A-B, cost dashboards.