Examples

The repo’s examples/ directory has 33 numbered scripts, each showing one feature from start to finish. Many run offline with a scripted or echo model; the rest need the API key noted next to them. Clone the repo and run them from its root.

# .env should contain OPENAI_API_KEY=sk-...
# Agent + retrieval + memory:
python examples/01_rag_pdf.py                       # default backend (unstructured)
python examples/01_rag_pdf.py --backend docling     # alt PDF backend
python examples/02_specialist_debate.py
python examples/03_multi_user_sessions.py
python examples/04_structured_outputs.py
python examples/05_memory_showcase.py
 
# Workflow primitives:
python examples/06_workflow_chain.py                # no API key required
python examples/07_workflow_route.py
python examples/08_workflow_loop.py
python examples/09_workflow_as_tool.py
python examples/10_workflow_architecture.py
python examples/11_workflow_custom_step.py
 
# Production observability (no API key required):
python examples/12_audit_log.py
python examples/13_telemetry.py
 
# Reasoning effort across providers (ANTHROPIC_API_KEY):
python examples/14_effort_dial.py
 
# Declarative config (no API key required):
python examples/15_config_file.py
 
# Shared workspace + prompt caching + living plan:
python examples/16_shared_workspace.py     # OPENAI_API_KEY
python examples/17_prompt_caching.py       # ANTHROPIC_API_KEY or OPENAI_API_KEY
python examples/18_living_plan.py          # no API key required
python examples/19_workspace_lifecycle.py  # no API key required
 
# Persistent subagents, OS sandbox, run-until-goal, document indexing:
python examples/20_persistent_subagents.py # no API key required
python examples/21_sandbox_os.py           # no API key required
python examples/22_run_until_goal.py
python examples/23_index_document.py
 
# Newer features, all offline unless noted (see the table below):
python examples/24_tool_search_code_mode.py
python examples/25_durable_resume.py
python examples/26_hitl_approvals.py
python examples/27_guardrails.py
python examples/28_eval_harness.py
python examples/29_resilience_governance.py
python examples/30_graph_memory_and_budget.py
python examples/31_model_routing.py
python examples/32_decisions.py            # live with TYPESAFE_API_KEY
python examples/33_decisions_images.py     # live with TYPESAFE_API_KEY

Each example that needs a key ships a graceful skip when OPENAI_API_KEY isn’t set: it prints a hint and exits 0 so a make examples-style runner doesn’t fail.

01. RAG over PDFs

Single-agent RAG over a folder of PDFs. Loader → chunker → vector store → retriever-as-tool → agent. About 100 lines.

Picks the backend at the CLI. --backend unstructured (default, Apache 2.0, what LangChain wraps) or --backend docling (MIT, IBM Research, 2026 best-in-class on native PDFs). Each backend lands in its own Chroma collection / persist directory (general_docs_<backend>) so swapping backends doesn’t require manual cache busting.

pip install 'loomflow[loader-pdf,vectorstore-chroma,openai]'           # default
pip install 'loomflow[loader-pdf-docling,vectorstore-chroma,openai]'   # for --backend docling

The example imports load_pdf directly to surface the backend choice. The auto-dispatch load(pdf) also works (uses the unstructured default). Pick whichever fits your code style.

See: End-to-end RAG tutorial for a guided walkthrough · PDF loader for the backend / strategy reference.

Read: examples/01_rag_pdf.py

02. Specialist debate

Five domain specialists (IT / physics / medicine / finance / law), each with their own folder of PDFs and their own Chroma collection, composed via Team.debate(...) with a synthesising judge agent.

See: Multi-Agent Debate.

Read: examples/02_specialist_debate.py

03. Multi-user sessions

Multi-user namespacing + conversation continuity on one shared Agent + InMemoryMemory. Demonstrates that user_id is a hard partition (Alice’s history never surfaces in Bob’s recall) and that reusing session_id rehydrates prior turns as real chat history. Also shows tools reading scope via get_run_context().

See: Memory · Tools.

Read: examples/03_multi_user_sessions.py

04. Structured outputs

Type-safe structured outputs. Define a Pydantic BaseModel, pass it as output_schema=, get a validated typed instance back on result.parsed. Demonstrates schema-driven extraction (a MeetingSummary with nested ActionItems, ISO dates, sentiment enum) from a raw meeting transcript.

from pydantic import BaseModel
from loomflow import Agent
 
class ActionItem(BaseModel):
    owner: str
    description: str
    due_date: str | None
 
