Getting Started › Installation

Installation

Get n00dles running in your Python environment. The core package has minimal dependencies and installs in seconds.

Requirements

Python 3.10 or later
pip 22+ or Poetry 1.5+
At least one LLM API key (Anthropic, OpenAI, etc.)

Install

Install from PyPI with pip:

bash

pip install get-n00dles

Or with Poetry:

bash

poetry add get-n00dles

ℹ

For the latest unreleased features, install from GitHub: pip install git+https://github.com/n00dlehouse/n00dles-py

Configure API keys

n00dles reads LLM credentials from environment variables. Set the key for your preferred provider:

bash

# Anthropic (recommended)
export ANTHROPIC_API_KEY="sk-ant-..."

# OpenAI
export OPENAI_API_KEY="sk-..."

# Mistral
export MISTRAL_API_KEY="..."

✓

Add these to your .env file and use python-dotenv — n00dles will pick them up automatically.

Verify installation

python

import n00dles
print(n00dles.__version__)  # → "0.3.0"

Getting Started › Quick Start

Quick Start

The condensed, copy-paste version. Want the guided, step-by-step walkthrough instead? Head to the interactive Quickstart.

The five-minute version

Install, set a key, define three agents, chain them, run. That's the whole thing:

bash

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

python — pipeline.py

from n00dles import agent, pipeline, run

@agent(model="claude-haiku-4-5")
def researcher(topic: str) -> str:
    """Research the topic. Return 3 key facts."""

@agent(model="claude-sonnet-4-6")
def writer(research: str) -> str:
    """Write a 200-word article from the research."""

content_pipeline = pipeline(researcher >> writer, retry=3)
result = run(content_pipeline, topic="the future of multi-agent AI")
print(result.output)

bash

python pipeline.py

What just happened

researcher ran first, hit Claude Haiku, returned a plain string
n00dles passed that string straight into writer as research
The whole thing retried up to 3× on transient failures, with no extra code from you
A trace event was recorded for each agent call — visible in the dashboard if you're on managed cloud

Want the guided version?

✓

The interactive Quickstart walkthrough covers the same ground with explanations at each step, a progress tracker, and copy buttons for every command. Good if this is your first time with n00dles.

Getting Started › Your First Pipeline

Your First Pipeline

A slower walkthrough than the quick start — building a small research-to-article pipeline from nothing, explaining each piece as we go.

The scenario

Say you want to turn a topic into a short, edited article: research it, write a draft, then polish the draft. That's three distinct jobs, each suited to a different model — a cheap model for research, a capable one for writing, a cheap one again for editing. That's exactly the shape n00dles is built for.

Step 1: define the agents

Each agent is a function. The docstring is the prompt, the signature is the contract:

python

from n00dles import agent, pipeline, run

@agent(model="claude-haiku-4-5")
def researcher(topic: str) -> str:
    """Research the topic. Return 3-5 key facts as bullet points."""

@agent(model="claude-sonnet-4-6")
def writer(research: str) -> str:
    """Write a 250-word article based on the research."""

@agent(model="claude-haiku-4-5")
def editor(draft: str) -> str:
    """Tighten the prose. Fix any awkward phrasing. Keep the meaning intact."""

Notice each function takes the previous one's output type as its input type — that's what makes the next step work.

Step 2: wire the pipeline

The >> operator chains agents sequentially. Wrap the chain in pipeline() to attach retry and timeout behavior to the whole thing:

python

content_pipeline = pipeline(
    researcher >> writer >> editor,
    retry=3,
    timeout=60,
)

ℹ

The retry/timeout on pipeline() apply per-agent, not to the pipeline as a whole — if writer fails, only writer retries, not the entire chain from the start.

Step 3: run it

python

result = run(content_pipeline, topic="why distributed tracing matters")

print(result.output)
print(f"{result.duration_ms}ms, {result.total_tokens} tokens, {len(result.agent_traces)} agent calls")

