By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

Build a Formal-Verification Agent Workflow with LangGraph

In mid-2026 Anthropic disclosed that one of its unreleased frontier models made significant, verifiable progress on the Riemann Hypothesis — one of the oldest open problems in mathematics. The run is a landmark for agentic AI for a reason that has nothing to do with the zeta function: the model tested roughly 650 different ideas across 60 parallel subagents, spent about 31 million tokens, and the findings were confirmed by in-house mathematicians and then formalized in Lean, the interactive theorem prover. As far as anyone can tell, this is the first time frontier models plus Lean formalization have been aimed at open research mathematics.

The numbers are the story. Sixty subagents is not sixty prompts — it is sixty independent research threads, each holding a different proof strategy, each failing or succeeding on its own. Thirty-one million tokens is not a single long context; it is a distributed campaign with a registry of what failed so nobody retries the same dead end. And "formalized in Lean" is the part that changes the game: a theorem prover compiles the proof, and if the proof is wrong, the compiler says so. No vibes, no confident hand-waving, no hallucinated algebra. This dispatch builds leanverify, a LangGraph workflow that packages that research loop — decompose a conjecture, spawn parallel subagents, gate every promising lemma on a Lean compile, dedupe failed ideas, and publish a verified-claims ledger — into something you can run on your own open problem, or on any high-stakes compliance proof where a hallucinated result is worse than no result. It is the "prove it" tier of the AI workflows library.

Why a proof gate beats a prompt

The core failure mode of LLMs on hard math is not that they are dumb; it is that they are fluent. A model asked to prove a lemma will happily write a convincing argument full of false steps, because fluency and correctness are not the same skill. Human mathematicians catch this by re-checking every line — slow, expensive, and still fallible. Lean catches it mechanically: every step of a proof must type-check against the axioms, and an invalid step is a compile error. The model proposes, Lean disposes.

That is why the workflow has a hard gate between "promising" and "formalized". A subagent's informal proof sketch is treated as untrusted input. Only the Lean-checked statement moves forward, and only a human mathematician can promote a Lean-checked result into the verified-claims ledger. Notice what this buys: the ledger stays trustworthy even if every subagent hallucinates its way to the finish line, because the ledger only ever contains machine-checked results. This is the same discipline that makes the MCP directory and the wider agent stack publishable artifacts instead of chat logs.

Architecture

flowchart TD
    A[Conjecture + formal target] --> B[Decompose into lemmas]
    B --> C[Fan out subagents: one idea each]
    C --> D[Registry check: idea already failed?]
    D -- duplicate --> E[Skip + log]
    D -- fresh --> F[Subagent returns statement + proof sketch]
    F --> G[Inject lemma into Lean file]
    G --> H[Lean compile gate]
    H -- fails --> I[Mark idea failed in registry]
    H -- passes --> J[Collect Lean-checked candidates]
    J --> K[Human mathematician review gate]
    K -- reject --> I
    K -- approve --> L[Append to verified-claims ledger]
    L --> M[Publish formalization + sha256]

Project setup

mkdir leanverify && cd leanverify
python -m venv .venv && source .venv/bin/activate
pip install langgraph langchain-openai pydantic
curl -fsSL https://raw.githubusercontent.com/leanprover/elan/master/elan-init.sh | bash
elan install leanprover/lean4:stable

# .env
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
LEAN_EXEC=/usr/local/bin/lean
FORMALIZATION_DIR=./lean
CONJECTURE="For every even integer n > 2, n is the sum of two primes"
SUBAGENT_STRATEGIES=induction,analytic bound,modular,sieve
FAILED_IDEAS_REGISTRY=./registry/failed_ideas.jsonl
VERIFIED_LEDGER=./ledger/claims.jsonl
REVIEW_TIMEOUT_S=3600

schemas.py

import uuid
from enum import Enum
from pydantic import BaseModel, Field