class MeetingSummary(BaseModel):
    title: str
    attendees: list[str]
    decisions: list[str]
    actions: list[ActionItem]
    sentiment: Literal["positive", "neutral", "negative"]
 
agent = Agent("Extract a structured summary.", model="gpt-4.1-mini")
result = await agent.run(transcript, output_schema=MeetingSummary)
 
summary: MeetingSummary = result.parsed   # validated, typed

Read: examples/04_structured_outputs.py

05. Memory showcase

Every memory backend behind one parameter. Walks through inmemory / sqlite / chroma / postgres / redis (Postgres/Redis skip gracefully without a DSN), demonstrates profile(user_id=) / forget(user_id=) / export(user_id=) GDPR ops, and shows the Consolidator extracting structured facts from raw chat episodes. The memory= parameter is the only thing that changes between backends.

See: Memory · GDPR ops.

Read: examples/05_memory_showcase.py

Workflow primitives

Each file is small (50–200 lines) and demonstrates one workflow pattern in isolation. Read them in order. Each builds on the previous one’s vocabulary.

06. Linear chain (no LLM)

Linear Workflow.chain([...]) of plain async functions. The simplest possible workflow shape. No LLM involved, no API key required. Touches RunContext propagation, WorkflowResult.visited, per_step introspection.

See: Workflow.chain.

Read: examples/06_workflow_chain.py

07. Classify + dispatch

Uses Workflow.route(classifier, {"a": agent_a, ...}) to classify the question with a tiny model, then dispatch it to a specialist Agent. Demonstrates “Agent as a workflow node” composition with developer-controlled branching.

See: Workflow.route · Composition.

Read: examples/07_workflow_route.py

08. Refinement loop (cycles)

Refinement loop with cycles: draft → review → judge → (revise → review → ... → END). Shows add_router with END sentinels, max_visits_per_node safety cap, and graceful cap-exceeded handling via try/except RuntimeError + the in-place state dict.

See: the cycles section of Explicit graph builder.

Read: examples/08_workflow_loop.py

09. Workflow as tool

Uses wf.as_tool(), the opposite composition direction: an open-ended customer-support Agent has a deterministic refund workflow available as a tool. Unified audit log shows agent’s tool_call AND workflow’s per-step entries under one user_id.

See: Composition (direction 2).

Read: examples/09_workflow_as_tool.py

10. Architecture inside a workflow

Agent with architecture="self-refine" inside a workflow chain. Demonstrates that workflow shape and agent architecture are orthogonal axes. The architecture is encapsulated inside the agent step; the workflow doesn’t see the internal draft → critique → refine iteration.

See: Architectures · Composition.

Read: examples/10_workflow_architecture.py

11. Custom step wrapping an Agent

Agent wrapped in a custom async def step. For when “just call agent.run(prev_output)” isn’t enough. Multi-field prompt formatting, capturing RunResult metadata (tokens, turns) into workflow state, post-processing the agent’s output.

See: The @step decorator.

Read: examples/11_workflow_custom_step.py

Production observability

The last two examples exercise the framework’s observability spine , no API key required, both run with in-memory backends so you can inspect the captured data directly.

12. Audit log (HMAC-signed, JSONL on disk)

Builds an Agent + a Workflow with a shared FileAuditLog, runs both, and inspects what was written. Five things this example covers:

• Two backends behind one protocol. InMemoryAuditLog for tests and notebooks, FileAuditLog for production.
• Per-user_id filtering. audit.query(user_id="alice") is a partitioned read, not a payload scan.
• HMAC tamper detection. verify_signature(entry, secret=) returns False for any mutation of the canonical payload, and for the wrong secret. Catches both tampering and secret-rotation mistakes. (To check a whole log, including deleted or reordered entries via the prev_hash chain, use await audit.verify(); see Verifying the whole log.)
• Restart recovery. A fresh FileAuditLog against the same path continues from the file’s last entry (taking a file lock per append), so new entries don’t collide even with several writers.
• Dict-config form for verbatim capture. Hand Agent or Workflow an audit_log={"scope_full": True, ...} dict and the resolver builds the right backend, wrapped in FullTranscriptAuditLog. Prompts, outputs, and full tool-result bodies all land in the log. The default is compliance-friendly: truncated prompts, no outputs recorded.

No API keys required. Uses EchoModel.

See: Audit log attribution.

Read: examples/12_audit_log.py

13. Telemetry (four sinks, no collector required)

Runs the same scripted agent against four different telemetry sinks that ship with Loomflow. No OpenTelemetry SDK, no collector deploy.

