pyagent-patterns — Composition, Guardrails & Recovery¶
Cross-cutting building blocks: compose patterns together, guard their inputs and outputs, recover from failures, and get pattern recommendations.
Composite¶
pyagent_patterns.composite.CompositePattern
¶
Bases: Pattern
Chain multiple patterns with escalation on quality failure.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
patterns
|
list[Pattern]
|
Ordered list of patterns to try. |
required |
quality_check
|
QualityCheckFn
|
Function that evaluates whether a pattern's result is acceptable. If it returns False, the next pattern is tried. |
always_pass
|
combine_results
|
bool
|
If True, passes previous output as context to next pattern. |
True
|
Recovery¶
pyagent_patterns.recovery.BoundedExecution
dataclass
¶
Wrap a pattern with resource limits.
Three-level recovery: 1. Retry with same pattern 2. Fallback to a cheaper/simpler pattern 3. Graceful degradation (return partial result)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pattern
|
Pattern
|
The primary pattern to execute. |
required |
fallback
|
Pattern | None
|
Optional simpler pattern to use on failure. |
None
|
max_retries
|
int
|
Maximum retry attempts before escalating. |
2
|
timeout_seconds
|
float
|
Maximum wall-clock time for the entire execution. |
300.0
|
max_tokens
|
int
|
Maximum total tokens before stopping. |
100000
|
run(task, context=None)
async
¶
Execute with bounded resources and three-level recovery.
pyagent_patterns.recovery.CircuitBreaker
¶
Prevent cascading failures in multi-agent systems.
When a pattern fails repeatedly, the circuit opens and rejects requests immediately. After a reset timeout, it enters half-open state and allows one test request through.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
failure_threshold
|
int
|
Number of consecutive failures before opening. |
3
|
reset_timeout_seconds
|
float
|
Seconds before transitioning from open to half-open. |
60.0
|
fallback_result
|
str
|
Result to return when circuit is open. |
'[Circuit Open] Service temporarily unavailable.'
|
execute(pattern, task, context=None)
async
¶
Execute a pattern through the circuit breaker.
Guardrails¶
pyagent_patterns.guardrails.GuardrailChain
¶
Chain multiple guardrails together. All must pass.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
guardrails
|
list[Guardrail]
|
List of guardrails to apply in order. |
required |
pyagent_patterns.guardrails.LengthGuard
¶
Bases: Guardrail
Reject messages exceeding a maximum length.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
max_chars
|
int
|
Maximum allowed characters. |
10000
|
truncate
|
bool
|
If True, truncate instead of rejecting. |
False
|
pyagent_patterns.guardrails.PIIGuard
¶
Bases: Guardrail
Detect and optionally redact personally identifiable information.
Detects: email addresses, phone numbers, SSNs, credit card numbers.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
redact
|
bool
|
If True, redact PII and pass. If False, reject on detection. |
True
|
pyagent_patterns.guardrails.ContentGuard
¶
Bases: Guardrail
Block content matching configurable deny patterns.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
deny_patterns
|
list[str] | None
|
List of regex patterns that should be blocked. |
None
|
deny_words
|
list[str] | None
|
List of exact words/phrases to block. |
None
|
Advisor¶
pyagent_patterns.advisor.PatternAdvisor
¶
Recommend patterns based on task description and constraints.
Usage
advisor = PatternAdvisor() rec = advisor.recommend("Write and review code", Constraints(quality=Quality.HIGH)) print(rec.pattern, rec.reason)
recommend(task, constraints=None)
¶
Recommend the best pattern for the given task and constraints.