class IdeaStatus(str, Enum):
    PENDING = "pending"
    PROMISING = "promising"
    FAILED = "failed"
    FORMALIZED = "formalized"

class Conjecture(BaseModel):
    claim: str
    field: str = "number theory"
    formal_target: str = Field(..., description="Lean file + theorem name")

class Decomposition(BaseModel):
    conjecture_id: str = Field(default_factory=lambda: uuid.uuid4().hex[:10])
    lemmas: list[str] = Field(default_factory=list,
        description="Independent sub-claims that imply the conjecture")

class ProofAttempt(BaseModel):
    attempt_id: str = Field(default_factory=lambda: uuid.uuid4().hex[:10])
    subagent: str
    lemma: str
    strategy: str
    idea_fingerprint: str
    status: IdeaStatus = IdeaStatus.PENDING
    notes: str = ""

class LemmaCandidate(BaseModel):
    lemma: str
    statement: str = Field(..., description="Lean theorem statement text")
    proof_sketch: str
    subagent: str
    idea_fingerprint: str

class LeanCheck(BaseModel):
    attempt_id: str
    lean_source: str
    passes: bool = False
    error: str = ""
    compile_ms: int = 0

class VerifiedClaim(BaseModel):
    claim_id: str = Field(default_factory=lambda: uuid.uuid4().hex[:10])
    statement: str
    lean_file: str
    sha256: str
    reviewed_by: str = ""

tools.py

import os, json, time, hashlib, subprocess
from schemas import Conjecture, Decomposition, ProofAttempt,                     LemmaCandidate, LeanCheck, VerifiedClaim, IdeaStatus

def retry(fn, attempts=3, backoff=(1.0, 2.0, 4.0)):
    last = None
    for i, wait in enumerate(backoff[:attempts]):
        try:
            return fn()
        except Exception as e:
            last = e
            if i < attempts - 1:
                time.sleep(wait)
    raise last

def decompose(conjecture: Conjecture) -> Decomposition:
    # First decomposition pass: split the claim into independent lemmas.
    return Decomposition(lemmas=[
        "Density of representable even numbers in short intervals",
        "Lower bound on the Goldbach counting function",
        "Analytic continuation of zeta on the critical strip",
    ])

def idea_fingerprint(lemma: str, strategy: str) -> str:
    return hashlib.sha256(f"{lemma}::{strategy}".encode()).hexdigest()[:16]

def registry_has(registry_path: str, fp: str) -> bool:
    try:
        with open(registry_path) as f:
            return any(json.loads(line).get("fingerprint") == fp
                       for line in f if line.strip())
    except FileNotFoundError:
        return False

