Cost per role in a LangGraph AI scientist that fans out

This builds an AI scientist in LangGraph 1.x against Claude: a planner turns a question into competing hypotheses, a probe role tests every hypothesis against every record in parallel, a referee adjudicates, and a synthesist writes the study up. At the constants this page ships with, one study is nine model calls. Raise two of those constants to the width a real study wants and the same code is 291 calls — and the instrumentation added in the last third of this page reports both as four rows, not as nine or 291.

What you will build

scientist.py, a single-file LangGraph app with four roles:

• planner — one call. Turns the question into at most HYPOTHESES competing explanations, each falsifiable against a single record.
• prober — one call per (hypothesis, record) pair. The multiplier. Reads one record, returns one line: SUPPORTS, REFUTES or SILENT.
• referee — one call. Adjudicates every hypothesis against the verdicts.
• synthesist — one call, on a stronger model. Writes the study.

The graph is four nodes and one fan-out. The planner's successor is a conditional edge that returns one Send per hypothesis, so the investigation branches run as separate tasks and their results merge back through a reducer. Inside each branch, the probes run on a small thread pool.

That is two fan-outs of different kinds in one program — one driven by the framework, one written by hand — and they behave differently once you attach attribution to them. That difference is the most useful thing on this page.

Each role is its own function, and the node that only orchestrates is not one of them. That matters later, because the instrumentation attaches to functions, and it is the better shape regardless: a node that inlines four model calls has nothing you can name, budget or move to a different model.

Written against LangGraph 1.x with langchain-anthropic, calling Anthropic's API directly rather than through a gateway. Models: claude-haiku-4-5 for the planner, the probes and the referee; claude-sonnet-4-6 for the synthesist.

Read this before you run it

The two constants that set the width start small on purpose:

HYPOTHESES = 3
PROBES_PER_HYPOTHESIS = 2

Nine calls. Everything this page says about scale — the 288 probe calls, the four rows, the budget — is what happens when you raise them to something like 24 and 12. Raise them deliberately, after the budget section, not before.

Prerequisites

Python 3.11 or newer.

pip install "langgraph>=1.0,<2" langchain-anthropic

The major is pinned so this page stays true as LangGraph moves. langchain-anthropic is left unpinned on purpose: install the one that matches your langgraph major.

You need an Anthropic API key. ChatAnthropic reads ANTHROPIC_API_KEY from the environment, so export it and no code has to touch it:

export ANTHROPIC_API_KEY=...

Step 1 — Fix the record store

Start with the corpus. A scientist with a search tool has an unbounded number of calls in it; a scientist handed a fixed record store has exactly hypotheses × records, which is a number you can multiply out before you run anything.

The records below are synthetic — invented for this page — but they are shaped like a real trial: one genuine cause, one plausible confound, and one record that quietly invalidates a control.

"""scientist.py — the record store an AI scientist is allowed to read."""

CORPUS = [
    {
        "id": "run-01",
        "text": "Cell A3, 25 C ambient, 1C charge. Capacity 100% to 91.4% over "
                "600 cycles. Internal resistance +6%.",
    },
    {
        "id": "run-02",
        "text": "Cell A4, 45 C ambient, 1C charge. Capacity 100% to 78.1% over "
                "600 cycles. Internal resistance +31%.",
    },
    {
        "id": "run-03",
        "text": "Cell B1, 25 C ambient, 2C charge. Capacity 100% to 84.7% over "
                "600 cycles. Internal resistance +19%.",
    },
    {
        "id": "run-04",
        "text": "Cell B2 held at 45 C for the calendar duration of the trial "
                "with no cycling at all. Capacity 100% to 96.2%.",
    },
    {
        "id": "run-05",
        "text": "Post-mortem teardown: lithium plating along the anode edge of "
                "A4 and B1. None visible on A3.",
    },
    {
        "id": "run-06",
        "text": "Charger firmware note: the 2C profile overshoots to 2.4C for "
                "the first 40 seconds of every charge.",
    },
    {
        "id": "run-07",
        "text": "Cell B3, 25 C ambient, 2C charge, firmware overshoot patched. "
                "Capacity 100% to 89.9% over 600 cycles.",
    },
    {
        "id": "run-08",
        "text": "Logger fault: rack 2, holding B1 and B2, recorded 25 C while "
                "its door stood open to a 31 C room for 40% of the trial.",
    },
]

QUESTION = (
    "Cells in this trial lost capacity at very different rates. What is "
    "driving the difference, and what should change before the next trial?"
)


def record_text(record: dict[str, str]) -> str:
    return f"[{record['id']}] {record['text']}"


def corpus_text() -> str:
    return "\n".join(record_text(r) for r in CORPUS)


if __name__ == "__main__":
    print(f"{len(CORPUS)} records, {len(corpus_text())} characters")

