pyagent-patterns — Base Classes

Core abstractions shared by every pattern.

pyagent_patterns.base.Message dataclass

A single message in an agent conversation.

Attributes:

Name	Type	Description
role	Role	The sender role (system, user, assistant, tool).
content	str	The text content of the message.
name	str | None	Optional agent name for multi-agent conversations.
metadata	dict[str, Any]	Arbitrary key-value metadata attached to the message.

pyagent_patterns.base.Agent dataclass

An LLM-backed agent with a name, system prompt, and callable.

Parameters:

Name	Type	Description	Default
name	str	Human-readable agent name.	required
llm	LLMCallable	The LLM callable to use for this agent.	required
system_prompt	str	Optional system prompt prepended to every call.	''
description	str	Description of the agent's purpose (for routing/selection).	''

Optional hooks (set via setter methods — no constructor change): _trace_bus: Emit agent_start/agent_end trace events. _context_ledger: Read context before LLM call; write output after. _compressor: Compress agent output before returning. _cost_tracker: Record cost per LLM call.

run(messages) async

Send messages to the LLM and return an assistant message.

When hooks are wired the execution order is: 1. Emit agent_start trace event 2. Prepend context from ledger as messages 3. Call LLM 4. Record cost 5. Compress output 6. Write result to context ledger 7. Emit agent_end trace event with timing/tokens

set_compressor(compressor)

Attach a MessageCompressor — compresses output before returning.

set_context(ledger)

Attach a ContextLedger — reads context before LLM, writes output after.

set_cost_tracker(tracker)

Attach a CostTracker — records cost after each LLM call.

set_trace_bus(bus)

Attach a TraceEventBus — emits agent_start/agent_end events on run().

pyagent_patterns.base.Pattern

Bases: ABC

Abstract base class for all multi-agent patterns.

Subclasses must implement _execute. The run method handles timing, context creation, and metadata collection.

Optional hooks

_trace_bus: Emit pattern_start/pattern_end trace events.

pattern_type abstractmethod property

Return the pattern type name (e.g., 'supervisor', 'debate').

run(task, context=None) async

Execute the pattern on the given task.

Parameters:

Name	Type	Description	Default
task	str	The user task or prompt.	required
context	Context | None	Optional existing context. Created automatically if None.	None

Returns:

Type	Description
Result	Result with output, messages, metadata, timing, and cost estimates.

set_trace_bus(bus)

Attach a TraceEventBus — emits pattern_start/pattern_end on run().

stream(task, context=None) async

Stream partial results as they become available.

Default implementation runs the full pattern and yields the result. Subclasses can override for true streaming.

pyagent_patterns.base.Context dataclass

Shared execution context for a pattern run.

Attributes:

Name	Type	Description
task	str	The original user task/prompt.
messages	list[Message]	Accumulated message history.
metadata	dict[str, Any]	Arbitrary shared state across agents.
parent_id	str | None	ID of the parent context (for nested patterns).

child(task=None)

Create a child context for nested pattern execution.

pyagent_patterns.base.Result dataclass

Outcome of a pattern execution.

Attributes:

Name	Type	Description
output	str	The final output text.
messages	list[Message]	All messages generated during execution.
metadata	dict[str, Any]	Pattern-specific metadata (rounds, consensus, votes, etc.).
duration_seconds	float	Wall-clock execution time.
token_estimate	int	Rough estimate of total tokens consumed.
cost_estimate	float	Rough estimate of total cost in USD.