The PyFly Agentic — Complete Tutorial

Use these capability snippets inside your application’s native runtime. Installation is through PyFly Agentic extras.

Copyright 2026 Firefly Software Foundation. Licensed under the Apache License 2.0.

From Zero to Expert — This tutorial walks you through every module in PyFly Agentic by building a real project from scratch: an Intelligent Document Processing (IDP) pipeline that classifies, digitises, extracts, validates, and assembles data from invoices.

Each chapter introduces a concept, explains why it exists, shows how it works with architecture diagrams, and immediately applies it to the IDP pipeline. By Chapter 18 you will have a production-grade GenAI application that uses agents, tools, prompts, reasoning patterns, content processing, memory, validation, pipelines, observability, explainability, experiments, multi-agent delegation, template agents, and a plugin system — all wired together.

---

Table of Contents

Part I — Foundation 1. Introduction — What, who, why, design principles, running example 2. Installation & Project Setup — UV, extras, AgenticConfig, model providers 3. Your First Agent — Agent, @agent, registry, context, lifecycle (diagram) 4. Tools — @agent_tool, ToolBuilder, guards, composition, built-ins, ToolKit, attaching tools to agents (diagram) 5. Prompts — Jinja2 templates, versioning, composition, validation, file loading

Part II — Intelligence

1. Reasoning Patterns — ReAct, CoT, Plan-and-Execute, Reflexion, Tree of Thoughts, Goal Decomposition, custom patterns (diagram)
2. Content Processing — Chunking, batch processing, compression, sliding window
3. Memory — Conversation memory, working memory, backends, MemoryManager, forking, integration points (diagram)
4. Validation & Quality of Service — Rules, OutputValidator, QoS checks, OutputReviewer

Part III — Orchestration & Operations

1. Pipeline — DAG engine, step executors, fan-out/fan-in, retries, PipelineContext (diagram)
2. Observability — Tracing, metrics, events, OpenTelemetry exporters
3. Explainability — Decisions, explanations, audit trail, reports

Part IV — Experimentation

1. Experiments — A/B testing, variant comparison, tracking
2. Lab — Interactive sessions, benchmarks, model comparison, eval datasets

Part V — Advanced

1. Template Agents — Summariser, classifier, extractor, conversational, router
2. Multi-Agent Delegation — Delegation router, strategies, memory forking (diagram)
3. Plugin System — Entry-point discovery, packaging agents/tools/patterns
4. Putting It All Together — Full IDP implementation, project structure, production checklist (full system diagram)

---

Part I — Foundation

---

Chapter 1: Introduction

What Is PyFly Agentic?

PyFly Agentic is a GenAI metaframework — it sits on top of Pydantic AI and provides the structure, patterns, and production-grade plumbing that Pydantic AI deliberately leaves to the application developer.

Think of Pydantic AI as the engine and PyFly Agentic as the car around it: the steering, brakes, GPS, dashboard, and everything else you need to actually drive to your destination.

Who Is It For?

• Backend engineers building GenAI features into existing applications.
• ML/AI engineers who want structured reasoning, validation, and observability out of the box.

The Four Design Principles

The framework is guided by four principles that show up in every module:

1. Protocol-driven contracts — Public APIs are Python Protocol classes or abstract base classes. You can swap or extend any component without touching framework internals.
2. Convention over configuration — Sensible defaults everywhere. One AgenticConfig object (a Pydantic validation model) centralises every knob. PyFly binds it from the effective configuration tree and environment overrides.
3. Layered composition — Modules are organised into layers (Core, Agent, Intelligence, Experimentation, Orchestration). Higher layers depend on lower layers, never the reverse.
4. Optional dependencies — Heavy libraries (embedding providers, vector store clients, storage backends) are declared as extras. The core framework imports them lazily so you only install what you use.

The Running Example: Intelligent Document Processing

Throughout this tutorial we build an IDP pipeline that processes invoices:

Raw Document → Classify → Digitise (OCR) → Extract Fields → Validate → Assemble Result

Every chapter teaches a framework concept and immediately applies it to a phase of this pipeline. By Chapter 18 you will have the complete, production-ready system.

---

Chapter 2: Installation & Project Setup

Prerequisites

• Python 3.12 or later.
• UV (recommended for dependency management): https://docs.astral.sh/uv/

Creating a New Project

# Use the reviewed PyFly checkout containing src/pyfly/agentic.
cd fireflyframework-pyfly
python -m venv .venv
source .venv/bin/activate
python -m pip install -e ".[agentic]"

This installs the framework with Pydantic AI ==2.59.0, its explicitly declared agent provider integrations, Pydantic, Jinja2, HTTPX, pricing metadata, and OpenTelemetry. Storage and document-processing libraries remain optional extras.

Run the installation commands from that same framework source root. Create the application package separately after installing the local framework dependency. See installation for source/release status.

Installing Extras

The framework provides optional extras for additional capabilities:

# Embedding providers (e.g. OpenAI / Azure)
python -m pip install -e ".[agentic,agentic-openai-embeddings]"

# Vector store backends
python -m pip install -e ".[agentic,agentic-vectorstores-chroma]"

# Memory persistence backends
python -m pip install -e ".[agentic,agentic-postgres]"

# Everything
python -m pip install -e ".[agentic,agentic-all]"

For our IDP project we will eventually use several of these, so install everything:

python -m pip install -e ".[agentic,agentic-all]"

Configuration with AgenticConfig

All framework settings live in a single typed Pydantic model bound from PyFly configuration; the native binder reads environment overrides prefixed with PYFLY_AGENTIC_:

from pyfly.agentic import AgenticConfig, get_config

# get_config() returns a configuration for the active runtime
config = get_config()
print(config.default_model) # Built-in fallback: "openai:gpt-4o"; overridden by your environment.
print(config.default_temperature) # None (no temperature forced; provider default)
print(config.max_retries) # 3

Set environment variables before startup. The following .env sample requires an explicit loader such as the one shown in quick start; AgenticConfig() does not load files or environment values by itself:

# .env

# --- Provider API keys (read from the environment) ---
OPENAI_API_KEY=sk-...
# ANTHROPIC_API_KEY=sk-ant-...
# GEMINI_API_KEY=...
# GROQ_API_KEY=gsk_...
# DEEPSEEK_API_KEY=...

# --- Framework settings ---
PYFLY_AGENTIC_DEFAULT_MODEL=openai-responses:gpt-6-luna
PYFLY_AGENTIC_LOG_LEVEL=DEBUG
PYFLY_AGENTIC_OBSERVABILITY_ENABLED=true

Here are the most commonly used configuration fields:

• default_model — Provider and model selector (e.g. "openai-responses:gpt-6-luna", "anthropic:claude-sonnet-5"). The built-in fallback remains "openai:gpt-4o" for compatibility; this tutorial sets it explicitly.
• default_temperature — Default sampling temperature. None (default) forces no temperature, so each provider uses its own default (some models, e.g. OpenAI o1/o3, reject an explicit temperature). When set, it's merged into an agent's settings only if the caller omits one.
• max_retries — Default retry count for agent runs.
• observability_enabled — Toggle OpenTelemetry instrumentation.
• Prompt files — Pass an explicit directory to PromptLoader.from_directory(path).
• default_chunk_size / default_chunk_overlap — Content chunking defaults.
• max_context_tokens — Maximum context window (default 128,000).
• Output validation — Configure explicit validators on the agent or pipeline.
• cost_tracking_enabled — Enable/disable usage and cost tracking.
• budget_limit_usd — Hard accounted budget limit in USD (BudgetExceededError on a priced overrun).
• cost_strict — When True, cost resolution raises UnknownModelCostError instead of returning None for models with no known pricing (default False).
• memory_backend — "in_memory", "file", "relational", "postgres", or "mongodb".
• memory_max_conversation_tokens — Token budget per conversation.
• encryption_enabled / encryption_key — Encrypt auto-composed working-fact content with typed JSON; metadata and conversations are outside coverage.

Configuration is owned by AgenticRuntime. get_config() resolves the active runtime; create a fresh runtime for each isolated application or test.

Model Providers & Authentication

Before you can run an agent against a real LLM, you need credentials for your model provider. PyFly Agentic delegates model communication entirely to Pydantic AI, which supports multiple providers out of the box.

Approach 1: Environment Variables (Recommended)

The simplest method — set the appropriate API key as an environment variable and use the "provider:model_name" string format:

Provider	Credentials	Model selector prefix
OpenAI	OPENAI_API_KEY	openai-responses: or openai-chat:
Anthropic	ANTHROPIC_API_KEY	anthropic:
Google Gemini	GEMINI_API_KEY	google:
Groq	GROQ_API_KEY	groq:
DeepSeek	DEEPSEEK_API_KEY	deepseek:
Mistral	MISTRAL_API_KEY	mistral:
AWS Bedrock	Host AWS credential chain	bedrock:
Ollama (local)	No key for a local instance	ollama:

Append the exact model ID enabled for your account or local server. Prefix support does not guarantee that a particular model is available. Azure deployment names and custom endpoints are configured through ModelSpec below. The framework and provider clients read credentials from the environment; ordinary agent code only needs the configured model. The following setup selects Responses with GPT-6 Luna:

# .env
OPENAI_API_KEY=sk-...
PYFLY_AGENTIC_DEFAULT_MODEL=openai-responses:gpt-6-luna

from pyfly.agentic.agents import Agent

# Uses OPENAI_API_KEY from the environment
agent = Agent(name="my-agent")

To switch providers, change the configured model and credential, then validate your ModelOptions and tool/output requirements against the new model:

# .env — switch to Anthropic
ANTHROPIC_API_KEY=sk-ant-...
PYFLY_AGENTIC_DEFAULT_MODEL=anthropic:claude-sonnet-5

Approach 2: PyFly Agentic model specifications

Use ModelSpec and ModelFactory for application-managed credentials and custom endpoints. Ordinary application code does not need provider SDK imports:

import os

from pyfly.agentic.agents import Agent
from pyfly.agentic.models import Credential, ModelFactory, ModelOptions, ModelSpec

spec = ModelSpec(
    provider="openai-responses",
    model="gpt-6-luna",
    credential=Credential.api_key(os.environ["OPENAI_API_KEY"]),
    options=ModelOptions(max_tokens=4096, reasoning="low", store_responses=False),
)
factory = ModelFactory()
agent = Agent(
    name="configured-agent",
    model=await factory.build(spec),
    model_options=spec.options,
)

Azure OpenAI uses your deployment name and endpoint. Set OPENAI_API_VERSION for a versioned endpoint; a deployment using /openai/v1 does not require a dated API version:

azure_spec = ModelSpec(
    provider="azure-responses",
    model=os.environ["AZURE_OPENAI_DEPLOYMENT"],
    base_url=os.environ["AZURE_OPENAI_ENDPOINT"],
    api_version=os.environ.get("OPENAI_API_VERSION"),
    credential=Credential.api_key(os.environ["AZURE_OPENAI_API_KEY"]),
    options=ModelOptions(max_tokens=4096),
)
azure_agent = Agent(
    name="azure-agent",
    model=await factory.build(azure_spec),
    model_options=azure_spec.options,
)

AWS Bedrock uses the host's AWS credential chain when no credential is passed:

bedrock_spec = ModelSpec(
    provider="bedrock",
    model=os.environ["BEDROCK_MODEL_ID"],
    region=os.environ["AWS_REGION"],
)
bedrock_agent = Agent(name="bedrock-agent", model=await factory.build(bedrock_spec))

A local OpenAI-compatible endpoint, such as Ollama's Chat API:

local_spec = ModelSpec(provider="ollama", model="llama3.2", base_url="http://localhost:11434/v1")
local_agent = Agent(name="local-agent", model=await factory.build(local_spec))

Model availability and endpoint support remain provider-specific. PyFly Agentic keeps legacy openai: and azure: selectors on Chat Completions; use openai-responses: or azure-responses: explicitly for Responses. GPT-6 Astra needs Responses for tools; Sol and Luna need ModelOptions(reasoning="none") for Chat tools. Responses supports reasoning with tools. See Models for capabilities and credentials.

Model controls and per-run overrides

Use PyFly Agentic's typed options on an agent or decorator and override individual fields for a run. Explicit None clears an inherited option; omitted fields keep it:

agent = Agent(name="portable", model_options=ModelOptions(max_tokens=4096))
result = await agent.run("Explain bounded retries.", model_options=ModelOptions(max_tokens=1024))

PyFly Agentic validates options against the selected model on every run and raises ModelOptionsError for unsupported controls. Avoid a global temperature when switching to reasoning models. Native model_settings= and preconfigured Pydantic AI Model objects remain advanced escape hatches; see Agents.

IDP Tie-In

For our IDP project, we will use environment variables. Make sure your .env contains the API key for whichever provider you choose:

# .env
OPENAI_API_KEY=sk-...
PYFLY_AGENTIC_DEFAULT_MODEL=openai-responses:gpt-6-luna
PYFLY_AGENTIC_OBSERVABILITY_ENABLED=true

---

Chapter 3: Your First Agent

Every GenAI application starts with a single question: "How do I talk to the model?" In raw Pydantic AI you create an Agent, give it a system prompt, and call run(). That works great for scripts — but the moment you need to register agents by name, share them across pipelines, delegation, and reasoning patterns, attach lifecycle hooks, or plug them into a larger system, you need a thin coordination layer on top.

That is exactly what Agent is. It wraps a Pydantic AI Agent and adds three things the framework relies on: an application-scoped registry (so any module can look up an agent by name), lifecycle management (init → warmup → shutdown), and metadata (tags, descriptions, correlation IDs). The underlying Pydantic AI agent does all the heavy lifting — model calls, tool dispatch, streaming — while Agent provides the scaffolding that makes it a team player in a larger system.

Agent System Architecture

The following diagram shows how Agent sits on top of Pydantic AI and connects to the rest of the framework:

flowchart TD
    HOST["Host application"] --> CREATE["Agent or agent decorator"]
    CREATE --> AGENT["Configured Agent"]
    AGENT -->|default registration| REGISTRY["AgentRegistry"]
    HOST -->|explicit startup and shutdown| LIFECYCLE["AgentLifecycle hooks"]
    HOST -->|per-run correlation| CONTEXT["AgentContext"]
    CONTEXT --> AGENT
    CONSUMERS["Pipelines, delegation, reasoning"] --> AGENT
    AGENT --> OPTIONS["Model selection and ModelOptions"]
    OPTIONS --> PAI["Pydantic AI<br/>requests, tools, streaming"]

Creating an Agent with the Decorator

The fastest way to create and register an agent is with the @agent decorator. You write a function that returns the system prompt, and the framework takes care of creating the agent, wiring the prompt, and registering it — all in one step:

from pyfly.agentic.agents import agent

# The decorator creates a Agent, uses this function as the dynamic
# instructions provider, and registers the agent in the active runtime’s AgentRegistry.
@agent(name="greeter")
def greeter_instructions(ctx):
    # This function is called at the start of every run.
    # You can inspect `ctx` to customise the prompt per-request.
    return "You are a friendly greeter. Say hello and ask how you can help."

What happens behind the scenes:

1. A Agent named "greeter" is created with the configured default model.
2. The decorated function becomes the agent's dynamic instructions provider — it is called at the start of every run and can use the context to customise the system prompt.
3. The agent is automatically registered in the active runtime’s AgentRegistry, so any module (pipelines, delegation routers) can look it up by name.

Creating an Agent with the Class

When you need full control — custom output types, explicit tool lists, or you prefer not to use decorators — instantiate Agent directly and register it yourself:

from pyfly.agentic.agents import Agent
from pyfly.agentic.agents.registry import agent_registry

# Create the agent with a static instructions string and a typed output.
# The `output_type` tells Pydantic AI to validate the LLM's response as a dict.
classifier = Agent(
    name="document_classifier",
    instructions=(
        "You are a document classification expert. "
        "Given a document, determine its type (invoice, receipt, contract, form), "
        "language, and page count. Return JSON with keys: doc_type, language, page_count."
    ),
    output_type=dict,
)

# Register it so other parts of the framework (pipelines, delegation) can find it.
agent_registry.register(classifier)

Running an Agent

Every agent supports three execution modes — pick the one that fits your context:

# Async — the standard choice for production services.
result = await classifier.run("Classify this document: Invoice from Acme Corp...")
print(result.output) # {"doc_type": "invoice", "language": "en", "page_count": 1}

# Synchronous — handy for scripts, notebooks, and quick experiments.
result = classifier.run_sync("Classify this document: ...")
print(result.output)

# Streaming — for real-time UI feedback where you want tokens as they arrive.
# streaming_mode is "buffered" (default, chunked) or "incremental" (token-by-token).
async with classifier.run_stream(
    "Classify this document: ...", streaming_mode="incremental"
) as stream:
    async for chunk in stream:
        print(chunk, end="", flush=True)

The Agent Registry

The AgentRegistry is a runtime-owned registry that maps agent names to Agent instances. This is the glue that lets any module — delegation routers, pipelines, reasoning patterns — discover and invoke agents without importing them directly:

from pyfly.agentic.agents.registry import agent_registry

# Retrieve an agent by name — returns the Agent or raises KeyError.
agent = agent_registry.get("document_classifier")