One record per probe call, not the whole corpus per probe call. That is the decision that makes the fan-out affordable: the probe prompt is a fixed system prompt plus roughly two lines, so widening the study multiplies a small number rather than a large one. It also gives every verdict a record id, which is what makes the referee's output checkable by a person.

Step 2 — The state, and the width in code

The reducer is the only part of the state that is not obvious. Branches run concurrently and each returns its own finding, so findings needs to say how concurrent writes combine — operator.add concatenates the lists instead of the last branch overwriting the rest.

# fragment
import operator
from typing import Annotated, TypedDict

HYPOTHESES = 3               # branches. A real study raises this to ~24.
PROBES_PER_HYPOTHESIS = 2    # calls per branch. A real study raises this to ~12.
BRANCH_CONCURRENCY = 3       # hypotheses investigated at once
PROBE_CONCURRENCY = 2        # probes in flight within one hypothesis


class Finding(TypedDict):
    hypothesis: str
    verdicts: list[str]
    missing: int


class ScientistState(TypedDict):
    question: str
    hypotheses: list[str]
    findings: Annotated[list[Finding], operator.add]
    adjudication: str
    report: str


class Investigation(TypedDict):
    """The payload one branch receives. Not the graph's state."""

    hypothesis: str

missing on Finding exists from the start, before any of this is instrumented. A probe can come back with nothing for ordinary reasons — a provider error, a timeout — and a referee that cannot tell "no evidence" from "evidence against" will adjudicate confidently on a gap. Counting what did not arrive is part of the science, not part of the monitoring.

The four width constants are the whole cost model of this program. Multiply the first two together and add three, and you have the call count for a study.

Step 3 — The planner and the fan-out

The planner decides how wide the study goes, so the cap lives in Python. A model asked for "at most three" will occasionally write five, and the slice is what makes that harmless — here it is also what stops a chatty planner from tripling your bill.

# fragment
import re

from langchain_anthropic import ChatAnthropic
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langgraph.types import Send

FAST_MODEL = "claude-haiku-4-5"

_BULLET = re.compile(r"^\s*(?:[-*•]|\d+[.)])\s*")


def _lines(text: str, limit: int) -> list[str]:
    cleaned = (_BULLET.sub("", line).strip() for line in text.splitlines())
    return [line for line in cleaned if line][:limit]


PLANNER_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a research planner. Given a question and the record store "
            "available, write competing hypotheses that could explain the "
            "observation. Each must be falsifiable against a single record. "
            "One per line, no numbering, no preamble. At most "
            "{max_hypotheses} hypotheses.",
        ),
        ("human", "Question: {question}\n\nRecords:\n{records}"),
    ]
)

planner_llm = ChatAnthropic(model=FAST_MODEL, max_tokens=512, temperature=0)
plan_chain = PLANNER_PROMPT | planner_llm | StrOutputParser()


def plan(state: ScientistState) -> dict:
    raw = plan_chain.invoke(
        {
            "question": state["question"],
            "records": corpus_text(),
            "max_hypotheses": HYPOTHESES,
        }
    )
    return {"hypotheses": _lines(raw, HYPOTHESES)}


def fan_out(state: ScientistState) -> list[Send]:
    return [Send("investigate", {"hypothesis": h}) for h in state["hypotheses"]]

Lines rather than JSON. A one-per-line contract degrades into a shorter list when the model ignores you; a malformed JSON object degrades into an exception, and a retry on a parse failure is a call you pay for twice.

fan_out is not a node. It is the function a conditional edge calls to decide where to go next, and returning a list of Send objects is how LangGraph is told to run the same node once per item with a different payload. It makes no model call, which is why it will not appear in the cost report later.

The planner is the one place the entire corpus goes into a prompt. That is fine at eight records and wrong at eight thousand; at a real corpus size you hand the planner an index and let the probes touch the records.

Step 4 — The prober, and a fan-out you write yourself

Each branch tests one hypothesis against PROBES_PER_HYPOTHESIS records. They do not depend on each other, so running them in sequence buys nothing but latency.

# fragment
from concurrent.futures import ThreadPoolExecutor

PROBE_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You test one hypothesis against one experimental record. Reply "
            "with exactly one line: SUPPORTS, REFUTES or SILENT, then a "
            "colon, then at most fifteen words quoting the part of the record "
            "you used. Answer SILENT unless the record bears on the "
            "hypothesis directly. Never speculate beyond the record.",
        ),
        ("human", "Hypothesis: {hypothesis}\n\nRecord: {record}"),
    ]
)

prober_llm = ChatAnthropic(model=FAST_MODEL, max_tokens=96, temperature=0)
probe_chain = PROBE_PROMPT | prober_llm | StrOutputParser()


