Get 2,500 events tracked for freeSign up now

Capsera
Start free
All tutorials
Muhammad Kumail15 min

Cost per round in a LangChain deep research agent

A planner, batched researchers over a fixed corpus, a gap-finder and a writer in LangChain 1.x on Claude, then per-agent cost and a budget on the fan-out.

Framework
LangChain 1.x
Provider
anthropic
Python
3.11+
Capsera
pip install capsera

This builds a deep research agent in LangChain 1.x against Claude: a planner turns one question into sub-questions, researchers answer them in parallel from a fixed corpus, a gap-finder decides whether a second round is needed, and a writer produces a cited brief. One question is four model calls at the floor and eleven at the cap, and where in that range a run lands is decided by two models rather than by your code. The last third of this page adds Capsera: one init() call to capture every request, four decorator lines to say which role made it, and a blocking budget on the role that multiplies.

What you will build

brief.py, a single-file agent with four roles and a supervisor:

  • planner — one call. Reads a cheap index of the corpus and the question, writes at most MAX_PER_ROUND sub-questions.
  • researcher — one call per open sub-question, run in parallel. Answers from the documents a code-side retriever hands it. The multiplier.
  • gap-finder — one call per round boundary. Reads the notes so far and either says ENOUGH or names the specific facts still missing, which is what buys a second round of researchers.
  • writer — one call, on the stronger model. Produces the brief with a document id on every factual sentence.

Retrieval is not a model call. A keyword retriever in Python picks the documents for each sub-question, and the citation check at the end is a regex, not a fifth role. Both are deliberate: a research agent that spends model calls on jobs a function can do has a cost curve that nothing in the prompt explains.

Each role is its own function. That matters later — it is the seam the instrumentation attaches to — and it is the better shape regardless, because a supervisor that inlines four model calls has nothing you can name, budget, or move to a different model.

Written against LangChain 1.x with langchain-anthropic, calling Anthropic's API directly rather than through a gateway. Models: claude-haiku-4-5 for the planner, the researchers and the gap-finder; claude-sonnet-5 for the writer.

Prerequisites

Python 3.11 or newer.

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

The major is pinned so this page stays true as LangChain moves. langchain-anthropic is left unpinned on purpose: install the one that matches your langchain 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 corpus, and retrieve from it in code

Start with the corpus, not the agent. A research agent pointed at a live search tool has an unbounded number of calls in it, because the model decides both how many searches to run and how much text each one drags into the next prompt. An agent handed a fixed set of documents has a call count you can multiply out before you run anything, which is what makes the rest of this page measurable.

The corpus below is a fixture — six short documents standing in for whatever your retriever returns. It is built so that no single document answers the question: the symptom is in one, the change that caused it is in another, and the reason that change matters is in a third. That is the only honest reason to build a research agent instead of making one model call.

"""brief.py — the evidence pack a deep research agent is allowed to read."""
import re

CORPUS = {
    "postmortem-2026-03-14": (
        "Checkout p99 latency rose from 410 ms to 2.6 s within twenty minutes "
        "of the gateway v9 rollout reaching the third region. Rolling the "
        "canary back restored p99 to 430 ms. Error rate never moved. Root "
        "cause was not determined before the incident was closed."
    ),
    "loadtest-checkout-2026-03-02": (
        "Pre-migration load test, checkout path, 1200 requests per second for "
        "thirty minutes against gateway v8. p50 88 ms, p99 410 ms, no errors. "
        "Run with the default upstream keepalive pool of 512 connections."
    ),
    "config-diff-gateway-v9": (
        "Config changes shipped with gateway v9:\n"
        "  upstream_keepalive_pool: 512 -> 0\n"
        "  tls_session_cache: on -> off\n"
        "  access_log_sampling: 1.0 -> 0.1\n"
        "The keepalive change was made to stop the connection leak reported "
        "against v8."
    ),
    "vendor-changelog-tls-9.2": (
        "Release 9.2 disables tls_session_cache by default because the shared "
        "cache was not partitioned per virtual host. Operators terminating TLS "
        "to upstreams should re-enable it, or expect a full handshake on every "
        "new upstream connection."
    ),
    "support-thread-4471": (
        "Twelve reports of slow checkout since 14 March, mostly mobile. "
        "Several customers say the first attempt hangs and a retry is fast. "
        "Nobody reports an error page. Closed as could-not-reproduce."
    ),
    "slo-checkout": (
        "Checkout is a tier-1 path. p99 must stay at or under 800 ms measured "
        "at the edge over a rolling hour. The monthly error budget covers "
        "availability only; latency breaches are reviewed individually."
    ),
}

QUESTION = (
    "Why did checkout p99 regress after the March gateway migration, and what "
    "should we change before resuming the rollout?"
)

_WORD = re.compile(r"[a-z0-9]+")
_STOP = frozenset(
    "the a an and or of to in on for is are was were it its this that with "
    "from at by we our not no".split()
)


def _terms(text: str) -> set[str]:
    return {w for w in _WORD.findall(text.lower()) if len(w) > 2 and w not in _STOP}


def doc_index() -> str:
    return "\n".join(
        f"{doc_id}: {' '.join(body.split()[:12])}..." for doc_id, body in CORPUS.items()
    )


