Advanced Agent Patterns¶

Import: from selectools.patterns import PlanAndExecuteAgent

Stability: beta

from selectools import Agent, AgentConfig, tool
from selectools.providers.stubs import LocalProvider
from selectools.patterns import ReflectiveAgent

@tool(description="No-op tool")
def noop(x: str) -> str:
    return x

provider = LocalProvider()

# Actor drafts content, critic evaluates and requests revisions
actor = Agent(
    tools=[noop],
    provider=provider,
    config=AgentConfig(max_iterations=1),
    system_prompt="You are a technical writer.",
)
critic = Agent(
    tools=[noop],
    provider=provider,
    config=AgentConfig(max_iterations=1),
    system_prompt="You are an editor. Say 'approved' when satisfied.",
)

agent = ReflectiveAgent(actor=actor, critic=critic, max_reflections=2)
result = agent.run("Explain what a vector database is in two sentences")

print(f"Final draft: {result.final_draft[:200]}")
print(f"Approved: {result.approved}")
print(f"Rounds: {result.total_rounds}")

Added in: v0.19.1 Module: src/selectools/patterns/ Import: from selectools.patterns import ... or from selectools import ...

Four production-ready multi-agent coordination patterns built on the v0.18.0 orchestration primitives. Each pattern wires up the AgentGraph topology for you — no graph-wiring required.

Pattern Overview¶

All patterns support .run() (sync) and .arun() (async).

PlanAndExecuteAgent¶

from selectools import Agent, OpenAIProvider
from selectools.patterns import PlanAndExecuteAgent

provider = OpenAIProvider()
planner = Agent(provider=provider, system_prompt="You are a planning agent.")
researcher = Agent(provider=provider, system_prompt="You are a research agent.")
writer = Agent(provider=provider, system_prompt="You are a writing agent.")

agent = PlanAndExecuteAgent(
    planner=planner,
    executors={"researcher": researcher, "writer": writer},
)
result = agent.run("Research LLM safety and write a 500-word blog post")
print(result.content)

How it works¶

1. The planner agent is called once to produce a JSON execution plan:

   [
     {"executor": "researcher", "task": "Find 3 key LLM safety concerns"},
     {"executor": "writer",     "task": "Write a blog post using the research"}
   ]
   

2. Each step's executor is called in sequence. Each step receives the accumulated output of previous steps as context.
3. The final executor's output becomes result.content.

Replanning on failure¶

agent = PlanAndExecuteAgent(
    planner=planner,
    executors={"researcher": researcher, "writer": writer},
    replanner=True,          # re-call planner if a step fails
    max_replan_attempts=2,   # limit replanning cycles
)

If a step raises an exception and replanner=True, the planner is re-called with the failure context to revise the remaining steps.

Result type¶

PlanAndExecuteAgent.run() returns a GraphResult:

Constructor reference¶

Planning-as-config (beta)¶

Any Agent can opt into the plan → (approve) → execute → synthesize flow without wiring PlanAndExecuteAgent manually — set AgentConfig(planning=PlanningConfig(...)):

from selectools import Agent, AgentConfig, PlanningConfig

config = AgentConfig(
    planning=PlanningConfig(
        enabled=True,
        provider=None,      # planner provider override (defaults to the agent's)
        model=None,         # planner model override (defaults to the agent's)
        auto_approve=True,  # False requires plan_approval_handler
        reasoning=True,     # surface the plan via result.reasoning
    )
)
agent = Agent(tools, provider=provider, config=config)
result = agent.run("Research X, then write a summary, and finally review it.")
print(result.reasoning)                      # the generated plan
print(result.trace.metadata["planning"])     # plan + steps_executed

Internally the agent clones itself into a planner and a single executor and delegates to PlanAndExecuteAgent, then runs one final synthesis call. The result is a normal AgentResult with usage aggregated across the planner, every step, and the synthesis call.

Complexity gate. Simple single-step inputs skip planning entirely. A cheap local heuristic scores the prompt (sequence connectives like "then" / "finally", numbered or bulleted lists, 3+ sentences, length over ~120 estimated tokens); planning triggers when the score reaches min_complexity (default 2). Set always=True to plan every input. Bare punctuation is deliberately not a signal: semicolons appear in pasted code far more often than as clause separators, so a prompt like "Refactor x = 1; y = 2 in my code" does not trigger planning.

Plan approval. With auto_approve=False, plan_approval_handler is required. It receives the structured plan (List[PlanStep]) and returns True (approve), False (reject — the agent falls back to a standard run with a one-time warning), or an edited List[PlanStep]. The handler must be sync: an async def handler raises TypeError when the plan is ready for approval (its coroutine could never equal True or a list, so it would otherwise look like a silent rejection).

Interplay. Streaming runs (astream(), or run() with a stream_handler) skip planning with a one-time UserWarning per agent. Structured output works: response_format is applied to the final synthesis call. enabled=False (or leaving planning unset) is a zero-overhead no-op.

Budgets. max_total_tokens / max_cost_usd bind across the whole planned flow: the planner and executor clones are seeded with the parent's running totals, and every sub-run's usage is merged back into the parent on every exit path (success, plan rejection, or a mid-flow exception), so a planned run cannot spend multiples of the configured cap. When a cap trips inside a step, the step's graceful "budget exceeded" message becomes that step's output and the plan continues — subsequent steps and the final synthesis call then trip the same cap before reaching the provider, so the final answer degrades to the budget message rather than raising.

Cancellation. The agent's cancellation_token is shared with the pattern and checked between steps, so remaining steps are skipped after a mid-plan cancellation — but the final synthesis call still runs, and its graceful "cancelled" message becomes the final answer (no exception is raised, matching the non-planned cancellation contract).