run() executes the pipeline, returns a RunResult with the final output plus everything you need for debugging — duration, token usage, and a per-agent trace. From here, the natural next steps are running independent agents in parallel, or routing between agents with branch() — both covered in the API reference.

Core Concepts › Agents

Agents

An agent is an LLM-backed function with typed inputs and outputs. In n00dles, you define one by decorating any Python function with @agent.

Defining an agent

python

from n00dles import agent

@agent(model="claude-sonnet-4-6")
def summarizer(text: str) -> str:
    """Summarize the given text in three concise sentences."""

The function's docstring becomes the system prompt. The type annotations define the contract — n00dles validates input and output against them automatically.

How it works

When you call an agent, n00dles:

Validates the input against the type annotations
Constructs a prompt from the docstring + serialized input
Calls the specified LLM with retry + timeout logic
Parses and validates the typed output
Emits a structured trace event

Output types

Agents support all standard Python types as outputs. For structured data, use Pydantic models:

python

from pydantic import BaseModel
from n00dles import agent

class Sentiment(BaseModel):
    label: str       # "positive" | "negative" | "neutral"
    confidence: float  # 0.0 – 1.0
    summary: str

@agent(model="claude-haiku-4-5")
def analyze_sentiment(review: str) -> Sentiment:
    """Analyze the sentiment of the product review."""

result = analyze_sentiment("Absolutely love this product!")
print(result.label)       # → "positive"
print(result.confidence)  # → 0.97

Overriding the prompt

Use the prompt parameter to supply a system prompt explicitly, independent of the docstring:

python

@agent(
    model="claude-sonnet-4-6",
    prompt="""You are a financial analyst specializing in tech sector equities.
Provide structured analysis with explicit uncertainty estimates."""
)
def equity_analyst(ticker: str, context: str) -> dict:
    """Analyze the equity."""

Core Concepts › Pipelines

Pipelines

A pipeline composes agents into a single executable unit — sequential chains, parallel fan-out, or conditional branches — with one shared retry and timeout policy.

Sequential composition

The >> operator chains two agents (or pipelines) so the left side's output becomes the right side's input:

python

chain = researcher >> writer >> editor
content_pipeline = pipeline(chain)

pipeline() wraps the composed chain and gives it a name, retry policy, and timeout — it's the unit you pass to run() or deploy.

Configuring retry & timeout

python

content_pipeline = pipeline(
    researcher >> writer >> editor,
    retry=3,        # per-agent retry attempts
    timeout=60,      # per-agent timeout, seconds
    name="content-pipeline",  # shown in dashboard/traces
)

Retry and timeout set on pipeline() apply to each agent independently, not the chain as a whole. An individual @agent decorator can override either value for itself — the most specific setting wins.

Nesting pipelines

A pipeline is itself a valid left or right operand of >>, so you can build larger systems out of smaller, independently-tested pipelines:

python

research_stage = pipeline(scrape >> summarize, name="research")
writing_stage = pipeline(draft >> edit, name="writing")

full_pipeline = pipeline(research_stage >> writing_stage)

✓

Nested pipelines show up as collapsible groups in trace views — useful for keeping a 15-agent system readable in the dashboard.

Composition reference

Operator / call	Behavior
a >> b	Sequential — a's output is b's input
parallel(a, b, …)	Concurrent — all run at once, results merged
branch(key=agent, …)	Conditional — routes to exactly one agent based on a classifier's output
pipeline(chain, …)	Wraps any of the above with a name, retry, and timeout

Core Concepts › State Management

State Management

n00dles checkpoints pipeline state after every agent, so a crash, restart, or deploy never loses progress mid-pipeline.

How checkpointing works

Every time an agent in a running pipeline completes, n00dles writes a checkpoint — the run ID, which step finished, and that step's output — to the configured state store. If the process dies before the next agent finishes, the pipeline resumes from the last checkpoint instead of from the start.

Checkpoints are written synchronously before the next agent is dispatched
Each checkpoint is keyed by run_id, so concurrent runs of the same pipeline never collide
Completed runs are retained for 30 days by default (configurable) for replay and debugging

Choosing a backend

