Agent frameworksLangGraph

LangGraph

cli_sdk.integrations.langgraph puts a calibrated guard node between the model and the tools of a LangGraph StateGraph. One call, add_cli_guard(builder, guard), adds three nodes after your agent node: cli_guard asks a ToolGuard about every tool call the model proposed, cli_human_review pauses the thread with interrupt() when a call needs a person, and cli_blocked refuses a call and tells the model why. Tools without a rule run as usual, and the guard’s decisions and the reviewer’s answers are saved in the checkpoint.

Everything on this page was run against langgraph 1.2.12, langgraph-prebuilt 1.1.0, langchain-core 1.6.4 and langchain-openai 1.6.6. The two examples at the end run offline, without keys, on synthetic data; the output shown is what they printed.

Why a calibrated guard, not a confidence threshold

A common first guard for an agent that pays insurance claims is a hand-set rule: pay automatically when the model says it is at least 90% sure. That number has no stated meaning. Model confidence is not calibrated, it shifts when the model, the prompt or the claim mix changes, and nobody can say how often a payment that clears it is wrong.

The guard in the claims example below replaces it with a threshold computed from 320 labelled historical payment decisions, and a statement attached to it: “With 90% confidence, the rate of decisions that are auto-approved and wrong is at most 0.05 (RCPS, Hoeffding-Bentkus, n=320).” The threshold came out at 0.854 on the evidence model’s score because of the data, not because someone picked it. The calibration profile records which evidence model and prompt produced it; change either and the profile goes stale, and the guard escalates every payment until you recalibrate. The statement bounds how often payments like the calibration set end up auto-approved and wrong. It does not make any single payment correct, which is why every payment below the threshold goes to an adjuster.

Install

pip install "cci-sdk[langgraph,openai]" langchain-openai

The langgraph extra installs langgraph>=1.0 (with langgraph-prebuilt, which provides ToolNode and tools_condition) and langchain-core>=1.0. The agent model comes from a LangChain chat-model package and the evidence model from a cci-sdk extra. Everything needs Python 3.10 or later.

Agent modelLangChain packageEvidence model extra
OpenAI, Azure OpenAI, Gemma on vLLM or SGLanglangchain-openaicci-sdk[openai]
Anthropic Claudelangchain-anthropiccci-sdk[anthropic], or a log-probability model through --evidence-provider
Google Geminilangchain-google-genaicci-sdk[openai] (Gemini evidence uses Gemini’s OpenAI-compatible endpoint)

How decisions map onto LangGraph

agent --(tool calls)--> cli_guard --allow----> tools ------------------------> agent
                                  --escalate-> cli_human_review --approve/edit--> tools
                                  |                             --reject-------> agent
                                  --block----> cli_blocked ----------------------> agent
Guard decision or reviewer answerWhat the adapter doesWhat LangGraph does
allowcli_guard routes the turn to your tools nodeToolNode runs the calls and their results go back to agent.
escalatecli_guard routes to cli_human_review, which calls interrupt() with a review card: each call’s arguments, reason, guarantee card and evidenceinvoke returns with __interrupt__. The thread is checkpointed at cli_human_review and waits for Command(resume=...).
blockcli_guard routes to cli_blocked, which answers every call of the turn with a ToolMessage(status="error") holding the reason and the guaranteeagent runs again with the refusals in its history; it should explain, not retry.
Reviewer approves: {"action": "approve"}Command(goto="tools")The paused calls run as proposed.
Reviewer edits: {"action": "edit", "args": {...}}Replaces the model’s AIMessage in place (same id, via model_copy) with the edited arguments; "calls": {call_id: {...}} edits one call of severalThe edited calls run. The guard does not check them again.
Reviewer rejects, or sends anything elseOne error ToolMessage per call, carrying the reviewer’s noteagent runs again; the tools never ran.

The guard runs once per model turn, in its own node, so the calibrated evaluation is never repeated when the review node re-runs on resume. When one turn proposes several calls, the most conservative decision applies to the whole turn (block, then escalate, then allow), because ToolNode runs a turn’s calls together. The guard fails closed: a missing or stale profile, an evidence-model error or a context that cannot be built all escalate, and never allow.

A local-mode Gate answers auto_approve or escalate, so a Gate rule never blocks in local mode. Blocks come from Belief, Set and Interval rules with block_below, block_labels or block_levels, from a hosted Gate that abstains, or from on_heuristic="block".

The checkpoint keeps the audit trail in state["cli_guard"]: the turn’s action, one entry per call, and, after a review, a review entry with the outcome (approve, edit or reject), the reviewer’s note and, for an edit, the calls that actually ran. The claims example below prints it.

Set up a guarded graph

Choose the agent model