def probe(hypothesis: str, record: dict[str, str]) -> str:
    verdict = probe_chain.invoke(
        {"hypothesis": hypothesis, "record": record_text(record)}
    )
    return f"{record['id']} {verdict.strip()}"


def investigate(state: Investigation) -> dict:
    records = CORPUS[:PROBES_PER_HYPOTHESIS]
    verdicts: list[str] = []
    missing = 0

    with ThreadPoolExecutor(max_workers=PROBE_CONCURRENCY) as pool:
        futures = [pool.submit(probe, state["hypothesis"], r) for r in records]
        for future in futures:
            try:
                verdicts.append(future.result())
            except Exception:
                missing += 1

    return {
        "findings": [
            {
                "hypothesis": state["hypothesis"],
                "verdicts": verdicts,
                "missing": missing,
            }
        ]
    }

max_tokens=96 on the prober is load-bearing, not tidiness. Output tokens are the expensive half of a call, this role runs hundreds of times, and the one line it is asked for fits easily. A prober allowed to write a paragraph is the same study at several times the price and a referee prompt several times longer.

investigate is the orchestrator of the branch and makes no model call itself — every call happens inside probe. Keep it that way. The alternative, inlining the chain invocation into the loop body, works and costs the same, and it leaves you with nothing to name when you want the probe's cost separated from everything else the branch might grow into later.

Step 5 — The referee and the synthesist

The referee sees every verdict from every branch at once, which is the only place in this program where a prompt grows with the width of the study. One line per verdict is what keeps that survivable at 288 probes.

# fragment
SUPPORTED_LIMIT = 3
DEEP_MODEL = "claude-sonnet-4-6"

REFEREE_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You adjudicate hypotheses against probe verdicts. Give one line "
            "per hypothesis: the hypothesis, then SUPPORTED, REFUTED or "
            "UNDECIDED, then the record ids that decided it. Mark UNDECIDED "
            "where evidence is missing rather than guessing. At most "
            "{max_supported} hypotheses may be marked SUPPORTED.",
        ),
        ("human", "Question: {question}\n\nFindings:\n{findings}"),
    ]
)

SYNTHESIST_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You write the results section of a short study. Four parts: what "
            "was asked, what the evidence shows, what is still unresolved, "
            "what to measure next. Use only the adjudication given. Do not "
            "speculate. Under 300 words.",
        ),
        ("human", "Question: {question}\n\nAdjudication:\n{adjudication}"),
    ]
)

referee_llm = ChatAnthropic(model=FAST_MODEL, max_tokens=1024, temperature=0)
synthesist_llm = ChatAnthropic(model=DEEP_MODEL, max_tokens=1024, temperature=0)

referee_chain = REFEREE_PROMPT | referee_llm | StrOutputParser()
synthesist_chain = SYNTHESIST_PROMPT | synthesist_llm | StrOutputParser()


def format_findings(findings: list[Finding]) -> str:
    blocks = []
    for finding in findings:
        lines = "\n".join(f"  {v}" for v in finding["verdicts"])
        if not lines:
            lines = "  (no evidence gathered)"
        if finding["missing"]:
            lines += f"\n  {finding['missing']} probe(s) returned nothing"
        blocks.append(f"- {finding['hypothesis']}\n{lines}")
    return "\n".join(blocks)


def adjudicate(state: ScientistState) -> dict:
    return {
        "adjudication": referee_chain.invoke(
            {
                "question": state["question"],
                "findings": format_findings(state["findings"]),
                "max_supported": SUPPORTED_LIMIT,
            }
        )
    }


def synthesise(state: ScientistState) -> dict:
    return {
        "report": synthesist_chain.invoke(
            {
                "question": state["question"],
                "adjudication": state["adjudication"],
            }
        )
    }

The synthesist is the only role on the stronger model, because it is the only one whose output a person reads. The other three produce input for another model. Whether that split is right is a judgement to revisit with real numbers rather than settle by guessing now, and the last section of this page is about getting those numbers.

Note that the referee is one call, not one per hypothesis. That is a choice: a per-hypothesis referee would be a second fan-out, a second thing to bound and a second thing to budget, in exchange for verdicts that never see each other. One referee that reads everything is both cheaper and better at spotting that two hypotheses are the same hypothesis.

Step 6 — Wire the graph

# fragment
from langgraph.graph import END, START, StateGraph

builder = StateGraph(ScientistState)
builder.add_node("plan", plan)
builder.add_node("investigate", investigate)
builder.add_node("adjudicate", adjudicate)
builder.add_node("synthesise", synthesise)

builder.add_edge(START, "plan")
builder.add_conditional_edges("plan", fan_out, ["investigate"])
builder.add_edge("investigate", "adjudicate")
builder.add_edge("adjudicate", "synthesise")
builder.add_edge("synthesise", END)