• InMemoryTelemetry: Accumulates CapturedSpan / CapturedMetric records in lists. Assert on them in tests directly.
• ConsoleTelemetry: Prints span lines (with nested-trace indentation) and metric lines to a stream as they happen. “Tail my agent in dev.”
• FileTelemetry: Append-only JSONL on disk. Each line is a structured record with parent_span_id linkage. Queryable offline with jq.
• MultiTelemetry: Fan-out. Watch live in stderr AND assert on the in-memory side after.

Also demonstrates histogram-vs-counter auto-dispatch. Metric names ending in _ms / _seconds / _bytes become histograms; everything else becomes a counter. One emit_metric() API, the right instrument under the hood.

Uses ScriptedModel so the run is deterministic. No API key. For production, swap the sink for OTelTelemetry; the agent code doesn’t change.

See: Telemetry.

Read: examples/13_telemetry.py

Reasoning effort

14. Effort dial across providers

Runs the same hard reasoning question (the classic 3L / 5L jug puzzle, minimum-steps variant) at low, medium, high and xhigh on Claude Opus 4.7, which accepts every level without clamping.

Five things this demonstrates:

• Dict-config form for model=. Set the model name, default effort, and strict_effort together in one dict: model={"name": "claude-opus-4-7", "effort": "medium"}. It follows the same shape as audit_log={...}, so the whole model setup lives in one place.
• Equivalent explicit kwargs. Agent(model="claude-opus-4-7", effort="medium") does the same thing. Top-level kwargs win when both forms are present, so you can layer environment overrides on top of a shared config dict.
• The dial actually moves cost / latency. Token usage printed at each tier. Output tokens grow with effort because the model spends more on internal thinking.
• Per-call override. run(..., effort="high") wins over the agent default for that specific run. Lets one Agent serve cheap chit-chat and occasional deep reasoning without spinning up two.
• strict_effort=True. Wiring effort to claude-haiku-3-5 (which supports neither effort nor thinking) raises EffortNotSupportedError. The wiring mistake surfaces immediately. The default warn-and-drop behaviour would let it pass silently. Same dict shape: model={"name": "claude-haiku-3-5", "effort": "high", "strict_effort": True}.

Requires ANTHROPIC_API_KEY. The dial works on any reasoning model; on others a level the model lacks is clamped to the nearest one it accepts.

See: Reasoning effort · Agent reference: effort.

Read: examples/14_effort_dial.py

Declarative config

15. Build an Agent from a TOML / dict config

Writes a complete agent.toml, builds an Agent from it with Agent.from_config(path), and asserts every backend resolved to the expected concrete class. Then does the same with Agent.from_dict(cfg) to show the in-memory form. Runs offline against EchoModel. No API key needed.

Five things this demonstrates:

• Agent.from_config(path). Reads a TOML file. One declaration covers model, memory, runtime, telemetry, audit log, permissions, budget, architecture, effort, skills, and MCP servers.
• Agent.from_dict(cfg). Same shape, parsed in memory. Useful when your config comes from Pydantic BaseSettings, a YAML file you already parsed, or a service-config endpoint.
• Backend tables. [memory] / [runtime] / [telemetry] / [audit_log] / [permissions] / [budget] each go through the same resolver Agent(...) uses for kwargs. String shorthand works too: telemetry = "memory", permissions = "strict".
• Arrays of tables. [[skills]] loads skill bundles by path. [[mcp]] connects to MCP servers (stdio or HTTP transport).
• Kwarg override. Real callables (tools, hooks, secret stores, retry policies) come through Python kwargs since TOML can’t express them. When a kwarg and a config entry collide, the kwarg wins. That’s the per-environment override path.

See: Config file (TOML / dict) · Agent reference: from_config.

Read: examples/15_config_file.py

Multi-agent workspace

16. Shared notebook for multi-agent teams

A research team (four specialists plus a synthesizer) collaborates on one question through a shared notebook. Each specialist runs in parallel and writes exactly one findings note. None of them sees the others’ transcripts. The synthesizer then reads everyone’s notes via list_notes / read_note and writes the final recommendation into the same notebook.

What this shows:

• No history sharing. Specialists run hermetically. Only their curated notes cross between agents, not raw transcripts.
• Synthesizer context stays small. It reads N ~100-token notes instead of N full transcripts.
• Auto-attribution. Each agent’s notes are tagged with its team role because the author identity is baked into the workspace tools’ closure. The agent never types its own name.
• Filesystem-mounted. WORKSPACE.md regenerates atomically. You can cat it during or after the run.