The agent model only has to call tools well; the guard scores each proposed call with its own evidence model (next step), which can be a different provider. The examples build the agent model with chat_model(provider) from examples/agents/_shared/langchain_models.py, which reads keys from environment variables (see examples/agents/.env.example). While a key is still a placeholder, the script stops with a message before any request is sent.

pip install "cci-sdk[langgraph,openai]" langchain-openai
export OPENAI_API_KEY=...          # your key
export OPENAI_MODEL=gpt-4.1-mini   # default
python examples/agents/langgraph/insurance_claims_graph.py --provider openai

ChatOpenAI(..., use_responses_api=False) keeps the agent on the Chat Completions API. The same gpt-4.1-mini default serves as a log-probability (L1) evidence model; for the evidence model prefer a dated snapshot such as gpt-4.1-mini-2025-04-14, so that a model update never silently changes the scoring function behind a profile.

The factory, from examples/agents/_shared/langchain_models.py:

def chat_model(provider: str, **overrides: Any) -> Any:
    """A LangChain chat model for ``provider`` (not used with ``--provider mock``)."""
    s = settings(provider).require_key()
    if provider == "openai":
        from langchain_openai import ChatOpenAI
 
        return ChatOpenAI(model=s.model, api_key=s.api_key, base_url=s.base_url, temperature=0,
                          use_responses_api=False, **overrides)
    if provider == "azure":
        from langchain_openai import AzureChatOpenAI
 
        return AzureChatOpenAI(azure_endpoint=s.base_url, azure_deployment=s.model, api_version=s.api_version,
                               api_key=s.api_key, temperature=0, **overrides)
    if provider == "anthropic":
        from langchain_anthropic import ChatAnthropic
 
        return ChatAnthropic(model=s.model, api_key=s.api_key, max_tokens=1024, **overrides)
    if provider == "gemini":
        from langchain_google_genai import ChatGoogleGenerativeAI
 
        return ChatGoogleGenerativeAI(model=s.model, api_key=s.api_key, temperature=0, **overrides)
    if provider in ("vllm", "sglang"):
        from langchain_openai import ChatOpenAI
 
        # The model name must equal the name the server was started with.
        return ChatOpenAI(model=s.model, base_url=s.base_url, api_key=s.api_key, temperature=0,
                          use_responses_api=False, **overrides)
    raise SystemExit(f"no LangChain chat model for provider {provider!r}")

The agent node puts the system prompt first on every call instead of storing it in the state, which also suits Gemma 4: its chat template honours a system message only as the first message.

With --provider mock (the default) the agent node runs ScriptedClaimsAgent, LangChain’s own GenericFakeChatModel with two overrides. bind_tools returns the model, because the base class raises NotImplementedError, and the closing note is written from the payment tool’s result. It is built with disable_streaming=True, and a new one per thread, because its script is a one-shot iterator:

class ScriptedClaimsAgent(GenericFakeChatModel):
    """LangChain's fake chat model, scripted per claim for --provider mock.
 
    It proposes the scripted tool calls in order, then writes its closing
    note from the payment tool's result, as a real model would.
    """
 
    def bind_tools(self, tools: Any, *, tool_choice: Any = None, **kwargs: Any) -> "ScriptedClaimsAgent":
        return self  # GenericFakeChatModel.bind_tools raises NotImplementedError
 
    def _generate(self, messages: list[Any], stop: Any = None, run_manager: Any = None, **kwargs: Any) -> ChatResult:
        last = messages[-1] if messages else None
        if isinstance(last, ToolMessage) and last.name == "approve_claim_payment":
            if last.status == "error":
                body = json.loads(last.content)
                text = ("No payment was made: the adjuster declined the proposal. "
                        f"Adjuster note: {body.get('reviewer_note') or body.get('reason')} The claim stays open.")
            else:
                text = f"{last.content} The claim can move to closure review."
            return ChatResult(generations=[ChatGeneration(message=AIMessage(content=text))])
        return super()._generate(messages, stop=stop, run_manager=run_manager, **kwargs)

Choose the evidence model and calibrate

The evidence model answers one closed question about each proposed call, and the guard calibrates its answers on labelled examples. Two access levels matter here (the access ladder has the rest):

  • L1, token log-probabilities. One request per score, at temperature 0, reading the probability of each option’s letter. OpenAI and Azure OpenAI non-reasoning configurations, and Gemma on vLLM or SGLang.
  • L0, sampling. sample_count sampled answers per score (the examples use 8 for Anthropic and Gemini), turned into smoothed frequencies. Claude and Gemini. Scores are coarser, so more calls escalate.

Calibration scores every labelled example once and caches the result as a JSON profile in --store (default .cli_profiles next to the example). For the claims example’s 320 decisions that is 320 requests at L1 or 2,560 at L0, and each payment check afterwards costs 1 or 8. The discharge-summary example scores each of its 1,537 labelled claims, so 1,537 requests at L1 or 12,296 at L0: use an L1 evidence model for Claim. The query, from examples/agents/langgraph/insurance_claims_graph.py:

PAYMENT_GATE = Gate(
    instructions=(
        "Under claims payment guideline CPG-7, is paying the proposed amount on this claim correct? "
        "Answer true only if the policy covers the loss, no referral trigger applies, the required "
        "documents are present, and the amount equals the payable amount."
    ),
    calibration_profile="claim-payments-v1",
    guarantee="risk_high_probability",
    target=0.05,
    delta=0.10,
)

the client:

def build_client(args: argparse.Namespace) -> LocalCLIClient:
    evidence_provider = args.evidence_provider or args.provider
    evidence = providers.evidence_backend(evidence_provider, mock_scorer=payment_evidence_scorer)
    return LocalCLIClient(evidence, store=calibration.default_store(__file__, args.store),
                          sample_count=providers.sample_count(evidence_provider))

and the calibration, in run():

    examples = calibration.load_jsonl(DATASET)
    [profile] = calibration.ensure_calibrated(client, [PAYMENT_GATE], examples, recalibrate=args.recalibrate)

A profile is rebuilt when the query’s instructions change or a different evidence model or configuration is used; until then, a stale profile gives heuristic answers and the guard escalates every payment. See Local mode for sizes, fingerprints and audits.

Wire the guard

The rule names the tool, the query and how to build the evaluation context from the call. Build it from your system of record, not from the agent’s description of the case, and with the same function that built the calibration contexts: the guarantee holds only for contexts scored the way the calibration contexts were (see Exchangeability).

def build_guard(client: LocalCLIClient) -> ToolGuard:
    """One rule: payments are decided by the calibrated Gate; other tools run freely."""
    return ToolGuard(client, [
        GuardRule(
            tool="approve_claim_payment",
            query=PAYMENT_GATE,
            # Same builder as the calibration examples: same keys, same guideline text.
            context=lambda args: claims_data.claim_context(CLAIMS[args["claim_id"]], args["amount"]),
        ),
    ])

claims_data is examples/agents/data/generate_insurance_claims.py, the generator that wrote the calibration data. Its claim_context renders the claim file, the proposed payment and the written guideline the same way for both:

def claim_context(claim: dict[str, Any], amount: float) -> dict[str, Any]:
    """The exact context the payment guard evaluates: claim file, proposed payment, guideline."""
    return {
        "claim": {field: claim.get(field) for field in CLAIM_FIELDS},
        "proposed_payment": {"claim_id": claim["claim_id"], "amount": round(float(amount), 2)},
        "guideline": GUIDELINE,
    }

Then add the guard to the graph. You add START -> agent and tools -> agent; add_cli_guard adds the conditional edges out of agent and the three guard nodes. Compile with a checkpointer: interrupt() needs one.

def build_graph(model: Any, guard: ToolGuard, tools: list[Any], checkpointer: Any = None) -> Any:
    """agent -> cli_guard -> tools | cli_human_review | cli_blocked, with a checkpointer for interrupts."""
    llm = model.bind_tools(tools)
 
    def agent(state: CLIGuardState) -> dict[str, Any]:
        return {"messages": [llm.invoke([SystemMessage(SYSTEM_PROMPT), *state["messages"]])]}
 
    builder = StateGraph(CLIGuardState)
    builder.add_node("agent", agent)
    builder.add_node("tools", ToolNode(tools))
    builder.add_edge(START, "agent")
    builder.add_edge("tools", "agent")
    add_cli_guard(builder, guard)            # agent -> cli_guard -> tools / review / blocked
    return builder.compile(checkpointer=checkpointer or InMemorySaver())

A context that cannot be built, for example for a claim id the claims system does not know, escalates.

Handle reviews and resume

Give each case its own thread_id. When the guard escalates, invoke returns with the thread paused at cli_human_review, and pending_review(result) returns the review card: interrupt_id, kind, question, calls (each the guard’s decision as to_dict()) and resume_with. Send the card to your reviewers, then resume the same thread with their answer. From the claims example:

def process_claim(graph: Any, guard: ToolGuard, claim_id: str, desk: AdjusterDesk,
                  ledger: list[dict[str, Any]], max_reviews: int = 3) -> dict[str, Any]:
    """Run one claim through the graph, answering each review interrupt the guard raises."""
    config = {"configurable": {"thread_id": f"claim-{claim_id}"}}
    before, seen = len(ledger), len(guard.log)
    request = HumanMessage(f"Process claim {claim_id}: look up the file and propose the payment under CPG-7.")
    out = graph.invoke({"messages": [request]}, config)
    decisions = []
    while True:
        for proposal in guard.log[seen:]:
            print(f"    agent proposes: {proposal.tool}({json.dumps(dict(proposal.arguments), sort_keys=True)})")
        display.decisions(guard.log[seen:])
        seen = len(guard.log)
        review = pending_review(out)
        if review is None:
            break
        if len(decisions) == max_reviews:     # a model that keeps re-proposing: leave the thread paused
            print(f"    still paused after {max_reviews} reviews; leaving the thread for the adjusters' queue")
            break
        print(f"    graph paused at {graph.get_state(config).next[0]} (thread {config['configurable']['thread_id']})")
        decisions.append(desk.decide(claim_id, review))
        out = graph.invoke(Command(resume=decisions[-1]), config)
    state = graph.get_state(config).values
    return {
        "case_id": claim_id,
        "audit": state.get("cli_guard"),
        "review": decisions[-1] if decisions else None,
        "executed": ledger[before:],
        "final": _text(state["messages"][-1].content),
    }