scientist = builder.compile()


def run_study(question: str) -> str:
    state = scientist.invoke(
        {"question": question, "findings": []},
        config={"max_concurrency": BRANCH_CONCURRENCY},
    )
    return state["report"]

add_conditional_edges("plan", fan_out, ["investigate"]) is the whole fan-out. The third argument tells the graph which nodes the edge can reach, so the branch target is registered even though no plain edge points at it. investigate then has one ordinary edge to adjudicate, and the reducer on findings is what makes that a join: the branches merge before the referee runs. max_concurrency in the run config bounds how many branches are in flight.

The whole file

"""scientist.py — an AI scientist: plan, probe wide, adjudicate, write up."""
import operator
import re
from concurrent.futures import ThreadPoolExecutor
from typing import Annotated, TypedDict

from langchain_anthropic import ChatAnthropic
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langgraph.graph import END, START, StateGraph
from langgraph.types import Send

FAST_MODEL = "claude-haiku-4-5"
DEEP_MODEL = "claude-sonnet-4-6"

HYPOTHESES = 3               # branches. A real study raises this to ~24.
PROBES_PER_HYPOTHESIS = 2    # calls per branch. A real study raises this to ~12.
BRANCH_CONCURRENCY = 3
PROBE_CONCURRENCY = 2
SUPPORTED_LIMIT = 3

CORPUS = [
    {
        "id": "run-01",
        "text": "Cell A3, 25 C ambient, 1C charge. Capacity 100% to 91.4% over "
                "600 cycles. Internal resistance +6%.",
    },
    {
        "id": "run-02",
        "text": "Cell A4, 45 C ambient, 1C charge. Capacity 100% to 78.1% over "
                "600 cycles. Internal resistance +31%.",
    },
    {
        "id": "run-03",
        "text": "Cell B1, 25 C ambient, 2C charge. Capacity 100% to 84.7% over "
                "600 cycles. Internal resistance +19%.",
    },
    {
        "id": "run-04",
        "text": "Cell B2 held at 45 C for the calendar duration of the trial "
                "with no cycling at all. Capacity 100% to 96.2%.",
    },
    {
        "id": "run-05",
        "text": "Post-mortem teardown: lithium plating along the anode edge of "
                "A4 and B1. None visible on A3.",
    },
    {
        "id": "run-06",
        "text": "Charger firmware note: the 2C profile overshoots to 2.4C for "
                "the first 40 seconds of every charge.",
    },
    {
        "id": "run-07",
        "text": "Cell B3, 25 C ambient, 2C charge, firmware overshoot patched. "
                "Capacity 100% to 89.9% over 600 cycles.",
    },
    {
        "id": "run-08",
        "text": "Logger fault: rack 2, holding B1 and B2, recorded 25 C while "
                "its door stood open to a 31 C room for 40% of the trial.",
    },
]

QUESTION = (
    "Cells in this trial lost capacity at very different rates. What is "
    "driving the difference, and what should change before the next trial?"
)

_BULLET = re.compile(r"^\s*(?:[-*•]|\d+[.)])\s*")


class Finding(TypedDict):
    hypothesis: str
    verdicts: list[str]
    missing: int


class ScientistState(TypedDict):
    question: str
    hypotheses: list[str]
    findings: Annotated[list[Finding], operator.add]
    adjudication: str
    report: str


class Investigation(TypedDict):
    hypothesis: str


def record_text(record: dict[str, str]) -> str:
    return f"[{record['id']}] {record['text']}"


def corpus_text() -> str:
    return "\n".join(record_text(r) for r in CORPUS)


def _lines(text: str, limit: int) -> list[str]:
    cleaned = (_BULLET.sub("", line).strip() for line in text.splitlines())
    return [line for line in cleaned if line][:limit]


PLANNER_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a research planner. Given a question and the record store "
            "available, write competing hypotheses that could explain the "
            "observation. Each must be falsifiable against a single record. "
            "One per line, no numbering, no preamble. At most "
            "{max_hypotheses} hypotheses.",
        ),
        ("human", "Question: {question}\n\nRecords:\n{records}"),
    ]
)

PROBE_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You test one hypothesis against one experimental record. Reply "
            "with exactly one line: SUPPORTS, REFUTES or SILENT, then a "
            "colon, then at most fifteen words quoting the part of the record "
            "you used. Answer SILENT unless the record bears on the "
            "hypothesis directly. Never speculate beyond the record.",
        ),
        ("human", "Hypothesis: {hypothesis}\n\nRecord: {record}"),
    ]
)