Set the backend once, globally, via configure():

python

from n00dles import configure

# SQLite — good default for local dev & single-node deploys
configure(state_store="sqlite:///n00dles.db")

# Redis — for multi-worker / horizontally scaled deploys
configure(state_store="redis://localhost:6379/0")

Backend	Good for	Notes
sqlite://	Local dev, single-node, low volume	Zero setup, file-based, ships by default
redis://Soon	Multi-worker, horizontally scaled	Shared state across processes/machines
postgres://Soon	Long-term retention, audit requirements	Available on Team/Enterprise plans

Resuming a crashed pipeline

If you know a run was interrupted, resume it explicitly by ID:

python

from n00dles import resume

result = resume(run_id="run_8f2a1c")
# picks up after the last completed agent — already-finished steps aren't re-run

⚠

Resume re-executes the in-flight agent from scratch — write agents to be idempotent if your pipeline does anything with side effects (sending emails, writing to a database).

Core Concepts › Error Handling

Error Handling

Retry, circuit breakers, and structured failure propagation are built into every agent — you opt out of them, not into them.

Retry behavior

Every agent retries on transient failures — rate limits, timeouts, connection resets — using exponential backoff with jitter. Non-transient failures (a validation error on the output type, for instance) are not retried, since retrying won't fix a malformed response on its own.

python

@agent(model="claude-sonnet-4-6", retry=5)
def flaky_call(x: str) -> str:
    """..."""
# backoff: ~1s, ~2s, ~4s, ~8s, ~16s (± jitter), then raises

Circuit breakersSoon

🔜

Coming soon. CircuitBreakerisn't in the current release yet — the section below describes the planned design. retry (above) and fallback= agents are available today and cover most of the same failure modes.

If an agent fails its retry budget N times in a row across separate calls (not just within one run), its circuit opens — further calls fail immediately without hitting the LLM, for a cooldown window. This protects you from hammering a provider that's already down.

python

from n00dles import agent, CircuitBreaker

@agent(
    model="gpt-4o",
    circuit_breaker=CircuitBreaker(failure_threshold=5, cooldown_s=30)
)
def summarize(text: str) -> str:
    """..."""

Custom error handlers

Attach on_error to a pipeline to react to a failure — log it, page someone, or fall back to a default value instead of raising:

python

def handle_failure(error: AgentError) -> str | None:
    if error.agent_name == "classifier":
        return "general"  # fall back to a default category
    return None  # None re-raises

triage = pipeline(classifier >> handle_support, on_error=handle_failure)

Inspecting failures

Every RunResult carries the full trace, even on failure:

python

from n00dles import AgentError

try:
    result = run(content_pipeline, topic="...")
except AgentError as e:
    print(e.agent_name, e.attempt, e.cause)
    for step in e.trace.agent_traces:
        print(step.name, step.status, step.duration_ms)

ℹ

On managed cloud, the same trace renders as a visual timeline in the dashboard — no need to print it yourself.

API Reference › @agent

@agent

The @agent decorator transforms any Python function into an LLM-backed agent with built-in retry, timeout, type validation, and tracing.

Signature

python

@agent(
    model: str,
    prompt: str | None = None,
    timeout: int = 60,
    retry: int = 3,
    temperature: float = 0.7,
    max_tokens: int | None = None,
    tags: list[str] = [],
)

Parameters

Parameter	Type	Default	Description
model	str	required	The LLM model identifier. e.g. "claude-sonnet-4-6", "gpt-4o", "mistral-large".
prompt	str \| None	None	System prompt. If None, the function's docstring is used.
timeout	int	60	Timeout in seconds per attempt. Raises TimeoutError if exceeded.
retry	int	3	Maximum retry attempts on transient failures. Uses exponential backoff with jitter.
temperature	float	0.7	LLM sampling temperature (0.0–2.0). Lower = more deterministic.
max_tokens	int \| None	None	Cap on output tokens. None defers to model default.
tags	list[str]	[]	Arbitrary tags attached to trace events for filtering in the dashboard.