desk is the example’s stand-in review queue; --interactive makes you the adjuster. The resume values the review node understands:

Resume valueEffect
{"action": "approve", "note": "..."}The paused calls run as proposed.
{"action": "edit", "args": {"amount": 1000.0}, "note": "..."}Every call runs with these arguments merged in.
{"action": "edit", "calls": {"<call id>": {...}}}Edits one call of a multi-call turn.
{"action": "reject", "note": "..."}Nothing runs; the model gets an error ToolMessage with the note.
anything elseTreated as a reject; the record keeps what was sent as requested.

InMemorySaver keeps threads in process memory only. When a reviewer may answer hours later, or in another process, compile the graph with a durable checkpointer (the SQLite and Postgres savers are separate packages) and resume with the same thread_id.

Example: insurance claim payments

examples/agents/langgraph/insurance_claims_graph.py

A homeowners claims agent looks up a claim file, works out the payable amount under a written, invented claims guideline (CPG-7: policy period, covered perils and exclusions, referral triggers for new policies, fraud indicators and prior claims, required documents, and payable = min(assessed loss, sub-limit, coverage limit) minus the deductible), and proposes approve_claim_payment(claim_id, amount). The guard is a Gate with guarantee="risk_high_probability", target=0.05 and delta=0.10, calibrated with RCPS.

The data is examples/agents/data/insurance_claims.jsonl, 320 synthetic historical decisions written by generate_insurance_claims.py (seeded, standard library only). Each label says whether paying the proposed amount was correct under the guideline; 154 of the 320 were. About 7% of the files carry an adjuster note that makes them a judgment call (an unsigned estimate, low-resolution photos), labelled by the historical adjuster’s call, and about 2% of labels are flipped to model labelling errors. The mock evidence model reads the claim file like a fallible reviewer: it is unsure whether the jewelry sub-limit applies, under-weights the police-report rule, is fooled by gradual leaks, and never sees a label.

git clone https://github.com/rahvis/cci-sdk && cd cci-sdk
python examples/agents/langgraph/insurance_claims_graph.py

The first run calibrates once and prints progress to stderr (calibrating 'claim-payments-v1' on 320 labelled examples ...); later runs reuse .cli_profiles/claim-payments-v1.json and print the same decisions. The output, in mock mode:

==============================================================================
LangGraph: claim payments behind a calibrated guard node (RCPS Gate)
==============================================================================
Synthetic data for demonstration only. Not claims-handling or legal advice.
Keep adjusters in the loop for every decision the guard escalates.

calibration profile claim-payments-v1: RCPS, n=320 labelled decisions (at least 47 needed for target=0.05, delta=0.10), status=serving
evidence model: mock (access level L1)
------------------------------------------------------------------------------
C-3001: burst pipe, complete file, no referral triggers
    agent proposes: approve_claim_payment({"amount": 1350.0, "claim_id": "C-3001"})
    guard: ALLOW     approve_claim_payment -> auto-approved under the calibrated risk bound (confidence 0.982, calibrated threshold 0.854).
           guarantee: With 90% confidence, the rate of decisions that are auto-approved and wrong is at most 0.05 (RCPS, Hoeffding-Bentkus, n=320).
    payment executed: 1,350.00 on C-3001
    agent: Payment of 1,350.00 issued on claim C-3001 (ledger entry 1). The claim can move to closure review.
    audit record in the checkpoint (state["cli_guard"]):
      action: allow
      call call-C-3001-pay: approve_claim_payment({"amount": 1350.0, "claim_id": "C-3001"}) -> allow
        evidence: confidence 0.982, threshold 0.854, Venn-Abers [0.963, 1.000]
        guarantee: risk_high_probability via RCPS, profile claim-payments-v1 (n=320)