REFEREE_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You adjudicate hypotheses against probe verdicts. Give one line "
            "per hypothesis: the hypothesis, then SUPPORTED, REFUTED or "
            "UNDECIDED, then the record ids that decided it. Mark UNDECIDED "
            "where evidence is missing rather than guessing. At most "
            "{max_supported} hypotheses may be marked SUPPORTED.",
        ),
        ("human", "Question: {question}\n\nFindings:\n{findings}"),
    ]
)

SYNTHESIST_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You write the results section of a short study. Four parts: what "
            "was asked, what the evidence shows, what is still unresolved, "
            "what to measure next. Use only the adjudication given. Do not "
            "speculate. Under 300 words.",
        ),
        ("human", "Question: {question}\n\nAdjudication:\n{adjudication}"),
    ]
)

planner_llm = ChatAnthropic(model=FAST_MODEL, max_tokens=512, temperature=0)
prober_llm = ChatAnthropic(model=FAST_MODEL, max_tokens=96, temperature=0)
referee_llm = ChatAnthropic(model=FAST_MODEL, max_tokens=1024, temperature=0)
synthesist_llm = ChatAnthropic(model=DEEP_MODEL, max_tokens=1024, temperature=0)

plan_chain = PLANNER_PROMPT | planner_llm | StrOutputParser()
probe_chain = PROBE_PROMPT | prober_llm | StrOutputParser()
referee_chain = REFEREE_PROMPT | referee_llm | StrOutputParser()
synthesist_chain = SYNTHESIST_PROMPT | synthesist_llm | StrOutputParser()


def plan(state: ScientistState) -> dict:
    raw = plan_chain.invoke(
        {
            "question": state["question"],
            "records": corpus_text(),
            "max_hypotheses": HYPOTHESES,
        }
    )
    return {"hypotheses": _lines(raw, HYPOTHESES)}


def fan_out(state: ScientistState) -> list[Send]:
    return [Send("investigate", {"hypothesis": h}) for h in state["hypotheses"]]


def probe(hypothesis: str, record: dict[str, str]) -> str:
    verdict = probe_chain.invoke(
        {"hypothesis": hypothesis, "record": record_text(record)}
    )
    return f"{record['id']} {verdict.strip()}"


def investigate(state: Investigation) -> dict:
    records = CORPUS[:PROBES_PER_HYPOTHESIS]
    verdicts: list[str] = []
    missing = 0

    with ThreadPoolExecutor(max_workers=PROBE_CONCURRENCY) as pool:
        futures = [pool.submit(probe, state["hypothesis"], r) for r in records]
        for future in futures:
            try:
                verdicts.append(future.result())
            except Exception:
                missing += 1

    return {
        "findings": [
            {
                "hypothesis": state["hypothesis"],
                "verdicts": verdicts,
                "missing": missing,
            }
        ]
    }


def format_findings(findings: list[Finding]) -> str:
    blocks = []
    for finding in findings:
        lines = "\n".join(f"  {v}" for v in finding["verdicts"])
        if not lines:
            lines = "  (no evidence gathered)"
        if finding["missing"]:
            lines += f"\n  {finding['missing']} probe(s) returned nothing"
        blocks.append(f"- {finding['hypothesis']}\n{lines}")
    return "\n".join(blocks)


def adjudicate(state: ScientistState) -> dict:
    return {
        "adjudication": referee_chain.invoke(
            {
                "question": state["question"],
                "findings": format_findings(state["findings"]),
                "max_supported": SUPPORTED_LIMIT,
            }
        )
    }


def synthesise(state: ScientistState) -> dict:
    return {
        "report": synthesist_chain.invoke(
            {
                "question": state["question"],
                "adjudication": state["adjudication"],
            }
        )
    }


builder = StateGraph(ScientistState)
builder.add_node("plan", plan)
builder.add_node("investigate", investigate)
builder.add_node("adjudicate", adjudicate)
builder.add_node("synthesise", synthesise)

builder.add_edge(START, "plan")
builder.add_conditional_edges("plan", fan_out, ["investigate"])
builder.add_edge("investigate", "adjudicate")
builder.add_edge("adjudicate", "synthesise")
builder.add_edge("synthesise", END)

scientist = builder.compile()


def run_study(question: str) -> str:
    state = scientist.invoke(
        {"question": question, "findings": []},
        config={"max_concurrency": BRANCH_CONCURRENCY},
    )
    return state["report"]


if __name__ == "__main__":
    print(run_study(QUESTION))

python scientist.py

At the shipped constants the probes only reach run-01 and run-02, so the study will honestly report that it cannot separate temperature from charge rate. Set PROBES_PER_HYPOTHESIS = 8 and it reaches run-06, run-07 and run-08, and the answer changes: the firmware overshoot is doing the damage and one of the controls was never at 25 C. That jump is the shape of the whole problem — the useful answer is on the other side of a much wider fan-out.

The application is finished. Nothing below changes what it does.