def doc_text(doc_id: str) -> str:
    return CORPUS[doc_id]


def retrieve(query: str, k: int) -> list[str]:
    wanted = _terms(query)
    scored = sorted(
        ((len(wanted & _terms(body)), doc_id) for doc_id, body in CORPUS.items()),
        key=lambda pair: (-pair[0], pair[1]),
    )
    return [doc_id for score, doc_id in scored[:k] if score]


if __name__ == "__main__":
    print(doc_index())

doc_index() is the cheap artefact that makes the expensive one avoidable. The planner sees six one-line stubs, not six documents, so the call that decides what to ask costs a fraction of the calls that do the answering. On a real corpus you would build the same index from titles and keep it to one line per document for the same reason.

The retriever is twelve lines of set intersection and it is the right tool here. Sub-questions are written by a model that has just read the index, so they reuse the corpus's own words, which is exactly the case keyword overlap handles. Reach for embeddings when the questions and the documents stop sharing vocabulary — and know that you are adding a per-question call to an embedding model to the count below when you do.

retrieve sorts on (-score, doc_id), so ties break on the id rather than on dictionary order, and it drops documents that scored zero instead of padding the prompt with text nobody asked for. Both matter for cost: k is a ceiling on documents per call, not a quota to fill.

Step 2 — Plan, with the breadth bound in code

The planner decides how wide the first round goes. So the cap lives in Python, not in the prompt — a model asked for "at most five" will occasionally write seven, and the slice is what makes that harmless.

# fragment
from langchain_anthropic import ChatAnthropic
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

FAST_MODEL = "claude-haiku-4-5"
MAX_PER_ROUND = 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]


PLAN_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You break a research question into sub-questions, each answerable "
            "from one or two documents. One question per line, no numbering and "
            "no commentary. Every question must be answerable from the corpus "
            "described in the index — never ask for data nobody has. At most "
            "{max_questions} questions.",
        ),
        ("human", "Question:\n{question}\n\nDocument index:\n{index}"),
    ]
)

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


def plan(question: str) -> list[str]:
    raw = plan_chain.invoke(
        {
            "question": question,
            "index": doc_index(),
            "max_questions": MAX_PER_ROUND,
        }
    )
    return _lines(raw, MAX_PER_ROUND)

Lines rather than JSON. A one-question-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.

"Never ask for data nobody has" is the instruction that pays for itself. Given a question about a latency regression, a planner with no idea what it can read will ask for flame graphs, per-region traces and a database slow-log — and each of those becomes a research call that retrieves nothing, answers UNANSWERED, and is billed in full. Showing the planner the index is how you stop paying for questions the corpus cannot touch.

Step 3 — Answer the open questions in parallel

Five sequential reads that do not depend on each other is five round trips of latency for nothing. batch() runs them concurrently and max_concurrency bounds how many are in flight.

# fragment
from typing import Any

RESEARCH_CONCURRENCY = 4
DOCS_PER_QUESTION = 2

RESEARCH_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You answer one narrow question from the documents you are given "
            "and from nothing else. Reply with at most three sentences, then a "
            "final line reading CITES: followed by the ids of the documents you "
            "used, comma separated. If the documents do not answer the "
            "question, reply with the single word UNANSWERED.",
        ),
        ("human", "Question: {question}\n\nDocuments:\n{sources}"),
    ]
)

researcher_llm = ChatAnthropic(model=FAST_MODEL, max_tokens=256, temperature=0)
research_chain = RESEARCH_PROMPT | researcher_llm | StrOutputParser()


def _sources_for(question: str) -> tuple[str, list[str]]:
    doc_ids = retrieve(question, DOCS_PER_QUESTION)
    body = "\n\n".join(f"[{d}]\n{doc_text(d)}" for d in doc_ids)
    return body or "(no document matched this question)", doc_ids


def research(questions: list[str], round_no: int) -> list[dict[str, Any]]:
    prepared = [(q, *_sources_for(q)) for q in questions]
    answers = research_chain.batch(
        [{"question": q, "sources": s} for q, s, _ in prepared],
        config={"max_concurrency": RESEARCH_CONCURRENCY},
    )

    notes: list[dict[str, Any]] = []
    for (question, _, doc_ids), answer in zip(prepared, answers, strict=True):
        text = answer.strip()
        if text.upper().startswith("UNANSWERED"):
            continue
        notes.append(
            {
                "question": question,
                "answer": text,
                "retrieved": doc_ids,
                "round": round_no,
            }
        )
    return notes

One question per research call, not the whole plan per call. That is what keeps widening the search affordable: the prompt is a fixed instruction plus at most DOCS_PER_QUESTION documents, so raising MAX_PER_ROUND multiplies a small number instead of growing a large one. It also means an UNANSWERED is a judgement about documents the model actually had in front of it, rather than a guess about the corpus as a whole.

max_tokens=256 on the researcher is load-bearing rather than tidiness. Output tokens are the expensive half of a call, this role runs once per open question, and the three sentences it is asked for fit easily. A researcher allowed to write an essay is the same search at several times the price, and it hands the writer a longer prompt too — twice, because the gap-finder reads the notes as well.

