A research brief writer

An agent that plans, gathers, judges, and writes — exercising every primitive and a couple of wrappers.

This recipe walks the canonical Bridle example end to end. The full file lives at bridle/examples/brief_writer.py. The shape:

Plan — produce a Plan of topics and angles.
Gather — for each topic, loop until enough sources are collected.
Judge — branch on whether the evidence is sufficient.
Write — produce a Brief.

It uses step, branch, loop, @agent, @tool, and the cache and retry wrappers.

The full agent

from __future__ import annotations

import argparse
from pydantic import BaseModel

import bridle
from bridle import agent, branch, cache, loop, retry, step, tool
from bridle.models.anthropic import install


class Topic(BaseModel):
    title: str
    angle: str


class Plan(BaseModel):
    topics: list[Topic]


class Source(BaseModel):
    url: str
    summary: str


class Brief(BaseModel):
    headline: str
    body: str


@tool
def search(query: str) -> list[str]:
    """Search the web. Returns up to 10 result URLs.

    This is a stub for the example — replace with a real search backend.
    """
    slug = query.lower().replace(" ", "-")[:40]
    return [f"https://example.com/{slug}/{i}" for i in range(5)]


@agent(input=str, output=Brief, model="claude-sonnet-4-6", token_budget=200_000)
def brief_writer(topic: str) -> Brief:
    plan = cache(step("draft a research plan with two distinct angles", schema=Plan, context=topic))

    sources: list[Source] = []
    for t in plan.topics:
        found = loop(
            f"gather distinct sources on {t.title}",
            schema=Source,
            until=lambda acc: len(acc) >= 2,
            tools=[search],
            max_iterations=4,
        )
        sources.extend(found)

    if not branch("is the evidence sufficient?", context=sources):
        return brief_writer(f"{topic} — go deeper on whatever's underdocumented")

    return retry(
        step("write the brief", schema=Brief, context=(topic, sources)),
        attempts=2,
    )


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("topic", nargs="?", default="the weather on Mars")
    args = parser.parse_args()

    install()
    brief = bridle.resolve(brief_writer(args.topic))

    print(f"# {brief.headline}\n")
    print(brief.body)


if __name__ == "__main__":
    main()

Run it:

ANTHROPIC_API_KEY=sk-ant-... python brief_writer.py "the weather on Mars"

Walking the agent

Schemas first. Four Pydantic models. Topic and Plan describe the planning step's output. Source is what each loop iteration returns. Brief is the final return value.

The search tool. Type-hinted, docstring-described, decorated with @tool. The body is a stub; in production you replace it with a real backend (see adding a real tool).

The agent declaration. input=str validates the first argument. output=Brief validates the return. model="claude-sonnet-4-6" becomes the default for every step inside. token_budget=200_000 caps the total tokens this run can spend.

Step 1: plan. A step with Plan as the schema. Wrapped in cache so re-running the agent on the same topic doesn't re-plan from scratch. The default cache key includes the topic, so two different topics get distinct entries.

Step 2: gather. A for loop in plain Python iterates the topics from the plan. For each, a loop primitive runs until the predicate fires — len(acc) >= 2 — using the search tool to find URLs and producing Source instances. max_iterations=4 is a hard ceiling; the loop raises LoopExhaustedError if it never satisfies the predicate.

Step 3: judge. A branch with the default bool schema. The if triggers resolution; the model produces True or False. On False, recurse with a sharper prompt — recursion is just a function call returning the inner agent's result.

Step 4: write. A step with Brief as the schema, wrapped in retry(..., attempts=2). If the model produces something that fails validation twice in a row, SchemaSatisfactionError bubbles. The retry handles transient model errors.

What you'd change in production

Replace the search tool. Wire it to a real search API. See adding a real tool.

Persist the cache. Swap the implicit MemoryCache for FileCache so plans survive restarts:

from bridle.cache.file import FileCache
bridle.set_cache(FileCache("./.bridle-cache"))

Trace it. Capture the trace and ship it to a log aggregator:

from bridle import Trace
from bridle.trace import set_active_trace

trace = Trace()
set_active_trace(trace)
brief = bridle.resolve(brief_writer("Mars weather"))
print(trace.to_jsonl())

Tighten the budget. token_budget=200_000 is generous. Lower it once you know your typical run.

Handle LoopExhaustedError. Right now an exhausted loop kills the agent. You may want to fall back to a partial brief:

from bridle import LoopExhaustedError

try:
    found = loop(..., max_iterations=4)
except LoopExhaustedError as exc:
    found = exc.accumulator   # use what we got

A research brief writer

The full agent

Walking the agent

What you'd change in production

Next

Adding a real tool

Caching

Traces

On this page