Seeing what it cost

What you do not have is any idea which of the four roles spent the money. The provider's invoice has one line for the API key, and the nine calls behind it are indistinguishable — same key, same account, three of the four on the same model. At 291 calls the invoice does not get more informative, only larger.

pip install capsera

At the top of scientist.py, above everything else:

# fragment
import os

import capsera

capsera.init(api_key=os.environ["CAPSERA_API_KEY"], endpoint="https://api.capsera.ai")

That is the capture step, and it is already complete. capsera.init() patches the provider clients that ChatAnthropic constructs underneath, so from that call on every request the program makes is recorded: the planner's, every probe in every branch, the referee's, the synthesist's. The prompts are untouched, the chains are untouched, the graph is untouched, and no traffic changed route — the SDK runs inside your process rather than in front of it, so there is no proxy, no base_url to swap and nothing new in the request path.

One addition for a script: a background thread ships events on an interval, so a process that exits immediately can exit before the queue drains.

# fragment
if __name__ == "__main__":
    print(run_study(QUESTION))
    capsera.shutdown()

capsera.shutdown() flushes pending events and stops the worker. A long-running service does not need it; a script, a cron job or a notebook cell does, and omitting it is the usual reason a first run reports nothing.

Run it now and you have a total, a per-model split and token counts per call. You also have every one of those events attributed to unknown, because nothing has said which role made it. unknown is a real row, not a dropped one — an unattributed cost is still a cost — but it is one row where you wanted four.

Name the roles

The calls are already captured. The decorators only say who:

# fragment
@capsera.langgraph_node("planner", team="ai-scientist", task_type="hypothesis-generation")
def plan(state: ScientistState) -> dict:
    ...  # body unchanged


@capsera.agent("prober", team="ai-scientist", task_type="evidence-probe")
def probe(hypothesis: str, record: dict[str, str]) -> str:
    ...  # body unchanged


@capsera.langgraph_node("referee", team="ai-scientist", task_type="adjudication")
def adjudicate(state: ScientistState) -> dict:
    ...  # body unchanged


@capsera.langgraph_node("synthesist", team="ai-scientist", task_type="write-up")
def synthesise(state: ScientistState) -> dict:
    ...  # body unchanged

Four added lines, and that is the entire diff for a program that can make 291 model calls. No function body changed, no call site changed, no chain or model construction changed, no argument threaded through the graph, no state field added. Every LLM call made while a decorated function is on the stack is attributed to that function's name, including calls made by the functions it calls and by the framework it invokes.

langgraph_node is @capsera.agent plus a record of the framework, so use it on graph nodes and agent on plain functions. That is the only difference, and it is why probe takes agent — it is a helper, not a node.

investigate is deliberately not decorated. It is a node, so langgraph_node would fit, but it makes no model call of its own: every call in the branch happens inside probe, which has its own scope. Decorating investigate would add a fifth row that can never have spend on it, and if you decorated it instead of probe you would get the same total under a name that describes the loop rather than the work. Decorate the function that makes the call. For a study-level total, use the team field — which is what team="ai-scientist" on all four decorators is for.

The fan-out you wrote yourself needs one more line

Two fan-outs, two behaviours, and only one of them is free.

The graph's fan-out is fine as it stands. plan returns hypotheses, the conditional edge issues one Send each, and LangGraph runs investigate per branch. The decorators sit on plan, adjudicate and synthesise — nodes LangGraph calls directly — so each scope is opened inside whichever worker is running that node, and the model call happens on the same thread a moment later. Nothing has to cross a thread boundary for that to work.

The pool inside investigate is the case to watch. @capsera.agent is on probe, and probe runs on a pool worker, so the same argument applies and those calls are attributed too. But the moment you move a decorator up — onto investigate, where it seems like it belongs — the scope is opened on the branch thread and the probes run somewhere else:

# fragment
import contextvars
from concurrent.futures import ThreadPoolExecutor


def losing_the_scope(hypothesis: str, records: list[dict[str, str]]) -> list[str]:
    # If the scope is opened out here, the workers do not see it.
    with ThreadPoolExecutor(max_workers=PROBE_CONCURRENCY) as pool:
        return list(pool.map(lambda r: probe(hypothesis, r), records))


def keeping_the_scope(hypothesis: str, records: list[dict[str, str]]) -> list[str]:
    # One copy of the caller's context per task, taken on the caller's thread.
    with ThreadPoolExecutor(max_workers=PROBE_CONCURRENCY) as pool:
        futures = [
            pool.submit(contextvars.copy_context().run, probe, hypothesis, r)
            for r in records
        ]
        return [f.result() for f in futures]