# Check existence before retrieval.
if agent_registry.has("document_classifier"):
    print("Agent is registered")

# List all registered agents (useful for admin dashboards / health checks).
for info in agent_registry.list_agents():
    print(f"{info.name} ({info.model})")

Agent Context

When a request arrives, you often need to carry metadata along the entire execution path — who sent the request, which experiment is active, what trace ID to log. AgentContext is that bag of request-scoped data:

from pyfly.agentic.agents.context import AgentContext

# Create a context with a correlation ID and arbitrary metadata.
# This context will be available in instructions providers and tool functions.
ctx = AgentContext(
    correlation_id="req-abc-123",
    metadata={"tenant": "acme-corp", "source": "email-inbox"},
)

The context is available inside dynamic instructions providers and tool functions, allowing you to tailor behaviour per-request without global state.

Lifecycle Management

Real-world agents often depend on external resources — database connections, model caches, file handles. AgentLifecycle gives you three hooks to manage them cleanly: init (one-time setup), warmup (pre-heat caches), and shutdown (release resources):

from pyfly.agentic.agents.lifecycle import AgentLifecycle

lifecycle = AgentLifecycle()

# Each hook accepts a sync or async callable.
lifecycle.on_init(lambda: print("Loading resources..."))
lifecycle.on_warmup(lambda: print("Warming up model cache..."))
lifecycle.on_shutdown(lambda: print("Releasing resources..."))

# In your application startup/shutdown sequence:
await lifecycle.run_init()
await lifecycle.run_warmup()
# ... application serves requests ...
await lifecycle.run_shutdown()

The Middleware Stack

Every Agent runs each call (run, run_sync, run_stream, run_with_reasoning) through a MiddlewareChain — a before/after pipeline whose hooks receive a MiddlewareContext (agent name, prompt, method, model, deps). Two middlewares are auto-wired: LoggingMiddleware is always added, and ObservabilityMiddleware is added when config.observability_enabled is true. You can attach more from pyfly.agentic.agents:

• PromptGuardMiddleware / OutputGuardMiddleware — block prompt-injection / leaky output.
• CostGuardMiddleware — raises BudgetExceededError when a budget is exceeded.
• CacheMiddleware (ResultCache) and PromptCacheMiddleware (CacheStatistics).
• ValidationMiddleware — validate structured output.
• RetryMiddleware — retry on failure (not auto-wired; rate-limit retry is built into run() itself).
• ExplainabilityMiddleware — record decisions to a TraceRecorder.

from pyfly.agentic.agents import Agent, PromptGuardMiddleware

agent = Agent(name="assistant")
agent.middleware.add(PromptGuardMiddleware())

For provider resilience, FallbackModelWrapper / run_with_fallback let an agent fall back to a secondary model, and CircuitBreakerMiddleware (Chapter 11) trips on repeated failures.

IDP Tie-In: The Document Classifier Agent

Time to build the first piece of our IDP pipeline. The document classifier receives a raw document (text or scanned image) and outputs structured metadata — document type, language, page count, and orientation. Every subsequent stage depends on this output.

from pyfly.agentic.agents import Agent

# The classifier is our pipeline's entry point — it decides how the document
# will be routed through digitisation, extraction, and validation.
classifier_agent = Agent(
    name="document_classifier",
    instructions=(
        "You are a document classification expert. Given a document (text or image), "
        "determine its type, language, and page count. "
        "Return JSON: {doc_type, language, page_count, orientation}."
    ),
    output_type=dict, # Pydantic AI will validate the LLM output as a dict
)

For multimodal input (e.g. scanned images), the framework provides typed content wrappers that ensure the LLM receives the content in the right format:

from pyfly.agentic.types import ImageUrl

# Pass a list of mixed content — text instructions + an image.
# The framework converts ImageUrl into the provider-specific multimodal format.
result = await classifier_agent.run([
    "Classify this document:",
    ImageUrl(url="data:image/png;base64,<base64-data>"),
])

Other supported types include AudioUrl, DocumentUrl, and VideoUrl.

---

Chapter 4: Tools

LLMs are powerful reasoners, but they cannot check a database, call an API, or read a file on their own. Tools bridge that gap: they are functions the model can call during a conversation to fetch data, trigger side-effects, or run computations.

Pydantic AI already supports tool functions, but PyFly Agentic wraps them with a richer layer: a protocol-based type system (ToolProtocol → AgentTool), guards that enforce validation, rate-limiting, and sandboxing before a tool executes (human-in-the-loop approval is separate — it pauses the run rather than rejecting it), composition primitives (sequential, fallback, conditional), an application-scoped registry for discovery, and a ToolKit that can convert framework tools into Pydantic AI tools for injection into any agent.

Tool System Architecture

The following diagram shows the full tool stack — from how you create tools at the top, through the guard chain, to how they reach an agent at the bottom:

flowchart TD
    BUILD["agent_tool, ToolBuilder, built-ins"] --> TOOL["PyFly Agentic tool"]
    TOOL -->|optional registration| REGISTRY["ToolRegistry"]
    TOOL --> KIT["ToolKit or direct tools list"]
    KIT --> AGENT["Agent"]
    AGENT --> SCHEMA["Typed schemas exposed to Pydantic AI"]
    SCHEMA --> CALL["Model requests a function tool"]
    CALL --> GUARDS["Guard chain before the handler"]
    GUARDS --> LISTENERS["before_call listeners"]
    LISTENERS --> HANDLER["Handler with optional timeout"]
    HANDLER --> EVENTS["after_call, on_error or on_pause"]
    EVENTS --> RESULT["Tool result or control signal"]

Creating a Tool with the Decorator

The fastest path — one decorator does everything:

from pyfly.agentic.tools import agent_tool

# @agent_tool creates an AgentTool, registers it in the active runtime ToolRegistry,
# and makes it discoverable by name for ToolKit grouping and agent bridging.
@agent_tool(name="lookup_vendor", description="Look up vendor details by name")
async def lookup_vendor(vendor_name: str) -> str:
    # In production, this would query a database or API.
    vendors = {"Acme Corp": "ID-001", "Globex": "ID-002"}
    return vendors.get(vendor_name, "Unknown vendor")

Creating a Tool with the Builder

When you need more control — or want to build tools programmatically at runtime — use the fluent ToolBuilder:

from pyfly.agentic.tools import ToolBuilder

async def fetch_exchange_rate(currency: str) -> float:
    """Read the fixed rates used by this invoice exercise, not live market quotes."""
    rates = {"USD": 1.0, "EUR": 0.85, "GBP": 0.73}
    return rates[currency.upper()]

# The builder pattern lets you set each property explicitly.
# Call .build() to produce the final AgentTool instance.
exchange_tool = (
    ToolBuilder("exchange_rate")
    .description("Get exchange rate for a currency code")
    .parameter("currency", str, description="Three-letter currency code")
    .handler(fetch_exchange_rate)
    .build()
)

Tool Guards

In production, you rarely want a tool to run unconditionally. Guards are decorators that wrap a tool's execution with hard, synchronous policy checks — input validation, rate-limiting, or filesystem sandboxing. They run before the handler (and optionally after), and they stack via CompositeGuard. (Human-in-the-loop approval, which pauses the run rather than rejecting it, is handled separately — see below.)

Validation Guard

Ensures that all required parameters are present before the tool handler runs:

from pyfly.agentic.tools import agent_tool, guarded
from pyfly.agentic.tools.guards import ValidationGuard

# The guard checks that all listed keys are present in kwargs.
@guarded(ValidationGuard(required_keys=["vendor_name"]))
@agent_tool(name="lookup_vendor", description="Look up vendor")
async def lookup_vendor(vendor_name: str) -> str:
    vendors = {"Acme Corp": "V-001", "Globex": "V-002"}
    return vendors.get(vendor_name, "Vendor not found")

Rate Limit Guard

Prevents a tool from being called too frequently — essential for expensive or metered external APIs:

from pyfly.agentic.tools.guards import RateLimitGuard

# Token-bucket limiter: 10 calls per 60-second sliding window.
@guarded(RateLimitGuard(max_calls=10, period_seconds=60))
@agent_tool(name="web_search", description="Search the web")
async def web_search(query: str) -> str:
    import os
    import httpx

    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(os.environ["SEARCH_ENDPOINT"], params={"q": query})
        response.raise_for_status()
        return response.text

Sandbox Guard

Restricts tool arguments via allow/deny regex patterns — useful for preventing path traversal, secret leakage, or access to dangerous locations:

from pyfly.agentic.tools.guards import SandboxGuard

# Allow paths under /tmp/uploads, deny path traversal (..) and .env files.
@guarded(SandboxGuard(
    allowed_patterns=[r"^/tmp/uploads"],
    denied_patterns=[r"\.\.", r"\.env"],
))
@agent_tool(name="read_file", description="Read a file")
async def read_file(path: str) -> str:
    import asyncio
    from pathlib import Path

    resolved = Path(path).resolve()
    resolved.relative_to(Path("/tmp/uploads").resolve())
    return await asyncio.to_thread(resolved.read_text, encoding="utf-8")

Human-in-the-loop approval

Human approval is not a guard — it pauses the run rather than failing it. Mark a tool with requires_approval=True; the agent run then pauses before the tool executes and the caller resumes with an approval decision. See Human-in-the-Loop Tool Approval for the full flow.

from pyfly.agentic.agents import Agent, is_deferred
from pyfly.agentic.tools import DeferredToolResults

records = {"42": {"title": "Draft invoice"}}

@agent_tool(name="delete_record", description="Delete a record from the local example store", requires_approval=True)
async def delete_record(record_id: str) -> str:
    removed = records.pop(record_id, None)
    return "Deleted" if removed is not None else "Record not found"

agent = Agent(name="approval-example", tools=[delete_record])

result = await agent.run("Delete record 42.")
if is_deferred(result):  # paused for sign-off
    approvals = {c.tool_call_id: True for c in result.output.approvals}  # True / ToolApproved / ToolDenied
    result = await agent.run(message_history=result.all_messages(),
                             deferred_tool_results=DeferredToolResults(approvals=approvals))

Composing Guards

Chain multiple guards with CompositeGuard — all must pass (AND semantics, evaluated in order, first failure short-circuits):

from pyfly.agentic.tools.guards import CompositeGuard, ValidationGuard, RateLimitGuard

guard = CompositeGuard(guards=[
    ValidationGuard(required_keys=["query"]),
    RateLimitGuard(max_calls=5, period_seconds=30),
])

Retryable Tools

The @retryable decorator wraps a tool's execute method with exponential-backoff retry logic — useful for tools that call flaky external APIs:

from pyfly.agentic.tools import agent_tool, retryable

@retryable(max_retries=3, backoff=1.0)
@agent_tool(name="call_erp", description="Fetch data from the ERP API")
async def call_erp(query: str) -> str:
    import os
    import httpx

    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(
            os.environ["ERP_ENDPOINT"], params={"q": query},
            headers={"Authorization": f"Bearer {os.environ['ERP_API_KEY']}"},
        )
        response.raise_for_status()
        return response.text

@retryable stacks with @guarded — guards run first, then retries wrap the handler:

@retryable(max_retries=2)
@guarded(RateLimitGuard(max_calls=10, period_seconds=60))
@agent_tool(name="web_search", description="Search the web")
async def web_search(query: str) -> str:
    import os
    import httpx

    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(os.environ["SEARCH_ENDPOINT"], params={"q": query})
        response.raise_for_status()
        return response.text

Tool Composition

Combine tools into higher-level operations:

from pyfly.agentic.tools import SequentialComposer, FallbackComposer, ConditionalComposer

# Sequential: output of one becomes the `input` kwarg to the next.
# First positional arg is the composed tool's name.
pipeline = SequentialComposer(
    "fetch-parse-validate",
    [fetch_tool, parse_tool, validate_tool],
    description="Fetch, parse, then validate",
)

# Fallback: try tools in order until one succeeds; raise if all fail.
resilient = FallbackComposer(
    "resilient-api",
    [primary_api, secondary_api, cache_lookup],
)

# Conditional: a router_fn receives kwargs and returns a key into tool_map.
def select_parser(**kwargs) -> str:
    return "json" if kwargs.get("format") == "json" else "text"

router = ConditionalComposer(
    "format-router",
    router_fn=select_parser,
    tool_map={"json": json_parser, "text": text_parser},
)

Built-In Tools

The framework ships with nine ready-to-use tools in tools/builtins/:

from pyfly.agentic.tools.builtins import (
    DateTimeTool, # Current date/time, timezone conversion
    JsonTool, # Parse, validate, extract, format JSON
    TextTool, # Word count, regex extract, truncate, replace
    CalculatorTool, # Safe math via AST parsing (no eval)
)

# Each tool follows ToolProtocol and can be registered and guarded
datetime_tool = DateTimeTool(default_timezone="America/New_York")
result = await datetime_tool.execute(action="now")

calculator = CalculatorTool()
result = await calculator.execute(expression="sqrt(144) + pi")

The full list: HttpTool, FileSystemTool, SearchTool, DatabaseTool, ShellTool, DateTimeTool, JsonTool, TextTool, CalculatorTool.

The Tool Registry

from pyfly.agentic.tools.registry import tool_registry

tool_registry.register(my_tool)
tool = tool_registry.get("my_tool")
all_tools = tool_registry.list_tools()

ToolKit

A ToolKit groups related tools and can convert them to Pydantic AI tools for direct injection into an agent:

from pyfly.agentic.tools import ToolKit
from pyfly.agentic.tools.builtins import DateTimeTool, CalculatorTool

datetime_tool = DateTimeTool()
calculator = CalculatorTool()

kit = ToolKit("utility-tools", [datetime_tool, calculator], description="Common utilities")

# Register all tools in the toolkit at once
from pyfly.agentic.tools import tool_registry
kit.register_all(tool_registry)

Attaching Tools to Agents

Pass PyFly Agentic tools or a ToolKit directly to Agent(tools=...). The framework adapts their schemas, guards, retries, and approval metadata to the runtime:

from pyfly.agentic.agents import Agent
from pyfly.agentic.tools import ToolKit, agent_tool
from pyfly.agentic.tools.builtins import CalculatorTool, DateTimeTool

@agent_tool(name="lookup_vendor", description="Look up a vendor in the local directory")
async def lookup_vendor(vendor_name: str) -> str:
    vendors = {"Acme Corp": "Vendor V-001; payment terms NET30", "Globex": "Vendor V-002; payment terms NET60"}
    return vendors.get(vendor_name, "Vendor not found in the directory")

kit = ToolKit("utilities", [lookup_vendor, CalculatorTool(), DateTimeTool()])
agent = Agent(name="helper", tools=[kit])

ToolRegistry is a catalogue for discovery; registration alone does not grant an agent access. Bind only the tools that agent needs. For native SDK integrations, kit.as_pydantic_tools(), kit.as_toolset(), and the .tool() / .tool_plain() decorator proxies remain available.

IDP Tie-In: OCR and Vendor Lookup Tools

For our IDP pipeline, we need tools the extraction agent can call. We define them with @agent_tool, group them into a ToolKit, and pass it through tools=[extraction_kit]. This is the pattern you will see end-to-end in Chapter 6 (reasoning patterns) and Chapter 18 (full IDP application).

Step 1 — Define the tools:

Configure OCR_ENDPOINT and OCR_API_KEY for an OCR service accepting {"image_base64": "base64 encoded image"} and returning {"text": "recognized text"}. The vendor directory below is local exercise data; missing vendors are reported explicitly.

from pyfly.agentic.tools import agent_tool, guarded
from pyfly.agentic.tools.guards import RateLimitGuard

@guarded(RateLimitGuard(max_calls=100, period_seconds=60))
@agent_tool(name="ocr_extract", description="Extract text from a document image via OCR")
async def ocr_extract(image_data: str) -> str:
    """POST a base64 image to the host's OCR endpoint and return its text field."""
    import os
    import httpx

    async with httpx.AsyncClient(timeout=30.0) as client:
        response = await client.post(
            os.environ["OCR_ENDPOINT"],
            headers={"Authorization": f"Bearer {os.environ['OCR_API_KEY']}"},
            json={"image_base64": image_data},
        )
        response.raise_for_status()
        text = response.json()["text"]
        if not isinstance(text, str) or not text.strip():
            raise ValueError("OCR endpoint returned no text")
        return text

@agent_tool(name="vendor_lookup", description="Look up vendor details from the ERP system")
async def vendor_lookup(vendor_name: str) -> str:
    vendors = {
        "Acme Corp": '{"id": "V-001", "tax_id": "US-12345", "payment_terms": "NET30"}',
        "Globex": '{"id": "V-002", "tax_id": "US-67890", "payment_terms": "NET60"}',
    }
    return vendors.get(vendor_name, '{"error": "Vendor not found"}')

Step 2 — Group into a ToolKit and attach to the extraction agent:

from pyfly.agentic.tools import ToolKit
from pyfly.agentic.tools.builtins import CalculatorTool
from pyfly.agentic.agents import Agent

extraction_kit = ToolKit(
    "idp-extraction",
    [ocr_extract, vendor_lookup, CalculatorTool()],
    description="Tools for IDP invoice extraction",
)

