End-to-end demo projects for openarmature. Each is a standalone
main.py you can run against any OpenAI-compatible LLM endpoint
(OpenAI's public API, vLLM, LM Studio, llama.cpp server, etc.).
Grouped by what they teach.
Classify a query with an LLM and route to one of two follow-up
nodes. Demonstrates: typed State with three reducer policies, the
OpenAIProvider from openarmature.llm, structured output via a
Pydantic class (response_schema=Classification -> Response.parsed
as a Classification instance), conditional routing on a parsed
field, and a compile-time observer.
Question-answering assistant: classify, then short-answer or
research-subgraph, then copy-edit. Demonstrates: conditional edges,
SubgraphNode, custom ProjectionStrategy, the merge reducer.
Compare two topics by running the same analysis subgraph on each.
Demonstrates: ExplicitMapping for reusing one compiled subgraph at
multiple parent sites with disjoint parent fields.
Question answering against a tiny baked-in document corpus, with two levels of subgraph nesting: outer coordinator -> doc-QA subgraph -> section-extract subgraph. A depth-aware observer prints the descent and return.
Summarize a batch of news headlines in parallel. Each per-headline
run goes through a summarize -> classify subgraph wrapped in
retry middleware (transient failures don't tank the batch) and
timing middleware (per-instance duration captured). Demonstrates:
add_fan_out_node with items_field mode, extra_outputs
collecting a parallel list, instance_middleware, concurrency cap.
Enrich an article with three independent analyses (summary,
sentiment, topic tags) running concurrently. Each analysis is a
separate subgraph with its own state schema. The sentiment branch
wraps its subgraph in retry middleware; the other two run bare.
Demonstrates: add_parallel_branches_node, BranchSpec per branch
with input/output projection, heterogeneous branch state schemas,
per-branch middleware.
Two independent analyses of a lunar-mission photograph (describe
the surface, describe the equipment) using versioned prompt
templates and a multimodal user message. Templates load from
FilesystemPromptBackend with a primary + fallback chain; both
renders are grouped under one observability PromptGroup so a
trace UI can render them as one logical unit. Image source
switches between ImageSourceURL and
ImageSourceInline(base64_data=...) via env var. Demonstrates:
PromptManager with composite backends, prompt fetch + render
with template variables, PromptGroup +
with_active_prompt_group, with_active_prompt nesting,
multimodal UserMessage carrying both text and image content
blocks.
Four-turn lunar-mission conversation with conversation memory
threaded through ChatPrompt + PlaceholderSegment. One turn
attaches a photograph; the agent processes it without changing
the chat shape. Demonstrates: ChatPrompt, ContentSegment,
PlaceholderSegment, history threading via the append reducer
on list[Message], conditional self-loop for multi-turn cycles.
Reaching an OpenAI request field openarmature does not model, and the
guardrails that stop you reaching one it does. A bulk telemetry-alert
classifier wants service_tier for the cheaper lane and logit_bias to
suppress a retired severity label; neither is portable, so neither has a
first-class field, and both ride in extras untouched. The refusals are the
interesting half: a field the mapping produced on this call cannot also come
from extras, because two sources of truth for one wire field is a bug you
want at the call site. Demonstrates: RuntimeConfig(extras=...) passthrough,
ProviderInvalidRequest on a managed-field collision, model / messages /
tools / tool_choice rejecting always so an extras tool array cannot skip
tool validation, and stop merging with stop_sequences rather than colliding
because both realize the same wire field. Runs against a stub transport, since
the point is the shape of the outbound request.
A lunar-mission assistant that calls local Python tools to answer
questions mixing fact recall and physics arithmetic. Defines two
tools (lookup_mission reading a baked-in record store,
compute_delta_v doing a Hohmann transfer), passes them to the
model via complete(tools=...), dispatches assistant.tool_calls
to the local functions, and feeds the results back as
ToolMessage entries. The agent loop is a graph cycle:
call_llm -> dispatch_tools -> call_llm via a conditional edge,
with a hard turn cap to prevent runaway loops. Demonstrates:
Tool definitions with JSON Schema parameters,
complete(tools=...), parsing ToolCall records,
ToolMessage(tool_call_id=...) round-trip, multi-turn
tool-calling loop as a graph cycle.
Retrieval-augmented answering over a lunar knowledge base, using the
two-stage retrieve-then-rerank pattern before generation. Batch-embeds a
small corpus once into an index (OpenAIEmbeddingProvider.embed over a
list, one vector per passage), then per query embeds the question, ranks
the corpus by cosine similarity for a broad shortlist, reranks that
shortlist with a cross-encoder (CohereRerankProvider.rerank) for
precision, and grounds an LLM answer in the reranked top passages. The
retrieve and rerank steps run as graph nodes, so their EmbeddingEvent /
RerankEvent reach any attached observer. Demonstrates:
OpenAIEmbeddingProvider and CohereRerankProvider from
openarmature.retrieval, the input_type query/document knob (a wire
no-op on symmetric OpenAI, meaningful on asymmetric providers), mapping
ScoredDocument results back by .index, and retrieval feeding an
OpenAIProvider answer node. Needs OPENAI_API_KEY and COHERE_API_KEY.
Extracting one structured mission record per lunar-landing report, under an
output-token cap that is too tight for a complete record. The model stops
mid-object, and the fragment that arrives is rejected at the schema boundary
exactly as a wrong-typed field would be. The cap is what the demo drives,
because it fails the same way on every endpoint; the wrong-typed field is the
more common failure in practice and how often you see it depends on whether
your endpoint enforces the schema during decoding. Both reach the same
exception, and the builder here handles both. Recovery needs two different
changes:
a caller-supplied builder quotes the model's own fragment and the reader's
objection back to it, and a per-attempt override raises the ceiling so the
retry has somewhere to put the answer. Demonstrates:
complete(response_schema=...) raising StructuredOutputInvalid instead of
returning a half-built object, LlmRetryConfig(reask=...) making that failure
retryable for one call, a builder reading exc.output_content,
exc.error_message, exc.finish_reason and exc.response_schema and
branching on which failure it got (the two arms need different information,
not just different wording, and the schema-mismatch arm sends the schema
itself, since error_message names only the first violation),
LlmRetryConfig(per_attempt_override=...) applying a config schedule to
retries only, the framework appending the model's reply and the correction as
an alternating transcript while authoring no prompt of its own, and reask
sharing the max_attempts budget with transient retries. Three postures:
MODE=off supplies neither and ends on the first unusable reply, MODE=nocap
supplies the correction but not the raised ceiling so every attempt is cut off
at the same place, and the default supplies both.
A lunar-mission planning pipeline that survives a simulated
mid-pipeline crash, then resumes the saved record under an
upgraded state schema. Phase one invokes a v1 graph against
MissionPlanStateV1; the SQLiteCheckpointer (JSON mode) writes
records to a temp DB synchronously after every node completes.
size_crew raises on its first call to simulate a transient
infrastructure failure; a second invoke with resume_invocation=
picks up cleanly. Phase two registers a v1->v2 migration
backfilling a new risk_assessment field, builds a v2 graph with
one new node, and resumes from the (now-completed) v1 invocation.
Demonstrates: SQLiteCheckpointer(serialization="json"),
with_checkpointer, save-on-completed-event, NodeException at
the invoke boundary, State.schema_version,
with_state_migration, invoke(resume_invocation=...).
Add observability to a draft -> review -> finalize pipeline
without changing any node code. Demonstrates: attach_observer,
NodeEvent, namespace chaining across subgraph boundaries,
function-shaped and class-shaped observers, plus the
OTelObserver running alongside the plain observer (same hook,
different backend).
Send LLM-call observability natively to Langfuse with a
prompt-linkage demonstration on a mission-briefing Q&A pipeline.
Demonstrates: LangfuseObserver attached at the graph level,
LangfusePromptBackend for prompt fetch, automatic
Generation -> Prompt link via observability_entities.
Dual OTel + Langfuse observers attached to one graph, caller hooks
deriving domain-shaped trace.input / trace.output from State,
built-in TimingMiddleware recording per-node duration via an
on_complete callback, multi-tenant caller-supplied metadata
propagating to both observers in one invoke() call. The
production-grade observability shape, end-to-end, with in-memory
captures so the demo prints what each backend would have ingested
without needing real production credentials.
All demos configure their LLM client via env vars; OpenAI public-API defaults shown:
| Env var | Default | Notes |
|---|---|---|
LLM_BASE_URL |
https://api.openai.com |
Host root only; the provider adds the path. |
LLM_MODEL |
gpt-4o-mini |
Any model the bound endpoint exposes. |
LLM_API_KEY |
(none) | Required; pass empty for local servers that don't authenticate. |
The Langfuse observer and the Langfuse prompt backend read the standard
Langfuse SDK variables when pointed at a live Langfuse account;
Langfuse() reads them automatically, so no credentials appear in the
example code:
| Env var | Notes |
|---|---|
LANGFUSE_PUBLIC_KEY |
From your Langfuse project settings. |
LANGFUSE_SECRET_KEY |
From your Langfuse project settings. |
LANGFUSE_BASE_URL |
Langfuse host (e.g. https://cloud.langfuse.com); the SDK also accepts LANGFUSE_HOST. |
# From the repo root, install the examples dep group:
uv sync --group examples
# The observer-hooks, multimodal-prompt, chat-with-multimodal,
# and production-observability demos also want the OTel SDK for
# their OTelObserver:
uv sync --group examples --all-extras
# Run any demo:
LLM_API_KEY=sk-... uv run python examples/hello-world/main.py
LLM_API_KEY=sk-... uv run python examples/routing-and-subgraphs/main.py "what year did the moon landing happen"For a local OpenAI-compatible server (vLLM, LM Studio, llama.cpp,
etc.), point LLM_BASE_URL at the host root (e.g. http://localhost:8000)
and set LLM_API_KEY to whatever value the server expects (often
empty or a placeholder).