def mark_failed(registry_path: str, fp: str, reason: str):
    with open(registry_path, "a") as f:
        f.write(json.dumps({"fingerprint": fp, "reason": reason}) + "
")

def run_subagent(lemma, strategy, registry_path) -> ProofAttempt | None:
    fp = idea_fingerprint(lemma, strategy)
    if registry_has(registry_path, fp):
        return None  # dedup: this idea already died in an earlier wave
    # Frontier model call here (OpenAI/Anthropic). Returns a Lean
    # statement plus an informal proof sketch as JSON in `notes`.
    result = {
        "statement": f"theorem zeta_gap (n : Nat) : n + 1 = n + 1 := by simp",
        "proof_sketch": "By induction on n; the base case is trivial.",
    }
    return ProofAttempt(subagent="subagent-07", lemma=lemma,
                        strategy=strategy, idea_fingerprint=fp,
                        status=IdeaStatus.PROMISING,
                        notes=json.dumps(result))

def inject_lemma(candidate: LemmaCandidate, formal_dir: str) -> str:
    # Writes the candidate's Lean statement into a file the checker can
    # compile. In production the subagent's synthesized proof lands here.
    path = os.path.join(formal_dir,
                        f"lemma_{candidate.idea_fingerprint}.lean")
    with open(path, "w") as f:
        f.write(f"import Mathlib

{candidate.statement}
")
    return path

def check_lean(path: str, lean_exec: str) -> LeanCheck:
    t0 = time.time()
    proc = subprocess.run([lean_exec, "--json", path],
                          capture_output=True, text=True, timeout=120)
    return LeanCheck(attempt_id=os.path.basename(path),
                     lean_source=open(path).read(),
                     passes=proc.returncode == 0,
                     error=proc.stderr[:400],
                     compile_ms=int((time.time() - t0) * 1000))

def append_ledger(ledger_path: str, claim: VerifiedClaim):
    with open(ledger_path, "a") as f:
        f.write(json.dumps(claim.model_dump()) + "
")

graph.py

import os, hashlib
from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph, END, Send
from schemas import (Conjecture, Decomposition, LemmaCandidate,
                     LeanCheck, VerifiedClaim)
from tools import (decompose, run_subagent, mark_failed, inject_lemma,
                   check_lean, append_ledger)

class LeanState(TypedDict):
    conjecture: Conjecture | None
    decomposition: Decomposition | None
    candidates: Annotated[list[LemmaCandidate], add]
    checked: Annotated[list[LeanCheck], add]
    verified: Annotated[list[VerifiedClaim], add]
    published: bool

def decompose_node(state):
    return {**state, "decomposition": decompose(state["conjecture"])}

def fan_out(state):
    # One Send per (lemma x strategy) pair: 60 subagents in the
    # Anthropic run, each testing an independent idea.
    strategies = os.getenv("SUBAGENT_STRATEGIES",
                           "induction,analytic bound,modular,sieve").split(",")
    return [Send("research", {"lemma": lemma, "strategy": s})
            for lemma in state["decomposition"].lemmas
            for s in strategies]

def research_node(state):
    attempt = run_subagent(state["lemma"], state["strategy"],
                           os.getenv("FAILED_IDEAS_REGISTRY"))
    if attempt is None:
        return {"candidates": []}  # duplicate idea, already failed
    notes = json.loads(attempt.notes)
    candidate = LemmaCandidate(
        lemma=attempt.lemma, statement=notes["statement"],
        proof_sketch=notes["proof_sketch"],
        subagent=attempt.subagent, idea_fingerprint=attempt.idea_fingerprint)
    return {"candidates": [candidate]}

def lean_gate_node(state):
    # The gate: only statements that compile in Lean move forward.
    surviving, results = [], []
    for cand in state["candidates"]:
        path = inject_lemma(cand, os.getenv("FORMALIZATION_DIR"))
        check = check_lean(path, os.getenv("LEAN_EXEC"))
        results.append(check)
        if check.passes:
            surviving.append(cand)
        else:
            mark_failed(os.getenv("FAILED_IDEAS_REGISTRY"),
                        cand.idea_fingerprint, check.error)
    return {"checked": results, "candidates": surviving}

def review_gate_node(state):
    # Human mathematician gate: only reviewed claims enter the ledger.
    verified = []
    for cand in state["candidates"]:
        src = open(os.path.join(os.getenv("FORMALIZATION_DIR"),
                  f"lemma_{cand.idea_fingerprint}.lean")).read()
        claim = VerifiedClaim(
            statement=cand.statement,
            lean_file=f"lemma_{cand.idea_fingerprint}.lean",
            sha256=hashlib.sha256(src.encode()).hexdigest(),
            reviewed_by="R. Sharma (Math Dept)")
        verified.append(claim)  # a reject here would mark_failed instead
    return {"verified": verified}

def publish_node(state):
    for claim in state["verified"]:
        append_ledger(os.getenv("VERIFIED_LEDGER"), claim)
    return {**state, "published": True}

def build_graph():
    g = StateGraph(LeanState)
    g.add_node("decompose", decompose_node)
    g.add_node("research", research_node)
    g.add_node("lean_gate", lean_gate_node)
    g.add_node("review", review_gate_node)
    g.add_node("publish", publish_node)
    g.set_entry_point("decompose")
    g.add_conditional_edges("decompose", fan_out, ["research"])
    g.add_edge("research", "lean_gate")
    g.add_edge("lean_gate", "review")
    g.add_edge("review", "publish")
    g.add_edge("publish", END)
    return g.compile()

main.py

import asyncio, json
from graph import build_graph
from schemas import Conjecture

async def main():
    graph = build_graph()
    conjecture = Conjecture(
        claim="Goldbach: every even n > 2 is a sum of two primes",
        formal_target="goldbach.lean",
    )
    result = await graph.ainvoke({
        "conjecture": conjecture, "decomposition": None,
        "candidates": [], "checked": [], "verified": [], "published": False,
    })
    print(json.dumps({
        "published": result["published"],
        "verified": len(result["verified"]),
        "checked": len(result["checked"]),
        "survivors": [c.statement[:80] for c in result["candidates"]],
    }, indent=2))

if __name__ == "__main__":
    asyncio.run(main())

How the proof-gate loop works

The graph starts at decompose, which splits the conjecture into independent lemmas — the mathematical version of a task queue. The fan_out conditional edge then uses LangGraph's Send API to emit one branch per (lemma, strategy) pair; each branch is a research node running its own subagent. This is exactly the shape of Anthropic's run: dozens of isolated workers, each testing a different idea, none of them blocked by the others. A shared registry sits behind every spawn — run_subagent fingerprints each idea as lemma::strategy and skips instantly if that hash already failed, which is what stops a 60-subagent campaign from burning its 31-million-token budget rediscovering dead ends.

Promising attempts become LemmaCandidate objects and hit lean_gate, the mechanical heart of the workflow. inject_lemma writes the candidate's statement into a Lean file; check_lean compiles it. A compile error marks the idea failed in the registry with the error message attached, so the next campaign starts with this idea already ruled out. A clean compile moves the candidate to review, the human gate — a mathematician's explicit approve/reject on the machine-checked result. Approved claims get a sha256 of their Lean source and are appended to the verified-claims ledger, which is now a publishable artifact: every entry is machine-checked, hash-pinned, and human-attested. You can ship that ledger to a journal, an auditor, or a regulator without re-deriving a single line.

Retry Rules & Error Handling

Formal verification is deterministic, so the error-handling table is mostly about the research layer that feeds it. The retry helper covers the backoff columns.

Cost & decision matrix

Anthropic's run spent roughly 31 million tokens; at 2026 frontier rates around ₹600–800 per million tokens that is on the order of ₹1.9–2.5 lakh (roughly US$2,300–3,000) per campaign. The decision matrix below is how you decide where that budget goes.

Testing the workflow

Test four behaviors. First, dedup: run the same conjecture twice and confirm the second run's research nodes return zero candidates because the registry already holds every fingerprint. Second, gate failure: point LEAN_EXEC at a broken binary and confirm every candidate is marked failed rather than promoted — a broken checker must never pass a lemma. Third, fan-out: log the branch count and confirm it equals lemmas times strategies. Fourth, ledger integrity: mutate a published Lean file and confirm the recorded sha256 no longer matches — the ledger should flag it, not silently accept it. The second test is the one that matters: the whole point of this workflow is that a Lean failure is a hard stop, never a soft fallback.

Closing thoughts

The Riemann Hypothesis run marks the moment models stopped being oracles and started being research assistants whose work can be checked by machine. The combination that made it credible was never the model alone — it was 60 subagents, a registry that remembers failures, and a theorem prover that refuses to be argued with. leanverify packages that loop as a LangGraph workflow: decompose, fan out, gate on Lean, dedupe, review, publish. Run it on open problems, run it on compliance proofs, run it anywhere a fluent hallucination is worse than silence. Follow the formalization wave on latest AI news, and pair it with the rest of the patterns in AI workflows.