extractor_agent = Agent(
    name="extractor",
    instructions="You are an invoice data extraction specialist.",
    tools=[extraction_kit], # PyFly Agentic adapts the toolkit
)

What happens next: In Chapter 6 we pass extractor_agent (with its tools already attached) to reasoning patterns like Plan-and-Execute and Reflexion. The pattern calls agent.run() internally — the tools are available because they were bound here. Chapter 18 shows the complete production module (idp/tools.py) with retries, guards, and the full ToolKit.

---

Chapter 5: Prompts

If tools are an agent's hands, prompts are its brain. The exact wording of a system prompt can mean the difference between extracting 60% of invoice fields and 98%. In production you need to version prompts (so you can A/B test), compose them (system + context + task), validate them (catch missing variables before runtime), and load them from files (so non-engineers can edit them).

The Prompts module provides all of this through a Jinja2-based template engine. Every template is a first-class object with a name, a version, declared variables, and a render method — not just a raw string.

Creating a Prompt Template

A PromptTemplate takes three positional arguments — name, system_template, and user_template — plus keyword-only version, description, required_variables, and metadata. Both templates are Jinja2 source:

from pyfly.agentic.prompts import PromptTemplate

extraction_prompt = PromptTemplate(
    "invoice_extraction",
    # system_template
    "You are an invoice data extraction specialist.\n"
    "Always return valid JSON matching the requested schema.",
    # user_template
    "Extract the following fields from the document text below:\n"
    "- invoice_number\n"
    "- vendor_name\n"
    "- total_amount (numeric)\n"
    "- due_date (ISO format)\n"
    "- line_items (list of {description, quantity, unit_price})\n\n"
    "Document text:\n{{ document_text }}",
    version="1.0.0",
    required_variables=["document_text"],
)

# render() returns a Prompt object with .system and .user fields (NOT a string).
prompt = extraction_prompt.render(document_text="Invoice #INV-001 from Acme Corp...")
print(prompt.system)  # rendered system_template
print(prompt.user)    # rendered user_template

Templates use Jinja2 syntax — {{ variable }}, {% if %}, {% for %}, filters, and macros all work. required_variables is validated at render time: rendering without a required variable raises PromptValidationError.

Versioning

The PromptRegistry supports multiple versions of the same template. This is crucial for A/B testing different prompt strategies:

from pyfly.agentic.prompts import PromptRegistry

registry = PromptRegistry()
registry.register(extraction_prompt_v1)
registry.register(extraction_prompt_v2)

# Get the latest version
latest = registry.get("invoice_extraction")

# Get a specific version
v1 = registry.get("invoice_extraction", "1.0.0")

Composition

Templates can be composed using three strategies:

Sequential Composition

Render templates in order and join them — useful for building system + context + task prompts. The system parts are joined together and the user parts are joined together, and render() returns a single Prompt:

from pyfly.agentic.prompts.composer import SequentialComposer

# By default, templates are joined with "\n\n". Override with the keyword-only `separator=`.
composer = SequentialComposer(
    [system_prompt, context_prompt, task_prompt],
    separator="\n\n",
)
prompt = composer.render(document_text="Invoice #INV-001...")  # -> Prompt(.system, .user)

Conditional Composition

Select a template based on a runtime condition. The condition_fn receives the render kwargs and returns a string key that maps into template_map:

from pyfly.agentic.prompts.composer import ConditionalComposer

# The condition function inspects the kwargs and returns a template key.
# Both args are positional: (condition_fn, template_map).
composer = ConditionalComposer(
    lambda **kwargs: "invoice" if kwargs.get("doc_type") == "invoice" else "generic",
    {
        "invoice": invoice_prompt,
        "generic": generic_prompt,
    },
)
prompt = composer.render(doc_type="invoice", document_text="...")  # -> Prompt

Merge Composition

Render templates and merge with a custom function — full control over how pieces combine:

from pyfly.agentic.prompts.composer import MergeComposer

# The merge_fn is applied separately to the system parts and to the user parts.
# Both args are positional: (templates, merge_fn).
composer = MergeComposer(
    [header, body, footer],
    lambda parts: "\n---\n".join(parts),
)
prompt = composer.render(document_text="...")  # -> Prompt

Validation

The PromptValidator checks rendered prompts against configurable constraints — token limits and required sections — catching problems before they reach the LLM:

from pyfly.agentic.prompts import PromptValidator

# Validate that the rendered prompt fits within 4,000 tokens
# and contains the required "valid JSON" section.
validator = PromptValidator(max_tokens=4000, required_sections=["valid JSON"])
prompt = extraction_prompt.render(document_text="Invoice #INV-001 from Acme Corp...")
# validate() takes a string — pass the rendered user (or system) text.
result = validator.validate(prompt.user)
if not result.valid:
    print(f"Prompt issues: {result.errors}")

Using Prompts with Agents

So far we've created templates, versioned them, and composed them — but none of that is useful until the rendered text reaches an agent. Here is how the two systems connect.

Direct rendering → agent.run() — The simplest path. Render a template and pass the rendered user text as the prompt. (render() returns a Prompt; an agent's run() takes the user text — the system prompt is usually set as the agent's instructions.)

from pyfly.agentic.agents import Agent
from pyfly.agentic.prompts import PromptTemplate

extraction_prompt = PromptTemplate(
    "invoice_extraction",
    "You are a precise invoice data extraction assistant.",  # system_template
    "Extract invoice_number, vendor_name, total_amount, due_date from:\n\n"
    "{{ document_text }}\n\nReturn valid JSON.",              # user_template
)

# Use the template's system text as the agent's instructions.
prompt = extraction_prompt.render(document_text=ocr_output)
agent = Agent(
    name="extractor", instructions=prompt.system, output_type=dict
)
result = await agent.run(prompt.user)
print(result.output) # {"invoice_number": "INV-001", ...}

Composed prompts → agent.run() — Use a composer when you need to assemble multiple templates (system instructions + context + task) into one prompt:

from pyfly.agentic.prompts.composer import SequentialComposer

# Each template provides system + user halves; here the work lives in the user half.
system = PromptTemplate("system", "You are a precise data extraction assistant.", "")
context = PromptTemplate("context", "", "Document type: {{ doc_type }}")
task = PromptTemplate("task", "", "Extract fields from:\n{{ document_text }}")

composer = SequentialComposer([system, context, task])
prompt = composer.render(doc_type="invoice", document_text=ocr_output)
result = await agent.run(prompt.user)

Reasoning patterns use prompts internally — Every reasoning pattern (Chapter 6) has named prompt slots backed by PromptTemplate instances. When a ReAct pattern calls agent.run(), it first renders its "thought" template, passes the result to the agent, and records the output in the trace. You can override any slot:

from pyfly.agentic.reasoning import ReActPattern

# Override the built-in thought prompt with your own template.
custom = PromptTemplate("my:thought", "You are a careful reasoner.", "Think about: {{ context }}")
pattern = ReActPattern(prompts={"thought": custom})

See Chapter 6 → Configurable Prompts for the full list of prompt slots per pattern.

Loading from Files

For large prompts or team workflows, store templates as files:

PromptLoader exposes three static factory methods — from_string(), from_file(), and from_directory():

from pyfly.agentic.prompts import PromptLoader

# From an inline string (name, system_template, user_template).
template = PromptLoader.from_string(
    "invoice_extraction",
    "You are an invoice parser.",
    "Extract fields from:\n{{ document_text }}",
)

# From a YAML file (keys map to PromptTemplate fields; name defaults to the file stem).
template = PromptLoader.from_file("prompts/invoice_extraction.yaml")

# Load an entire directory (defaults to the *.j2 glob).
all_templates = PromptLoader.from_directory("prompts/")

When a name is not supplied, the loader infers it from the file stem (e.g. invoice_extraction.yaml becomes "invoice_extraction").

IDP Tie-In: Versioned Extraction Prompts

For our IDP pipeline, we create versioned prompts that can be A/B tested later:

from pyfly.agentic.prompts import PromptTemplate, PromptRegistry

prompt_registry = PromptRegistry()

# Version 1: Simple extraction
extraction_v1 = PromptTemplate(
    "idp_extraction",
    "You are an invoice parser. Return valid JSON.",  # system_template
    "Extract these fields from the invoice:\n"          # user_template
    "- invoice_number, vendor_name, total_amount, due_date, line_items\n\n"
    "Text: {{ document_text }}",
    version="1.0.0",
    required_variables=["document_text"],
)

# Version 2: More structured with examples
extraction_v2 = PromptTemplate(
    "idp_extraction",
    "You are an expert invoice parser. Extract structured data and return valid JSON.",
    "Required fields:\n"
    " invoice_number: string (format: INV-NNNN)\n"
    " vendor_name: string\n"
    " total_amount: float\n"
    " due_date: string (ISO 8601)\n"
    " line_items: list of {description: str, quantity: int, unit_price: float}\n\n"
    "Example output:\n"
    '{"invoice_number": "INV-0001", "vendor_name": "Example", "total_amount": 100.0, '
    '"due_date": "2026-01-01", "line_items": [{"description": "Widget", "quantity": 2, '
    '"unit_price": 50.0}]}\n\n'
    "Document text:\n{{ document_text }}",
    version="2.0.0",
    required_variables=["document_text"],
)

prompt_registry.register(extraction_v1)
prompt_registry.register(extraction_v2)

# In production, select the version based on experiment configuration
template = prompt_registry.get("idp_extraction") # Returns v2 (latest)
prompt = template.render(document_text=ocr_output)  # -> Prompt(.system, .user)

---

Part II — Intelligence

---

Chapter 6: Reasoning Patterns

Here's a hard truth about LLMs: if you throw a complex question at an agent in a single prompt, you're gambling. Sometimes the model nails it; sometimes it hallucinated half the answer. Reasoning patterns fix that by giving the agent a structured way to think before it answers — loops of thinking, acting, observing, and reflecting, each step recorded so you can see exactly what happened.

The framework ships six patterns out of the box (ReAct, Chain of Thought, Plan-and-Execute, Reflexion, Tree of Thoughts, Goal Decomposition). Each one is a different strategy for the same problem: how do you turn "figure this out" into a repeatable, observable, debuggable process?

The Architecture

AbstractReasoningPattern provides a shared Template Method loop: step counting, trace recording, max-steps enforcement, and optional output review. ReAct, Chain of Thought, Plan-and-Execute, and Reflexion customize its hooks. Tree of Thoughts and Goal Decomposition override execution while reusing structured-output and trace helpers:

_reason(state) → Generate a thought ("what do I think?")
_act(state) → Perform an action ("what should I do?")
_observe(state, act) → Process the action's result ("what happened?")
_should_continue() → Decide whether to loop again ("am I done?")
_extract_output() → Produce the final answer ("what's the result?")

Here is how all the pieces fit together:

flowchart TD
    PATTERN["Reasoning pattern"] --> MODE{"Execution strategy"}
    MODE -->|ReAct, CoT, Plan-and-Execute, Reflexion| LOOP["Base execute loop"]
    LOOP --> REASON["reason, early-stop check, act, observe"]
    REASON --> CONTINUE{"Continue below max_steps?"}
    CONTINUE -->|yes| LOOP
    CONTINUE -->|finished| OUTPUT["Extract output"]
    CONTINUE -->|limit reached| ERROR["ReasoningStepLimitError"]
    MODE -->|Tree of Thoughts| TREE["Generate, evaluate and select a branch"]
    MODE -->|Goal Decomposition| GOALS["Plan phases and execute tasks"]
    TREE --> OUTPUT
    GOALS --> OUTPUT
    OUTPUT --> REVIEW["Optional output reviewer"]
    REVIEW --> RESULT["ReasoningResult with trace"]

How Reasoning Patterns Use Agent Tools

An important thing to understand: reasoning patterns don't manage tools. They manage thinking. When a pattern needs to interact with the outside world (search a database, call an API, do math), it delegates to agent.run() — and the agent's configured Pydantic AI tools handle the rest.

This means you must attach tools to the agent before passing it to a pattern. Here is the canonical flow, using the framework's tool system end-to-end:

from pyfly.agentic.agents import Agent
from pyfly.agentic.reasoning import ReActPattern
from pyfly.agentic.tools import ToolKit, agent_tool
from pyfly.agentic.tools.builtins import CalculatorTool

@agent_tool(name="vendor_lookup", description="Look up a vendor in the local directory")
async def vendor_lookup(vendor_name: str) -> str:
    vendors = {"Acme Corp": '{"id": "V-001", "tax_id": "US-12345"}'}
    return vendors.get(vendor_name, '{"error": "Vendor not found"}')

extraction_tools = ToolKit("extraction-tools", [vendor_lookup, CalculatorTool()])
extractor = Agent(name="reasoning-extractor", tools=[extraction_tools])
react = ReActPattern(max_steps=5)
result = await react.execute(extractor, "Look up Acme Corp and calculate 100 plus 8 percent tax.")

The runtime flow is: Pattern → agent.run(prompt) → Pydantic AI → LLM decides to call tools → tool results → back to pattern loop.

You can also skip the framework tool system and pass plain Pydantic AI tool functions directly to Agent(tools=[...]) — see Chapter 4 → Attaching Tools to Agents for framework and advanced integration options.

Architecture: Why Tools Are Explicitly Bound to Agents

You may wonder: "If @agent_tool already registers a tool in the active runtime ToolRegistry, why doesn't every agent automatically see every tool?"

This is by design. The framework has two separate layers with different purposes:

1. ToolRegistry (discovery layer) — An application-scoped catalog of all tools in the system. Used for admin dashboards, documentation generation, plugin discovery, and pipeline wiring. Think of it as a phone book: it lists everything that exists.

2. Agent tool binding (execution layer) — The specific set of tools an agent can invoke during an LLM run. Only tools explicitly passed via Agent(tools=[]) or agent.tool() are available to the model.

Why explicit binding?

• Security — An agent that processes user input should not have access to ShellTool or delete_record. Implicit auto-injection of all registered tools would create a dangerous attack surface.
• Predictability — When you read an agent definition, you can see exactly which tools it can call. No surprises from a plugin that registered a tool at import time.
• Cost control — Each tool in a Pydantic AI agent's tool list adds tokens to the system prompt. Injecting 50 tools when an agent only needs 3 wastes tokens and confuses the model.
• Principle of least privilege — Each agent gets the minimum set of tools required for its task, not the maximum set available.

Use ToolKit to curate which subset of registered tools each agent receives:

from pyfly.agentic.tools import ToolKit
from pyfly.agentic.tools.registry import tool_registry

# Curate: only extraction-related tools for this agent
extraction_kit = ToolKit("extraction", [
    tool_registry.get("vendor_lookup"),
    tool_registry.get("calculate"),
])

extractor = Agent(
    name="extractor",
    tools=[extraction_kit],
)

Memory in Reasoning Patterns

Reasoning patterns can also access the framework's memory system. When you pass a MemoryManager via the memory keyword argument, it becomes available in the pattern's internal state as state["memory"]. This lets pattern hooks read and write working memory during iterations:

from pyfly.agentic.reasoning import PlanAndExecutePattern
from pyfly.agentic.memory import MemoryManager

memory = MemoryManager(working_scope_id="extraction-session")

# Store a fact that the pattern's hooks can read
memory.set_fact("doc_type", "invoice")

# The memory kwarg is forwarded into state["memory"] inside the pattern
pattern = PlanAndExecutePattern(max_steps=15)
result = await pattern.execute(
    extractor_agent,
    "Extract invoice fields from the OCR text.",
    memory=memory, # Available as state["memory"] in _reason(), _act(), etc.
)

In pipelines, this happens automatically — ReasoningStep passes PipelineContext.memory to the pattern (see Chapter 10).

Structured Output Models

Patterns use typed Pydantic models instead of raw text. This is a big deal: instead of parsing free-form strings for magic words like "FINISH", the model returns structured objects with explicit fields. No more fragile regex parsing:

from pyfly.agentic.reasoning.models import (
    ReasoningThought,
    ReasoningAction,
    ReasoningPlan,
    PlanStepDef,
    StepStatus,
    ReflectionVerdict,
    BranchEvaluation,
    GoalDecompositionResult,
    GoalPhase,
)

# A structured thought — is_final replaces magic phrases like "FINISH"
thought = ReasoningThought(
    content="The invoice number is INV-2026-001",
    is_final=True,
    final_answer="INV-2026-001",
    confidence=0.95,
)

# A structured plan with step tracking
plan = ReasoningPlan(
    goal="Extract all invoice fields",
    steps=[
        PlanStepDef(id="s1", description="Find invoice number"),
        PlanStepDef(id="s2", description="Find vendor name", dependencies=["s1"]),
        PlanStepDef(id="s3", description="Find amount and date", dependencies=["s1"]),
    ],
)
# Each step has a status: pending → running → completed/failed/skipped

The Six Built-In Patterns

1. ReAct (Reasoning + Acting)

Interleaves thinking and tool use. Each iteration produces a ReasoningThought; when is_final=True, the loop stops.

from pyfly.agentic.reasoning import ReActPattern

react = ReActPattern(max_steps=5)
result = await react.execute(my_agent, "What is the total on invoice INV-001?")
print(result.output)
print(f"Steps taken: {len(result.trace.steps)}")