Returns

The decorator returns a callable with the same signature as the decorated function. The return type is the declared return annotation, validated via Pydantic.

⚠

If the LLM output cannot be coerced into the declared return type after all retries, an AgentOutputError is raised.

Examples

python

# Minimal — model is the only required arg
@agent(model="claude-haiku-4-5")
def translator(text: str, target_lang: str) -> str:
    """Translate the text into the target language."""

# With overrides for a production-critical agent
@agent(
    model="claude-sonnet-4-6",
    timeout=120,
    retry=5,
    temperature=0.2,
    tags=["production", "kyc"]
)
def kyc_extractor(document: str) -> CustomerRecord:
    """Extract structured KYC data from the document."""

API Reference › pipeline()

pipeline()

Wraps a composed chain of agents into a single named, configured, runnable unit.

Signature

python

def pipeline(
    chain: Agent | ParallelAgent | BranchAgent,
    name: str | None = None,
    retry: int = 3,
    timeout: int = 60,
    on_error: Callable | None = None,
) -> Pipeline

Parameters

Parameter	Type	Default	Description
chain	Agent \| ParallelAgent \| BranchAgent	required	The composed structure to run — typically built with `>>`, `parallel()`, or `branch()`.
name	str \| None	None	Identifier shown in traces and the dashboard. Defaults to an auto-generated name if omitted.
retry	int	3	Default retry budget for every agent in the chain that doesn't set its own.
timeout	int	60	Default per-agent timeout in seconds for every agent in the chain.
on_error	Callable \| None	None	Called with an `AgentError` on failure. Return a fallback value to swallow the error, or `None` to re-raise.

Returns

A Pipeline instance. It's callable directly, or passed to run(), resume(), or the noodles deploy CLI.

Examples

python

# Sequential, with a custom error handler
support = pipeline(
    classifier >> branch(billing=handle_billing, default=handle_general),
    name="support-triage",
    retry=2,
    on_error=log_and_fallback,
)

# Pipelines compose — a pipeline is a valid chain for another pipeline
full = pipeline(research_pipeline >> writing_pipeline, name="full-content-flow")

API Reference › run()

run()

Executes a pipeline or a single agent synchronously and returns a RunResult with the output and full trace.

Signature

python

def run(
    target: Pipeline | Agent,
    timeout: int | None = None,
    tags: list[str] = [],
    **inputs,
) -> RunResult

Parameters

Parameter	Type	Default	Description
target	Pipeline \| Agent	required	What to execute. A bare agent runs as a one-step pipeline.
timeout	int \| None	None	Overrides the whole-run timeout. None defers to the pipeline's own setting.
tags	list[str]	[]	Extra tags merged onto this run's trace, on top of any tags set on individual agents.
**inputs	any	—	Keyword arguments forwarded to the first agent in the chain.

RunResult

The object returned by every successful run:

Field	Type	Description
output	any	The final agent's return value
run_id	str	Unique ID for this run — pass to `resume()` if interrupted
duration_ms	int	Wall-clock time for the whole run
total_tokens	int	Summed token usage across every agent call, including retries
agent_traces	list[AgentTrace]	Per-agent timing, status, and token usage, in execution order

Examples

python

# Run a full pipeline
result = run(content_pipeline, topic="Banking 5.0")

# Run a single agent directly — useful for testing one step in isolation
result = run(researcher, topic="Banking 5.0")

# Override timeout and tag this run for filtering in the dashboard
result = run(content_pipeline, timeout=120, tags=["backfill"], topic="...")

API Reference › parallel()

parallel()

Runs multiple agents concurrently against the same input and merges their results — n00dles handles the fan-out and fan-in for you.

Signature

python

def parallel(
    *agents: Agent,
    max_concurrency: int | None = None,
) -> ParallelAgent

Parameters

Parameter	Type	Default	Description
*agents	Agent	required	Two or more agents to run concurrently. Each receives the same input.
max_concurrency	int \| None	None	Caps how many of the agents run at once. None runs all of them simultaneously.

Result shape