Capsera keeps the attribution stack in a ContextVar, and a ContextVar does not cross a thread boundary on its own — it crosses when the code starting the thread copies the context. LangGraph's own fan-out does that. A bare ThreadPoolExecutor does not, and the calls record as unknown with no error and no warning to tell you. Measured on LangGraph with langchain 1.3.14 and capsera 0.3.0.

Copy the context once per task, in the caller, as above. A single Context cannot be entered from two threads at the same time, so reusing one copy across a pool raises at runtime rather than losing attribution quietly — which is at least the better failure.

Async needs nothing extra. A ContextVar is snapshotted per task, so branches awaited concurrently under asyncio.gather keep their own attribution.

The safe rule for this program: put the decorator on the function that makes the call, and if you ever hand work to a thread you started yourself, hand it a copied context too.

Roles, not invocations

At this width the tempting mistake is to make the name unique. capsera.tag() takes a runtime string, so it is two lines to write this:

# fragment
import capsera


def probe_the_wrong_way(hypothesis: str, index: int, record: dict[str, str]) -> str:
    # One agent per invocation: 288 rows, none of which exists on the next run.
    with capsera.tag(f"prober-{index}-{record['id']}"):
        return probe_chain.invoke(
            {"hypothesis": hypothesis, "record": record_text(record)}
        )

Do not. Two things break. The dashboard becomes 288 rows of one call each, which answers no question anybody asked — nobody wants to know what prober-17-run-04 cost. And an agent budget is scoped to an agent name, so a name that only existed during one study can never accumulate against a limit; the role you actually wanted to cap has no row at all.

Name roles. Three hundred invocations of prober is one row with a call count next to it, which is both readable and governable. The per-invocation axis has its own fields on the same event — customer_id and cost_center are parameters on @capsera.agent and @capsera.langgraph_node, and session_id is one on capsera.tag — so you can slice by who a study was for without fragmenting the agent dimension that budgets and cost attribution depend on.

tag() is still the right tool for a genuine sub-scope inside a function: a second adjudication pass on a different model, costed separately from the first. It is the wrong tool for naming an iteration.

Put a budget on the role that multiplies

Three of the four roles make exactly one call per study no matter how wide it goes. The prober makes hypotheses × records, and the first of those numbers comes from a model. That is multi-agent amplification in one sentence, and it is why the cap belongs on the prober rather than on the study.

Create the budget in the dashboard with scope agent and the agent set to prober — the same string as the decorator. That string is the whole join between the two halves of this page: a budget can only act on a boundary the events carry, and spend attributed to unknown cannot be governed by an agent budget, which is the practical reason attribution comes first. Pick the amount from a week of your own prober spend rather than a figure from a tutorial, and set the action to block.

Enforcement is on by default (enable_budget_enforcement=True in init()) and the check runs before the provider call. When the prober is over its budget, BudgetExceededError is raised inside your process and the request never leaves it, so no tokens are spent on it. This is pre-call enforcement, and the cost of it is one round trip to Capsera per call — at 288 probes, 288 round trips.

Then decide what a blocked probe means for the study. The choice is real, because it is your loop that collects the results:

# fragment
import contextvars
from concurrent.futures import ThreadPoolExecutor

import capsera


def investigate(state: Investigation) -> dict:
    records = CORPUS[:PROBES_PER_HYPOTHESIS]
    verdicts: list[str] = []
    missing = 0
    blocked = None

    with ThreadPoolExecutor(max_workers=PROBE_CONCURRENCY) as pool:
        futures = [
            pool.submit(contextvars.copy_context().run, probe, state["hypothesis"], r)
            for r in records
        ]
        for future in futures:
            try:
                verdicts.append(future.result())
            except capsera.BudgetExceededError as exc:
                blocked = exc
                missing += 1
            except Exception:
                missing += 1

    if blocked is not None and not verdicts:
        # A hypothesis with no evidence behind it is not a finding.
        raise blocked

    return {
        "findings": [
            {
                "hypothesis": state["hypothesis"],
                "verdicts": verdicts,
                "missing": missing,
            }
        ]
    }

Some probes blocked and some returned: keep going, and let missing tell the referee the evidence was incomplete — the referee's prompt already says to mark those UNDECIDED rather than guess. Every probe blocked: raise, because a branch with nothing in it would be adjudicated as a real negative result.

LangGraph surfaces a node's exception out of invoke(), so that raise ends the study. Give it one place to land:

# fragment
import capsera


def run_study(question: str) -> str | None:
    """Returns None when the prober budget ended the study."""
    try:
        state = scientist.invoke(
            {"question": question, "findings": []},
            config={"max_concurrency": BRANCH_CONCURRENCY},
        )
    except capsera.BudgetExceededError as exc:
        print(f"prober budget reached, no report produced: {exc}")
        return None
    return state["report"]