Dropping UNANSWERED answers rather than keeping them is a choice with a cost attached: those calls are already paid for, and the only thing they buy you is the knowledge that the question was unanswerable. Their number is the signal to watch, because a planner producing them is a planner asking about a corpus it cannot see.

zip(..., strict=True) because pairing questions to answers by position is only safe if the lengths agree, and a silent truncation here would attach an answer to the wrong question — which then gets cited in a brief.

round_no is stamped on every note. It costs nothing now and it is the thing you want when a brief turns out to rest entirely on follow-up questions the first round never thought to ask.

Step 4 — Find the gaps, and bound the depth

This is the role that makes the agent a research agent rather than a fan-out. It reads the notes and says what is still missing, and that answer is what buys another round of researchers.

It is also where the whole thing turns into a runaway loop if you let it. Research, find gaps, research the gaps, find more gaps is a cycle with no natural end: there is always one more question available, and every lap is a fresh fan-out. Two bounds, both in Python, both readable as constants: MAX_ROUNDS caps depth and MAX_QUESTIONS caps the total number of sub-questions across every round.

# fragment
MAX_ROUNDS = 2
MAX_QUESTIONS = 8

GAP_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You decide whether a set of research notes answers the original "
            "question. If they do, reply with the single word ENOUGH. "
            "Otherwise write one question per line, each naming a specific "
            "missing fact and each answerable from the corpus in the index. "
            "Never repeat a question that has already been answered, and never "
            "ask for a document that is not in the index.",
        ),
        (
            "human",
            "Question:\n{question}\n\nNotes so far:\n{notes}\n\n"
            "Document index:\n{index}",
        ),
    ]
)

gap_llm = ChatAnthropic(model=FAST_MODEL, max_tokens=384, temperature=0)
gap_chain = GAP_PROMPT | gap_llm | StrOutputParser()


def format_notes(notes: list[dict[str, Any]]) -> str:
    return "\n\n".join(f"Q: {n['question']}\nA: {n['answer']}" for n in notes)


def find_gaps(question: str, notes: list[dict[str, Any]]) -> list[str]:
    raw = gap_chain.invoke(
        {
            "question": question,
            "notes": format_notes(notes),
            "index": doc_index(),
        }
    )
    if raw.strip().upper().startswith("ENOUGH"):
        return []
    return _lines(raw, MAX_PER_ROUND)

The gap-finder is asked for the questions, not for how many to run. It returns everything it named, capped only at MAX_PER_ROUND, and the supervisor decides how many of those it can afford. That split is what makes the next section possible: the questions you could not afford are a list you have, so a run cut short by the cap can say what it does not know rather than reading like a complete brief.

"Never ask for a document that is not in the index" repeats the planner's instruction because the gap-finder is in a worse position to obey it. It has just read notes that stop short of an answer, and the obvious next move — ask for the data that would settle it — is exactly the move the corpus cannot support.

Step 5 — Write the brief, and check the citations in Python

Do not ask a model to verify its own citations. A second call that reviews the first one's references costs as much as the writer and is graded by the same faculty that produced them. The check is a regex over the ids the notes actually retrieved, it is exact, and it is free.

# fragment
WRITER_MODEL = "claude-sonnet-5"

BRIEF_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You write a research brief from notes. Structure: one paragraph "
            "answering the question, then the changes you recommend as a short "
            "list, then a final line reading OPEN: naming what is still "
            "unsettled. Every factual sentence ends with a document id in "
            "square brackets, taken from the ids listed as available. Claim "
            "nothing the notes do not contain. The OPEN line must include every "
            "question listed as never researched.",
        ),
        (
            "human",
            "Question:\n{question}\n\nNotes:\n{notes}\n\n"
            "Available document ids:\n{available}\n\n"
            "Questions that were never researched:\n{unresolved}",
        ),
    ]
)

writer_llm = ChatAnthropic(model=WRITER_MODEL, max_tokens=1024, temperature=0)
brief_chain = BRIEF_PROMPT | writer_llm | StrOutputParser()

_CITATION = re.compile(r"\[([a-z0-9\-.]+)\]")


def write_brief(
    question: str, notes: list[dict[str, Any]], unresolved: list[str]
) -> str:
    available = sorted({d for n in notes for d in n["retrieved"]})
    return brief_chain.invoke(
        {
            "question": question,
            "notes": format_notes(notes),
            "available": ", ".join(available),
            "unresolved": "\n".join(unresolved) or "none",
        }
    )


def unsupported_citations(brief: str, notes: list[dict[str, Any]]) -> list[str]:
    seen = {d for n in notes for d in n["retrieved"]}
    return sorted({c for c in _CITATION.findall(brief) if c not in seen})

The writer is handed the ids that were retrieved, not the corpus index. A writer that can see a document id it has no note about will cite it, and the citation will look correct — the id is real, it just was not read on this run. unsupported_citations is what catches the case anyway, and it returns a list rather than raising, because a brief with one bad reference is worth reading with that reference flagged.

The writer 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.

Step 6 — The supervisor

One function, and every branch the agent can take is visible in it.