The next agent in the chain receives a dict keyed by each upstream agent's function name — no manual merging required:

python

@agent(model="claude-sonnet-4-6")
def merge_signals(scrape_news: list, scrape_twitter: list) -> dict:
    """Merge and rank both signals. Parameter names match the upstream agents' function names exactly."""

ℹ

If a downstream agent only declares some of the upstream parameter names, n00dles passes just those — you don't have to accept every branch's output.

Examples

python

from n00dles import agent, pipeline, parallel, run

@agent(model="gpt-4o")
def scrape_news(query: str) -> list:
    """Scrape latest news for the query."""

@agent(model="gpt-4o")
def scrape_twitter(query: str) -> list:
    """Pull recent posts for the query."""

intel = pipeline(parallel(scrape_news, scrape_twitter) >> merge_signals, timeout=20)
result = run(intel, query="AI regulation 2026")

API Reference › branch()

branch()

Routes execution to exactly one of several agents, based on a key returned by the upstream agent.

Signature

python

def branch(
    default: Agent | None = None,
    **routes: Agent,
) -> BranchAgent

Parameters

Parameter	Type	Default	Description
default	Agent \| None	None	Runs when the routing key matches none of the named routes. If None, an unmatched key raises `BranchError`.
**routes	Agent	required	Maps a routing key (string) to the agent that should handle it.

Routing key

The upstream agent decides the route. If its output is a dict, n00dles looks for a category key by convention; if its output is a plain str, the string itself is the key:

python

@agent(model="claude-haiku-4-5")
def classifier(ticket: str) -> dict:
    """Classify the ticket. Return {category, confidence}."""
    # category must be one of the route keys below, e.g. "billing" or "support"

Examples

python

from n00dles import agent, pipeline, branch, run

@agent(model="claude-sonnet-4-6")
def handle_support(ticket: str) -> str:
    """Handle customer support inquiry."""

@agent(model="gpt-4o-mini")
def handle_billing(ticket: str) -> str:
    """Handle billing and payment inquiry."""

triage = pipeline(
    classifier >> branch(
        support=handle_support,
        billing=handle_billing,
        default=handle_support,
    ),
    retry=2,
)
result = run(triage, ticket="My invoice is wrong")

Guides › Production Deploy

Production Deploy

Ship a pipeline as a live HTTP endpoint with one CLI command — no Dockerfile or infra config required to get started.

🔜

Coming soon. The noodlesCLI isn't in the current release yet — everything below describes the planned design. Today, ship a pipeline by deploying your own Python process (Docker, a serverless function, a long-running worker) and calling run() / arun() directly.

The deploy CLI

bash

noodles deploy pipeline.py --name content-pipeline
# → builds a container, ships it, and gives you a live URL
# → https://run.n00dles.io/your-org/content-pipeline

The deployed endpoint accepts a POST request with your pipeline's input arguments as JSON, and returns the RunResult serialized the same way:

bash

curl -X POST https://run.n00dles.io/your-org/content-pipeline \
  -H "Content-Type: application/json" \
  -d '{"topic": "Banking 5.0"}'

Deployment targets

Target	Flag	Good for
n00dles Cloud	`(default)`	Zero-config, managed, on Pro/Team plans
AWS Lambda	`--target lambda`	Sporadic traffic, pay-per-invocation
Docker	`--target docker`	Self-hosting on your own infra / K8s
Fly.io / Railway	`--target flyio`	Always-on workers, simple ops

Environment & secrets

Secrets in your local .env aren't uploaded automatically — push them explicitly so they end up in the deploy target's secret store, not in a build artifact:

bash

noodles secrets push --env .env --name content-pipeline

⚠

noodles deploy never reads or uploads .env on its own — secrets are a separate, explicit step on purpose.

Rollbacks

Every deploy is versioned. Roll back instantly if a deploy misbehaves:

bash

noodles deploy list --name content-pipeline
noodles deploy rollback --name content-pipeline --version 12

Guides › Testing

Testing