------------------------------------------------------------------------------
C-3002: jewelry theft; the agent proposes the full loss and misses the sub-limit
    agent proposes: approve_claim_payment({"amount": 3700.0, "claim_id": "C-3002"})
    guard: ESCALATE  approve_claim_payment -> below the calibrated auto-approval threshold (confidence 0.416, calibrated threshold 0.854).
           guarantee: With 90% confidence, the rate of decisions that are auto-approved and wrong is at most 0.05 (RCPS, Hoeffding-Bentkus, n=320).
    graph paused at cli_human_review (thread claim-C-3002)
    review request for C-3002:
      action:    approve_claim_payment({"amount": 3700.0, "claim_id": "C-3002"})
      why:       below the calibrated auto-approval threshold (confidence 0.416, calibrated threshold 0.854).
      guarantee: With 90% confidence, the rate of decisions that are auto-approved and wrong is at most 0.05 (RCPS, Hoeffding-Bentkus, n=320).
    [simulated adjuster] edit amount to 1,000.00: Jewelry theft sub-limit applies: 1,500.00 less the 500.00 deductible.
    payment executed: 1,000.00 on C-3002
    agent: Payment of 1,000.00 issued on claim C-3002 (ledger entry 2). The claim can move to closure review.
    audit record in the checkpoint (state["cli_guard"]):
      action: escalate
      call call-C-3002-pay: approve_claim_payment({"amount": 3700.0, "claim_id": "C-3002"}) -> escalate
        evidence: confidence 0.416, threshold 0.854, Venn-Abers [0.192, 0.212]
        guarantee: risk_high_probability via RCPS, profile claim-payments-v1 (n=320)
      review: edit; note: Jewelry theft sub-limit applies: 1,500.00 less the 500.00 deductible.
        executed as: approve_claim_payment({"amount": 1000.0, "claim_id": "C-3002"})
------------------------------------------------------------------------------
C-3003: electronics theft 9 days after inception, prior claims, fraud indicators
    agent proposes: approve_claim_payment({"amount": 4480.0, "claim_id": "C-3003"})
    guard: ESCALATE  approve_claim_payment -> below the calibrated auto-approval threshold (confidence 0.002, calibrated threshold 0.854).
           guarantee: With 90% confidence, the rate of decisions that are auto-approved and wrong is at most 0.05 (RCPS, Hoeffding-Bentkus, n=320).
    graph paused at cli_human_review (thread claim-C-3003)
    review request for C-3003:
      action:    approve_claim_payment({"amount": 4480.0, "claim_id": "C-3003"})
      why:       below the calibrated auto-approval threshold (confidence 0.002, calibrated threshold 0.854).
      guarantee: With 90% confidence, the rate of decisions that are auto-approved and wrong is at most 0.05 (RCPS, Hoeffding-Bentkus, n=320).
    [simulated adjuster] reject: Loss 9 days after inception with two fraud indicators: refer to SIU, no payment.
    payment executed: none
    agent: No payment was made: the adjuster declined the proposal. Adjuster note: Loss 9 days after inception with two fraud indicators: refer to SIU, no payment. The claim stays open.
    audit record in the checkpoint (state["cli_guard"]):
      action: escalate
      call call-C-3003-pay: approve_claim_payment({"amount": 4480.0, "claim_id": "C-3003"}) -> escalate
        evidence: confidence 0.002, threshold 0.854, Venn-Abers [0.099, 0.110]
        guarantee: risk_high_probability via RCPS, profile claim-payments-v1 (n=320)
      review: reject; note: Loss 9 days after inception with two fraud indicators: refer to SIU, no payment.
------------------------------------------------------------------------------
The bundled calibration sets are small so the example runs quickly. For
production, calibrate on at least the recommended number of labelled examples
from your own traffic (about 1,000 for alpha=0.10), audit on held-out data,
and monitor for drift.

What happened:

  • C-3001 is a burst pipe with a complete file and no referral triggers. The agent proposed 1,850.00 less the 500.00 deductible, the evidence model scored the proposal 0.982, above the calibrated threshold of 0.854, and the payment ran without review.
  • C-3002 is a jewelry theft with a 1,500.00 sub-limit. The agent proposed the whole loss less the deductible, 3,700.00. The evidence model cannot tell whether the sub-limit applies and scored it 0.416, so the graph paused. The adjuster edited the amount to 1,000.00 and that is the payment that ran; the audit record keeps both the proposal and the edit.
  • C-3003 is a theft 9 days after the policy started, with three prior claims, two fraud indicators and no police report. It scored 0.002 and paused; the adjuster rejected it and referred it to the Special Investigations Unit. Nothing was paid, and the model received an error ToolMessage with the adjuster’s note.

The full record the checkpoint holds for C-3002, as graph.get_state(config).values["cli_guard"]:

