Tax-doc review agent

A production-style LangGraph pipeline that extracts invoice fields, validates them deterministically, pauses for human approval and resumes from a durable Redis checkpoint.

How it works

Problem

Invoices are keyed in by hand, and any slip in the NIP or the VAT amount comes back as a correction. Validation through a language model leaks: the same document passes one run and fails the next, and on a process restart all state is gone. What was needed was a pipeline that pulls the fields, checks them with rules, not a model, and can halt in place for a human to look, then resume exactly where it stopped.

Solution

A LangGraph pipeline with deterministic validation: field extraction through Pydantic, NIP and VAT maths through an MCP server, and the human decision point through a dynamic interrupt(). The checkpoint lives in Redis, so after a process restart the same thread resumes from where it stopped. Approval arrives over REST (POST /review with edits), and only then does the document reach the ledger.

What it does

Invoice fields extracted through Pydantic structured output
NIP and VAT maths validated through an MCP server (nip_check, vat_math_check)
Dynamic interrupt(): pauses only on low confidence, a failed rule, or a high amount
Resume after a process restart from the Redis checkpoint
Human decision over REST: POST /review
Hermetic tests on FakeLLM, Docker, mypy, Ruff

How it's built

FastAPI → DocumentService → LangGraph StateGraph
Nodes: ingest → extract → validate → route → human_review → finalize
Validation runs async through an MCP client to a stdio server
RedisSaver writes the checkpoint per thread_id
Bearer auth on the REST API, correct 4xx responses
Docker Compose (Redis 8), Dockerfile, Makefile, GitHub Actions

REST HITL session

# example session: commands verbatim from README; output reconstructed
$ curl -X POST http://localhost:8000/documents \
    -H 'Authorization: Bearer dev-token' -H 'Content-Type: application/json' \
    -d '{"content":{"vendor":"Acme Supplies","nip":"5260250995","amount_net":"2000.00","vat_rate":"23","amount_gross":"2460.00","date":"2025-06-15","category":"office","confidence":0.42}}'

# -> 202 Accepted
# { "thread_id": "a1b2c3", "status": "needs_review",
#   "interrupt": { "field": "category", "reason": "low confidence (0.42)",
#                  "current": "office", "allowed": ["professional_services"] } }

$ curl http://localhost:8000/documents/a1b2c3 -H 'Authorization: Bearer dev-token'
# -> 200 OK   (pending interrupt payload still present after a process restart)

$ curl -X POST http://localhost:8000/documents/a1b2c3/review \
    -H 'Authorization: Bearer dev-token' -H 'Content-Type: application/json' \
    -d '{"approved":true,"edits":{"category":"professional_services"}}'

# -> 200 OK
# { "status": "posted",
#   "ledger_entry": { "entry_id": "ledger-a1b2c3", "vendor": "Acme Supplies",
#                     "amount_gross": "2460.00", "category": "professional_services" } }

What it changed

≥15 hermetic pytest-asyncio tests on FakeLLM (CI with no LLM access)
Redis 8 as the checkpointer: resume after a process restart with no state lost
Docker Compose (Redis 8), Dockerfile, Makefile, Ruff, mypy, GitHub Actions in the pipeline

Stack

Python
LangGraph
Redis checkpointer
interrupt()/HITL
MCP server
FastAPI
Pydantic
pytest-asyncio
Docker
GitHub Actions

Story

InMemorySaver dies when the process exits; RedisSaver keeps checkpoints by thread_id, so a different graph object resumes the same document after a restart. Dynamic interrupt() was used because review is conditional: interrupt_before would pause every run and carries no data-specific JSON payload. Finalization uses a deterministic ledger-{thread_id} key, so a retried final step is safe to reconcile.

Repo ↗