When to use: Tasks that need external information (tool calls, searches, API lookups) combined with reasoning.

2. Chain of Thought

The agent reasons step by step through a problem. Each step is a ReasoningThought.

from pyfly.agentic.reasoning import ChainOfThoughtPattern

cot = ChainOfThoughtPattern(max_steps=10)
result = await cot.execute(my_agent, "Calculate the VAT on $1,234.56 at 20%.")

When to use: Math, logic puzzles, multi-step deductions where the agent needs to show its work.

3. Plan-and-Execute

The agent first generates a ReasoningPlan with PlanStepDef items, then executes each step with status tracking (pending → running → completed/failed). Supports replanning when steps fail.

from pyfly.agentic.reasoning import PlanAndExecutePattern

planner = PlanAndExecutePattern(max_steps=15, allow_replan=True)
result = await planner.execute(
    extractor_agent,
    "Extract all fields from this invoice:\n" + ocr_text,
)

When to use: Complex, multi-step tasks where you want visibility into progress and the ability to recover from failures.

4. Reflexion

Adds a self-evaluation loop. After generating output, the agent critiques itself via a ReflectionVerdict. If is_satisfactory=False, the issues and suggestions are fed back into a retry prompt.

from pyfly.agentic.reasoning import ReflexionPattern

reflexion = ReflexionPattern(max_steps=3)
result = await reflexion.execute(my_agent, "Write a unit test for binary search.")

When to use: Tasks where quality matters and the agent can meaningfully self-correct (writing code, generating reports, data extraction).

5. Tree of Thoughts

Explores multiple reasoning branches and evaluates each with a BranchEvaluation (score + reasoning). The highest-scoring branch is selected.

from pyfly.agentic.reasoning import TreeOfThoughtsPattern

tot = TreeOfThoughtsPattern(branching_factor=3)
result = await tot.execute(my_agent, "Design an API for a task management system.")

When to use: Open-ended problems with multiple valid approaches where you want to explore and compare alternatives.

6. Goal Decomposition

Breaks a high-level goal into structured GoalPhase objects, each with concrete tasks.

from pyfly.agentic.reasoning import GoalDecompositionPattern

decomposer = GoalDecompositionPattern(max_steps=20)
result = await decomposer.execute(my_agent, "Build an IDP pipeline for invoice processing.")

When to use: Strategic planning, project decomposition, breaking down large goals into actionable phases.

Configurable Prompts

Every pattern uses PromptTemplate instances for its LLM calls. You can override any prompt by passing a prompts dict:

from pyfly.agentic.prompts.template import PromptTemplate
from pyfly.agentic.reasoning import ReActPattern

custom_thought = PromptTemplate(
    "my:react:thought",
    "You are a careful reasoner.",            # system_template
    "Think carefully about: {{ context }}",   # user_template
    required_variables=["context"],
)
pattern = ReActPattern(prompts={"thought": custom_thought})

Each pattern has named prompt slots:

• ReAct: "thought"
• Chain of Thought: "step"
• Plan-and-Execute: "plan", "execute_step", "replan"
• Reflexion: "critique", "retry"
• Tree of Thoughts: "branch", "evaluate"
• Goal Decomposition: "decompose", "plan_phase"

All built-in prompts are registered in the active runtime’s prompt_registry under the reasoning: namespace (e.g. "reasoning:react:thought").

Reasoning Pipeline

Chain patterns sequentially — the output of one becomes the input to the next:

from pyfly.agentic.reasoning import ReasoningPipeline, PlanAndExecutePattern, ReActPattern

pipeline = ReasoningPipeline([
    PlanAndExecutePattern(max_steps=15),
    ReActPattern(max_steps=10),
])
result = await pipeline.execute(my_agent, "Analyse the invoice and verify the totals.")

Reasoning Trace

Every pattern records its work in a ReasoningTrace. The trace contains typed steps:

• ThoughtStep — The agent's reasoning.
• ActionStep — A tool invocation.
• ObservationStep — The result of an action.
• ReflectionStep — Self-evaluation.
• PlanStep — A plan or sub-goal.

for step in result.trace.steps:
    print(step.kind, step.model_dump(exclude_none=True))

Creating a Custom Pattern

Extend AbstractReasoningPattern and override its hooks. This implementation asks the agent for an answer, then asks it to check and correct that answer in a second call. The trace records the two operations; independent domain validation can be added through an OutputReviewer.

from pyfly.agentic.reasoning.base import AbstractReasoningPattern
from pyfly.agentic.reasoning.trace import ObservationStep, ThoughtStep

class VerifyAndCorrectPattern(AbstractReasoningPattern):
    def __init__(self):
        super().__init__("verify_and_correct", max_steps=1)

    async def _reason(self, state):
        draft = await state["agent"].run(str(state["input"]))
        state["draft"] = str(draft.output)
        return ThoughtStep(content="Generated a draft for verification.")

    async def _act(self, state):
        review = await state["agent"].run(
            f"Check the draft against the task. Return a corrected final answer.\n"
            f"Task: {state['input']}\nDraft: {state['draft']}"
        )
        state["output"] = review.output
        return ObservationStep(content=str(review.output), source="verification")

Register it to make it available framework-wide:

from pyfly.agentic.reasoning.registry import reasoning_registry

reasoning_registry.register("verify_and_correct", VerifyAndCorrectPattern)

IDP Tie-In: Extraction with Plan-and-Execute + Reflexion

Now let's put reasoning patterns to work in our invoice pipeline. In Chapter 4 we defined IDP tools (ocr_extract, vendor_lookup) and grouped them into extraction_kit. We also created extractor_agent with those tools attached via tools=[extraction_kit]. Here we pass that agent — tools and all — to reasoning patterns.

The extraction phase is the hardest part — we need to find invoice numbers, vendor names, amounts, and line items from messy OCR text. A single-shot prompt might miss something, so we use Plan-and-Execute to break it into steps, and Reflexion as a safety net when validation catches errors:

from pyfly.agentic.reasoning import PlanAndExecutePattern, ReflexionPattern

# ── Recall from Chapter 4 ───────────────────────────────────────────
# extractor_agent = Agent(
# name="extractor",
# tools=[extraction_kit], # ocr_extract, vendor_lookup, CalculatorTool
# )
# The tools are already bound — reasoning patterns call agent.run() internally,
# so the agent can invoke any of its tools during each reasoning step.
# ─────────────────────────────────────────────────────────────────────

# Phase 3: Systematic extraction.
# Plan-and-Execute generates a plan ("find invoice number", "find vendor", ...)
# and executes each step with status tracking. If a step fails, it can replan.
extraction_pattern = PlanAndExecutePattern(max_steps=15, allow_replan=True)
extraction_result = await extraction_pattern.execute(
    extractor_agent, # Tools already attached in Ch4 via tools=[extraction_kit]
    f"Extract invoice fields from:\n{ocr_text}",
)

# Phase 4: If validation catches errors (e.g. invalid date format), use Reflexion
# to let the agent critique its own output and self-correct.
if not validation_passed:
    reflexion = ReflexionPattern(max_steps=2)
    corrected = await reflexion.execute(
        extractor_agent,
        f"Previous extraction had errors: {errors}. Re-extract from:\n{ocr_text}",
    )

Architecture recap: Reasoning patterns never see tools directly. They receive an agent (which owns its tools) and call agent.run(). This is why tools must be bound to the agent before passing it to a pattern — see the "Attaching Tools to Agents" section in Chapter 4 and the full idp/tools.py module in Chapter 18.

---

Chapter 7: Content Processing

Here's a problem you'll hit fast: your 50-page invoice PDF produces OCR text that is 200,000 tokens long — but your model's context window is 128K, and you're paying per token. You can't just shove the entire document into one call.

The Content module solves this with a three-stage pipeline:

1. Chunk — Split oversized content into overlapping pieces that each fit within a token budget. TextChunker handles text (by token, sentence, or paragraph), DocumentSplitter handles multi-page documents, and ImageTiler handles large images.
2. Process — Send chunks through an agent concurrently via BatchProcessor. This is where you do per-chunk OCR cleanup, translation, or summarisation.
3. Compress — Merge the results back into a single context that fits the downstream agent's window. ContextCompressor supports three strategies: truncation (cheap, lossy), summarisation (LLM-based, preserves meaning), and map-reduce (chunk → summarise each → merge).

Together they let you process documents of any size without losing information or blowing up your bill.

Content Processing Pipeline

graph LR
    subgraph "1. Chunk"
        TC["TextChunker"]
        DS["DocumentSplitter"]
        IT["ImageTiler"]
    end

    subgraph "2. Process"
        BP["BatchProcessor\n(concurrent agent calls)"]
    end

    subgraph "3. Compress"
        CC["ContextCompressor"]
        TR["TruncationStrategy"]
        SM["SummarizationStrategy"]
        MR["MapReduceStrategy"]
    end

    subgraph "Utilities"
        TE["TokenEstimator"]
        SW["SlidingWindowManager"]
    end

    TC --> BP
    DS --> BP
    IT --> BP
    BP --> CC
    CC --> TR
    CC --> SM
    CC --> MR
    TE -.-> TC
    TE -.-> CC
    TE -.-> SW

Text Chunking

TextChunker splits text into overlapping chunks using one of three strategies:

from pyfly.agentic.content.chunking import TextChunker

chunker = TextChunker(
    chunk_size=4000, # Max tokens per chunk
    chunk_overlap=200, # Overlap between consecutive chunks
    strategy="paragraph", # "token" | "sentence" | "paragraph"
)

chunks = chunker.chunk(long_document_text)
for chunk in chunks:
    print(f"Chunk {chunk.index}/{chunk.total_chunks}: {len(chunk.content)} chars")
    # chunk.metadata contains additional info

Each chunk is a Chunk model with content, index, total_chunks, overlap_tokens, and an open metadata dict.

Strategy guide: - "token" — Split by estimated token count. Best for arbitrary text. - "sentence" — Split at sentence boundaries. Best for natural language. - "paragraph" — Split at paragraph boundaries (\n\n). Best for structured docs.

Document Splitter

DocumentSplitter handles multi-document inputs, splitting at page breaks (\f) or horizontal rules (---):

from pyfly.agentic.content.chunking import DocumentSplitter

splitter = DocumentSplitter(min_length=50)
segments = splitter.split(multi_page_text)

Image Tiler

For large images that exceed a VLM's pixel budget, ImageTiler computes tile coordinates:

from pyfly.agentic.content.chunking import ImageTiler

tiler = ImageTiler(tile_width=1024, tile_height=1024, overlap=128)
tiles = tiler.compute_tiles(image_width=4096, image_height=6144)
# Each tile is a Chunk with metadata: x, y, width, height, row, col

Batch Processing

BatchProcessor sends chunks through an agent concurrently. By default it returns a list of per-chunk result strings; pass a result_aggregator callable to merge them into any shape you like:

from pyfly.agentic.content.chunking import BatchProcessor

processor = BatchProcessor(concurrency=4)
results = await processor.process(ocr_agent, chunks)  # -> list[str]

Context Compression

When concatenated chunks still exceed the extraction agent's context window, compress them:

Truncation Strategy

Hard-cuts the text at a token limit:

from pyfly.agentic.content.compression import ContextCompressor, TruncationStrategy

compressor = ContextCompressor(strategy=TruncationStrategy())
compressed = await compressor.compress(full_text, max_tokens=8000)

Summarization Strategy

Uses an LLM agent to intelligently summarise:

from pyfly.agentic.content.compression import SummarizationStrategy

compressor = ContextCompressor(
    strategy=SummarizationStrategy(summary_agent)
)
compressed = await compressor.compress(full_text, max_tokens=8000)

Map-Reduce Strategy

Chunks the text, summarises each chunk in parallel, then merges the summaries:

from pyfly.agentic.content.compression import MapReduceStrategy

compressor = ContextCompressor(
    strategy=MapReduceStrategy(summary_agent)
)
compressed = await compressor.compress(full_text, max_tokens=8000)

Sliding Window Manager

Maintains a sliding window over a stream of messages, keeping total token usage within budget:

from pyfly.agentic.content.compression import SlidingWindowManager

window = SlidingWindowManager(max_tokens=8000)
window.add("First OCR page output...")
window.add("Second OCR page output...")
current_context = window.get_context() # Only recent items that fit

Token Estimator

Estimate token counts without an API call:

from pyfly.agentic.content.compression import TokenEstimator

estimator = TokenEstimator() # Default ratio: 1.33 tokens per word
tokens = estimator.estimate("This is a test sentence.")

Binary Ingestion (content.binary)

The text pipeline above assumes you already have text. For real documents — PDFs, Office files, images, archives, emails — the content.binary submodule (installed via the [agentic-binary] extra) turns raw caller bytes into one or more normalised BinaryArtifact rows ready for OCR/extraction. BinaryNormalizer dispatches by sniffed media type and delegates to pluggable handlers: PdfGuard (PDF sanitisation), ImageNormalizer, OfficeConverter (build via build_office_converter, backed by GotenbergConverter, LibreOfficeConverter, or NoOpOfficeConverter), ArchiveUnpacker, and EmailUnpacker.

from pyfly.agentic.content.binary import (
    BinaryNormalizer, BinaryConfig, build_office_converter, sniff_media_type,
)

config = BinaryConfig()  # caps, toggles
normalizer = BinaryNormalizer(config=config, office=build_office_converter(config))

# normalise() is async and never returns empty.
artifacts = await normalizer.normalise(raw_bytes, filename="invoice.pdf")
for art in artifacts:
    print(art.media_type, art.kind, art.page_count)

Handlers are injected so a host can swap implementations (e.g. Gotenberg vs LibreOffice) without touching the framework.

IDP Tie-In: Processing Large Documents

In our IDP pipeline, the OCR phase may produce text that exceeds the extraction agent's context window. Here's how we handle it:

from pyfly.agentic.content.chunking import TextChunker, BatchProcessor
from pyfly.agentic.content.compression import ContextCompressor, MapReduceStrategy

# Step 1: Chunk the raw OCR output
chunker = TextChunker(chunk_size=3000, chunk_overlap=200, strategy="paragraph")
chunks = chunker.chunk(raw_ocr_text)

# Step 2: If needed, process chunks in parallel through OCR cleanup agent.
# By default process() returns a list of strings (one per chunk).
processor = BatchProcessor(concurrency=4)
cleaned_chunks = await processor.process(ocr_cleanup_agent, chunks)

# Step 3: Compress for the extraction agent (compress() is async — await it).
full_text = "\n".join(cleaned_chunks)
compressor = ContextCompressor(
    strategy=MapReduceStrategy(summary_agent)
)
compressed_text = await compressor.compress(full_text, max_tokens=8000)

---

Chapter 8: Memory

Without memory, every agent.run() call starts from scratch — the agent has no idea what happened in the previous turn, what the pipeline figured out two steps ago, or what facts a delegated sub-agent discovered. That's fine for one-shot tasks, but real applications need context: multi-turn conversations, session state, pipeline variables. The Memory module gives your agents a brain that persists across calls.

There are two kinds of memory here: - Conversation memory — the actual chat history ("what was said"), automatically trimmed to stay within your token budget. - Working memory — a key-value scratchpad ("what we know") for storing facts like doc_type = "invoice" that multiple agents or pipeline steps can read.

Architecture

flowchart TD
    CONSUMERS["Agents, delegation, reasoning, pipelines"] --> MANAGER["MemoryManager"]
    MANAGER --> HISTORY["ConversationMemory<br/>typed messages held in process"]
    HISTORY --> BUDGET["Token budget and optional summarization"]
    HISTORY <-->|export and import| SNAPSHOT["Host-persisted conversation snapshot"]
    MANAGER --> FACTS["WorkingMemory<br/>scope-specific facts"]
    FACTS --> CONTRACT["MemoryStore protocol"]
    CONTRACT --> LOCAL["InMemoryStore, FileStore, SQLiteStore"]
    CONTRACT --> DB["Native Data: relational, PostgreSQL, MongoDB"]
    CONTRACT --> CUSTOM["Custom backend"]

The system has four layers:

1. ConversationMemory — Token-aware, per-conversation chat history. Wraps Pydantic AI's message_history and drops the oldest turns when you exceed budget.
2. WorkingMemory — Scoped key-value scratchpad for facts and intermediate state.
3. MemoryStore — Pluggable persistence (InMemoryStore, FileStore, or yours).
4. MemoryManager — Facade that composes conversation + working memory behind a single API. This is the object you wire into agents, pipelines, and patterns.

Quick Start

Here's the simplest way to give an agent a memory — attach a MemoryManager and pass a conversation_id to each call:

from pyfly.agentic.agents import Agent
from pyfly.agentic.memory import MemoryManager

# Create a memory manager with a 32K token budget for conversation history.
# When the history exceeds this, the oldest turns get dropped automatically.
memory = MemoryManager(max_conversation_tokens=32_000)

# Wire the memory into the agent. From now on, every run() call
# can participate in a persistent conversation.
agent = Agent(name="assistant", memory=memory)

# Start a new conversation — this returns a unique conversation ID.
conv_id = memory.new_conversation()

# Turn 1: the agent has no history yet.
result1 = await agent.run("What is Python?", conversation_id=conv_id)

