A Production Stack
for Multi-Agent LLM Systems
Declare, execute, remember, and observe — think Kubernetes for agent orchestration. Four architecture pillars that take you from a YAML blueprint to production-grade observability and distributed tracing for multi-agent systems.
Declare your system in YAML
Define agents, workflows, providers and contracts in one file. BlueprintCompiler validates, compiles, diffs, and tests the spec with no boilerplate Python required.
pyagent-blueprint
# customer-support.yaml
api_version: pyagent/v1
metadata: { name: customer-support, version: "1.0.0" }
providers:
fast: { provider: anthropic, model: claude-haiku-3-5-20241022 }
expert: { provider: anthropic, model: claude-sonnet-4-20250514 }
agents:
classifier: { provider: fast, prompt: "Classify the request." }
specialist: { provider: expert, prompt: "Handle professionally." }
workflows:
main:
pattern: supervisor
agents: { classifier: classifier, routes: { billing: specialist } }
pyagent validate customer-support.yaml
pyagent diff v1.yaml v2.yaml
from pyagent_patterns.orchestration import Pipeline
from pyagent_patterns.base import Agent
from pyagent_providers import AnthropicLLM
from pyagent_router.middleware import RouterMiddleware
router = RouterMiddleware(model_registry={
"claude-haiku": AnthropicLLM("claude-haiku-3-5-20241022"),
"claude-sonnet": AnthropicLLM("claude-sonnet-4-20250514"),
})
pipeline = Pipeline(stages=[
router.wrap(Agent("extractor", AnthropicLLM("claude-sonnet-4-20250514"))),
router.wrap(Agent("summarizer", AnthropicLLM("claude-sonnet-4-20250514"))),
])
result = asyncio.run(pipeline.run("Summarise the quarterly report..."))
18 patterns, one consistent API
A library of design patterns for multi-agent systems — Pipeline, Supervisor, Debate, Fan-Out, ReAct and 13 more. Not sure which to use? Each page explains when to reach for it, so you can pick the right design pattern for your multi-agent workflow. Difficulty-based routing picks the right model per task, and compression middleware enforces token budgets across the pipeline.
pyagent-patterns
pyagent-providers
pyagent-router
pyagent-compress
Three-tier memory, trust-aware
Working memory for the current task, session memory across turns, semantic memory for long-term retrieval. Every item carries a trust level, sensitivity tier and expiry, with PII redaction built in.
pyagent-context
from pyagent_context import ContextLedger, ContextItem, TrustLevel, Sensitivity
ledger = ContextLedger()
ledger.append(ContextItem(
content="Customer ID: C-10482, tier: premium, since 2022",
source="crm",
trust=TrustLevel.VERIFIED,
sensitivity=Sensitivity.INTERNAL,
))
graph.wire_context(ledger) # shared across all agents
from pyagent_trace.events import TraceEventBus
bus = TraceEventBus()
graph.wire_trace(bus) # attaches to every agent in the compiled graph
pyagent apply customer-support.yaml
pyagent simulate customer-support.yaml main "I need a refund"
pyagent dashboard
# → http://localhost:8080 (traces · costs · governance)
Every call traced, every cost tracked
Full observability in multi-agent systems: distributed tracing across every agent, pattern, and provider. OTel spans, Langfuse export, record/replay for debugging, and a web dashboard with trace explorer, cost analytics, governance scoring, and provider health monitoring.
pyagent-trace
pyagent-studio
Why PyAgent?
The only framework purpose-built across all four production pillars.
Quick Start
pip install pyagent-blueprint
# pipeline.yaml
api_version: pyagent/v1
metadata: { name: doc-pipeline, version: "1.0.0" }
agents:
extractor: { prompt: "Extract the key facts as bullet points." }
summarizer: { prompt: "Summarise the input in 3 sentences." }
workflows:
main:
pattern: pipeline
agents: { stages: [extractor, summarizer] }
import asyncio
from pyagent_blueprint import load_blueprint, BlueprintCompiler
graph = BlueprintCompiler().compile(load_blueprint("pipeline.yaml"))
result = asyncio.run(graph.run("main", "Process this document"))
print(result.output)
pip install pyagent-patterns
import asyncio
from pyagent_patterns.base import Agent, MockLLM
from pyagent_patterns.orchestration import Pipeline
llm = MockLLM(responses=["Extracted: key facts", "Summary: concise version"])
pipeline = Pipeline(stages=[
Agent("extractor", llm),
Agent("summarizer", llm),
])
result = asyncio.run(pipeline.run("Process this document"))
print(result.output) # "Summary: concise version"
pip install pyagent-all
import asyncio
from pyagent_blueprint import load_blueprint, BlueprintCompiler
from pyagent_trace.events import TraceEventBus
bus = TraceEventBus()
graph = BlueprintCompiler().compile(load_blueprint("blueprint.yaml"))
graph.wire_trace(bus)
result = asyncio.run(graph.run("main", "Analyse Q3 revenue trends"))
# Then: pyagent dashboard --trace traces/runs.jsonl
Four pillars, one coherent stack
Blueprint compiles to RuntimeGraph, which orchestrates execution, manages context, and streams every event to the observability layer.
BlueprintCompiler transforms the spec into a runnable RuntimeGraph2ContextLedger maintains three-tier memory across agents and turns5CompressMiddleware trims inter-agent token transfer and enforces budgets6TraceEventBus collects events from agents, patterns, and providers718 Patterns
Where to start
New to multi-agent systems?
Start with the tutorial, then explore the pattern library one pattern at a time.
Adding to an existing codebase?
Drop patterns in alongside your existing LLM setup, then layer in routing and context.
Building for production?
Use all four pillars: Blueprint for versioning and CI, then Observability for monitoring.