Building a deep research pipeline in 50 lines

I stream myself building things with n00dles every Thursday, and the single most requested build is some version of "research assistant that doesn't just Google one thing and call it done." So let's build a real one: it scrapes the web and academic papers at the same time, synthesizes both into key findings, and writes a structured report — in well under 50 lines, no framework ceremony, and runnable by the time you finish reading this.

What we're building

Four agents, three steps:

• scrape_web and scrape_papers run at the same time — they don't depend on each other, so there's no reason to make one wait for the other.
• synthesize waits for both, then merges them into a short list of findings.
• write_report takes those findings and turns them into something you'd actually hand to someone.

That's the entire shape of the pipeline. No orchestration framework needed beyond n00dles itself, no separate task queue, no glue code to keep the two scrapers in sync.

Setup

pip install get-n00dles
export ANTHROPIC_API_KEY="sk-ant-..."
export OPENAI_API_KEY="sk-..."

We're mixing providers on purpose in this tutorial — GPT-4o for the web scraper, Claude for everything else — because n00dles wraps litellm under the hood, and switching providers per agent is just a different string in the model= argument. There's no reason every agent in a pipeline needs to use the same model; pick whichever one is actually best (or cheapest) for each specific job.

Step 1: the two scrapers

Every n00dles agent is a plain Python function with a docstring and type hints — the docstring becomes the system prompt, the type hints become the validated I/O contract. No subclassing, no chain object to construct:

from n00dles import agent, pipeline, parallel, run

@agent(model="gpt-4o")
def scrape_web(topic: str) -> list:
    """Find and scrape the 10 most relevant recent articles on the topic."""

@agent(model="claude-haiku-4-5")
def scrape_papers(topic: str) -> list:
    """Find the 5 most relevant academic papers on the topic."""

Both take a topic: str and return a list. n00dles instructs the model to respond in JSON matching that shape and validates the response before it ever reaches the next step — if the model returns something that doesn't parse as a list, you get a clear AgentOutputError instead of a downstream agent silently choking on malformed input three steps later.

Step 2: synthesize, in parallel

This is the part people are usually surprised is this short. To run scrape_web and scrape_papers at the same time instead of one after the other, wrap them in parallel():

@agent(model="claude-sonnet-4-6")
def synthesize(scrape_web: list, scrape_papers: list) -> str:
    """Synthesize web articles and papers into 5-8 key findings."""

Look closely at the parameter names: scrape_web and scrape_papers— exactly matching the two upstream agents' function names. That's how synthesize gets both fan-out results without any manual wiring. When you compose parallel(scrape_web, scrape_papers) >> synthesize, n00dles runs both scrapers concurrently, then hands synthesizea dict keyed by each one's name — matched automatically to its parameter names.

Step 3: write the report

@agent(model="claude-sonnet-4-6")
def write_report(synthesize: str) -> str:
    """Write a structured report with an executive summary and findings."""

Same pattern again: a single-param agent following another single-output step just receives that output directly — no parameter-name gymnastics needed when there's only one upstream node to read from.

Step 4: wire it and run it

research = pipeline(
    parallel(scrape_web, scrape_papers) >> synthesize >> write_report,
    name="deep-research",
    timeout=90,
)

result = run(research, topic="agentic AI in regulated industries")
print(result.output)

That's the whole pipeline. timeout=90sets a 90-second budget per node that doesn't set its own — generous enough for a scraping call, tight enough to fail fast if a provider hangs. Run it, and result.output is the final report from write_report.

Inspecting what actually happened

run()doesn't just give you the final string — it gives you a RunResult with a full trace of every agent call:

print(result.duration_ms)    # wall-clock time for the whole run
print(result.total_tokens)    # summed across every call, including retries
for t in result.agent_traces:
    print(t.name, t.status, t.duration_ms, "ms")

If you compare duration_ms against the sum of scrape_web and scrape_papers's individual durations, you'll see the parallel fan-out actually paid off — the total run time tracks the slower of the two scrapers, not the sum of both.

Leveling up: structured findings

Returning str from synthesize works, but if you want write_report — or anything downstream — to reason about findings programmatically instead of re-parsing prose, swap in a Pydantic model:

from pydantic import BaseModel

class Finding(BaseModel):
    claim: str
    confidence: float
    sources: list[str]

class Findings(BaseModel):
    items: list[Finding]

@agent(model="claude-sonnet-4-6")
def synthesize(scrape_web: list, scrape_papers: list) -> Findings:
    """Synthesize web articles and papers into 5-8 key findings."""

Nothing else in the pipeline needs to change. n00dles sees the Pydantic return type, instructs the model to respond with matching JSON, and validates it before write_report ever sees it — so by the time your downstream code runs, synthesize's output is guaranteed to be a real Findings object with a real list[Finding], not a string you have to hope was formatted correctly.

The full script

Here's everything together, structured output included, right at the 50-line mark promised in the title:

from pydantic import BaseModel
from n00dles import agent, pipeline, parallel, run


class Finding(BaseModel):
    claim: str
    confidence: float
    sources: list[str]


class Findings(BaseModel):
    items: list[Finding]


@agent(model="gpt-4o")
def scrape_web(topic: str) -> list:
    """Find and scrape the 10 most relevant recent articles on the topic."""


@agent(model="claude-haiku-4-5")
def scrape_papers(topic: str) -> list:
    """Find the 5 most relevant academic papers on the topic."""


@agent(model="claude-sonnet-4-6")
def synthesize(scrape_web: list, scrape_papers: list) -> Findings:
    """Synthesize web articles and papers into 5-8 key findings."""


@agent(model="claude-sonnet-4-6")
def write_report(synthesize: Findings) -> str:
    """Write a structured report with an executive summary and findings."""


research = pipeline(
    parallel(scrape_web, scrape_papers) >> synthesize >> write_report,
    name="deep-research",
    timeout=90,
)

if __name__ == "__main__":
    result = run(research, topic="agentic AI in regulated industries")
    print(result.output)
    print(f"{result.total_tokens} tokens, {result.duration_ms:.0f}ms")

Where to take this next

A few natural next steps, all of which slot into this same structure without restructuring anything: add a third parallel source (forums, a Slack export, an internal knowledge base) by adding it to the parallel() call and giving synthesize a matching parameter name; add a fallback= agent on scrape_web in case one provider has a bad day; or route the final report through a branch() based on urgency or topic category. Each of those is a one-line addition, not a restructure — which is, honestly, the entire point of building it this way.