# Turn 2: under the hood, Agent loads the message_history from Turn 1
# and passes it to Pydantic AI, so the model sees the retained conversation within its configured token budget.
result2 = await agent.run("What about its type system?", conversation_id=conv_id)
# result2 receives the retained messages from the first turn.

Conversation Memory

ConversationMemory manages chat history per conversation ID. It enforces a token budget by dropping the oldest turns (FIFO):

from pyfly.agentic.memory import ConversationMemory

conv_mem = ConversationMemory(max_tokens=16_000)
cid = conv_mem.new_conversation()

# After an agent run, store the turn
conv_mem.add_turn(
    cid,
    user_prompt="Hello",
    assistant_response="Hi there!",
    raw_messages=result.new_messages(),
)

# Before the next run, get trimmed history
history = conv_mem.get_message_history(cid)

With memory attached and a conversation_id, PyFly Agentic loads history unless explicit message_history was supplied and records completed turns automatically. This is process-local history; persist snapshots with conversation export/import when needed.

Working Memory

WorkingMemory is a scoped key-value store for facts and intermediate state:

from pyfly.agentic.memory import WorkingMemory

wm = WorkingMemory(scope_id="idp-session-42")
wm.set("doc_type", "invoice")
wm.set("vendor", "Acme Corp")

print(wm.get("doc_type")) # "invoice"
print(wm.to_dict()) # {"doc_type": "invoice", "vendor": "Acme Corp"}

# Render as a context block for prompt injection
print(wm.to_context_string())
# Working Memory:
# - doc_type: invoice
# - vendor: Acme Corp

Multiple WorkingMemory instances can share a backend while maintaining independent namespaces:

from pyfly.agentic.memory import InMemoryStore, WorkingMemory

store = InMemoryStore()
agent_a_mem = WorkingMemory(store=store, scope_id="agent_a")
agent_b_mem = WorkingMemory(store=store, scope_id="agent_b")

agent_a_mem.set("key", "from A")
agent_b_mem.set("key", "from B")
assert agent_a_mem.get("key") == "from A" # Isolated

Storage Backends

These backends persist working facts. They do not automatically persist ConversationMemory; see the memory persistence diagram.

InMemoryStore

Dict-backed, fast, non-persistent. Ideal for testing and short-lived sessions:

from pyfly.agentic.memory import InMemoryStore
store = InMemoryStore()

FileStore

JSON file persistence. Each namespace gets its own file:

from pyfly.agentic.memory import FileStore
store = FileStore(base_dir=".firefly_memory")

SQLiteStore

Single-file SQLite persistence — durable and queryable without a separate server:

from pyfly.agentic.memory import SQLiteStore
store = SQLiteStore(path=".firefly_memory/memory.db")

For native database memory choose relational, postgres or mongodb and configure the application's PyFly Data resources. Auto-configuration borrows those resources; standalone URL/pool settings now fail with migration guidance. relational can use SQLite with agentic-relational; postgres preserves firefly_memory.memory_entries, and Mongo preserves firefly_memory.entries.

# runtime is injected from the owning application context.
await runtime.memory.async_set_fact("document_type", "invoice")
assert await runtime.memory.async_get_fact("document_type") == "invoice"

Await MemoryManager.async_* / WorkingMemory.async_* in native async code. They execute on the application loop and join the caller's unit of work. Same-loop blocking sync calls fail promptly; another thread may bridge to the running owner loop with transaction context. Closing an owned adapter leaves its borrowed datasource/client open. Data disposes its resources after Agentic drains. The complete SQLite transaction example demonstrates commit/rollback without a provider request. Existing global-entry-ID schemas/indexes require explicit compound namespace/ID migrations before startup; no destructive DDL or payload rewrite occurs automatically.

Conversation history remains process-local unless exported/imported. The optional create_llm_summarizer() uses the configured model or an explicit model=; it does not itself turn history into persistent database records. See memory lifecycle and native migration.

Custom Backends

Implement the MemoryStore protocol:

from pyfly.agentic.memory import MemoryStore, MemoryEntry
from pyfly.agentic.memory.store import InMemoryStore

class PrefixedStore:
    def __init__(self, prefix: str, backend: MemoryStore) -> None:
        self._prefix = prefix
        self._backend = backend

    def _namespace(self, namespace: str) -> str:
        return f"{len(self._prefix)}:{self._prefix}:{namespace}"

    def save(self, namespace: str, entry: MemoryEntry) -> None:
        self._backend.save(self._namespace(namespace), entry)

    def load(self, namespace: str) -> list[MemoryEntry]:
        return self._backend.load(self._namespace(namespace))

    def load_by_key(self, namespace: str, key: str) -> MemoryEntry | None:
        return self._backend.load_by_key(self._namespace(namespace), key)

    def delete(self, namespace: str, entry_id: str) -> None:
        self._backend.delete(self._namespace(namespace), entry_id)

    def clear(self, namespace: str) -> None:
        self._backend.clear(self._namespace(namespace))

store = PrefixedStore("invoice-service", InMemoryStore())
store.save("working", MemoryEntry(key="vendor", content="Acme Corp"))
assert store.load_by_key("working", "vendor").content == "Acme Corp"

MemoryManager

The MemoryManager facade is the object you attach to agents, delegation routers, and pipelines:

from pyfly.agentic.memory import MemoryManager

mgr = MemoryManager(
    max_conversation_tokens=32_000,
    working_scope_id="main-session",
)

# Conversation
cid = mgr.new_conversation()
mgr.add_turn(cid, "hello", "hi", raw_messages)
history = mgr.get_message_history(cid)

# Working memory
mgr.set_fact("doc_type", "invoice")
mgr.get_fact("doc_type") # "invoice"

Forking

When delegating to a sub-agent or branching a pipeline, fork() creates a child manager that shares conversation memory but gets independent working memory. This means a delegated agent can store its own facts without overwriting the parent's:

# Parent memory has facts about the overall session
mgr.set_fact("session_id", "abc-123")

# Fork for a sub-agent — it gets its own working memory scope
# but can still read the same conversation history.
child = mgr.fork(working_scope_id="sub-agent-classify")
child.set_fact("classification", "invoice") # Only visible in child
assert mgr.get_fact("classification") is None # Parent is unaffected
assert child.get_fact("session_id") is None # Child has its own scope

Memory Integration Points

Memory flows through the framework in four ways. Understanding these is key to building applications where context is never lost:

1. Agent integration — When Agent has a MemoryManager attached, calling run(prompt, conversation_id=cid) automatically loads conversation history as message_history before the call and stores new messages after it. You never touch message_history manually.

2. Delegation integration — DelegationRouter automatically forks memory when routing to a sub-agent, so the delegated agent gets its own working-memory scope while sharing conversation context.

3. Pipeline integration — When you create PipelineContext(memory=mgr), both AgentStep and ReasoningStep propagate the memory to the agent and pattern respectively. Facts stored in one step are readable in subsequent steps.

4. Reasoning integration — Pass memory=mgr as a keyword argument to pattern.execute(agent, input, memory=mgr). The memory object becomes available as state["memory"] inside all pattern hooks (_reason, _act, etc.), so custom patterns can read and write working memory during reasoning iterations.

Configuration

Memory settings via environment variables:

export PYFLY_AGENTIC_MEMORY_BACKEND=in_memory
export PYFLY_AGENTIC_MEMORY_MAX_CONVERSATION_TOKENS=128000
export PYFLY_AGENTIC_MEMORY_SUMMARIZE_THRESHOLD=10
export PYFLY_AGENTIC_MEMORY_FILE_DIR=.firefly_memory

IDP Tie-In: Carrying Facts Across Pipeline Steps

Now let's see why memory matters for our IDP pipeline. When the classifier figures out that a document is an invoice, the extractor needs to know that — it selects different prompts for invoices vs. receipts. Working memory is the bridge:

from pyfly.agentic.memory import MemoryManager

# One memory manager for the entire pipeline run.
# Every step can read and write facts here.
memory = MemoryManager(working_scope_id="idp-run-001")

# Phase 1: The classifier stores what it learned.
# These facts persist across all subsequent pipeline steps.
classification = {"doc_type": "invoice", "language": "en", "page_count": 2}
memory.set_fact("doc_type", classification["doc_type"])
memory.set_fact("language", classification["language"])
memory.set_fact("page_count", classification["page_count"])

# Phase 3: The extractor reads what the classifier discovered.
# This works because both steps share the same MemoryManager.
doc_type = memory.get_fact("doc_type") # "invoice"
# Now we can pick the invoice-specific extraction prompt.

---

Chapter 9: Validation & Quality of Service

You've built an agent that extracts invoice data — but how do you trust the output? LLMs are probabilistic: they can hallucinate an invoice number that doesn't exist, format a date as "January 15^th" instead of ISO 8601, or return an amount as a string instead of a number. In production, bad data propagates downstream and causes real damage.

The Validation module gives you a two-layer defence:

• Layer 1 — Structural validation catches deterministic errors. You define rules (regex, range, format, enum, or custom functions) for each field, group them into an OutputValidator, and run them against the extracted dict or Pydantic model. This is fast, cheap (no LLM call), and catches the most common failures: wrong date formats, out-of-range amounts, invalid invoice number patterns.

• Layer 2 — Quality of Service (QoS) catches statistical errors that rules can't see. ConfidenceScorer asks the LLM to self-rate its own output. ConsistencyChecker runs the same prompt multiple times and measures agreement — low agreement suggests hallucination. GroundingChecker verifies that extracted values actually appear in the source document (no LLM needed). QoSGuard composes all three into a single pass/fail gate.

Both layers can be wired into an OutputReviewer that wraps an agent call with automatic retry: if structural validation or QoS fails, the reviewer feeds the errors back to the agent and re-runs — up to max_retries times.

Output Validation Rules

Rules are composable predicates that check a single field value:

from pyfly.agentic.validation.rules import (
    RegexRule,
    FormatRule,
    RangeRule,
    EnumRule,
    CustomRule,
    FieldValidator,
)

# Regex: field value must match a pattern.
# First arg is always the field name.
inv_rule = RegexRule("invoice_number", r"^INV-\d{4,10}$")

# Format: field matches a named format (email, url, date, uuid, iso_date)
date_rule = FormatRule("due_date", "iso_date")

# Range: numeric field within bounds
amount_rule = RangeRule("total_amount", min_value=0.01, max_value=10_000_000)

# Enum: field value is one of an allowed set
status_rule = EnumRule("status", ["approved", "rejected", "pending"])

# Custom: any predicate function
custom_rule = CustomRule(
    "vendor_name",
    lambda v: isinstance(v, str) and len(v.strip()) > 0,
    description="Value must be a non-empty string",
)

Field Validator

Validates a single field against one or more rules:

validator = FieldValidator("invoice_number", [inv_rule])
# validate() takes the field's value directly, not a dict.
results = validator.validate("INV-001234")
assert all(r.passed for r in results)

Output Validator

Validates an entire structured output (dict or Pydantic model). The constructor takes a dict mapping field names to lists of rules:

from pyfly.agentic.validation.rules import OutputValidator

validator = OutputValidator({
    "invoice_number": [RegexRule("invoice_number", r"^INV-\d{4,10}$")],
    "vendor_name": [RegexRule("vendor_name", r".{2,}")],
    "total_amount": [RangeRule("total_amount", min_value=0.01, max_value=10_000_000)],
    "due_date": [FormatRule("due_date", "iso_date")],
    "status": [EnumRule("status", ["approved", "rejected", "pending"])],
})

# validate() accepts a dict, Pydantic model, or any object with __dict__.
report = validator.validate(extracted_data)
if not report.valid:
    for err in report.errors:
        print(f" {err.field_name}: {err.message}")

Quality of Service (QoS)

The QoS module provides post-generation quality checks. Each checker requires an agent (for LLM-based evaluation) or operates purely on text (grounding).

Confidence Scorer

Asks the agent to self-evaluate its output confidence on a 0.0–1.0 scale:

from pyfly.agentic.validation.qos import ConfidenceScorer

# The scorer needs an agent — it sends a self-evaluation prompt to the LLM.
scorer = ConfidenceScorer(evaluator_agent)
score = await scorer.score("The invoice total is $1,234.56") # 0.0–1.0

Consistency Checker

Runs the same prompt multiple times and measures word-level agreement across outputs (Jaccard similarity):

from pyfly.agentic.validation.qos import ConsistencyChecker

# The checker needs an agent and a run count. It runs the prompt num_runs times.
checker = ConsistencyChecker(extractor_agent, num_runs=3)
score, outputs = await checker.check("What is the capital of France?")
# score: float (1.0 = all answers agree), outputs: list[str]
print(f"Consistency: {score:.2f} across {len(outputs)} runs")

Grounding Checker

Verifies that extracted field values actually appear in the source document — no agent needed, purely text-based:

from pyfly.agentic.validation.qos import GroundingChecker

checker = GroundingChecker(min_grounding_ratio=0.8)

# check() takes the source text and a dict of field_name → extracted_value.
score, field_map = checker.check(
    source_text="Invoice #INV-001: Total Amount: $1,234.56. Vendor: Acme Corp.",
    extracted_fields={"total": "$1,234.56", "vendor": "Acme Corp", "id": "INV-999"},
)
# score: float (fraction of grounded fields), field_map: dict[str, bool]
print(f"Grounding: {score:.2f}") # 0.67 (2 of 3 grounded)
print(f"Ungrounded: {[k for k, v in field_map.items() if not v]}") # ["id"]

QoS Guard

Composes all checks into a single gate. You build the individual checkers and pass them in:

from pyfly.agentic.validation.qos import (
    QoSGuard, ConfidenceScorer, ConsistencyChecker, GroundingChecker,
)

# Build the individual checkers first.
guard = QoSGuard(
    confidence_scorer=ConfidenceScorer(evaluator_agent),
    consistency_checker=ConsistencyChecker(extractor_agent, num_runs=3),
    grounding_checker=GroundingChecker(),
    min_confidence=0.8,
    min_consistency=0.6,
    min_grounding=0.8,
)

# evaluate() takes the output string plus optional context for each check.
result = await guard.evaluate(
    str(extracted_data),
    prompt="Extract fields from the invoice.",
    source_text=ocr_text,
    extracted_fields={"total": "$1,234.56", "vendor": "Acme Corp"},
)
# result is a QoSResult with: .passed, .confidence, .consistency_score,
# .grounding_score, .details (dict with per-check info)
if result.passed:
    print("Quality check passed")
else:
    print(f"Failed: confidence={result.confidence:.2f}")

Output Reviewer

The OutputReviewer closes the loop between generation and validation. It wraps an agent call with schema parsing + rule validation + automatic retry. When the LLM produces output that fails Pydantic parsing or validation rules, the reviewer automatically retries with a feedback prompt describing exactly what was wrong.

Basic Usage

from pydantic import BaseModel, Field
from pyfly.agentic.validation import OutputReviewer

class InvoiceData(BaseModel):
    vendor: str
    amount: float = Field(ge=0)
    date: str
    invoice_number: str | None = None

reviewer = OutputReviewer(output_type=InvoiceData, max_retries=3)
result = await reviewer.review(
    agent,
    "Extract invoice data from: Acme Corp, $1,234, 2026-01-15",
)
print(result.output) # InvoiceData(vendor="Acme Corp", amount=1234.0, ...)
print(result.attempts) # 1 if first try succeeded, 2+ if retries needed

With Validation Rules

Combine schema parsing with field-level rules:

from pyfly.agentic.validation import OutputReviewer, OutputValidator, EnumRule

validator = OutputValidator({"vendor": [EnumRule("vendor", ["Acme Corp", "Globex"])]})
reviewer = OutputReviewer(
    output_type=InvoiceData,
    validator=validator,
    max_retries=2,
)

With Reasoning Patterns

Attach a reviewer to any reasoning pattern to validate the final output:

from pyfly.agentic.reasoning import ReActPattern
from pyfly.agentic.validation import OutputReviewer

reviewer = OutputReviewer(output_type=InvoiceData, max_retries=2)
pattern = ReActPattern(reviewer=reviewer)
result = await pattern.execute(agent, "Extract invoice data from the document.")
# Output is automatically validated and retried if needed

ReviewResult

The result contains:

• output — The validated output.
• attempts — Total attempts (1 = first try succeeded).
• validation_report — The final ValidationReport if a validator was used.
• retry_history — List of RetryAttempt objects with attempt number, raw output, and error messages.

Rubric Reviewer (LLM-as-judge)

Where OutputReviewer enforces a schema and deterministic rules, RubricReviewer evaluates free-form output against a list of natural-language criteria using a separate grader agent. When criteria are unmet, it sends a revision prompt back to the generator and loops (up to max_iterations), returning a ReviewResult:

from pyfly.agentic.validation import RubricReviewer

reviewer = RubricReviewer(
    rubric=[
        "All five invoice fields are present.",
        "Amounts are numeric, not strings.",
        "The due date is ISO 8601.",
    ],
    grader=evaluator_agent,  # optional; a default grader is created otherwise
    max_iterations=3,
)
result = await reviewer.review(extractor_agent, f"Extract fields from:\n{ocr_text}")