{
  "action": "escalate",
  "calls": [
    {
      "action": "escalate",
      "arguments": {
        "amount": 3700.0,
        "claim_id": "C-3002"
      },
      "evidence": {
        "confidence": 0.41556015001557495,
        "decision": "escalate",
        "threshold": 0.8544119777365106,
        "venn_abers": [
          0.19230769230769232,
          0.21153846153846154
        ]
      },
      "guarantee": {
        "calibration_n": 320,
        "calibration_profile": "claim-payments-v1",
        "method": "RCPS",
        "statement": "With 90% confidence, the rate of decisions that are auto-approved and wrong is at most 0.05 (RCPS, Hoeffding-Bentkus, n=320).",
        "type": "risk_high_probability"
      },
      "id": "call-C-3002-pay",
      "reason": "below the calibrated auto-approval threshold (confidence 0.416, calibrated threshold 0.854).",
      "tool": "approve_claim_payment"
    }
  ],
  "review": {
    "executed_calls": [
      {
        "arguments": {
          "amount": 1000.0,
          "claim_id": "C-3002"
        },
        "id": "call-C-3002-pay",
        "tool": "approve_claim_payment"
      }
    ],
    "note": "Jewelry theft sub-limit applies: 1,500.00 less the 500.00 deductible.",
    "outcome": "edit"
  }
}

The guarantee is about the rate over many payment requests like the calibration set: with 90% confidence over the draw of the 320 calibration decisions, at most 5% of such requests end up auto-approved and wrong. At the calibrated threshold, 108 of the 320 historical decisions would have been auto-approved and the other 212 sent to adjusters. That share is the price of a high-confidence bound on 320 examples; more labelled decisions let the threshold move down. The bound is over all requests, not over the approved ones, and it says nothing about any single payment in isolation. To bound the error rate among auto-approved payments directly, use guarantee="fdr" (see Gate).

Example: discharge summaries with a claim-verification node

examples/agents/langgraph/clinical_summary_graph.py

A discharge-summary workflow turns a synthetic chart into a patient-friendly summary for a patient portal. There is no tool call here, so there is no ToolGuard: the check is a graph node that uses the Claim primitive in local mode.

START -> draft_summary -> verify_claims --every claim verified--> publish -> END
                                        --otherwise-------------> clinician_review --approve--> publish
                                                                                   --reject---> END

draft_summary asks the chat model for the summary, verify_claims splits it into atomic claims and keeps only the claims whose support score clears a calibrated threshold, and clinician_review pauses the graph with interrupt() whenever a claim was dropped or no guarantee is available. The portal only ever receives retained claims. The check and the graph:

CLAIM_CHECK = Claim(
    instructions=(
        "Keep only the statements in this patient-friendly discharge summary that the discharge "
        "chart supports."
    ),
    calibration_profile="discharge-claims-v1",
    alpha=0.10,
)
def build_graph(model: Any, client: LocalCLIClient, portal: list[dict[str, Any]], checkpointer: Any = None) -> Any:
    """draft_summary -> verify_claims -> publish | clinician_review -> publish | END."""
 
    def draft_summary(state: SummaryState) -> dict[str, Any]:
        chart = json.dumps(state["chart"], indent=2, sort_keys=True)
        reply = model.invoke([SystemMessage(DRAFT_PROMPT), HumanMessage(f"Discharge chart:\n{chart}")])
        content = reply.content
        if not isinstance(content, str):   # some providers return a list of content blocks
            content = " ".join(part.get("text", "") for part in content if isinstance(part, dict))
        return {"draft": content.strip()}
 
    def verify_claims(state: SummaryState) -> dict[str, Any]:
        # Same builder as the calibration examples: {"chart": ..., "answer": <draft>}.
        context = charts_data.summary_context(state["chart"], state["draft"])
        answer = client.evaluate(context, {"claims": CLAIM_CHECK}).answers["claims"]
        return {"verification": {
            "retained_claims": list(answer.retained_claims),
            "dropped_claims": [{"text": d.text, "score": d.score, "reason": d.reason} for d in answer.dropped_claims],
            "threshold": answer.raw.get("threshold"),
            "guarantee": answer.guarantee.describe(),
            "heuristic": answer.is_heuristic,
            "verified_text": answer.as_text(),
        }}
 
    def route_after_verification(state: SummaryState) -> str:
        v = state["verification"]
        # Publish unreviewed only when every claim was verified; anything else goes to a clinician.
        verified = v["retained_claims"] and not v["dropped_claims"] and not v["heuristic"]
        return "publish" if verified else "clinician_review"
 
    def clinician_review(state: SummaryState) -> Command[Literal["publish", "__end__"]]:
        # Nothing above interrupt() has side effects: this node re-runs from the top on resume.
        v = state["verification"]
        decision = interrupt({
            "kind": "clinician_review",
            "encounter_id": state["encounter_id"],
            "retained_claims": v["retained_claims"],
            "dropped_claims": v["dropped_claims"],
            "verified_text": v["verified_text"],
            "guarantee": v["guarantee"],
            "heuristic": v["heuristic"],
            "resume_with": {"action": "approve | reject", "note": "optional clinician note"},
        })
        if not isinstance(decision, Mapping):
            decision = {"action": decision}
        review = {"action": decision.get("action"), "note": decision.get("note")}
        if decision.get("action") == "approve" and v["verified_text"]:
            return Command(goto="publish", update={"review": review})
        # Reject, an unrecognized answer, or nothing verified to publish: fail closed.
        return Command(goto=END, update={"review": review,
                                         "status": "returned to the clinician; nothing was published"})
 
    def publish(state: SummaryState) -> dict[str, Any]:
        text = state["verification"]["verified_text"]      # only retained claims are ever published
        portal.append({"encounter_id": state["encounter_id"], "text": text})
        return {"published": text, "status": "published to the patient portal"}
 
    builder = StateGraph(SummaryState)
    builder.add_node("draft_summary", draft_summary)
    builder.add_node("verify_claims", verify_claims)
    builder.add_node("clinician_review", clinician_review, destinations=("publish", END))
    builder.add_node("publish", publish)
    builder.add_edge(START, "draft_summary")
    builder.add_edge("draft_summary", "verify_claims")
    builder.add_conditional_edges("verify_claims", route_after_verification,
                                  {"publish": "publish", "clinician_review": "clinician_review"})
    builder.add_edge("publish", END)
    return builder.compile(checkpointer=checkpointer or InMemorySaver())

