Advanced Agent Patterns¶

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}")

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)

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
)

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

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]}...")

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]}")

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

import asyncio

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

asyncio.run(main())