You can also load the rubric from a Markdown bullet list with RubricReviewer.from_rubric_file("rubric.md").

IDP Tie-In: Validating Extracted Invoice Data

For our IDP pipeline, we combine structural validation, QoS checks, and the output reviewer:

from pyfly.agentic.validation.rules import (
    OutputValidator, FieldValidator, RegexRule, FormatRule, RangeRule,
)
from pyfly.agentic.validation.qos import QoSGuard
from pyfly.agentic.validation import OutputReviewer
from pydantic import BaseModel, Field

# Define the expected output schema
class InvoiceExtraction(BaseModel):
    invoice_number: str
    vendor_name: str
    total_amount: float = Field(ge=0)
    due_date: str
    line_items: list[dict]

# Structural validation rules — dict of field_name -> list of rules
invoice_validator = OutputValidator({
    "invoice_number": [RegexRule("invoice_number", r"^INV-\d{4,10}$")],
    "vendor_name": [RegexRule("vendor_name", r".{2,}")],
    "total_amount": [RangeRule("total_amount", min_value=0.01, max_value=10_000_000)],
    "due_date": [FormatRule("due_date", "iso_date")],
})

# Output reviewer with schema + rules + retry
reviewer = OutputReviewer(
    output_type=InvoiceExtraction,
    validator=invoice_validator,
    max_retries=3,
)

# QoS guard — compose individual checkers with thresholds
from pyfly.agentic.validation.qos import (
    QoSGuard, ConfidenceScorer, ConsistencyChecker, GroundingChecker,
)

qos = QoSGuard(
    confidence_scorer=ConfidenceScorer(evaluator_agent),
    consistency_checker=ConsistencyChecker(extractor_agent, num_runs=2),
    grounding_checker=GroundingChecker(),
    min_confidence=0.8,
    min_consistency=0.6,
    min_grounding=0.8,
)

---

Part III — Orchestration & Operations

---

Chapter 10: Pipeline

So far we have individual agents (classify, digitise, extract), tools (OCR, vendor lookup), reasoning patterns (Plan-and-Execute, Reflexion), validation rules, and memory. Each piece works in isolation — but a real IDP system needs to wire them together into a single, reliable flow: classify → digitise → extract → validate → assemble.

The Pipeline module models your processing flow as a graph where nodes are processing steps and edges define data flow. For acyclic graphs without runtime routers, the engine starts a node as soon as its dependencies resolve. Cycles and runtime routers use a frontier scheduler with a recursion limit. The engine also handles configured retries, timeouts, and conditional execution.

Pipeline Execution Architecture

The following diagram shows how the pipeline engine executes a DAG:

flowchart TD
    BUILDER["PipelineBuilder"] --> DAG["Graph and dependencies"]
    DAG --> ENGINE["PipelineEngine"]
    ENGINE --> MODE{"Runtime routers or cycles?"}
    MODE -->|no| READY["Schedule any dependency-ready nodes"]
    READY --> EXECUTE["Run independent nodes concurrently"]
    EXECUTE --> UPDATE["Update results and release dependents"]
    UPDATE --> MORE{"Pending nodes?"}
    MORE -->|yes| READY
    MODE -->|yes| FRONTIER["Follow runtime targets<br/>bounded by recursion_limit"]
    FRONTIER --> STEP["Execute nodes and apply state reducers"]
    STEP --> NEXT{"More runtime targets?"}
    NEXT -->|yes| FRONTIER
    NEXT -->|no| RESULT["PipelineResult<br/>outputs, state, trace and duration"]
    MORE -->|no| RESULT

Core Concepts

A pipeline is a graph where:

• Nodes are processing steps (call an agent, run a reasoning pattern, execute a function).
• Edges define data flow and execution order.
• The engine schedules independent, dependency-ready nodes concurrently, or follows runtime targets for graphs with routers and cycles.

The Pipeline Builder

The fluent PipelineBuilder is the recommended way to construct pipelines:

from pyfly.agentic.pipeline.builder import PipelineBuilder
from pyfly.agentic.pipeline.steps import AgentStep, CallableStep, ReasoningStep

engine = (
    PipelineBuilder("invoice-pipeline")
    .add_node("classify", AgentStep(classifier_agent))
    .add_node("extract", AgentStep(extractor_agent))
    .add_node("validate", CallableStep(validate_fn))
    .chain("classify", "extract", "validate")
    .build()
)

result = await engine.run(inputs="<invoice text>")

Step Executors

The built-in executors (all implementing StepExecutor) cover most scenarios:

• AgentStep — Runs a Agent with the input as prompt.
• ReasoningStep — Runs a reasoning pattern through an agent.
• CallableStep — Wraps any async function (context, inputs) -> output.
• FanOutStep — Deprecated list-splitting helper. Use a state-based router returning list[Send] for dynamic fan-out and reducers to collect results.
• FanInStep — Merges outputs from multiple upstream nodes.
• BranchStep — Deprecated; use a state-based builder's .branch(...) method.
• BatchLLMStep — Runs an agent over a batch of inputs concurrently.
• EmbeddingStep — Embeds text via a BaseEmbedder (see Embeddings & Vector Stores).
• RetrievalStep — Retrieves nearest neighbours from a vector store: RetrievalStep(store, *, embedder=None, top_k=5, input_key="input").

Parallel Execution (Fan-Out / Fan-In)

Dispatch one page to each worker invocation with Send, then collect the outputs with a state reducer. This continues the configured ocr_agent from earlier chapters:

from typing import Annotated
from pydantic import BaseModel, Field
from pyfly.agentic.pipeline import PipelineBuilder, Send, extend


class PageState(BaseModel):
    pages: list[str] = Field(default_factory=list)
    page: str = ""
    results: Annotated[list[str], extend] = Field(default_factory=list)
    document: str = ""


async def prepare_pages(state: PageState) -> dict:
    if not state.pages:
        raise ValueError("At least one page is required")
    return {"pages": state.pages}


def dispatch_pages(state: PageState) -> list[Send]:
    return [Send("read_page", {"page": page}) for page in state.pages]


async def read_page(state: PageState) -> dict:
    result = await ocr_agent.run(state.page)
    return {"results": [str(result.output)]}


async def assemble_pages(state: PageState) -> dict:
    return {"document": "\n\n".join(state.results)}

engine = (
    PipelineBuilder("parallel-ocr", state=PageState)
    .add_node(prepare_pages).add_node(read_page).add_node(assemble_pages)
    .branch(prepare_pages, dispatch_pages)
    .add_edge(read_page, assemble_pages)
    .build()
)
result = await engine.invoke(PageState(pages=["Page 1: Invoice INV-42", "Page 2: Total USD 120"]))
print(result.state.document)

Conditional Execution

Gate nodes with a condition function. If the condition returns False, the node is skipped:

from pyfly.agentic.pipeline.dag import DAGNode

dag.add_node(DAGNode(
    node_id="ocr",
    step=AgentStep(ocr_agent),
    condition=lambda ctx: ctx.metadata.get("needs_ocr", False),
))

Retries and Timeouts

Configure per-node resilience:

engine = (
    PipelineBuilder("robust-pipeline")
    .add_node("extract", AgentStep(extractor_agent), retry_max=3, timeout_seconds=30.0)
    .build()
)

Retries use linear backoff. On exhaustion, the node fails and the pipeline reports success=False.

Pipeline Context

PipelineContext is the shared data bus flowing through the DAG:

from pyfly.agentic.pipeline.context import PipelineContext
from pyfly.agentic.memory import MemoryManager

memory = MemoryManager(working_scope_id="invoice-run-42")

ctx = PipelineContext(
    inputs=document_bytes,
    metadata={"source": "email", "tenant": "acme-corp"},
    memory=memory,
)
result = await engine.run(context=ctx)

It carries: inputs, metadata, correlation_id, and results from completed upstream nodes. When a MemoryManager is attached, AgentStep and ReasoningStep automatically propagate memory to agents and patterns.

Pipeline Result

PipelineResult aggregates all outcomes:

result = await engine.run(context=ctx)

if result.success:
    print(result.final_output)
    print(f"Completed in {result.total_duration_ms:.0f} ms")
else:
    print(f"Failed nodes: {result.failed_nodes}")

# Inspect individual nodes
for node_id, node_result in result.outputs.items():
    print(f" {node_id}: {'ok' if node_result.success else 'FAILED'}")

# Execution trace for observability
for entry in result.execution_trace:
    print(f" [{entry.node_id}] {entry.status} ({entry.duration_ms:.0f} ms)")

Manual DAG Construction

For full control, build the DAG directly:

from pyfly.agentic.pipeline.dag import DAG, DAGNode, DAGEdge
from pyfly.agentic.pipeline.engine import PipelineEngine

dag = DAG("my-pipeline")
dag.add_node(DAGNode(node_id="step_a", step=my_step))
dag.add_node(DAGNode(node_id="step_b", step=other_step))
dag.add_edge(DAGEdge(source="step_a", target="step_b"))

engine = PipelineEngine(dag)
result = await engine.run(inputs="hello")

Embeddings & Vector Stores

EmbeddingStep and RetrievalStep build on two reusable framework modules. The embeddings package ships BaseEmbedder/EmbedderRegistry with 8 provider backends (OpenAI, Azure, Cohere, Google, Mistral, Voyage, Bedrock, Ollama). The vectorstores package ships BaseVectorStore with 6 backends — InMemoryVectorStore, ChromaVectorStore, PineconeVectorStore, QdrantVectorStore, PgVectorVectorStore, and SqliteVecVectorStore — plus a scoping layer (ScopedVectorStore, TenantScopedVectorStore, scope_namespace, parse_scope_namespace) for multi-tenant isolation:

from pyfly.agentic.vectorstores import InMemoryVectorStore
from pyfly.agentic.pipeline import RetrievalStep

store = InMemoryVectorStore()          # async upsert / search / search_text / delete
retrieve = RetrievalStep(store, top_k=5, input_key="input")

The framework ships these as building blocks; it does not bundle a turnkey RAG/corpus agent. BaseEmbedder, EmbedderRegistry, BaseVectorStore, and InMemoryVectorStore are re-exported from the top-level pyfly.agentic package.

State-Based Pipelines, Checkpointing & Audit Logs

Beyond the port-based DAG above, PipelineBuilder has a state-based mode: pass a Pydantic state= model and nodes become async (state) -> dict | None | Pause | Send | list[Send] functions over a typed shared state. Branching is a single .branch(source, router) call, and state reducers (append, extend, merge_dict, replace) control how each node's returned dict is merged into the state.

Two control signals shape the flow: Pause suspends a run (resume later), and Send dispatches dynamic fan-out work. For durability, pass a Checkpointer (FileCheckpointer(root=...) writes CheckpointRecords) so a failed run can resume or start mid-pipeline. An AuditLog records every node transition — choose FileAuditLog, LoggingAuditLog, OtelAuditLog, or QueryableAuditLog (each entry is an AuditEntry):

from pyfly.agentic.pipeline import (
    PipelineBuilder, FileCheckpointer, OtelAuditLog,
)

engine = PipelineBuilder(
    "stateful-idp",
    state=IdpState,                       # a pydantic.BaseModel
    checkpointer=FileCheckpointer(root=".checkpoints"),
    audit_log=OtelAuditLog(),
).build()

IDP Tie-In: Wiring the Complete Pipeline

Here's our IDP pipeline as a DAG with all five phases:

from pyfly.agentic.pipeline.builder import PipelineBuilder
from pyfly.agentic.pipeline.steps import AgentStep, ReasoningStep, CallableStep
from pyfly.agentic.pipeline.context import PipelineContext
from pyfly.agentic.reasoning import PlanAndExecutePattern
from pyfly.agentic.memory import MemoryManager

# Step functions for non-agent nodes
async def validate_step(context, inputs):
    extracted = inputs.get("input", {})
    report = invoice_validator.validate(extracted)
    context.metadata["validation_report"] = report
    return {"extracted": extracted, "valid": report.valid, "errors": report.errors}

async def assemble_step(context, inputs):
    data = inputs.get("input", {})
    return {
        "document_type": context.metadata.get("doc_type", "unknown"),
        "extracted_fields": data.get("extracted", {}),
        "validation_passed": data.get("valid", False),
    }

# Build the DAG
idp_pipeline = (
    PipelineBuilder("idp-pipeline")
    .add_node("classify", AgentStep(classifier_agent))
    .add_node("digitise", AgentStep(ocr_agent), retry_max=2, timeout_seconds=60)
    .add_node(
        "extract",
        ReasoningStep(PlanAndExecutePattern(max_steps=15), extractor_agent),
    )
    .add_node("validate", CallableStep(validate_step))
    .add_node("assemble", CallableStep(assemble_step))
    .chain("classify", "digitise", "extract", "validate", "assemble")
    .build()
)

# Execute
memory = MemoryManager(working_scope_id="idp-session")
ctx = PipelineContext(
    inputs=document_bytes,
    metadata={"source": "email-inbox", "tenant_id": "acme-corp"},
    memory=memory,
)
result = await idp_pipeline.run(context=ctx)

if result.success:
    print(f"Document type: {result.final_output['document_type']}")
    print(f"Extracted fields: {result.final_output['extracted_fields']}")
else:
    print(f"Pipeline failed at: {result.failed_nodes}")

---

Chapter 11: Observability

Your pipeline is running in production, processing thousands of invoices a day. Then latency spikes. Or accuracy drops. Or a customer reports a missing field. Without observability you're flying blind — you have no idea which agent is slow, which tool is failing, or how many tokens you're burning.

The Observability module wraps OpenTelemetry and gives you three primitives out of the box: tracing (distributed spans across agents, tools, and pipeline steps), metrics (counters and histograms for tokens, latency, cost, errors, and reasoning depth), and events (structured logs for significant occurrences). When observability is enabled, the framework instruments agent runs automatically — you get spans for free.

Tracing

AgentTracer wraps the OpenTelemetry Tracer and adds GenAI-specific attributes. It exposes purpose-built context managers — agent_span(agent_name, *, model=..., **attrs), tool_span(...), reasoning_span(...) — plus a generic custom_span(name, **attrs), an event(name, **attrs) annotation helper, and a static set_error(span, error):

from pyfly.agentic.observability import AgentTracer

tracer = AgentTracer(service_name="idp-service")

with tracer.agent_span("classifier", model="openai:gpt-4o") as span:
    result = await classifier_agent.run("Classify this document")
    span.set_attribute("tokens.total", 150)

# Or a generic span with arbitrary attributes:
with tracer.custom_span("agent.run", phase="classify") as span:
    result = await classifier_agent.run("Classify this document")
    span.set_attribute("output.type", type(result.output).__name__)

The @traced Decorator

Automatically create a span around any function:

from pyfly.agentic.observability import traced

@traced(name="classify_document")
async def classify_document(text: str) -> dict:
    return await classifier_agent.run(text)

Metrics

AgentMetrics records GenAI-specific OpenTelemetry instruments via purpose-built methods (each takes keyword args like agent=, model=, operation=, pattern= — not a generic labels= dict):

from pyfly.agentic.observability import AgentMetrics

metrics = AgentMetrics(service_name="idp-service")

metrics.record_tokens(150, agent="classifier", model="openai:gpt-4o")
metrics.record_prompt_tokens(100, agent="classifier", model="openai:gpt-4o")
metrics.record_completion_tokens(50, agent="classifier", model="openai:gpt-4o")
metrics.record_latency(142.5, operation="classify", agent="classifier")
metrics.record_cost(0.0021, agent="classifier", model="openai:gpt-4o")
metrics.record_error(operation="classify", agent="classifier", error_type="Timeout")
metrics.record_reasoning_depth(4, pattern="react")

The @metered Decorator

@metered's first parameter is operation (it records latency, and an error on exception). Use a positional string or operation=:

from pyfly.agentic.observability import metered

@metered("extraction")
async def extract_fields(text: str) -> dict:
    return await extractor_agent.run(text)

Events

AgentEvents emits structured events (logged as JSON-serialisable dicts) via typed methods — there is no generic emit():

from pyfly.agentic.observability import AgentEvents

events = AgentEvents()
events.agent_started("classifier", model="openai:gpt-4o")
events.agent_completed("classifier", tokens=150, latency_ms=250)
events.tool_executed("vendor_lookup", success=True, latency_ms=12)
events.reasoning_step("react", step=1, step_type="thought")
events.agent_error("classifier", error="timeout")

Exporter Configuration

The framework emits spans and metrics purely through the OpenTelemetry API; it does not configure the OTel SDK or any exporters itself. The host application owns OTel SDK and exporter setup — wire up your TracerProvider, MeterProvider, and the exporters (OTLP collector, console, etc.) however your deployment requires. Supply application-owned providers to AgenticRuntime when explicit ownership is required. Runtime services do not install a competing global SDK. See native telemetry ownership.

Configuration via environment variables:

export PYFLY_AGENTIC_OBSERVABILITY_ENABLED=true
export PYFLY_AGENTIC_LOG_LEVEL=DEBUG

Usage Tracking & Cost Estimation

The framework automatically tracks token usage and estimates cost for every agent run, reasoning step, and pipeline execution. UsageTracker accumulates UsageRecord objects with input/output tokens, cost, latency, and model details.

