ragent

<div align="center"> # 🧠 ragent ### *A guardrail-first policy assistant that knows when to give up and call a human* 🙋‍♂️ *Multi-turn LangGraph agent · hybrid RAG · frustration-aware escalation · async ingestion pipeline*        </div> --- ## 👋 What even is this thing? `ragent` answers "what does the policy say about X?" questions over your own PDFs/DOCX policy docs — but it's built around two questions most weekend-project RAG bots don't bother asking: 1. 🧯 **"Is the model actually making this up?"** → input/output guardrails + a lexical grounding check + an LLM judge + RAGAS, so hallucinations get caught instead of shipped. 2. 😤 **"Is the user about to lose their mind?"** → a cumulative, regex-scored frustration meter that auto-escalates to a human the moment things cross a threshold — no vibes-based LLM judgment call required. It's exposed three ways off the *same* agent graph: | Surface | Use it for | |---|---| | 🌐 **FastAPI** (`/api/v1/ask`, `/ask/stream`, `/ingest`) | Talking to it over HTTP, streaming tokens via SSE | | 🔌 **MCP server** (`src/mcp_server.py`) | Plugging policy search into Claude Desktop or any MCP client | | 🐍 **Direct Python** (`ask()`, `ask_with_session()`) | Embedding the agent straight into another service | Built as a personal deep-dive into LangGraph state machines, hybrid retrieval, and "what does production-grade actually require" — so expect a few 🚧 rough edges alongside the parts that are genuinely solid. --- ## 🗺️ Table of Contents - [🏗️ Architecture](#️-architecture) - [🔁 Diagram Zoo (learning reference)](#-diagram-zoo-learning-reference) - [Request lifecycle — sync `/ask`](#1-request-lifecycle--sync-ask) - [Async ingestion via Celery + Redis](#2-async-ingestion-via-celery--redis) - [Frustration score state machine](#3-frustration-score-state-machine) - [Middleware pipeline](#4-middleware-pipeline) - [🧰 Tech Stack & Why](#-tech-stack--why) - [💡 Core Design Decisions](#-core-design-decisions) - [📁 Module Map](#-module-map) - [⚙️ Configuration Reference](#️-configuration-reference) --- ## 🏗️ Architecture ```mermaid flowchart TB subgraph Clients["🚪 Entry Points"] A1["FastAPI\n/ask, /ask/stream"] A2["MCP Server (stdio)\nsearch_policies / list_policy_documents"] A3["Direct Python API\nask() / ask_with_session()"] end subgraph MW["🛡️ HTTP Middleware Stack"] RID["RequestID"] --> RL["RateLimit\n(sliding window / IP)"] --> CORS["CORS"] --> LOG["Structured JSON logging"] end subgraph Core["🧠 Master Agent — LangGraph StateGraph"] direction TB IG["input_guard\nPII + prompt-injection filter"] FC["frustration_check\nregex signal scan + score update"] RT["router\nLLM structured-output classifier"] PA["policy_agent (sub-graph)"] OOS["out_of_scope_handler"] OG["output_guard\nPII redaction + grounding check"] HE["human_escalation 🆘\nticket stub"] end subgraph SubGraph["📚 policy_agent sub-graph"] direction TB TQ["transform_query"] --> RD["retrieve"] --> GN["generate\n(ChatAnthropic)"] end subgraph RAGLayer["🔍 RAG Layer"] VS[("Chroma\npersistent + singleton")] EMB["Embeddings\nHF BGE (local) or OpenAI"] BM["BM25Retriever\nkeyword search"] ENS["EnsembleRetriever\n0.6 semantic / 0.4 BM25"] end subgraph AsyncPipe["⚙️ Async Ingestion Pipeline"] REDIS[("Redis\nbroker + result backend")] WORKER["Celery worker\nload → chunk → embed → store"] end A1 --> MW --> IG A3 --> IG A2 -.direct retriever call, skips guardrails.-> ENS IG -->|blocked| END1(["END"]) IG --> FC FC -->|score ≥ 0.80| HE FC -->|below threshold| RT RT -->|intent=policy| PA RT -->|intent=out_of_scope| OOS PA --> OG --> END2(["END"]) OOS --> END2 HE --> END2 PA -.contains.-> SubGraph RD --> ENS --> VS ENS --> BM VS --> EMB A1 -."POST /ingest".-> REDIS --> WORKER --> VS CKPT[("SqliteSaver\n(fallback: MemorySaver)")] Core -. persists frustration_score,\nescalation_requested, per session_id .- CKPT ``` --- ## 🔁 Design Zoo A handful of extra diagrams purely so future-me (or anyone poking around this repo) can see *how* each moving part behaves, not just that it exists. ### 1. Request lifecycle — sync `/ask` ```mermaid sequenceDiagram participant C as Client participant MW as Middleware participant API as FastAPI route participant SEM as asyncio.Semaphore participant G as master_agent (LangGraph) participant IG as input_guard participant FR as frustration_check participant RT as router (LLM) participant PA as policy_agent participant VS as Chroma participant LLM as ChatAnthropic participant OG as output_guard C->>MW: POST /ask {query, session_id?} MW->>MW: assign request_id, check rate limit MW->>API: forward request API->>SEM: acquire (max_concurrent_llm_calls) API->>G: ainvoke(turn_input, thread_id=session_id)\nwrapped in asyncio.wait_for(timeout) G->>IG: check_input(state) alt PII or injection detected IG-->>G: error + canned final_response G-->>API: return blocked response else clean IG-->>G: sanitized_query G->>FR: detect_frustration_signals + update score alt score ≥ 0.80 FR-->>G: escalation_requested G->>G: human_escalation_node 🆘 G-->>API: ticket-reference message else FR-->>G: continue G->>RT: classify(sanitized_query) RT-->>G: intent (policy | out_of_scope) alt intent == policy G->>PA: run sub-graph PA->>VS: retriever.invoke(query) VS-->>PA: top-k Document chunks PA->>LLM: generate(context + question) LLM-->>PA: generated_response PA-->>G: state update G->>OG: check_output(state) OG-->>G: final_response + guardrail_flags else out_of_scope G->>G: out_of_scope_handler_node end G-->>API: AskResponse end end API->>SEM: release API-->>C: {response, guardrail_flags, sources?, session_id} ``` The **streaming** twin (`/ask/stream`) runs the identical graph via `astream_events(..., version="v2")`, filters for `on_chat_model_stream` events scoped to the `generate` node only (so router/judge tokens never leak to the client), and emits SSE with a final `done: true` frame. ### 2. Async ingestion via Celery + Redis Swapped `BackgroundTasks` for a real queue. 🎉 ```mermaid sequenceDiagram participant U as You participant API as POST /ingest participant R as Redis (broker) participant W as Celery worker participant VS as Chroma U->>API: {"directory": "data/policies", "clear": false} API->>R: ingest_documents.delay(dir, clear) R-->>API: task_id API-->>U: 202 Accepted {task_id, status: "queued"} Note over W,R: worker polls Redis, picks up the task R->>W: dequeue ingest_documents W->>W: update_state(step="clearing", pct=0) opt clear=true W->>VS: delete_collection() end W->>W: update_state(step="loading", pct=20) W->>W: load_directory() → chunk with RecursiveCharacterTextSplitter W->>W: update_state(step="embedding", pct=50) W->>VS: add_documents(chunks) W->>W: update_state(step="done", pct=100) W-->>R: result {chunks_written, cleared} U->>API: GET /ingest/status/{task_id} API->>R: AsyncResult(task_id).state / .info R-->>API: {state, pct, step, chunks_written} API-->>U: IngestStatusResponse ``` **Why this over `BackgroundTasks`:** crash-safe redelivery (`task_acks_late=True`), automatic retry with exponential backoff on transient I/O errors, real progress percentages instead of "check the logs," and the ability to add more worker processes without touching the API process at all. ### 3. Frustration score state machine The escalation logic, drawn as what it actually is — a one-way ratchet with a trapdoor. 😅 ```mermaid stateDiagram-v2 [*] --> Calm: session starts (score = 0.0) Calm --> Mildly_Annoyed: score += weight\n(e.g. "that's wrong" = +0.15) Mildly_Annoyed --> Frustrated: score += weight\n(e.g. "this is ridiculous" = +0.30) Frustrated --> Escalated: score ≥ 0.80 🆘 Mildly_Annoyed --> Calm: no new signals\n(score never decreases,\nbut also never auto-cools) Calm --> Escalated: single strong signal\n(e.g. "connect me to a human" = +0.40,\ncan combine with prior turns) Escalated --> [*]: human_escalation_node\nticket issued, graph ends turn note right of Escalated Score is cumulative across the whole session (persisted via SqliteSaver, keyed by session_id). It only ever goes up within a turn — there's no decay mechanism today. end note ``` ### 4. Middleware pipeline Every request runs this gauntlet before it ever touches the agent graph: ```mermaid flowchart LR Req(["Incoming request"]) --> RID["🏷️ RequestIDMiddleware\nassigns/echoes X-Request-ID"] RID --> RLM["🚦 RateLimitMiddleware\nsliding window per client IP"] RLM -->|"limit exceeded"| R429(["429 + Retry-After"]) RLM -->|"under limit"| CORS["🌍 CORSMiddleware"] CORS --> TIME["⏱️ Request timing log\nmethod, path, status, duration_ms"] TIME --> ROUTE["Route handler\n(/ask, /ingest, /health...)"] ROUTE --> JSONLOG["📋 JSONLogFormatter\nstructured stdout for log aggregators"] ``` ### 5. Concurrency control — the semaphore dance Per-worker throttling so one uvicorn process doesn't fire 200 concurrent LLM calls at Anthropic: ```mermaid flowchart TB subgraph Worker["Single uvicorn worker process"] SEM["asyncio.Semaphore(max_concurrent_llm_calls)"] R1["Request A"] -->|acquire| SEM R2["Request B"] -->|acquire| SEM R3["Request C"] -.waits.-> SEM SEM --> LLMCall["ChatAnthropic.ainvoke()"] end Note["Total concurrency across the fleet =\nmax_concurrent_llm_calls × num_workers\n(the semaphore is per-process, not global)"] ``` --- ## 🧰 Tech Stack & Why | Layer | Choice | Why 🤔 | |---|---|---| | **Orchestration** | LangGraph `StateGraph` | Explicit, inspectable state machine instead of an implicit agent loop — routing/escalation logic is testable as plain functions independent of any LLM call. | | **LLM** | Claude via `langchain-anthropic` | One provider client reused for generation, routing, scope detection, and judging — swappable per-node by model name. | | **Web framework** | FastAPI + Uvicorn | Native async matches LangGraph's async invocation path; free OpenAPI docs at `/docs`; first-class SSE for token streaming. | | **Vector store** | Chroma, singleton + threading lock | Zero-ops embedded vector DB; singleton avoids N embedding-model loads and SQLite "database is locked" errors under concurrent requests. | | **Embeddings** | HuggingFace BGE (`all-MiniLM-L6-v2`) locally, OpenAI opt-in | Local by default = no second paid API + no extra network hop on every retrieval. | | **Keyword retrieval** | `rank-bm25` + custom `BM25Retriever` | Lexical fallback for exact-term queries dense embeddings under-rank. | | **Hybrid fusion** | `EnsembleRetriever`, weights `[0.6, 0.4]` | RRF-style fusion of semantic + keyword without hand-rolling it. | | **Doc parsing** | `unstructured[pdf,docx]` | Handles messy real-world PDFs/DOCX (tables, headers) better than naive text extraction. | | **Chunking** | `RecursiveCharacterTextSplitter` | Splits on paragraph → line → sentence → word, preserving semantic units. | | **Structured LLM output** | Pydantic + `.with_structured_output()` | Router intent, scope decision, judge scores are typed models — no hand-parsed JSON. | | **Multi-turn state** | `SqliteSaver` checkpointer (fallback `MemorySaver`), keyed by `thread_id=session_id` | Frustration score + escalation flag survive a server restart now. | | **Async task queue** | Celery + Redis | Crash-safe, retryable, horizontally-scalable ingestion — see the [ingestion diagram](#2-async-ingestion-via-celery--redis). | | **CLI** | Typer + Rich | `ingest --file/-f` or `--dir/-d`, with `--dry-run` and a pretty summary table — nicer than raw `argparse` prints. | | **Frustration detection** | Hand-written weighted regex bank, no LLM call | Deterministic, sub-millisecond, zero-cost per turn — this signal gates on *every single message*, so an LLM call here would be wasteful. | | **Evaluation** | Custom LLM-judge + RAGAS | Two independent eval paths for faithfulness/relevance/completeness/hallucination and standardized retrieval metrics. | | **Observability** | LangSmith (opt-in) + structured JSON logs + request IDs | Tracing when you want it, greppable/aggregatable logs always. | | **Rate limiting** | In-memory sliding-window middleware | Simple per-process throttle; documented TODO to move to Redis `INCR` so multi-worker deployments share one limit. | | **Tool exposure** | MCP server (stdio) | Any MCP client (Claude Desktop, etc.) can call `search_policies`/`list_policy_documents` directly — deliberately narrower than the guarded HTTP path. | | **Packaging** | `pyproject.toml` optional-dependency groups (`api`, `mcp`, `eval`, `openai`, `dev`) | Install only what you need. | | **Linting/typing** | `ruff` + `mypy --strict` | Fast lint/import-sort + strict typing on a `TypedDict`-heavy state model. | --- ## 💡 Core Design Decisions ### 🛡️ Guardrails live *in* the graph, not in HTTP middleware `input_guard` and `output_guard` are LangGraph nodes, so the same checks apply whether you hit the agent over HTTP, embed it directly in Python, or (partially) via MCP. Trade-off: a blocked request still costs one node execution — there's no free "reject before graph starts" path. ### 😤 Frustration is cumulative and one-directional `update_frustration_score()` never subtracts — once you're annoyed, the score only climbs within the session (see the [state machine](#3-frustration-score-state-machine)). Crossing `ESCALATION_THRESHOLD = 0.80` routes to `human_escalation_node` **regardless of what the router would've decided** — a deterministic, regex-computed circuit breaker sitting on top of an otherwise LLM-driven pipeline. ### 🧭 Structured routing that fails open `router_node` classifies intent via `.with_structured_output(RouterDecision)`. If the router call throws, the code **defaults to `"policy"`** rather than failing the whole request — availability over strict scope enforcement, by explicit design. ### 🥈 Hybrid retrieval is opt-in `build_retriever()` only builds the BM25 + semantic ensemble when `use_bm25=True` **and** a corpus is supplied. It's there and ready, but wiring a corpus into `retrieve_node` at call time is what flips it on. ### 🧵 Output grounding via cheap lexical overlap, not embeddings `_is_grounded()` tokenizes retrieved context + generated response (words >4 chars), and flags low grounding if a >20-word response shares <3 words with the retrieved vocabulary. Deliberately lightweight — a full embedding-similarity check would cost an extra model call on *every* response. ### 📦 Celery + Redis over fire-and-forget background tasks Documented directly in `ingest.py`'s docstring — the switch bought crash recovery, real status polling, automatic retry with backoff, horizontal scaling, and process isolation (the old approach shared the API's event loop; the new one runs in a totally separate worker process). ### 🔒 Chroma as a locked singleton `get_vectorstore()` uses a double-checked lock so 20 concurrent requests share one Chroma client instead of spinning up 20 (20x the embedding-model RAM, 20 SQLite file handles, guaranteed "database is locked" errors). ### 🚦 Per-process semaphore + timeout around every LLM call `/ask` acquires an `asyncio.Semaphore(max_concurrent_llm_calls)` before invoking the graph, wrapped in `asyncio.wait_for(agent_timeout_seconds)`. The semaphore is explicitly **per-worker-process** (documented in `dependencies.py`) — total fleet concurrency is `max_concurrent_llm_calls × num_workers`, so the per-worker value should be tuned against your actual Anthropic rate limit. ### 🩺 Two-tier health checking `/health` is a pure liveness probe for pod-restart decisions. `/status` actually opens the Chroma collection and reports `degraded` (not a 5xx) if it can't — because the service can still limp along answering from parametric knowledge even with retrieval down. ### 🔌 MCP server as a narrower surface `mcp_server.py` calls the retriever directly, skipping guardrails and frustration tracking entirely — MCP tools are meant to hand raw context to an *external* LLM's own reasoning loop, not re-expose the fully-guarded conversational agent. --- ## 📁 Module Map ``` ragent/ ├── main.py # uvicorn entrypoint ├── scripts/ │ └── ingest_data.py # 🖥️ Typer CLI: --file/--dir, --clear, --dry-run, Rich summary table └── src/ ├── agents/ │ ├── state.py # AgentState TypedDict — shared schema across every graph node │ ├── master_agent.py # top-level StateGraph: guardrails, frustration, routing, escalation │ ├── policy_agent.py # sub-graph: transform_query → retrieve → generate │ └── __init__.py ├── api/ │ ├── app.py # FastAPI factory (lifespan, middleware wiring, routers) │ ├── middleware.py # 🆕 RequestID, RateLimit (sliding window), JSON log formatter │ ├── dependencies.py # verify_api_key, get_llm_semaphore, get_request_id │ ├── schema.py # Pydantic request/response models │ └── routes/ │ ├── query.py # POST /ask, POST /ask/stream (SSE) — semaphore + timeout wrapped │ ├── ingest.py # POST /ingest (Celery), GET /ingest/status/{task_id} │ ├── ingest_bgtask.py # 🗄️ archived: the old BackgroundTasks version, kept for reference │ └── health.py # GET /health (liveness), GET /status (readiness) ├── worker/ │ ├── celery_app.py # 🆕 Celery config: JSON serialization, late acks, single-prefetch │ └── tasks.py # 🆕 ingest_documents task — retry w/ backoff, progress reporting ├── guardrails/ │ ├── input_guard.py # PII (CC/SSN) + prompt-injection regex filter │ ├── output_guard.py # PII redaction + lexical grounding heuristic │ ├── frustration_detector.py # weighted regex signal bank + cumulative score │ └── scope_detector.py # standalone LLM in/out-of-scope classifier ├── rag/ │ ├── document_processor.py # UnstructuredFileLoader + RecursiveCharacterTextSplitter │ ├── embeddings.py # local (HF BGE) or OpenAI embedding provider, lru_cached │ ├── vectorstore.py # 🆕 Chroma singleton (thread-locked) + retriever factory │ └── retriver.py # BM25Retriever + EnsembleRetriever hybrid builder ├── evaluation/ │ ├── judge_llm.py # structured-output LLM judge: 4-dimension scoring │ └── ragas_eval.py # RAGAS harness (faithfulness, answer relevance, context precision) ├── observability/ │ └── tracing.py # LangSmith env-var setup, run URL builder ├── mcp_server.py # stdio MCP server: search_policies, list_policy_documents └── config/ └── settings.py # pydantic-settings, .env-backed ``` --- ## ⚙️ Configuration Reference All settings load via `pydantic-settings` from environment variables / a `.env` file (case-insensitive). | Setting | Default | Notes | |---|---|---| | `anthropic_api_key` | `""` | Required for any LLM call to succeed | | `llm_model` | `claude-opus-4-8` | Shared across generation, routing, scope detection, judging | | `llm_max_tokens` / `llm_temperature` | `4096` / `0.1` | Low temperature favors grounded answers over creativity | | `embedding_provider` / `embedding_model` | `local` / `all-MiniLM-L6-v2` | `local` (HF BGE, CPU) or `openai` | | `chroma_presist_dir` / `chroma_collection_name` | `./data/chroma-db` / `policies` | Persistent vector store location | | `retriver_top_k` | `6` | Top-k for retrieval | | `use_bm25` | `False` | Enables hybrid retrieval once a corpus is wired in | | `chunk_size` / `chunk_overlap` | `100` / `200` | Splitter config | | `langsmith_api_key` / `langsmith_project` / `langsmith_tracing` | — / `ragent` / `False` | `langsmith_enabled` gates on both key + flag | | `redis_url` | `redis://localhost:6317` | 🆕 Celery broker + result backend | | `max_concurrent_llm_calls` | `10` | 🆕 Per-worker semaphore size | | `agent_timeout_seconds` | `60` | 🆕 Hard timeout around `master_agent.ainvoke()` | | `rate_limit_per_minute` | `60` | 🆕 Sliding-window limit per client IP | | `cors_origin` | `["*"]` | 🆕 CORS allow-list | | `api_secret_key` | `""` | 🆕 Bearer token for `verify_api_key` — unset = auth disabled | --- <div align="center"> *Built as a hands-on exploration of guardrail-first agent design — issues, PRs, and "hey this regex is wrong" comments welcome.* ✨ </div>