Adding Human Review to LangGraph Workflows

LangGraph makes it straightforward to build stateful, multi-step AI agents. One thing it doesn’t provide out of the box is a clean way to pause a running graph and wait for a human decision — especially when that decision needs to happen asynchronously, outside the process.

Relayna fills that gap. In this tutorial we’ll walk through a complete, runnable example: a PDF invoice approval workflow where an AI agent extracts invoice data, submits it for human review via a magic link, and branches based on the reviewer’s decision.

The full source is on GitHub: github.com/redlin/relayna-examples

What We’re Building

The workflow has four stages:

1. Extract — Claude reads the PDF and pulls out structured invoice data (vendor, amount, line items, due date)
2. Upload — the PDF is uploaded to Relayna’s asset storage
3. Review — a checkpoint is created with the PDF and extracted data attached; the reviewer gets a magic link
4. Branch — on approval, payment is processed; on rejection, the workflow stops; on needs-changes, corrections are applied and a new checkpoint is created

PDF Invoice
    │
    ▼
extract_invoice_data   (GPT-4o reads PDF text → structured JSON)
    │
    ▼
upload_pdf_to_relayna  (PDF → Relayna asset storage)
    │
    ▼
create_review_checkpoint ◄─────────────────────────┐
    │                                               │
    ▼                                               │
poll_for_decision      (blocks until human decides) │
    │                                               │
    ├── approved      → handle_approved → done      │
    ├── rejected      → handle_rejected → done      │
    ├── expired       → handle_expired  → done      │
    └── needs_changes → handle_needs_changes ───────┘

The State

Every LangGraph node reads from and writes to a shared TypedDict. One important note: checkpoint_id is a reserved key in LangGraph (used internally for its own checkpointing system). We use review_checkpoint_id instead.

from typing import Optional
from typing_extensions import TypedDict

class InvoiceState(TypedDict):
    invoice_path: str
    extracted_data: dict
    extraction_error: Optional[str]
    asset_id: Optional[str]
    review_checkpoint_id: Optional[str]   # NOT checkpoint_id — that's reserved by LangGraph
    review_url: Optional[str]
    status: Optional[str]
    decision_comment: Optional[str]
    revision_count: int
    max_revisions: int
    result: Optional[str]

Calling the Relayna API

Relayna exposes a simple REST API. We wrap it in a thin httpx client. The key detail is trust_env=False — without it, httpx picks up any HTTP_PROXY environment variable and routes localhost traffic through it, causing timeouts.

import httpx

class RelaynaClient:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url.rstrip("/")
        self._http = httpx.Client(
            headers={
                "Authorization": f"Bearer {api_key}",
                "Accept": "application/json",
            },
            trust_env=False,  # don't route through system proxy
            timeout=120.0,
        )

    def upload_asset(self, file_path: str) -> str:
        """Upload a PDF and return the asset UUID."""
        with open(file_path, "rb") as f:
            response = self._http.post(
                f"{self.base_url}/api/assets/upload",
                files={"file": (Path(file_path).name, f, "application/pdf")},
                data={"purpose": "invoice", "ttl_seconds": "86400"},
            )
        response.raise_for_status()
        return response.json()["asset"]["id"]

    def create_checkpoint(self, title, instructions, summary, items, **kwargs):
        """Create a review checkpoint and return (id, review_url)."""
        payload = {"title": title, "instructions": instructions,
                   "summary": summary, "items": items, **kwargs}
        response = self._http.post(
            f"{self.base_url}/api/checkpoints", json=payload
        )
        response.raise_for_status()
        data = response.json()
        return data["checkpoint"]["id"], data["review_url"]

    def get_status(self, review_checkpoint_id: str):
        """Poll the lightweight status endpoint."""
        response = self._http.get(
            f"{self.base_url}/api/checkpoints/{review_checkpoint_id}/status"
        )
        response.raise_for_status()
        return response.json()

The Key Nodes

Extracting Invoice Data

We use pypdf to extract text from the PDF, then pass it to GPT-4o. OpenAI doesn’t accept raw PDF bytes, so text extraction is the right approach here.

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from pypdf import PdfReader

def extract_invoice_data(state: InvoiceState) -> dict:
    reader = PdfReader(state["invoice_path"])
    pdf_text = "\n".join(page.extract_text() or "" for page in reader.pages)

    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    response = llm.invoke([HumanMessage(content=(
        "Extract these fields from the invoice and return ONLY valid JSON:\n"
        '{"invoice_number", "vendor_name", "due_date", "currency", "total", "line_items": [...]}\n\n'
        f"Invoice text:\n\n{pdf_text}"
    ))])

    extracted = json.loads(response.content.strip())
    return {"extracted_data": extracted}

Creating the Review Checkpoint