from pyfly.agentic.observability import default_usage_tracker

# After running agents, inspect accumulated usage
summary = default_usage_tracker.get_summary()
print(f"Total tokens: {summary.total_tokens}")
print(f"Total cost: ${summary.total_cost_usd:.4f}")
print(f"Requests: {summary.total_requests}")

# Filter by agent or pipeline correlation ID
agent_summary = default_usage_tracker.get_summary_for_agent("extractor")
pipeline_summary = default_usage_tracker.get_summary_for_correlation("run-123")

Cost is computed by a resolver chain (observability/cost_resolvers.py). The default chain DEFAULT_RESOLVERS tries provider_reported_cost (uses cost the provider already reported) then genai_prices_cost (prices via the bundled genai-prices data). The entry point is resolve_cost:

from pyfly.agentic.observability import resolve_cost, CostContext

cost = resolve_cost(
    CostContext(model="openai:gpt-4o", input_tokens=1000, output_tokens=500)
)
# Returns the USD cost, or None when no resolver can price the model.

When config.cost_strict=True, an unpriceable model raises UnknownModelCostError instead of returning None. UsageTracker applies this chain automatically for every recorded run.

Budget Enforcement

Set a hard budget limit; a warning is logged when costs exceed it:

export PYFLY_AGENTIC_BUDGET_LIMIT_USD=10.00

Pipeline results include aggregated usage automatically:

result = await engine.run(context=ctx)
if result.usage:
    print(f"Pipeline cost: ${result.usage.total_cost_usd:.4f}")

Automatic Integration

With observability_enabled, agent middleware creates spans and emits start events. With cost_tracking_enabled, UsageTracker records completed model requests and its configured sinks emit metrics and completion events. The host configures the OpenTelemetry SDK and exporters. See the Observability guide for these separate controls.

IDP Tie-In: Instrumenting the Pipeline

from pyfly.agentic.observability import AgentTracer, traced

tracer = AgentTracer(service_name="idp-service")

@traced(name="idp.process_document")
async def process_document(document_bytes: bytes) -> dict:
    with tracer.custom_span("idp.run") as span:
        ctx = PipelineContext(inputs=document_bytes)
        result = await idp_pipeline.run(context=ctx)
        span.set_attribute("idp.success", result.success)
        # Per-agent token, latency, and cost metrics are recorded automatically
        # when observability is enabled; inspect them via default_usage_tracker.
        return result.final_output

---

Chapter 12: Explainability

In regulated industries — finance, healthcare, legal — "the model said so" is not an acceptable answer. Auditors, compliance officers, and customers need to see why the agent classified a document as an invoice, why it chose one vendor name over another, and what alternatives it considered. The Explainability module provides four building blocks: a trace recorder that captures every decision, an explanation generator that turns raw records into natural-language narratives, an append-only audit trail, and a report builder that compiles everything into markdown or JSON.

Trace Recorder

TraceRecorder captures decision records during execution. The method is record(category, *, agent="", detail=None, input_summary="", output_summary=""), and recorded items are exposed via the .records property:

from pyfly.agentic.explainability import TraceRecorder

recorder = TraceRecorder()
recorder.record(
    "reasoning_step",
    agent="extractor",
    detail={"chosen": "regex_match", "alternatives": ["llm_extraction", "template_match"]},
    input_summary="raw OCR text",
    output_summary="invoice_number=INV-2026-001",
)

Each DecisionRecord has these fields: timestamp, category, agent, detail, input_summary, and output_summary.

Explanation Generator

Transforms raw decision records into a natural-language narrative:

from pyfly.agentic.explainability import ExplanationGenerator

generator = ExplanationGenerator()
explanation = generator.generate(recorder.records)
print(explanation)
# Multi-line "Decision Explanation" narrative walking each record chronologically.

Audit Trail

An append-only log. Each entry captures an actor, action, resource, and outcome:

from pyfly.agentic.explainability import AuditTrail

trail = AuditTrail()
trail.append("extractor", "field_extraction", resource="invoice_number", outcome="success")

# Inspect or export
print(len(trail))            # number of entries
print(trail.entries)         # list[AuditEntry]
print(trail.export_json())   # JSON string

Report Builder

Compile records into a structured ExplainabilityReport, then render it:

from pyfly.agentic.explainability import ReportBuilder

builder = ReportBuilder(title="Invoice Extraction Report")
report = builder.build(recorder.records)   # -> ExplainabilityReport

# Render via static helpers
markdown = ReportBuilder.to_markdown(report)   # for documentation/review
json_data = ReportBuilder.to_json(report)      # for programmatic consumption

IDP Tie-In: Audit Trail for Invoice Extraction

In our IDP pipeline, we record why each field was extracted the way it was:

from pyfly.agentic.explainability import (
    TraceRecorder, ExplanationGenerator, AuditTrail, ReportBuilder,
)

recorder = TraceRecorder()

# During extraction, record decisions
recorder.record(
    "field_extraction",
    agent="field_extractor",
    detail={"chosen": "INV-2026-001", "alternatives": ["INV-2026-01", "2026-001"]},
    output_summary="Matched the INV-NNNN pattern with highest confidence.",
)
recorder.record(
    "field_extraction",
    agent="field_extractor",
    detail={"chosen": "1234.56", "alternatives": ["1,234.56", "$1234.56"]},
    output_summary="Normalised currency format to numeric value.",
)

# Generate explanation
generator = ExplanationGenerator()
explanation = generator.generate(recorder.records)

# Audit trail — append (actor, action) entries
trail = AuditTrail()
trail.append("field_extractor", "extract_invoice_number", resource="INV-2026-001")
trail.append("field_extractor", "extract_amount", resource="1234.56")
print(trail.export_json())

# Build and render the report
report = ReportBuilder(title="Invoice Extraction Report").build(recorder.records)
print(ReportBuilder.to_markdown(report))

---

Part IV — Experimentation

---

Chapter 13: Experiments

You've built a working extraction agent — but is GPT-4o the right model? Would Claude give better results? What about lowering the temperature from 0.5 to 0.1? Answering these questions by hand (run each variant, eyeball the output, repeat) doesn't scale. The Experiments module gives you a structured way to define variants, run them against the same inputs, collect metrics (latency, token usage, output length), and compare results — all in a few lines of code.

Defining an Experiment

An Experiment holds variants plus a dataset of test inputs. Each Variant carries name, model, temperature, and a parameters dict:

from pyfly.agentic.experiments import Experiment, Variant

experiment = Experiment(
    name="extraction_model_comparison",
    hypothesis="Claude Haiku produces more accurate extractions than GPT-4o on invoice extraction.",
    variants=[
        Variant(name="gpt4o", model="openai:gpt-4o"),
        Variant(name="claude", model="anthropic:claude-haiku-4-5"),
    ],
    dataset=[
        "Extract fields from: Invoice #INV-001, Acme Corp, $500, 2026-01-15",
        "Extract fields from: Invoice #INV-002, Globex, $1,200, 2026-02-28",
    ],
)

Running an Experiment

ExperimentRunner.run(experiment, agent_factory, *, context=None) runs every variant against the experiment's dataset. The second positional argument is an agent_factory callable (variant) -> agent that builds an agent configured for that variant. It returns a list[VariantResult]:

from pyfly.agentic.experiments import ExperimentRunner
from pyfly.agentic.agents import Agent
from pyfly.agentic.models import ModelOptions

def make_agent(variant):
    return Agent(
        name=f"extractor-{variant.name}",
        model=variant.model,
        model_options=ModelOptions(temperature=variant.temperature),
        instructions="You are an invoice data extraction specialist.",
    )

runner = ExperimentRunner()
results = await runner.run(experiment, make_agent)

Tracking Results

ExperimentTracker stores VariantResults in memory with optional JSON persistence (pass storage_path=). ExperimentRunner already records each result into its tracker; you can also record manually and export:

from pyfly.agentic.experiments import ExperimentTracker

tracker = ExperimentTracker(storage_path="./experiment_results.json")
for result in results:
    tracker.record(result)

# Query and export
subset = tracker.get_by_experiment("extraction_model_comparison")
print(tracker.export_json())

Comparing Variants

from pyfly.agentic.experiments import VariantComparator

comparator = VariantComparator()
metrics = comparator.compare(results)
print(comparator.summary(results))
# "gpt4o: avg_latency=1.2s, avg_output_len=145 | claude: avg_latency=0.9s, ..."

IDP Tie-In: Finding the Best Extraction Model

This experiment deliberately compares sampling on models that support temperature. For reasoning models, leave temperature unset or select a supported non-reasoning mode; ModelOptions validates the combination.

experiment = Experiment(
    name="idp_extraction_ab_test",
    hypothesis="Lower temperature improves IDP invoice extraction accuracy.",
    variants=[
        Variant(name="gpt4o", model="openai:gpt-4o", temperature=0.1),
        Variant(name="gpt4o_warm", model="openai:gpt-4o", temperature=0.5),
        Variant(name="claude", model="anthropic:claude-haiku-4-5"),
    ],
    dataset=test_invoices,
)

def make_agent(variant):
    return Agent(
        name=f"extractor-{variant.name}",
        model=variant.model,
        model_options=ModelOptions(temperature=variant.temperature),
        instructions="You are an invoice data extraction specialist.",
    )

results = await ExperimentRunner().run(experiment, make_agent)
print(VariantComparator().summary(results))

---

Chapter 14: Lab

The Experiments module helps you compare variants — but sometimes you just need a sandbox to poke at an agent interactively, run quick benchmarks, compare side-by-side outputs, or evaluate accuracy against a labelled dataset. The Lab module is that sandbox. It provides interactive sessions (REPL-like), benchmarking helpers, model comparison tables, evaluation datasets, and pluggable scorers — everything you need to iterate on agent quality before going to production.

Interactive Sessions

Test an agent conversationally:

from pyfly.agentic.lab import LabSession

session = LabSession(name="extraction-dev", agent=extractor_agent)
response = await session.interact("Extract fields from: Invoice #INV-001...")
print(response)

# Review history
for entry in session.history:
    print(f"[{entry.timestamp}] {entry.prompt} → {entry.response}")

Benchmarking

Measure agent performance across a set of prompts:

from pyfly.agentic.lab import Benchmark

bench = Benchmark(inputs=[
    "Extract from: Invoice #INV-001, Acme Corp, $500",
    "Extract from: Invoice #INV-002, Globex, $1,200",
    "Extract from: Invoice #INV-003, Initech, $3,456.78",
])
result = await bench.run(extractor_agent)
print(f"Avg latency: {result.avg_latency_ms:.1f} ms")
print(f"P95 latency: {result.p95_latency_ms:.1f} ms")

Side-by-Side Comparison

Compare multiple agents on the same prompts:

from pyfly.agentic.lab import ModelComparison

comparison = ModelComparison(prompts=[
    "Extract from: Invoice #INV-001, Acme Corp, $500",
])
entries = await comparison.compare({
    "gpt4o": extractor_gpt4o,
    "claude": extractor_claude,
})
for entry in entries:
    for agent_name, response in entry.responses.items():
        print(f" {agent_name}: {response}")

Datasets

Manage test inputs and expected outputs:

from pyfly.agentic.lab import EvalDataset, EvalCase

dataset = EvalDataset(cases=[
    EvalCase(
        input="Extract from: Invoice #INV-001, Acme Corp, $500, 2026-01-15",
        expected_output='{"invoice_number": "INV-001", "vendor_name": "Acme Corp", "total_amount": 500.0}',
    ),
])

# Or load from a JSON file
dataset = EvalDataset.from_json("test_data/invoices.json")

Evaluators

Run an agent against a dataset with a pluggable scorer:

from pyfly.agentic.lab import EvalOrchestrator

# Custom scorer for extraction accuracy
def extraction_scorer(expected: str, actual: str) -> float:
    import json
    try:
        exp = json.loads(expected)
        act = json.loads(actual)
        matching = sum(1 for k in exp if exp.get(k) == act.get(k))
        return matching / len(exp) if exp else 0.0
    except (json.JSONDecodeError, AttributeError):
        return 0.0

orchestrator = EvalOrchestrator(scorer=extraction_scorer)
report = await orchestrator.evaluate(extractor_agent, dataset)
print(f"Accuracy: {report.avg_score:.1%} across {report.total_cases} cases")

IDP Tie-In: Benchmarking Extraction Accuracy

# Load a dataset of real invoices with expected outputs
dataset = EvalDataset.from_json("test_data/invoices.json")

# Benchmark with custom extraction scorer
orchestrator = EvalOrchestrator(scorer=extraction_scorer)
report = await orchestrator.evaluate(extractor_agent, dataset)
print(f"Extraction accuracy: {report.avg_score:.1%}")

---

Part V — Advanced

---

Chapter 15: Template Agents

By now you've written several agents from scratch — classifier, extractor, OCR. Each time you had to think about the system prompt, output type, and registration. But many agent patterns are universal: summarise text, classify into categories, extract structured data, hold a conversation, route to sub-agents. The framework ships five template agents as factory functions that encode best practices for each pattern. You provide the domain-specific bits (categories, schemas, personality), and the factory handles prompt engineering, output typing, and registry registration.

Summarizer

from pyfly.agentic.agents.templates import create_summarizer_agent

agent = create_summarizer_agent(
    max_length="short", # concise | short | medium | detailed
    style="technical", # professional | casual | technical | academic
    output_format="bullets", # paragraph | bullets | numbered
)
result = await agent.run("Long invoice description text here...")

Classifier

Returns a structured ClassificationResult with category, confidence, and reasoning:

from pyfly.agentic.agents.templates import create_classifier_agent

agent = create_classifier_agent(
    categories=["invoice", "receipt", "contract", "form"],
    descriptions={
        "invoice": "Bills requesting payment for goods or services",
        "receipt": "Proof of payment or purchase",
        "contract": "Legal agreements between parties",
        "form": "Fillable forms and applications",
    },
)
result = await agent.run("Invoice from Acme Corp, Amount Due: $1,234.56")
# result.output → ClassificationResult(category="invoice", confidence=0.95, ...)

Extractor

Extracts structured data into a user-provided Pydantic model:

from pydantic import BaseModel
from pyfly.agentic.agents.templates import create_extractor_agent

class Invoice(BaseModel):
    vendor: str
    amount: float
    date: str
    invoice_number: str | None = None

agent = create_extractor_agent(
    Invoice,
    field_descriptions={
        "vendor": "The company that issued the invoice",
        "amount": "Total monetary amount",
    },
)
result = await agent.run("Invoice from Acme Corp, $1,234.56, 2026-01-15")
# result.output → Invoice(vendor="Acme Corp", amount=1234.56, ...)

Conversational Agent

Memory-enabled multi-turn assistant:

from pyfly.agentic.agents.templates import create_conversational_agent
from pyfly.agentic.memory import MemoryManager

memory = MemoryManager(max_conversation_tokens=32_000)
agent = create_conversational_agent(
    personality="friendly and concise",
    domain="accounts payable",
    memory=memory,
)

cid = memory.new_conversation()
result = await agent.run("I need help with invoice INV-001.", conversation_id=cid)
result = await agent.run("What's the payment status?", conversation_id=cid)

Router Agent

Intent-based routing to child agents:

from pyfly.agentic.agents.templates import create_router_agent

agent = create_router_agent(
    agent_map={
        "invoice_processor": "Handles invoice extraction and validation",
        "receipt_processor": "Handles receipt scanning and categorisation",
        "support": "General questions about the IDP system",
    },
    fallback_agent="support",
)
result = await agent.run("Process this invoice from Acme Corp")
# result.output → RoutingDecision(target_agent="invoice_processor", confidence=0.92, ...)

Common Parameters

All template factories accept:

• name — Agent name for the registry (sensible defaults provided).
• model — LLM model string; falls back to framework default.
• extra_instructions — Text appended to the system prompt.
• tools — Additional tools to attach.
• auto_register — Set to False to skip registry registration.

IDP Tie-In: Using Templates Instead of Hand-Rolled Agents

Replace our manual agents with templates for cleaner code:

from pyfly.agentic.agents.templates import (
    create_classifier_agent,
    create_extractor_agent,
)

# Phase 1: Use the built-in classifier template
classifier_agent = create_classifier_agent(
    categories=["invoice", "receipt", "contract", "form"],
    name="document_classifier",
)

# Phase 3: Use the built-in extractor template
extractor_agent = create_extractor_agent(
    InvoiceExtraction,
    field_descriptions={
        "invoice_number": "Format: INV-NNNN",
        "vendor_name": "Company that issued the invoice",
        "total_amount": "Total amount due",
        "due_date": "Payment due date in ISO format",
    },
    name="field_extractor",
)

---

Chapter 16: Multi-Agent Delegation

Not every document is an invoice. Your IDP system might receive receipts, contracts, and forms — each requiring a specialised agent with different prompts, tools, and validation rules. Instead of building one mega-agent that tries to do everything, you build specialised agents and let a delegation router decide which one handles each request. The router picks an agent based on a strategy (round-robin for load balancing, capability-based for expertise matching), delegates the work, and optionally forks memory so the sub-agent gets its own working-memory scope.

Delegation Architecture