Step exceptions. PlanAndExecuteAgent swallows per-step exceptions: the step is marked failed and execution continues (the adapter configures no replanner). A provider error that would propagate out of a normal run() is therefore absorbed when planning engages — the failed step just contributes no output to the synthesis prompt. The planner call and the synthesis call are not protected; exceptions there propagate to the caller (with all sub-run usage already merged into agent.usage).

Known gaps. result.trace is the synthesis run's trace (the plan is attached under trace.metadata["planning"]); per-step traces are not merged because PlanAndExecuteAgent does not aggregate traces. When the agent has no memory, a planned run does not update the parent's in-process _history (the clones run memory-less; the turn is persisted to memory only when one is attached).

See examples/109_planning_as_config.py for a fully offline demo.

ReflectiveAgent¶

from selectools.patterns import ReflectiveAgent

actor = Agent(provider=provider, system_prompt="You are a technical writer.")
critic = Agent(provider=provider, system_prompt="You are an editor. Give feedback. Say 'approved' when satisfied.")

agent = ReflectiveAgent(actor=actor, critic=critic, max_reflections=3)
result = agent.run("Write a concise explanation of transformer attention")

print(result.final_draft)   # final approved draft
print(result.approved)      # True if critic said "approved"
print(result.total_rounds)  # number of actor-critic cycles

How it works¶

Each round: 1. Actor receives the task (round 0) or the task + previous draft + critique (round N). 2. Critic evaluates the draft and provides feedback. 3. If the critic's response contains stop_condition (default: "approved"), the loop ends.

The loop also ends when max_reflections is reached regardless of approval.

Per-round records¶

for rnd in result.rounds:
    print(f"Round {rnd.round_number}: approved={rnd.approved}")
    print(f"  Draft:   {rnd.draft[:100]}...")
    print(f"  Critique: {rnd.critique[:100]}...")

Result type — ReflectiveResult¶

Constructor reference¶

DebateAgent¶

from selectools.patterns import DebateAgent

optimist = Agent(provider=provider, system_prompt="You argue in favour of the proposal.")
skeptic  = Agent(provider=provider, system_prompt="You argue against the proposal.")
judge    = Agent(provider=provider, system_prompt="You synthesize debate arguments objectively.")

agent = DebateAgent(
    agents={"optimist": optimist, "skeptic": skeptic},
    judge=judge,
    max_rounds=2,
)
result = agent.run("Should we rewrite our monolith as microservices?")

print(result.conclusion)      # judge's synthesis
print(result.total_rounds)    # 2

for rnd in result.rounds:
    for position, argument in rnd.arguments.items():
        print(f"[{position}] {argument[:200]}")

How it works¶

1. Each debate round: every agent is called in order. Rounds 2+ include the prior round's transcript so agents can respond to each other.
2. After all rounds, the judge receives the full transcript and synthesizes a conclusion.

Result type — DebateResult¶

DebateRound:

Constructor reference¶

TeamLeadAgent¶

from selectools.patterns import TeamLeadAgent

lead       = Agent(provider=provider, system_prompt="You are a project lead.")
researcher = Agent(provider=provider, system_prompt="You find and summarize information.")
writer     = Agent(provider=provider, system_prompt="You write clear, concise reports.")

# Sequential — subtasks run one after another, each sees prior results
agent = TeamLeadAgent(lead=lead, team={"researcher": researcher, "writer": writer},
                      delegation_strategy="sequential")

# Parallel — subtasks run simultaneously via AgentGraph fan-out
agent = TeamLeadAgent(lead=lead, team={"researcher": researcher, "writer": writer},
                      delegation_strategy="parallel")

# Dynamic (default) — lead reviews after each result and may reassign
agent = TeamLeadAgent(lead=lead, team={"researcher": researcher, "writer": writer},
                      delegation_strategy="dynamic", max_reassignments=2)

result = agent.run("Produce a competitive analysis of the top 3 LLM frameworks")
print(result.content)
print(result.total_assignments)  # total task executions including reassignments

Delegation strategies¶

How the lead delegates¶

The lead agent generates a JSON subtask plan:

[
  {"assignee": "researcher", "task": "Find the top 3 LLM frameworks"},
  {"assignee": "writer",     "task": "Write the competitive analysis"}
]

In dynamic mode, after all pending subtasks complete, the lead reviews the work log and decides whether to synthesize or reassign:

{
  "complete": false,
  "reassignments": [{"assignee": "researcher", "task": "Also compare pricing models"}],
  "synthesis": ""
}

Result type — TeamLeadResult¶

Subtask:

Constructor reference¶

Async Usage¶

All patterns support await agent.arun(prompt):

import asyncio

async def main():
    result = await agent.arun("Write a technical blog post about vector databases")
    print(result.content)

asyncio.run(main())

Choosing a Pattern¶

Need typed step-by-step execution with named specialists?
  → PlanAndExecuteAgent

Need iterative quality improvement with self-critique?
  → ReflectiveAgent

Need to explore a decision from multiple viewpoints?
  → DebateAgent

Need to decompose a large task across a team?
  → TeamLeadAgent (parallel for speed, dynamic for adaptability)

See Also¶

• Orchestration — AgentGraph, routing, parallel execution, HITL
• Supervisor — SupervisorAgent with 4 built-in strategies
• Pipeline — Composable pipelines with @step and | operator
• Examples: 70_plan_and_execute.py, 71_reflective_agent.py, 72_debate_agent.py, 73_team_lead_agent.py

Related Examples¶