Uses gpt-4.1-mini, so the demo is cheap (~$0.02 per run). Requires OPENAI_API_KEY.

See: Workspace.

Read: examples/16_shared_workspace.py

17. Prompt caching

Runs the same big system prompt twice, back to back, with prompt_caching=True. The second run shows non-zero cached_tokens_in and a markedly lower cost_usd.

Covers the boolean form on both Anthropic and OpenAI, the dict form for advanced control (ttl="1h", cache_key), and the per-provider behavior (Anthropic injects cache_control markers; OpenAI is automatic and the flag buys accurate accounting plus routing).

Requires ANTHROPIC_API_KEY or OPENAI_API_KEY.

See: Prompt caching.

Read: examples/17_prompt_caching.py

18. Living plan

Walks a living_plan=True agent through the TodoWrite discipline: commit a plan with plan_write, do work with a tool, rewrite the plan to mark the step done with a finding, emit a final message. After the run it inspects the workspace and confirms the plan mirrored to a kind="plan" note.

Also shows pre-seeding: pass a constructed LivingPlan so the next run starts with a plan already in place. Uses a ScriptedModel, so it runs offline. No API key.

See: Living plan.

Read: examples/18_living_plan.py

19. Workspace lifecycle

Exercises all eight workspace lifecycle features in one offline script: namespacing, versioning, archive, questions, semantic search (with a tiny deterministic stub embedder), citation tracking plus outcome attribution, relevance-aware search, and citation-aware prune().

Fully offline. Uses InMemoryWorkspace. No API key.

See: Workspace lifecycle.

Read: examples/19_workspace_lifecycle.py

Persistent subagents, sandbox, autonomy, ingest

20. Persistent subagents

Every Team.* builder registers each worker with a persistent ID (derived from its role and instructions, so it is the same in every process) and a stable session per user, so workers accumulate conversation memory across handoffs, rounds, and multiple coordinator.run() calls without one user’s worker session leaking into another’s. The example inspects what the Team builder stamps onto the coordinator under both modes: the default persistent registry and the persistent_subagents=False opt-out, where each spawn starts cold. Runs offline via EchoModel. No API key.

See: Team facade.

Read: examples/20_persistent_subagents.py

21. OS-level sandbox

OSSandbox runs each tool call in a child process wrapped by the platform’s kernel isolation facility, so a tool cannot read or write outside its declared roots (apart from the system libraries, the Python installation and the tool’s own package, plus its skill’s directory for a skill tool) or reach the network. The agent’s own workspace and living-plan tools stay in-process beside it. The child gets a minimal environment, not the parent’s API keys, and returns its result as JSON. Backends auto-select: Seatbelt (sandbox-exec) on macOS, Bubblewrap (bwrap) on Linux, and a graceful degrade to FilesystemSandbox(SubprocessSandbox) elsewhere. The active backend is reported via OSSandbox.mode. Run it as a module (python -m examples.21_sandbox_os) so the demo tools are importable in the sandbox child. No API key.

See: Sandboxing.

Read: examples/21_sandbox_os.py

22. Run until goal

Agent(run_until=...) keeps re-prompting the agent until a small, fast checker model confirms a measurable stop condition holds. The loop is bounded three ways: max_iterations, max_no_progress, and max_cost_usd. Accepts a bare condition string or a dict that names the checker and the bounds. Runs offline with ScriptedModel (a worker that takes two passes and a checker that says NOT_DONE then DONE). No API key.

See: Goal loops (run_until).

Read: examples/22_run_until_goal.py

23. Index a document

index_document(path, store) collapses the load → chunk → add plumbing for RAG ingest. Build the vector store once with its embedder, then call index_document to grow the index and store.search to query it. Uses the zero-key HashEmbedder + InMemoryVectorStore so it runs out of the box; swap in OpenAIEmbedder() + a persistent ChromaVectorStore for a real index with no other code change. No API key.

See: RAG.

Read: examples/23_index_document.py

Examples 24–34: newer features

One feature per script, in the same style as the earlier ones. All of them run offline with scripted models; 32 and 33 also run live against TypeSafe’s Jev when TYPESAFE_API_KEY is set.

Sample data

The image-bearing examples (01, 02) generate small sample PDFs on first run (via reportlab) and cache them under examples/data/. The on-disk Chroma indices are also cached, so subsequent runs only re-execute the agent loop against the model.