# fragment
def research_brief(question: str) -> dict[str, Any]:
    questions = plan(question)
    if not questions:
        raise RuntimeError("the planner produced no sub-question")

    notes: list[dict[str, Any]] = []
    unresolved: list[str] = []
    asked = 0

    # Runs at least once, so round_no is always bound below.
    for round_no in range(MAX_ROUNDS):
        notes.extend(research(questions, round_no))
        asked += len(questions)
        if round_no == MAX_ROUNDS - 1:
            break
        gaps = find_gaps(question, notes)
        allowance = min(MAX_PER_ROUND, MAX_QUESTIONS - asked)
        questions, unresolved = gaps[:allowance], gaps[allowance:]
        if not questions:
            break

    if not notes:
        raise RuntimeError("no sub-question could be answered from the corpus")

    brief = write_brief(question, notes, unresolved)
    return {
        "brief": brief,
        "rounds": round_no + 1,
        "questions_asked": asked,
        "unresolved": unresolved,
        "unsupported": unsupported_citations(brief, notes),
    }

allowance is where the two caps meet. MAX_PER_ROUND bounds how wide any one fan-out can be and MAX_QUESTIONS - asked bounds the total, so a gap-finder that names five follow-ups after a five-question first round gets three of them and the other two land in unresolved. Neither number is negotiable by a model, which is the only kind of bound worth having.

Count the calls: one planner, one to five researchers, one gap-finder, zero to three more researchers, one writer. Four at the floor and eleven at the cap. Both of those numbers come from your constants; nothing in your file decides where between them a given question lands.

The whole file

"""brief.py — a deep research agent: plan, research, find gaps, write."""
import re
from typing import Any

from langchain_anthropic import ChatAnthropic
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

FAST_MODEL = "claude-haiku-4-5"
WRITER_MODEL = "claude-sonnet-5"

MAX_PER_ROUND = 5
MAX_ROUNDS = 2
MAX_QUESTIONS = 8
RESEARCH_CONCURRENCY = 4
DOCS_PER_QUESTION = 2

CORPUS = {
    "postmortem-2026-03-14": (
        "Checkout p99 latency rose from 410 ms to 2.6 s within twenty minutes "
        "of the gateway v9 rollout reaching the third region. Rolling the "
        "canary back restored p99 to 430 ms. Error rate never moved. Root "
        "cause was not determined before the incident was closed."
    ),
    "loadtest-checkout-2026-03-02": (
        "Pre-migration load test, checkout path, 1200 requests per second for "
        "thirty minutes against gateway v8. p50 88 ms, p99 410 ms, no errors. "
        "Run with the default upstream keepalive pool of 512 connections."
    ),
    "config-diff-gateway-v9": (
        "Config changes shipped with gateway v9:\n"
        "  upstream_keepalive_pool: 512 -> 0\n"
        "  tls_session_cache: on -> off\n"
        "  access_log_sampling: 1.0 -> 0.1\n"
        "The keepalive change was made to stop the connection leak reported "
        "against v8."
    ),
    "vendor-changelog-tls-9.2": (
        "Release 9.2 disables tls_session_cache by default because the shared "
        "cache was not partitioned per virtual host. Operators terminating TLS "
        "to upstreams should re-enable it, or expect a full handshake on every "
        "new upstream connection."
    ),
    "support-thread-4471": (
        "Twelve reports of slow checkout since 14 March, mostly mobile. "
        "Several customers say the first attempt hangs and a retry is fast. "
        "Nobody reports an error page. Closed as could-not-reproduce."
    ),
    "slo-checkout": (
        "Checkout is a tier-1 path. p99 must stay at or under 800 ms measured "
        "at the edge over a rolling hour. The monthly error budget covers "
        "availability only; latency breaches are reviewed individually."
    ),
}

QUESTION = (
    "Why did checkout p99 regress after the March gateway migration, and what "
    "should we change before resuming the rollout?"
)

_WORD = re.compile(r"[a-z0-9]+")
_BULLET = re.compile(r"^\s*(?:[-*•]|\d+[.)])\s*")
_CITATION = re.compile(r"\[([a-z0-9\-.]+)\]")
_STOP = frozenset(
    "the a an and or of to in on for is are was were it its this that with "
    "from at by we our not no".split()
)

PLAN_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You break a research question into sub-questions, each answerable "
            "from one or two documents. One question per line, no numbering and "
            "no commentary. Every question must be answerable from the corpus "
            "described in the index — never ask for data nobody has. At most "
            "{max_questions} questions.",
        ),
        ("human", "Question:\n{question}\n\nDocument index:\n{index}"),
    ]
)

RESEARCH_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You answer one narrow question from the documents you are given "
            "and from nothing else. Reply with at most three sentences, then a "
            "final line reading CITES: followed by the ids of the documents you "
            "used, comma separated. If the documents do not answer the "
            "question, reply with the single word UNANSWERED.",
        ),
        ("human", "Question: {question}\n\nDocuments:\n{sources}"),
    ]
)

GAP_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You decide whether a set of research notes answers the original "
            "question. If they do, reply with the single word ENOUGH. "
            "Otherwise write one question per line, each naming a specific "
            "missing fact and each answerable from the corpus in the index. "
            "Never repeat a question that has already been answered, and never "
            "ask for a document that is not in the index.",
        ),
        (
            "human",
            "Question:\n{question}\n\nNotes so far:\n{notes}\n\n"
            "Document index:\n{index}",
        ),
    ]
)