BudgetExceededError is the only exception the SDK raises deliberately; everything else fails open. Do not catch it and retry immediately — the budget will still be exceeded. Narrowing the study and rerunning, queueing until the period rolls over, and failing are all defensible; for a batch study, failing usually is.

Three limits to know before you rely on this as a hard ceiling.

The blast radius is the role, not the run. A budget on prober stops prober calls. The referee and the synthesist are different agents with no budget on them, so in the partial case they still run and still cost — the study finishes, on thinner evidence, and pays for the write-up either way. That is usually what you want at this shape, and it is worth deciding on purpose rather than discovering: if a study built on a tenth of the intended evidence is worthless to you, add the guard that ends it rather than leaving the last two roles to spend on nothing.

A wide fan-out can overshoot. The check reads budget state computed from delivered events, and events are delivered in batches every 500 ms (flush_interval_ms), so a burst of concurrent calls can each pass a check taken before any of them was recorded. Here the burst is BRANCH_CONCURRENCY × PROBE_CONCURRENCY — six at the shipped constants, and whatever you raise them to afterwards. Set the budget below a figure you cannot exceed, or lower the two concurrency constants; they are the ones that decide the overshoot, not the two width constants.

It fails open, on purpose. If the pre-call check does not complete within budget_check_timeout (1 second by default), the call is allowed. A network problem between your service and Capsera should not halt production traffic. Of the three actions, block and margin downgrade act at runtime; throttle records the decision but does not delay the call. The full behaviour is in budgets and enforcement.

What this run actually cost

No dollar figures on this page: this program has not been run against a real key, and a number invented here would not be yours anyway — it depends on your corpus, how many hypotheses your planner writes, and today's prices. What is worth predicting is the shape of the result.

Four agent rows — planner, prober, referee, synthesist — under one team, ai-scientist, with a cost per run for the study as a whole. Four rows at nine calls and four rows at 291, which is the only reason a report on a system this wide is readable at all. The call count next to prober is the number that moved.

Read the prober row as cost per call and call count, not as a total. They fail differently and they have different fixes. Call count climbing with cost per call flat is a planner writing more hypotheses, or a corpus that grew; cost per call climbing with the count flat is a prompt that got longer or a max_tokens somebody raised. The total alone cannot tell you which, and at this width the two are worth several times more than the total is.

Which row is largest is not predictable in advance, and that is the point of measuring it rather than reasoning about it. The synthesist makes one call on the more expensive model; the prober makes hundreds of small ones on the cheaper one. Where the crossover falls depends on the width and on how long your records are, and it decides whether the next thing to change is the synthesist's model or PROBES_PER_HYPOTHESIS.

Every probe call sends the same system prompt in front of a different record. Cache read share will read zero, because nothing in this code sets an Anthropic cache_control breakpoint, so that prefix is billed as fresh input on all 288 calls. That is a lever this code leaves on the table and it is invisible on an invoice: prompt caching that never engaged. Turning on enable_prompt_analysis=True in init() puts a number on how often one system-prompt hash repeats, which is the evidence for whether it is worth restructuring the prompt to earn a cache hit.

Each event also carries the file and line that made the call, which points at scientist.py rather than into langgraph, so a row you did not expect traces back to code rather than to a model name.

Questions this page answers

How do I track cost per agent in a LangGraph app? Call capsera.init() once, which patches the provider clients ChatAnthropic constructs underneath, so every call in the process is already recorded. Then put @capsera.langgraph_node("<name>") on each node that makes a call, and @capsera.agent("<name>") on any plain function that makes one. The decorator adds who made the call; it does not change whether the call is captured.

Does Capsera attribution survive a LangGraph fan-out? Yes, when the decorator is on the node, because the scope is opened inside whichever worker runs the node and nothing has to cross a thread boundary. A hand-rolled fan-out inside a node is the case to watch: Capsera keeps the attribution stack in a ContextVar, and a bare ThreadPoolExecutor starts its workers with a fresh context, so those calls record as unknown with no error. Submit contextvars.copy_context().run instead. Measured on LangGraph with langchain 1.3.14 and capsera 0.3.0.

Should each agent invocation get its own name in Capsera? No. Name roles, not invocations. Three hundred invocations of one prober role produce one readable row you can budget; three hundred generated names produce three hundred rows that do not exist on the next run, and an agent budget cannot accumulate against a name that only lived for one study. The per-invocation axis has its own fields on the same event: customer_id, cost_center and session_id.

Can a budget stop one LangGraph node without killing the run? Yes, if you decide that inside the node. A blocking budget scoped to one agent raises BudgetExceededError before the provider call, so no tokens are spent on that role. Catch it where the fan-out results are collected and the rest of the graph still runs on partial evidence; let it out of the node and LangGraph surfaces it from invoke() and the study ends. Capping one role caps that role, not the roles downstream of it.