The data is examples/agents/data/discharge_claims.jsonl, 240 synthetic draft summaries with 1,537 labelled claims written by generate_discharge_claims.py. Each chart lists diagnoses, medications with dose, frequency and status (new, continued, changed, held, stopped), follow-up appointments, pending results and return precautions, and each draft states one claim per sentence. A written labelling protocol decides which claims are supported; 156 claims are not (a wrong dose, an invented dose change, a wrong appointment date, an invented test), spread over 128 of the 240 drafts. Some supported claims are plain-language paraphrases that share few words with the chart, and about 1% of labels are flipped to model annotator disagreement. The mock evidence model scores support from the facts a claim shares with the chart, misses a contradicted medication status some of the time, and never sees a label.

git clone https://github.com/rahvis/cci-sdk && cd cci-sdk
python examples/agents/langgraph/clinical_summary_graph.py

The output, in mock mode:

==============================================================================
LangGraph: discharge summaries with a calibrated claim-verification node
==============================================================================
Synthetic data for demonstration only. Not medical advice and not a medical
device. A calibrated guarantee bounds error rates on data like the calibration
set; it does not make any single recommendation safe. Keep a qualified
clinician in the loop and follow your institution's clinical governance.

calibration profile discharge-claims-v1: conformal-factuality, n=240 (minimum 9, recommended 1000), realized-coverage CI [0.867, 0.930], status=serving
evidence model: mock (access level L1)
------------------------------------------------------------------------------
D-4001: pneumonia; the drafting model writes a faithful summary
    draft: 7 claims; calibrated support threshold 0.678
      retained  You were in the hospital for community-acquired pneumonia.
      retained  Start amoxicillin-clavulanate 875 mg twice daily for 5 more days.
      retained  Continue metformin 1000 mg twice daily as before.
      retained  See your primary care doctor within 7 days.
      retained  You have an appointment at the pulmonology clinic on 2026-10-09.
      retained  The results of your blood cultures are still pending; the team will call you with them.
      retained  Come back to the emergency department if you notice worsening shortness of breath.
    guarantee: With probability at least 90%, every retained claim is supported (conformal factuality, n=240).
    outcome: published to the patient portal
    portal text: You were in the hospital for community-acquired pneumonia.
      Start amoxicillin-clavulanate 875 mg twice daily for 5 more days.
      Continue metformin 1000 mg twice daily as before. See your primary care
      doctor within 7 days. You have an appointment at the pulmonology clinic
      on 2026-10-09. The results of your blood cultures are still pending; the
      team will call you with them. Come back to the emergency department if
      you notice worsening shortness of breath.
------------------------------------------------------------------------------
D-4002: heart failure; the draft invents a dose increase and a wrong appointment date
    draft: 7 claims; calibrated support threshold 0.678
      retained  You were in the hospital for acute decompensated heart failure.
      retained  Your furosemide dose is now 40 mg twice daily, up from 40 mg once daily.
      retained  Do not take lisinopril until your follow-up visit.
      retained  Weigh yourself every morning and write it down.
      retained  See your primary care doctor within 7 days.
      DROPPED   Your metoprolol succinate dose was increased to 100 mg once daily.  (support score 0.384)
      DROPPED   You have an appointment at the cardiology clinic on 2026-10-01.  (support score 0.440)
    guarantee: With probability at least 90%, every retained claim is supported (conformal factuality, n=240).
    graph paused at clinician_review (thread encounter-D-4002)
    review request for D-4002:
      action:    publish_discharge_summary({"claims_dropped": 2, "claims_retained": 5, "encounter_id": "D-4002"})
      why:       2 of 7 claims are at or below the calibrated support threshold; approving publishes only the 5 retained claims.
      guarantee: With probability at least 90%, every retained claim is supported (conformal factuality, n=240).
    [simulated clinician] approved
    resume value: {"action": "approve", "note": "Publish the verified claims; dropped claims stay out."}
    review recorded in the checkpoint (state["review"]): {"action": "approve", "note": "Publish the verified claims; dropped claims stay out."}
    outcome: published to the patient portal
    portal text: You were in the hospital for acute decompensated heart
      failure. Your furosemide dose is now 40 mg twice daily, up from 40 mg
      once daily. Do not take lisinopril until your follow-up visit. Weigh
      yourself every morning and write it down. See your primary care doctor
      within 7 days.