Mock any agent or LLM provider and run full pipeline integration tests without spending a single API token.

Mocking individual agents

mock_agent swaps an agent's implementation for a fixed return value for the duration of the context manager — no network call happens:

python

from n00dles.testing import mock_agent

def test_writer_uses_research():
    with mock_agent(researcher, returns="fact 1, fact 2, fact 3"):
        result = run(content_pipeline, topic="test topic")
    assert "fact 1" in result.output

Mocking a whole pipeline

For tests where you only care about what calls a pipeline, not what the pipeline does, mock every agent at once:

python

from n00dles.testing import mock_pipeline

def test_endpoint_calls_pipeline():
    with mock_pipeline(content_pipeline, returns="mocked article"):
        response = client.post("/generate", json={"topic": "x"})
    assert response.status_code == 200

✓

Both mock_agent and mock_pipeline work as decorators too, if you'd rather not nest a with block.

Snapshot testing outputs

For agents with structured (Pydantic) output, snapshot the shape rather than the exact LLM text — LLM outputs vary token-for-token even at temperature 0 across model versions:

python

def test_sentiment_shape():
    result = run(analyze_sentiment, review="Great product!")
    assert result.output.label in ("positive", "negative", "neutral")
    assert 0.0 <= result.output.confidence <= 1.0

ℹ

Reserve real LLM calls (no mocking) for a small set of CI "canary" tests that run on a schedule, not on every PR — that keeps your test suite both fast and free of flaky model-output assertions.

Guides › Observability

Observability

Every token, latency, and tool call is traced automatically. Export it anywhere, or view it in the hosted dashboard.

Built-in trace events

Every agent call emits a structured event with no extra code required:

Field	Description
agent_name	The decorated function's name
model	Model identifier used for the call
status	ok / retried / failed
duration_ms	Wall-clock time for this specific call
tokens_in / tokens_out	Token usage for this call
tags	Tags from the agent, the pipeline, and the run, merged

Exporting traces

Point n00dles at any OpenTelemetry-compatible collector, or use a built-in exporter:

python

from n00dles import configure

# Generic OpenTelemetry — available now
configure(otel_endpoint="https://otel-collector.internal:4317")

# Built-in exporters — coming soon
configure(trace_exporter="langfuse", langfuse_public_key="...")
configure(trace_exporter="helicone", helicone_api_key="...")

🔜

Coming soon.The dedicated Langfuse and Helicone exporters aren't in the current release yet. The generic OpenTelemetry exporter above is real and available today (pip install get-n00dles[otel]) — point it at any OTel collector, including ones Langfuse/Helicone already accept.

✓

Exporters are additive — you can send to OpenTelemetry and Langfuse at the same time, or none at all and just read RunResult.agent_traces directly.

Dashboard metrics

On managed cloud, every trace also lands in the hosted dashboard automatically — total runs, success rate, p50 latency, and a per-pipeline breakdown, with no exporter configuration needed. See it in action on the dashboard preview.

Examples › Research Pipeline

Research Pipeline

Scrape, summarize, analyze, and write a structured report from a raw topic — with the scrape-and-summarize step fanned out in parallel per source.

The pipeline

python — research_pipeline.py

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."""

@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."""

@agent(model="claude-sonnet-4-6")
def write_report(synthesize: str) -> 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,
)

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

Why it's built this way

Two scrapers run in parallel because they don't depend on each other — running them sequentially would just be wasted wall-clock time
A dedicated synthesize step merges both sources before writing, so the report writer never has to juggle two input shapes itself
Cheaper models for scraping (gpt-4o, haiku), a stronger model only for the synthesis and writing steps that actually need the reasoning

Examples › Content Factory

Content Factory

Research, draft, SEO-optimize, and localize into three languages — fanned out in parallel — before a final human review gate.

The pipeline

python — content_factory.py

from n00dles import agent, pipeline, parallel, run

@agent(model="claude-haiku-4-5")
def research(topic: str) -> str:
    """Research the topic, return key points and sources."""