BRIEF_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You write a research brief from notes. Structure: one paragraph "
            "answering the question, then the changes you recommend as a short "
            "list, then a final line reading OPEN: naming what is still "
            "unsettled. Every factual sentence ends with a document id in "
            "square brackets, taken from the ids listed as available. Claim "
            "nothing the notes do not contain. The OPEN line must include every "
            "question listed as never researched.",
        ),
        (
            "human",
            "Question:\n{question}\n\nNotes:\n{notes}\n\n"
            "Available document ids:\n{available}\n\n"
            "Questions that were never researched:\n{unresolved}",
        ),
    ]
)

planner_llm = ChatAnthropic(model=FAST_MODEL, max_tokens=512, temperature=0)
researcher_llm = ChatAnthropic(model=FAST_MODEL, max_tokens=256, temperature=0)
gap_llm = ChatAnthropic(model=FAST_MODEL, max_tokens=384, temperature=0)
writer_llm = ChatAnthropic(model=WRITER_MODEL, max_tokens=1024, temperature=0)

plan_chain = PLAN_PROMPT | planner_llm | StrOutputParser()
research_chain = RESEARCH_PROMPT | researcher_llm | StrOutputParser()
gap_chain = GAP_PROMPT | gap_llm | StrOutputParser()
brief_chain = BRIEF_PROMPT | writer_llm | StrOutputParser()


def _terms(text: str) -> set[str]:
    return {w for w in _WORD.findall(text.lower()) if len(w) > 2 and w not in _STOP}


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]


def doc_index() -> str:
    return "\n".join(
        f"{doc_id}: {' '.join(body.split()[:12])}..." for doc_id, body in CORPUS.items()
    )


def doc_text(doc_id: str) -> str:
    return CORPUS[doc_id]


def retrieve(query: str, k: int) -> list[str]:
    wanted = _terms(query)
    scored = sorted(
        ((len(wanted & _terms(body)), doc_id) for doc_id, body in CORPUS.items()),
        key=lambda pair: (-pair[0], pair[1]),
    )
    return [doc_id for score, doc_id in scored[:k] if score]


def _sources_for(question: str) -> tuple[str, list[str]]:
    doc_ids = retrieve(question, DOCS_PER_QUESTION)
    body = "\n\n".join(f"[{d}]\n{doc_text(d)}" for d in doc_ids)
    return body or "(no document matched this question)", doc_ids


def format_notes(notes: list[dict[str, Any]]) -> str:
    return "\n\n".join(f"Q: {n['question']}\nA: {n['answer']}" for n in notes)


def plan(question: str) -> list[str]:
    raw = plan_chain.invoke(
        {
            "question": question,
            "index": doc_index(),
            "max_questions": MAX_PER_ROUND,
        }
    )
    return _lines(raw, MAX_PER_ROUND)


def research(questions: list[str], round_no: int) -> list[dict[str, Any]]:
    prepared = [(q, *_sources_for(q)) for q in questions]
    answers = research_chain.batch(
        [{"question": q, "sources": s} for q, s, _ in prepared],
        config={"max_concurrency": RESEARCH_CONCURRENCY},
    )

    notes: list[dict[str, Any]] = []
    for (question, _, doc_ids), answer in zip(prepared, answers, strict=True):
        text = answer.strip()
        if text.upper().startswith("UNANSWERED"):
            continue
        notes.append(
            {
                "question": question,
                "answer": text,
                "retrieved": doc_ids,
                "round": round_no,
            }
        )
    return notes


def find_gaps(question: str, notes: list[dict[str, Any]]) -> list[str]:
    raw = gap_chain.invoke(
        {
            "question": question,
            "notes": format_notes(notes),
            "index": doc_index(),
        }
    )
    if raw.strip().upper().startswith("ENOUGH"):
        return []
    return _lines(raw, MAX_PER_ROUND)


def write_brief(
    question: str, notes: list[dict[str, Any]], unresolved: list[str]
) -> str:
    available = sorted({d for n in notes for d in n["retrieved"]})
    return brief_chain.invoke(
        {
            "question": question,
            "notes": format_notes(notes),
            "available": ", ".join(available),
            "unresolved": "\n".join(unresolved) or "none",
        }
    )


def unsupported_citations(brief: str, notes: list[dict[str, Any]]) -> list[str]:
    seen = {d for n in notes for d in n["retrieved"]}
    return sorted({c for c in _CITATION.findall(brief) if c not in seen})


def research_brief(question: str) -> dict[str, Any]:
    questions = plan(question)
    if not questions:
        raise RuntimeError("the planner produced no sub-question")

    notes: list[dict[str, Any]] = []
    unresolved: list[str] = []
    asked = 0

    # Runs at least once, so round_no is always bound below.
    for round_no in range(MAX_ROUNDS):
        notes.extend(research(questions, round_no))
        asked += len(questions)
        if round_no == MAX_ROUNDS - 1:
            break
        gaps = find_gaps(question, notes)
        allowance = min(MAX_PER_ROUND, MAX_QUESTIONS - asked)
        questions, unresolved = gaps[:allowance], gaps[allowance:]
        if not questions:
            break

    if not notes:
        raise RuntimeError("no sub-question could be answered from the corpus")

    brief = write_brief(question, notes, unresolved)
    return {
        "brief": brief,
        "rounds": round_no + 1,
        "questions_asked": asked,
        "unresolved": unresolved,
        "unsupported": unsupported_citations(brief, notes),
    }