flowchart TD
    REQUEST["Prompt"] --> ROUTER["DelegationRouter"]
    ROUTER --> STRATEGY["RoundRobin, Capability, ContentBased,<br/>CostAware, or composed strategy"]
    STRATEGY --> DECISION["Ranked RoutingDecision"]
    DECISION --> EXECUTE["execute selects the first candidate"]
    MEMORY["Optional MemoryManager"] --> FORK["fork with delegation scope"]
    EXECUTE -->|router has memory| FORK
    FORK --> AGENT["Selected agent"]
    EXECUTE -->|no router memory| AGENT
    AGENT --> RUN["agent.run"]

Delegation Router

from pyfly.agentic.agents.delegation import DelegationRouter, RoundRobinStrategy

router = DelegationRouter(
    agents=[agent_a, agent_b, agent_c],
    strategy=RoundRobinStrategy(),
)
result = await router.route("Process this document.")

Round Robin Strategy

Distributes requests evenly across a pool of agents. Useful for load balancing when all agents have equivalent capabilities:

from pyfly.agentic.agents.delegation import RoundRobinStrategy

strategy = RoundRobinStrategy()
router = DelegationRouter([agent_1, agent_2, agent_3], strategy)
# Request 1 → agent_1, Request 2 → agent_2, Request 3 → agent_3, Request 4 → agent_1...

Capability-Based Strategy

Selects the first agent whose tags include a required capability:

from pyfly.agentic.agents.delegation import CapabilityStrategy

strategy = CapabilityStrategy(required_tag="invoice_extraction")
router = DelegationRouter([invoice_agent, receipt_agent], strategy)
result = await router.route("Extract invoice data.")
# → Routed to invoice_agent (which has the "invoice_extraction" tag)

Other Strategies

The framework ships 7 delegation strategies, all importable from pyfly.agentic.agents.delegation (and re-exported from pyfly.agentic.agents):

• RoundRobinStrategy — even load balancing.
• CapabilityStrategy — match by required tag.
• ContentBasedStrategy — route by keywords in the request content.
• CostAwareStrategy — prefer the lowest-cost capable agent.
• ChainStrategy — run a fixed chain of agents in order.
• FallbackStrategy — try agents in order until one succeeds.
• WeightedStrategy — weighted random selection.

Memory with Delegation

When a MemoryManager is attached, delegated agents receive a forked memory scope:

from pyfly.agentic.memory import MemoryManager

memory = MemoryManager(working_scope_id="main")
router = DelegationRouter([agent_a, agent_b], RoundRobinStrategy(), memory=memory)
result = await router.route("Process this.")
# The delegated agent gets its own working memory scope

IDP Tie-In: Routing Documents to Specialised Agents

from pyfly.agentic.agents.delegation import DelegationRouter, CapabilityStrategy

# Specialised agents for different document types
invoice_agent = create_extractor_agent(InvoiceSchema, name="invoice_extractor")
receipt_agent = create_extractor_agent(ReceiptSchema, name="receipt_extractor")
contract_agent = create_extractor_agent(ContractSchema, name="contract_extractor")

# Route based on capability
strategy = CapabilityStrategy(required_tag="invoice")
router = DelegationRouter(
    agents=[invoice_agent, receipt_agent, contract_agent],
    strategy=strategy,
)

---

Chapter 17: Plugin System

As your application grows, you'll want to share agents, tools, and reasoning patterns across projects — or let third-party teams contribute their own. The Plugin module uses Python's standard entry-point mechanism so that any installed package can register components that are discovered and loaded automatically at startup. No manual imports, no central configuration file — just install the package and go.

Discovering Plugins

from pyfly.agentic.plugin import PluginDiscovery

result = PluginDiscovery.discover_all()
print(f"Loaded {len(result.successful)} plugins, {len(result.failed)} failed")

Creating a Plugin

In your package's pyproject.toml, declare entry points under the framework's groups:

[project.entry-points."pyfly.agentic.agents"]
my_agent = "my_package.agents:MySpecialAgent"

[project.entry-points."pyfly.agentic.tools"]
my_tool = "my_package.tools:MyCustomTool"

[project.entry-points."pyfly.agentic.reasoning_patterns"]
my_pattern = "my_package.reasoning:MyCustomPattern"

The three entry-point groups are:

• pyfly.agentic.agents — Custom agents.
• pyfly.agentic.tools — Custom tools.
• pyfly.agentic.reasoning_patterns — Custom reasoning patterns.

On discovery, the framework loads each entry point and registers it in the appropriate registry.

Configuration

# Enable auto-discovery on startup (default: True)
# PYFLY_AGENTIC_PLUGIN_AUTO_DISCOVER=true

IDP Tie-In: Packaging IDP as a Plugin

You can package the entire IDP pipeline as a plugin that self-registers when installed:

# In idp_plugin/pyproject.toml
[project.entry-points."pyfly.agentic.agents"]
document_classifier = "idp_plugin.agents:classifier_agent"
field_extractor = "idp_plugin.agents:extractor_agent"
ocr_agent = "idp_plugin.agents:ocr_agent"

After uv add idp-plugin, calling PluginDiscovery.discover_all() registers all three agents automatically.

---

Chapter 18: Putting It All Together

You've learned every module in PyFly Agentic, each in isolation. Now it's time to see how they all fit together in a single, production-grade application. The diagram below shows the full system architecture — every layer, every connection:

Full System Architecture

flowchart TD
    HOST["Host application<br/>credentials, serving and shutdown"] --> FLOWS["Pipelines, workflows and delegation"]
    HOST --> AGENT["Agent"]
    FLOWS --> REASON["Optional reasoning and review"]
    FLOWS --> AGENT
    REASON --> AGENT
    TEMPLATES["Template factories"] --> AGENT
    AGENT --> MODELS["ModelSpec and ModelOptions<br/>provider and API selection"]
    MODELS --> PAI["Pydantic AI<br/>model requests and tool dispatch"]
    PAI --> TOOLS["PyFly Agentic tools<br/>guards, listeners and approval"]
    AGENT --> MEMORY["Conversation history and working facts"]
    AGENT --> MW["Configured middleware"]
    MW --> TELEMETRY["Host-configured observability"]
    FLOWS --> CONTENT["Content, embeddings and retrieval"]
    DEV["Lab and experiments"] --> AGENT
    HOST --> AUDIT["Explicit audit and explainability recording"]
    HOST --> CONFIG["Configuration, registries and plugin discovery"]

Let's assemble the complete IDP pipeline using everything we've learned. This is the full, production-ready implementation.

Project Structure

idp-service/
├── pyproject.toml
├── .env
├── prompts/
│ ├── classification.jinja2
│ └── extraction.jinja2
├── src/
│ └── idp_service/
│ ├── __init__.py
│ ├── agents.py # Agent definitions
│ ├── tools.py # Tool definitions
│ ├── pipeline.py # Pipeline wiring
│ ├── validation.py # Validation rules
│ └── main.py # In-process entry point
└── tests/
    └── test_pipeline.py

Configuration (.env)

PYFLY_AGENTIC_DEFAULT_MODEL=openai-responses:gpt-6-luna
PYFLY_AGENTIC_MAX_RETRIES=3
PYFLY_AGENTIC_OBSERVABILITY_ENABLED=true
PYFLY_AGENTIC_MEMORY_BACKEND=file
PYFLY_AGENTIC_MEMORY_FILE_DIR=.firefly_memory
PYFLY_AGENTIC_DEFAULT_CHUNK_SIZE=4000
PYFLY_AGENTIC_VALIDATION_ENABLED=true

Agents (agents.py)

from pydantic import BaseModel, Field
from pyfly.agentic.agents.templates import (
    create_classifier_agent,
    create_extractor_agent,
    create_summarizer_agent,
)
from pyfly.agentic.agents import Agent

# Output schema for extraction
class InvoiceData(BaseModel):
    invoice_number: str
    vendor_name: str
    total_amount: float = Field(ge=0)
    due_date: str
    line_items: list[dict] = []

# Phase 1: Document classifier
classifier_agent = create_classifier_agent(
    categories=["invoice", "receipt", "contract", "form", "other"],
    name="document_classifier",
)

# Phase 2: OCR/digitisation agent
ocr_agent = Agent(
    name="ocr_agent",
    instructions="Extract all text from this document. Preserve layout and structure.",
)

# Phase 2.5: Summariser for compression
summary_agent = create_summarizer_agent(
    name="doc_summariser",
    max_length="medium",
    style="technical",
)

# Phase 3: Field extractor — with tools from tools.py attached
from .tools import extraction_kit

extractor_agent = create_extractor_agent(
    InvoiceData,
    field_descriptions={
        "invoice_number": "Format: INV-NNNN or similar",
        "vendor_name": "Company that issued the invoice",
        "total_amount": "Total monetary amount due",
        "due_date": "Payment due date (ISO 8601)",
        "line_items": "List of line items with description, quantity, unit_price",
    },
    name="field_extractor",
    tools=[extraction_kit], # PyFly Agentic adapts the toolkit
)

Tools (tools.py)

Tools live in their own module. Each @agent_tool call creates an AgentTool, registers it in the active runtime’s ToolRegistry, and returns the instance. The ToolKit bundles them for agent injection via tools=[extraction_kit]:

from pyfly.agentic.tools import agent_tool, guarded, retryable, ToolKit
from pyfly.agentic.tools.guards import RateLimitGuard, ValidationGuard
from pyfly.agentic.tools.builtins import CalculatorTool

# OCR tool — rate-limited because the upstream API is metered
@retryable(max_retries=2, backoff=1.0)
@guarded(RateLimitGuard(max_calls=100, period_seconds=60))
@agent_tool(name="ocr_extract", description="Extract text from a document image via OCR")
async def ocr_extract(image_data: str) -> str:
    """POST a base64 image to the host's OCR endpoint and return its text field."""
    import os
    import httpx

    async with httpx.AsyncClient(timeout=30.0) as client:
        response = await client.post(
            os.environ["OCR_ENDPOINT"],
            headers={"Authorization": f"Bearer {os.environ['OCR_API_KEY']}"},
            json={"image_base64": image_data},
        )
        response.raise_for_status()
        text = response.json()["text"]
        if not isinstance(text, str) or not text.strip():
            raise ValueError("OCR endpoint returned no text")
        return text

# Vendor lookup — validates that vendor_name is present
@guarded(ValidationGuard(required_keys=["vendor_name"]))
@agent_tool(name="vendor_lookup", description="Look up vendor details from the ERP system")
async def vendor_lookup(vendor_name: str) -> str:
    vendors = {
        "Acme Corp": '{"id": "V-001", "tax_id": "US-12345", "payment_terms": "NET30"}',
        "Globex": '{"id": "V-002", "tax_id": "US-67890", "payment_terms": "NET60"}',
    }
    return vendors.get(vendor_name, '{"error": "Vendor not found"}')

# Calculator — a built-in tool, no decorator needed
calculator = CalculatorTool()

# Bundle the tools the extractor agent needs.
# Pass the toolkit directly to Agent(tools=[extraction_kit]).
extraction_kit = ToolKit(
    "extraction-tools",
    [ocr_extract, vendor_lookup, calculator],
    description="Tools available during invoice field extraction",
)

Validation (validation.py)

from pyfly.agentic.validation.rules import (
    OutputValidator, RegexRule, FormatRule, RangeRule,
)
from pyfly.agentic.validation.qos import (
    QoSGuard, ConfidenceScorer, ConsistencyChecker, GroundingChecker,
)
from pyfly.agentic.validation import OutputReviewer
from .agents import InvoiceData, extractor_agent

invoice_validator = OutputValidator({
    "invoice_number": [RegexRule("invoice_number", r"^INV-\d{4,10}$")],
    "vendor_name": [RegexRule("vendor_name", r".{2,}")],
    "total_amount": [RangeRule("total_amount", min_value=0.01, max_value=10_000_000)],
    "due_date": [FormatRule("due_date", "iso_date")],
})

reviewer = OutputReviewer(
    output_type=InvoiceData,
    validator=invoice_validator,
    max_retries=3,
)

qos_guard = QoSGuard(
    confidence_scorer=ConfidenceScorer(extractor_agent),
    consistency_checker=ConsistencyChecker(extractor_agent, num_runs=2),
    grounding_checker=GroundingChecker(),
    min_confidence=0.8,
    min_consistency=0.6,
    min_grounding=0.8,
)

Pipeline (pipeline.py)

from pyfly.agentic.pipeline.builder import PipelineBuilder
from pyfly.agentic.pipeline.steps import AgentStep, ReasoningStep, CallableStep
from pyfly.agentic.pipeline.context import PipelineContext
from pyfly.agentic.reasoning import PlanAndExecutePattern, ReflexionPattern
from pyfly.agentic.content.chunking import TextChunker, BatchProcessor
from pyfly.agentic.content.compression import ContextCompressor, MapReduceStrategy
from pyfly.agentic.memory import MemoryManager
from pyfly.agentic.observability import traced
from .agents import classifier_agent, ocr_agent, summary_agent, extractor_agent
from .validation import invoice_validator, reviewer

async def validate_step(context, inputs):
    extracted = inputs.get("input", {})
    report = invoice_validator.validate(extracted)
    context.metadata["validation_report"] = report
    if not report.valid:
        # Self-correct with Reflexion
        reflexion = ReflexionPattern(max_steps=2)
        corrected = await reflexion.execute(
            extractor_agent,
            f"Fix these errors: {report.errors}. Original data: {extracted}",
        )
        return {"extracted": corrected.output, "valid": True, "errors": []}
    return {"extracted": extracted, "valid": True, "errors": []}

async def assemble_step(context, inputs):
    data = inputs.get("input", {})
    return {
        "document_type": context.metadata.get("doc_type", "unknown"),
        "language": context.metadata.get("language", "unknown"),
        "extracted_fields": data.get("extracted", {}),
        "validation_passed": data.get("valid", False),
    }

idp_pipeline = (
    PipelineBuilder("idp-pipeline")
    .add_node("classify", AgentStep(classifier_agent))
    .add_node("digitise", AgentStep(ocr_agent), retry_max=2, timeout_seconds=60)
    .add_node(
        "extract",
        ReasoningStep(PlanAndExecutePattern(max_steps=15, allow_replan=True), extractor_agent),
    )
    .add_node("validate", CallableStep(validate_step))
    .add_node("assemble", CallableStep(assemble_step))
    .chain("classify", "digitise", "extract", "validate", "assemble")
    .build()
)

@traced(name="idp.process")
async def process_document(document_bytes: bytes, metadata: dict | None = None) -> dict:
    memory = MemoryManager(working_scope_id="idp-session")
    ctx = PipelineContext(
        inputs=document_bytes,
        metadata=metadata or {},
        memory=memory,
    )
    result = await idp_pipeline.run(context=ctx)
    return result.final_output if result.success else {"error": result.failed_nodes}

Entry Point (main.py)

PyFly Agentic is a pure in-process library: it serves no port and consumes no broker. Your host service owns serving and calls process_document directly. The host also owns OTel SDK and exporter configuration; the framework emits spans and metrics through the OpenTelemetry API, so they flow through whatever providers the host has set up:

import asyncio

from pyfly.agentic.auto_configuration import bind_agentic_config
from pyfly.agentic.runtime import AgenticRuntime
from pyfly.core import Config


async def main(document_bytes: bytes, filename: str) -> dict:
    async with AgenticRuntime(config=bind_agentic_config(Config())) as runtime:
        with runtime.activate():
            # Import registrations inside this application's runtime.
            from .pipeline import process_document

            return await process_document(
                document_bytes,
                metadata={"filename": filename, "source": "host-service"},
            )


if __name__ == "__main__":
    with open("invoice.pdf", "rb") as fh:
        print(asyncio.run(main(fh.read(), "invoice.pdf")))

To expose this over HTTP or wire it to a message broker, embed process_document in your host service's framework of choice — the agent library stays in-process.

Production Checklist

Before deploying to production, verify:

• Configuration — All PYFLY_AGENTIC_* environment variables are set.
• Model access — API keys for your LLM provider are configured.
• Observability — observability_enabled is on, and your host service has configured the OTel SDK/exporters so framework spans and metrics flow to your backend.
• Memory persistence — memory_backend is set to "file" (or a custom backend) for durability.
• Validation — OutputValidator rules match your business requirements.
• QoS thresholds — min_confidence and num_runs (ConsistencyChecker) are tuned for your use case.
• Retry limits — Pipeline nodes have appropriate retry_max and timeout_seconds.
• Experiments — You've A/B tested your prompt and model variants.
• Audit trail — Explainability is enabled for regulated workloads.

---

Next Steps

Congratulations — you now know every module in PyFly Agentic. Here are some paths to explore further:

• Dive deeper — Each chapter links to a detailed module guide in docs/.
• Read the source — The framework is fully typed and well-documented in code.
• Run the tests — uv run pytest runs 1,300+ tests across ~128 files that exercise every module.
• Build your own — Extend AbstractReasoningPattern, implement MemoryStore, create custom tools, or write a plugin.

Module Reference

• Architecture
• Agents
• Template Agents
• Tools
• Prompts
• Reasoning Patterns
• Content Processing
• Memory
• Validation
• Pipeline
• Observability
• Explainability
• Experiments
• Lab
• Use Case: IDP