Skip to content

Getting Started

Installation

# Core patterns only (zero LLM deps)
pip install pyagent-patterns

# With routing
pip install pyagent-router

# With compression
pip install pyagent-compress

# With tracing (requires opentelemetry)
pip install pyagent-trace

# Everything
pip install pyagent-all

Core Concepts

1. Message

The fundamental unit of communication between agents.

from pyagent_patterns.base import Message

user_msg = Message.user("What is the capital of France?")
system_msg = Message.system("You are a geography expert.")
assistant_msg = Message.assistant("The capital of France is Paris.")

2. Agent

An agent wraps an LLM callable with a name and system prompt.

from pyagent_patterns.base import Agent, MockLLM

# For testing: MockLLM returns canned responses
llm = MockLLM(responses=["Paris"])
agent = Agent("geography_expert", llm, system_prompt="You are a geography expert.")

# For production: any async callable (str) → str
import asyncio
result = asyncio.run(agent.run([Message.user("Capital of France?")]))
print(result.content)  # "Paris"

3. Pattern

A Pattern orchestrates one or more agents to solve a task.

from pyagent_patterns.orchestration import Pipeline

pipeline = Pipeline(stages=[
    Agent("extract", llm),
    Agent("summarize", llm),
])

result = asyncio.run(pipeline.run("Process this document"))
print(result.output)           # Final output string
print(result.messages)         # All intermediate messages
print(result.metadata)         # Pattern-specific metadata
print(result.duration_seconds) # Execution time

4. Result

Every pattern.run() returns a Result with: - output: Final string output - messages: All intermediate Message objects - metadata: Pattern-specific data (rounds, routes, scores, etc.) - duration_seconds: Wall-clock execution time - token_estimate: Estimated total tokens used

Your First Pattern

import asyncio
from pyagent_patterns.base import Agent, MockLLM
from pyagent_patterns.resolution import SelfReflection

llm = MockLLM(responses=[
    "def fibonacci(n): return n if n <= 1 else fibonacci(n-1) + fibonacci(n-2)",
    "Critique: naive recursion, O(2^n). Needs memoization.",
    "from functools import lru_cache\n@lru_cache\ndef fibonacci(n): return n if n <= 1 else fibonacci(n-1) + fibonacci(n-2)",
    "APPROVED — O(n) with memoization",
])

pattern = SelfReflection(agent=Agent("coder", llm), max_rounds=3)
result = asyncio.run(pattern.run("Write an efficient Fibonacci function"))
print(result.output)
print(f"Improved in {result.metadata['rounds']} rounds")

Pattern Advisor

Not sure which pattern to use? Let the advisor decide:

from pyagent_patterns.advisor import PatternAdvisor, Constraints, Quality

advisor = PatternAdvisor()
rec = advisor.recommend("Write and review a legal contract", Constraints(quality=Quality.HIGH))
print(f"Use: {rec.pattern} — {rec.reason}")
print(f"Estimated calls: {rec.estimated_calls}, Cost: {rec.estimated_cost_range}")

Adding Hooks (Optional)

Agents support opt-in hooks for tracing, context memory, compression, and cost tracking — zero overhead when not wired:

from pyagent_trace.events import TraceEventBus
from pyagent_trace import CostTracker
from pyagent_context import ContextLedger
from pyagent_compress import MessageCompressor

bus = TraceEventBus()
bus.subscribe(lambda e: print(f"[{e.event_type}] {e.agent_name}"))

agent = (
    Agent("analyst", llm, system_prompt="Analyse data.")
    .set_trace_bus(bus)                              # emit trace events
    .set_context(ContextLedger())                    # read/write context
    .set_compressor(MessageCompressor(0.5))          # compress output
    .set_cost_tracker(CostTracker(event_bus=bus))    # track costs
)

result = asyncio.run(agent.run("What are the key trends?"))
# Console prints: [agent_start] analyst → [agent_end] analyst

All hooks are None by default — existing code works identically without them.

Next Steps