if __name__ == "__main__":
    result = research_brief(QUESTION)
    print(result["brief"])
    print(
        f"\n{result['rounds']} round(s), {result['questions_asked']} question(s) asked"
    )
    if result["unsupported"]:
        print(f"unsupported citations: {', '.join(result['unsupported'])}")

python brief.py

You get a brief with a document id on each sentence, a count of rounds and questions, and a list of any citations the notes do not support. The interesting part is which documents it reaches: an answer that names the keepalive pool and the TLS session cache together has combined three documents no one of which says it, and an answer that stops at "the rollout caused it" has stayed inside the postmortem. Both are real outcomes of an agent shaped like this, and the difference between them is the gap-finder.

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 four to eleven calls behind it are indistinguishable — same key, same account, three of the four roles on the same model.

pip install capsera

At the top of brief.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 client that ChatAnthropic constructs underneath, so from that call on every request the agent makes is recorded: the planner's, each researcher's inside batch(), the gap-finder's, the writer's. The prompts are untouched, the chains are untouched, batch() 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__":
    result = research_brief(QUESTION)
    print(result["brief"])
    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 decorator only says who:

# fragment
@capsera.agent("planner", team="research-desk", task_type="decomposition")
def plan(question: str) -> list[str]:
    ...  # body unchanged


@capsera.agent("researcher", team="research-desk", task_type="evidence-lookup")
def research(questions: list[str], round_no: int) -> list[dict[str, Any]]:
    ...  # body unchanged


@capsera.agent("gap-finder", team="research-desk", task_type="sufficiency")
def find_gaps(question: str, notes: list[dict[str, Any]]) -> list[str]:
    ...  # body unchanged


@capsera.agent("writer", team="research-desk", task_type="brief")
def write_brief(
    question: str, notes: list[dict[str, Any]], unresolved: list[str]
) -> str:
    ...  # body unchanged

Four added lines, and that is the entire diff. No function body changed, no call site changed, no model construction changed, no argument threaded through the supervisor. Every LLM call made while a decorated function is on the stack is attributed to that agent, including calls made by the functions it calls and by the framework it invokes.

The fan-out keeps its identity. research does its work in research_chain.batch(...), which dispatches to worker threads rather than calling on the thread you were on, and the researcher scope is still what gets recorded. Capsera holds the attribution stack in a ContextVar and LangChain's batch() dispatches through a context-copying executor, so the scope the caller was in is the scope the worker sees — verified empirically on LangChain 1.3 with capsera 0.3.0. batch() also produces one event per underlying call rather than one per batch, so five open questions are five researcher events (coverage table).

The credit there belongs to LangChain, and it is worth knowing where the guarantee stops. A ContextVar does not cross a thread boundary on its own — it crosses when the code starting the thread copies the context, which batch() does and a hand-rolled pool does not. Fan out the questions yourself with a bare ThreadPoolExecutor and the same decorator records unknown instead of researcher, with no error to tell you. Measured on the same versions:

# fragment
import contextvars
from concurrent.futures import ThreadPoolExecutor

# Loses the scope: workers start with a fresh context.
with ThreadPoolExecutor(max_workers=RESEARCH_CONCURRENCY) as pool:
    answers = list(pool.map(research_one, questions))

# Keeps it: hand each worker a copy of the caller's context.
ctx = contextvars.copy_context()
with ThreadPoolExecutor(max_workers=RESEARCH_CONCURRENCY) as pool:
    answers = list(pool.map(lambda q: ctx.run(research_one, q), questions))

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

Leave the supervisor undecorated. research_brief makes no LLM call of its own; every call inside it happens while one of the four roles is on the stack, and the innermost scope is the one recorded. Decorating it would add a row that can never have spend on it. For a per-question total, use the team field — which is what team="research-desk" on all four decorators is for.

Cost the follow-up round separately

Depth is the part of this agent whose cost you cannot see from the role rows alone. A run that stopped after one round and a run that went two both report as researcher, so the row grows and nothing in it says the growth came from follow-up questions rather than from a wider first pass. capsera.tag() opens a sub-scope inside a function you have already decorated, which is exactly this case:

# fragment
import capsera


@capsera.agent("researcher", team="research-desk", task_type="evidence-lookup")
def research(questions: list[str], round_no: int) -> list[dict[str, Any]]:
    if round_no == 0:
        return _research(questions, round_no)
    # A follow-up round is the same agent answering questions the first round
    # created. Cost the depth separately from the breadth.
    with capsera.tag(
        "researcher", team="research-desk", task_type="evidence-followup"
    ):
        return _research(questions, round_no)

_research is the original body, moved down one level and otherwise unchanged. The follow-up keeps the agent name, so a budget scoped to researcher still counts every round, and it changes the task type, so the share of research spend that came from depth is a number you can read instead of a suspicion.

Use tag() for a genuine sub-scope like this one. It is the wrong tool for naming an iteration: tagging each research call with its sub-question would give you one row per question, none of which exists on the next run, and an agent budget cannot accumulate against a name that lived for one brief. The per-invocation axis has its own fields on the same event — customer_id and cost_center on @capsera.agent, session_id on capsera.tag — so you can slice by which team or which corpus a brief was for without fragmenting the agent dimension that budgets and cost attribution depend on.