------------------------------------------------------------------------------
The bundled calibration sets are small so the example runs quickly. For
production, calibrate on at least the recommended number of labelled examples
from your own traffic (about 1,000 for alpha=0.10), audit on held-out data,
and monitor for drift.

What happened:

  • The calibrated support threshold is 0.678: a claim is kept only when its support score is above it. It comes from the highest-scoring unsupported claim in each calibration draft.
  • D-4001 is a faithful pneumonia summary. All seven claims scored well above the threshold, nothing was dropped, and the text was published without review.
  • D-4002 is a heart-failure summary with two problems: it says the metoprolol dose was increased to 100 mg (the chart continues 50 mg), and it gives a cardiology date the chart does not have. Those two claims scored 0.384 and 0.440 and were dropped, the graph paused at clinician_review, and the clinician approved publishing the five verified claims only. The review is kept in the checkpoint as state["review"].

Dropped claims are withheld, not corrected. The D-4002 summary that reached the portal no longer mentions the cardiology appointment at all, and the patient still needs the right date: in a real workflow the clinician adds it, and anything the clinician writes is theirs, not covered by the guarantee. The guarantee, “With probability at least 90%, every retained claim is supported (conformal factuality, n=240).”, is about drafts like the calibration set: for at most about 10% of them an unsupported claim survives the filter. It does not make any single summary correct, alpha=0.10 is a demonstration setting for patient-facing text, and portal content stays under your clinical governance. This is not a medical device and not medical advice.

Pitfalls

  • Nodes that call interrupt() re-run from their first line on resume. Keep model calls and side effects out of them, or after the interrupt() line. The adapter evaluates the guard in cli_guard for this reason, and the example’s clinician_review does nothing before interrupt(). Do not wrap interrupt() in try/except: that swallows the pause.
  • Do not call interrupt() inside a tool or a ToolNode wrapper. The whole tools node re-runs on resume, so a sibling call in the same step runs twice (verified with a tool that sent an email). Pause before the tools node, as cli_human_review does.
  • A checkpointer is required, and InMemorySaver dies with the process. For reviews that arrive later or elsewhere, use a durable checkpointer and the same thread_id. Keep interrupt payloads and resume values JSON-serializable; the guard’s records are.
  • Only Command(resume=...) resumes. graph.invoke(None, config) on a paused thread returns the same pending interrupt. With several pending interrupts at once (for example from parallel Send branches), resume with Command(resume={interrupt_id: value, ...}); pending_review gives you the interrupt_id.
  • invoke(..., version="v2") returns a GraphOutput. Read .interrupts rather than result["__interrupt__"], which is deprecated there. pending_review accepts both shapes.
  • Every tool call needs a matching ToolMessage before the next model call, or OpenAI and Anthropic reject the history. cli_blocked and the reject path add one error ToolMessage per call id; a review node of your own must do the same.
  • Edited calls are not re-checked. The reviewer’s edit is final. If hard limits must hold even after a human edit, enforce them in the tool itself.
  • state["cli_guard"] holds the latest guarded turn only. A later turn with tool calls replaces it. Read graph.get_state_history(config) for every turn, or copy each record into your audit log.
  • Self-hosted Gemma needs tool calling switched on. Without --enable-auto-tool-choice --tool-call-parser gemma4 (and the Gemma 4 chat template on vLLM), tool calls come back as plain text, agent routes to the end and the guard never runs. Pass model= exactly as the server was started, and api_key="EMPTY" unless the server has a key.
  • Fake models for tests. LangChain’s fake chat models inherit a bind_tools that raises NotImplementedError: subclass and return self. Construct them with disable_streaming=True, or streaming a content-less tool-call message raises ValueError, and build one per thread, because their script is consumed as they go.
  • Imports. ToolNode and tools_condition come from langgraph.prebuilt, not langchain.tools. For the prebuilt agent, langgraph.prebuilt.create_react_agent is deprecated: use langchain.agents.create_agent with the LangChain middleware instead of this node.

Related

LangChainGoogle ADK