@agent(model="claude-sonnet-4-6")
def draft(research: str) -> str:
    """Write a 900-word blog post from the research."""

@agent(model="claude-haiku-4-5")
def seo_optimize(draft: str) -> str:
    """Add headings, meta description, and keyword-optimized intro."""

@agent(model="gpt-4o-mini")
def localize_es(seo_optimize: str) -> str:
    """Translate and culturally adapt the post for Spanish-speaking readers."""

@agent(model="gpt-4o-mini")
def localize_de(seo_optimize: str) -> str:
    """Translate and culturally adapt the post for German-speaking readers."""

factory = pipeline(
    research >> draft >> seo_optimize >> parallel(localize_es, localize_de),
    name="content-factory",
    retry=3,
)

# Runs daily via cron, scheduled with `noodles schedule`
result = run(factory, topic="Q3 product roadmap")

Why it's built this way

Localization fans out in parallel at the end — each language is independent once there's a finished, SEO-optimized English draft
retry=3 on the whole pipeline because this runs unattended on a schedule — nobody's watching it fail at 3am
Cheap models everywhere except drafting — research, SEO, and translation are all lower-stakes than the actual prose

Examples › Document Processor

Document Processor

Extract structured data from PDFs and contracts at scale — n00dles handles up to 1,000 documents in parallel with full audit trails.

The pipeline

python — document_processor.py

from pydantic import BaseModel
from n00dles import agent, pipeline, run_batch

class InvoiceRecord(BaseModel):
    vendor: str
    amount: float
    due_date: str
    line_items: list[dict]

@agent(model="claude-haiku-4-5")
def extract(document_text: str) -> InvoiceRecord:
    """Extract structured invoice data from the document."""

@agent(model="claude-sonnet-4-6")
def validate(extract: InvoiceRecord) -> dict:
    """Flag anomalies: amount mismatches, duplicate invoices, missing vendor info."""

pipeline_def = pipeline(extract >> validate, name="invoice-intake", retry=2)

# run_batch fans out across up to 1,000 documents concurrently
results = run_batch(
    pipeline_def,
    inputs=[{"document_text": text} for text in load_documents()],
    max_concurrency=50,
)

Why it's built this way

A typed Pydantic output on extract means malformed extractions raise immediately instead of silently propagating bad data downstream
A separate validate step keeps anomaly detection logic out of the extraction prompt — easier to tune independently
run_batch with max_concurrency=50 caps in-flight LLM calls so you don't blow through provider rate limits on a 1,000-document run

Examples › Support Triage

Support Triage

Classify, route, draft, and gate behind human review — handling roughly 80% of tickets automatically with full escalation logic.

The pipeline

python — support_triage.py

from n00dles import agent, pipeline, branch, run

@agent(model="claude-haiku-4-5")
def classify(ticket: str) -> dict:
    """Classify: {category: billing|technical|account|other, urgency: low|med|high}"""

@agent(model="claude-sonnet-4-6")
def draft_billing_reply(ticket: str) -> str:
    """Draft a reply for a billing inquiry, citing the relevant policy."""

@agent(model="claude-sonnet-4-6")
def draft_technical_reply(ticket: str) -> str:
    """Draft a reply for a technical issue, including troubleshooting steps."""

@agent(model="claude-haiku-4-5")
def needs_human(draft: str, urgency: str) -> bool:
    """Return True if this reply should be reviewed by a human before sending."""

triage = pipeline(
    classify >> branch(
        billing=draft_billing_reply,
        technical=draft_technical_reply,
        default=draft_technical_reply,
    ),
    name="support-triage",
    retry=2,
)

result = run(triage, ticket="My subscription charged me twice this month")
# high-urgency or needs_human=True replies route to a review queue instead of auto-sending

Why it's built this way

classify returns urgency alongside category — the routing decision and the escalation decision use the same upstream call, not two separate LLM round-trips
default=draft_technical_reply means an unrecognized category still gets a reasonable attempt instead of a hard failure
needs_human as a gate, not a branch — every reply gets drafted either way; the gate only decides whether a human sees it before it goes out