Put a budget on the role that multiplies

The planner makes one call per run. The gap-finder makes one per round boundary, and MAX_ROUNDS decides how many of those there can be. The writer makes one. All three of those counts are in your file. The researcher's is not, and it is worse than the usual case: it is breadth multiplied by depth, and both factors are chosen by a model. The planner decides how many questions the first round asks; the gap-finder decides whether there is a second round and what it carries. Point this agent at a corpus of four hundred documents instead of six and raise the caps to the width that finds anything, and the researcher is the only row that moves. That is multi-agent amplification in one sentence, and it is why the cap belongs on the researcher rather than on the run.

Create the budget in the dashboard with scope agent and the agent set to researcher — 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 a per-agent budget. Pick the amount from a week of your own research 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 researcher 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: the check happens where your code is, not in front of the provider, and a blocked call costs you the check rather than the completion.

Then decide what a blocked research call means for the brief. batch() raises on the first failure and abandons the remaining inputs unless you tell it otherwise, so this is a decision you make rather than one you inherit:

# fragment
import capsera


def _research(questions: list[str], round_no: int) -> list[dict[str, Any]]:
    prepared = [(q, *_sources_for(q)) for q in questions]
    answers = research_chain.batch(
        [{"question": q, "sources": s} for q, s, _ in prepared],
        config={"max_concurrency": RESEARCH_CONCURRENCY},
        return_exceptions=True,
    )

    notes: list[dict[str, Any]] = []
    blocked = 0
    for (question, _, doc_ids), answer in zip(prepared, answers, strict=True):
        if isinstance(answer, capsera.BudgetExceededError):
            blocked += 1
            continue
        if isinstance(answer, BaseException):
            raise RuntimeError(f"researching {question!r} failed: {answer}")
        text = answer.strip()
        if text.upper().startswith("UNANSWERED"):
            continue
        notes.append(
            {
                "question": question,
                "answer": text,
                "retrieved": doc_ids,
                "round": round_no,
            }
        )
    if blocked and not notes:
        raise capsera.BudgetExceededError(
            f"researcher budget reached, {blocked} question(s) unread in round {round_no}"
        )
    return notes

return_exceptions=True asks batch() to hand failures back as items in the result list instead of raising on the first one, which is what lets you tell a budget block from a flaky call and count the questions that went unread. Without it, one blocked call ends the round and the remaining questions are never even attempted.

A round that came back empty raises; a round that came back partial does not. That asymmetry is the judgement call, and it is worth arguing with. A brief has somewhere to put ignorance — the OPEN line, and the unresolved list that feeds it — so a partial round is a brief that says what it could not check. A brief with no notes at all has nothing to say and no way to say so, and the writer is the expensive call, so paying for it to produce a paragraph of hedging is the worst outcome available.

That leaves the supervisor to carry the questions the budget stopped, which is one branch on the loop it already has:

# fragment
import capsera


def research_brief(question: str) -> dict[str, Any]:
    questions = plan(question)
    if not questions:
        raise RuntimeError("the planner produced no sub-question")

    notes: list[dict[str, Any]] = []
    unresolved: list[str] = []
    asked = 0

    for round_no in range(MAX_ROUNDS):
        try:
            notes.extend(research(questions, round_no))
        except capsera.BudgetExceededError:
            if not notes:
                raise
            # An earlier round is on the record. Write from it, and say what
            # this round never got to ask.
            unresolved = unresolved + questions
            break
        asked += len(questions)
        if round_no == MAX_ROUNDS - 1:
            break
        gaps = find_gaps(question, notes)
        allowance = min(MAX_PER_ROUND, MAX_QUESTIONS - asked)
        questions, unresolved = gaps[:allowance], gaps[allowance:]
        if not questions:
            break

    if not notes:
        raise RuntimeError("no sub-question could be answered from the corpus")

    brief = write_brief(question, notes, unresolved)
    return {
        "brief": brief,
        "questions_asked": asked,
        "unresolved": unresolved,
        "unsupported": unsupported_citations(brief, notes),
    }

The questions the budget blocked join the ones the cap already dropped, in the same list, and go to the writer through the same prompt variable. There is no second code path for "we ran out of money" — it is the path the agent already had for "we ran out of allowance", which is the only reason this degrade is cheap enough to be worth doing.

The refusal still needs one place to land:

# fragment
import capsera


def run_once(question: str) -> dict[str, Any] | None:
    """Returns None when a budget stopped the run before any evidence existed."""
    try:
        return research_brief(question)
    except capsera.BudgetExceededError as exc:
        print(f"researcher budget reached, no brief produced: {exc}")
        return None

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 MAX_PER_ROUND and rerunning, queueing the question until the period rolls over, and failing are all defensible; for an agent that answers questions from a queue, 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 researcher stops research calls. The planner, the gap-finder and the writer are different agents with no budget on them, so a run that gets one usable round still reaches the writer on the stronger model and still pays for it. That is the right shape here — the researcher is the row that grows with the corpus — but it is worth deciding on purpose rather than discovering: if what you want capped is the writer's model spend, that is a second budget on writer, not a side effect of this one.