The checkpoint attaches both the original PDF (as an asset item) and the extracted JSON data, so the reviewer can verify the AI’s work side by side.

def create_review_checkpoint(state: InvoiceState) -> dict:
    client = RelaynaClient.from_env()
    extracted = state["extracted_data"]

    review_checkpoint_id, review_url = client.create_checkpoint(
        title=f"Invoice Review: {extracted['vendor_name']} — {extracted['currency']} {extracted['total']}",
        instructions=(
            "Please review this invoice carefully:\n\n"
            "1. **Check the PDF** — verify vendor identity, amounts, and dates.\n"
            "2. **Review extracted data** — confirm the AI-extracted fields are correct.\n"
            "3. Choose: **Approve**, **Reject**, or **Request Changes**.\n\n"
            "Your comment will be returned to the agent."
        ),
        summary=f"Invoice from {extracted['vendor_name']} for {extracted['currency']} {extracted['total']}, due {extracted.get('due_date')}.",
        items=[
            {"item_type": "asset", "asset_id": state["asset_id"], "label": "Invoice PDF", "position": 0},
            {"item_type": "json", "label": "AI-Extracted Data", "content_json": extracted, "position": 1},
        ],
        ttl_seconds=86400,
    )

    print(f"\n  Review URL: {review_url}\n")
    return {"review_checkpoint_id": review_checkpoint_id, "review_url": review_url}

Polling for a Decision

The poll_for_decision node calls the lightweight status endpoint every 15 seconds until the human acts. The terminal statuses are approved, rejected, needs_changes, and expired.

def poll_for_decision(state: InvoiceState) -> dict:
    client = RelaynaClient.from_env()
    terminal = {"approved", "rejected", "needs_changes", "expired", "cancelled"}

    while True:
        data = client.get_status(state["review_checkpoint_id"])
        if data["status"] in terminal:
            return {"status": data["status"], "decision_comment": data.get("decision_comment")}
        time.sleep(15)

Alternatively, set a callback_url when creating the checkpoint and Relayna will POST the decision to your server — no polling needed.

Wiring the Graph

from langgraph.graph import StateGraph, START, END

builder = StateGraph(InvoiceState)

builder.add_node("extract_invoice_data", extract_invoice_data)
builder.add_node("upload_pdf_to_relayna", upload_pdf_to_relayna)
builder.add_node("create_review_checkpoint", create_review_checkpoint)
builder.add_node("poll_for_decision", poll_for_decision)
builder.add_node("handle_approved", handle_approved)
builder.add_node("handle_rejected", handle_rejected)
builder.add_node("handle_needs_changes", handle_needs_changes)
builder.add_node("handle_expired", handle_expired)

builder.add_edge(START, "extract_invoice_data")
builder.add_edge("extract_invoice_data", "upload_pdf_to_relayna")
builder.add_edge("upload_pdf_to_relayna", "create_review_checkpoint")
builder.add_edge("create_review_checkpoint", "poll_for_decision")

builder.add_conditional_edges(
    "poll_for_decision",
    lambda s: s["status"],
    {
        "approved": "handle_approved",
        "rejected": "handle_rejected",
        "needs_changes": "handle_needs_changes",
        "expired": "handle_expired",
    },
)

builder.add_edge("handle_approved", END)
builder.add_edge("handle_rejected", END)
builder.add_edge("handle_expired", END)

# needs_changes loops back for another review round
builder.add_conditional_edges(
    "handle_needs_changes",
    lambda s: "create_review_checkpoint" if s["revision_count"] < s["max_revisions"] else "handle_rejected",
    {"create_review_checkpoint": "create_review_checkpoint", "handle_rejected": "handle_rejected"},
)

graph = builder.compile()

Running It

git clone https://github.com/redlin/relayna-examples.git
cd relayna-examples/langgraph-invoice-review

uv venv && uv pip install -e .
source .venv/bin/activate

cp .env.example .env
# Fill in RELAYNA_API_KEY and OPENAI_API_KEY

python scripts/generate_invoice.py   # creates a sample invoice.pdf
python main.py --invoice invoice.pdf

The agent prints a review URL. Open it — no login required. Approve, reject, or request changes. The agent detects your decision and completes the workflow.

What the Reviewer Sees

Relayna renders both items side by side in the browser: the original PDF and the structured JSON extracted by the AI. The reviewer can read the actual invoice, check it against the extracted data, and leave a comment with their decision.

When the reviewer requests changes, the handle_needs_changes node uses the LLM to merge their comment into the extracted data, then creates a fresh checkpoint for a second round of review.

Try It Yourself

The complete, runnable example is at github.com/redlin/relayna-examples. It includes a sample invoice PDF generator so you don’t need a real invoice to get started.