A parallel round can overshoot. The check reads budget state computed from delivered events, and events are delivered in batches on an interval (flush_interval_ms, 500 ms by default), so a burst of concurrent calls can each pass a check taken before any of them was recorded. Here the burst is RESEARCH_CONCURRENCY — four as shipped, and whatever you raise it to when the corpus gets bigger. Set the budget below a figure you cannot exceed, or lower the concurrency; it is the concurrency that decides the overshoot, not MAX_PER_ROUND.

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 agent 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 sub-questions your planner writes, and today's prices. What is worth predicting is the shape of the result.

Four agent rows — planner, researcher, gap-finder, writer — under one team, research-desk, with a cost per run for the question as a whole, and the researcher row split by task type into evidence-lookup and evidence-followup.

Read each row as call count and cost per call, not as a total. Three things move on this agent and each has a different fix. The researcher's call count moving in evidence-lookup means the planner is asking more questions. The same count moving in evidence-followup means the gap-finder is less often satisfied, which is the more expensive of the two because a follow-up round also pays for the gap-finder call that authorised it. And the researcher's cost per call moving means the retriever matched bigger documents, which is a property of your corpus rather than of your prompt.

Which row is largest is not predictable in advance, and that is the point of measuring it rather than reasoning about it. The writer makes one call on the more expensive model with every note in the prompt; the researchers make up to eight small ones on the cheaper model. Where the crossover falls depends on how wide you set the caps and on how long the notes get, and it decides whether the next thing to change is the writer's model or the width of the search.

The gap-finder is the row worth watching for a reason the others are not: its input is the notes, so its cost per call grows with every round it authorises. It is cheap when it matters least and dearest when it matters most, which is an argument for reading its cost per call rather than its count.

The researcher calls in a round send text that repeats. Every one carries the same instruction, and the retriever will hand the same document to several sub-questions when several sub-questions are about the same thing. The cache read share column is what tells you whether any of that repetition was billed twice, and it is a number to read rather than one to predict: a shared prefix has to be long and identical to be worth anything, and putting the instruction and the documents before the question is the change that earns it. Turning on enable_prompt_analysis=True in init() puts a figure on how often one system-prompt hash repeats, which is the evidence for whether that restructuring is worth doing: prompt caching that never engaged.

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

Questions this page answers

How do I get per-agent cost for a LangChain deep research agent? Call capsera.init() once, which patches the provider client ChatAnthropic constructs underneath, so every call the agent makes is already recorded. Then put @capsera.agent("<role>") on each role function — the planner, the researcher, the gap-finder, the writer. The decorator says who made the call; it does not change whether the call is captured, and no function body, call site or model construction changes.

Which agent in a deep research pipeline should carry the budget? The researcher, because it is the only role whose call count is set by two models rather than by your code. The planner decides how wide the first round goes and the gap-finder decides whether there is a second one, so researcher calls are breadth multiplied by depth. The planner, the gap-finder and the writer are one call each per round.

How do I stop a deep research agent from running unbounded rounds? Put both bounds in Python, not in a prompt. MAX_ROUNDS caps depth and MAX_QUESTIONS caps the total number of sub-questions across all rounds, so the gap-finder can name follow-ups it will not get. Then pass the questions it named but you did not research to the writer, so a run cut short by the cap says what it does not know instead of reading like a complete brief.

Why does the same research question cost a different amount each run? Three numbers move and none of them is a line you wrote. The planner chooses how many sub-questions the first round asks. The gap-finder chooses whether there is a second round and how many questions it carries. And the retriever decides which documents each research call carries, so input tokens track the size of whichever documents matched. Read call count and cost per call per role, not the total.

Questions this page answers

How do I get per-agent cost for a LangChain deep research agent?
Call capsera.init() once, which patches the provider client ChatAnthropic constructs underneath, so every call the agent makes is already recorded. Then put @capsera.agent("<role>") on each role function — the planner, the researcher, the gap-finder, the writer. The decorator says who made the call; it does not change whether the call is captured, and no function body, call site or model construction changes.
Which agent in a deep research pipeline should carry the budget?
The researcher, because it is the only role whose call count is set by two models rather than by your code. The planner decides how wide the first round goes and the gap-finder decides whether there is a second one, so researcher calls are breadth multiplied by depth. The planner, the gap-finder and the writer are one call each per round.
How do I stop a deep research agent from running unbounded rounds?
Put both bounds in Python, not in a prompt. MAX_ROUNDS caps depth and MAX_QUESTIONS caps the total number of sub-questions across all rounds, so the gap-finder can name follow-ups it will not get. Then pass the questions it named but you did not research to the writer, so a run cut short by the cap says what it does not know instead of reading like a complete brief.
Why does the same research question cost a different amount each run?
Three numbers move and none of them is a line you wrote. The planner chooses how many sub-questions the first round asks. The gap-finder chooses whether there is a second round and how many questions it carries. And the retriever decides which documents each research call carries, so input tokens track the size of whichever documents matched. Read call count and cost per call per role, not the total.

Give every agent an identity, a budget, and hard limits.

One line of code. Anthropic, OpenAI, and Google Gemini.

See pricing