Agent Orchestration

Agent Orchestration

Comprehensive patterns for building and coordinating AI agents -- from single-agent reasoning loops to multi-agent systems and framework selection. Each category has individual rule files in rules/ loaded on-demand.

CC native `/workflows` (2.1.154): Claude Code now ships dynamic workflows — ask Claude to create a workflow and it orchestrates tens-to-hundreds of agents in the background; view runs with /workflows. This is complementary to the patterns here: use CC /workflows for large-scale, fire-and-forget background fan-out (you check back later); use the bounded foreground Agent Teams / Task-tool patterns below when ≤8 agents must coordinate within a single skill invocation via shared memory (handoff files, mesh messaging). Different scale, not a replacement.

>

Ask only when genuinely blocked (CC 2.1.154): CC now reserves the multiple-choice question prompt for decisions it genuinely cannot make itself, rather than asking when it already has enough context to proceed. When orchestrating agents, don't gate progress on an AskUserQuestion the lead can resolve from available context — reserve prompts for true branch points (irreversible actions, missing requirements). This complements ork's voice-friendly decision guidance.

Quick Reference

Total: 10 rules across 4 categories

Quick Start

# ReAct agent loop
async def react_loop(question: str, tools: dict, max_steps: int = 10) -> str:
    history = REACT_PROMPT.format(tools=list(tools.keys()), question=question)
    for step in range(max_steps):
        response = await llm.chat([{"role": "user", "content": history}])
        if "Final Answer:" in response.content:
            return response.content.split("Final Answer:")[-1].strip()
        if "Action:" in response.content:
            action = parse_action(response.content)
            result = await tools[action.name](*action.args)
            history += f"\nObservation: {result}\n"
    return "Max steps reached without answer"

# Supervisor with fan-out/fan-in
async def multi_agent_analysis(content: str) -> dict:
    agents = [("security", security_agent), ("perf", perf_agent)]
    tasks = [agent(content) for _, agent in agents]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return await synthesize_findings(results)

Agent Loops

Patterns for autonomous LLM reasoning: ReAct (Reasoning + Acting), Plan-and-Execute with replanning, self-correction loops, and sliding-window memory management.

Key decisions: Max steps 5-15, temperature 0.3-0.7, memory window 10-20 messages.

Multi-Agent Coordination

Fan-out/fan-in parallelism, supervisor routing with dependency ordering, conflict resolution (confidence-based or LLM arbitration), result synthesis, and CC Agent Teams (mesh topology for peer messaging in CC 2.1.33+).

Key decisions: 3-8 specialists, parallelize independent agents, use Task tool (star) for simple work, Agent Teams (mesh) for cross-cutting concerns.

Alternative Frameworks

CrewAI hierarchical crews with Flows (1.8+), OpenAI Agents SDK handoffs and guardrails (0.12+), Microsoft Agent Framework (AutoGen + SK merger), GPT-5.2-Codex for long-horizon coding, and AG2 for open-source flexibility.

Key decisions: Match framework to team expertise + use case. LangGraph for state machines, CrewAI for role-based teams, OpenAI SDK for handoff workflows, MS Agent for enterprise compliance.

Multi-Scenario

Orchestrate a single skill across 3 parallel scenarios (simple/medium/complex) with progressive difficulty scaling (1x/3x/8x), milestone synchronization, and cross-scenario result aggregation.

Key decisions: Free-running with checkpoints, always 3 scenarios, 1x/3x/8x exponential scaling, 30s/90s/300s time budgets.

Key Decisions

Common Mistakes

• No step limit in agent loops (infinite loops)
• No memory management (context overflow)
• No error isolation in multi-agent (one failure crashes all)
• Note (CC 2.1.161): parallel tool calls now fail independently — a failed Bash no longer cancels siblings in the same batch. This caveat still applies at the agent-orchestration level, not to tool batches; claude agents rows now show done/total for fanned-out work.
• Note (CC 2.1.157): claude agents honors the agent field in settings.json for dispatched sessions; --agent <name> overrides it — pin the agent type explicitly when dispatching.
• Missing synthesis step (raw agent outputs not useful)
• Mixing frameworks in one project (complexity explosion)
• Using Agent Teams for simple sequential work (use Task tool)
• Sequential instead of parallel scenarios (defeats purpose)

Related Skills

• ork:langgraph - LangGraph workflow patterns (supervisor, routing, state)
• function-calling - Tool definitions and execution
• ork:task-dependency-patterns - Task management with Agent Teams workflow

Capability Details

react-loop

Keywords: react, reason, act, observe, loop, agent Solves:

• Implement ReAct pattern
• Create reasoning loops
• Build iterative agents

plan-execute

Keywords: plan, execute, replan, multi-step, autonomous Solves:

• Create plan then execute steps
• Implement replanning on failure
• Build goal-oriented agents

supervisor-coordination

Keywords: supervisor, route, coordinate, fan-out, fan-in, parallel Solves:

• Route tasks to specialized agents
• Run agents in parallel
• Aggregate multi-agent results

agent-debate

Keywords: debate, conflict, resolution, arbitration, consensus Solves:

• Resolve agent disagreements
• Implement LLM arbitration
• Handle conflicting outputs

result-synthesis

Keywords: synthesize, combine, aggregate, merge, summary Solves:

• Combine outputs from multiple agents
• Create executive summaries
• Score confidence across findings

crewai-patterns

Keywords: crewai, crew, hierarchical, delegation, role-based, flows Solves:

• Build role-based agent teams
• Implement hierarchical coordination
• Use Flows for event-driven orchestration

autogen-patterns

Keywords: autogen, microsoft, agent framework, teams, enterprise, a2a Solves:

• Build enterprise agent systems
• Use AutoGen/SK merged framework
• Implement A2A protocol

framework-selection

Keywords: choose, compare, framework, decision, which, crewai, autogen, openai Solves:

• Select appropriate framework
• Compare framework capabilities
• Match framework to requirements

scenario-orchestrator

Keywords: scenario, parallel, fan-out, difficulty, progressive, demo Solves:

• Run skill across multiple difficulty levels
• Implement parallel scenario execution
• Aggregate cross-scenario results

scenario-routing

Keywords: route, synchronize, milestone, checkpoint, scaling Solves:

• Route tasks by difficulty level
• Synchronize at milestones
• Scale inputs progressively

Framework Selection Checklist

Choose the right multi-agent framework.

Requirements Analysis

• [ ] Use case clearly defined
• [ ] Complexity level assessed (single vs multi-agent)
• [ ] State management needs identified
• [ ] Human-in-the-loop requirements defined
• [ ] Observability needs documented

Framework Evaluation

LangGraph

• [ ] Need complex stateful workflows
• [ ] Require persistence and checkpoints
• [ ] Want streaming support
• [ ] Need human-in-the-loop
• [ ] Already using LangChain ecosystem

CrewAI

• [ ] Role-based collaboration pattern
• [ ] Hierarchical team structure
• [ ] Agent delegation needed
• [ ] Quick prototyping required
• [ ] Built-in memory preferred

OpenAI Agents SDK

• [ ] OpenAI-native ecosystem
• [ ] Handoff pattern fits use case
• [ ] Need built-in guardrails
• [ ] Want OpenAI tracing
• [ ] Simpler agent definition preferred

Microsoft Agent Framework

• [ ] Enterprise compliance requirements
• [ ] Using Azure ecosystem
• [ ] Need A2A protocol support
• [ ] Want AutoGen+SK merger features
• [ ] Long-term Microsoft support preferred

AG2 (Community AutoGen)

• [ ] Open-source flexibility priority
• [ ] Community-driven development OK
• [ ] AutoGen familiarity exists
• [ ] Custom modifications needed

Technical Considerations

• [ ] Team expertise with framework
• [ ] Framework maturity level acceptable
• [ ] Community support adequate
• [ ] Documentation quality sufficient
• [ ] Production readiness validated

Integration Assessment

• [ ] Observability tool compatibility (Langfuse, etc.)
• [ ] LLM provider compatibility
• [ ] Existing codebase integration
• [ ] Testing framework support
• [ ] CI/CD pipeline compatibility

Risk Mitigation

• [ ] Fallback strategy defined
• [ ] Framework lock-in assessed
• [ ] Migration path understood
• [ ] Version update strategy
• [ ] Community health evaluated

Decision Documentation

• [ ] Framework choice documented
• [ ] Rationale recorded
• [ ] Alternatives considered listed
• [ ] Trade-offs acknowledged
• [ ] Review date scheduled

Multi-Agent Orchestration Checklist

Architecture

• [ ] Define agent responsibilities
• [ ] Plan communication patterns
• [ ] Set coordination strategy
• [ ] Design failure handling

Agent Design

• [ ] Single responsibility per agent
• [ ] Clear input/output contracts
• [ ] Independent operation
• [ ] Stateless when possible

Communication

• [ ] Message format definition
• [ ] Async message passing
• [ ] Result aggregation
• [ ] Error propagation

Coordination

• [ ] Central orchestrator
• [ ] Task queue management
• [ ] Priority handling
• [ ] Deadlock prevention

Monitoring

• [ ] Agent health checks
• [ ] Task completion tracking
• [ ] Performance metrics
• [ ] Error rates

Architectural Patterns for Multi-Scenario Orchestration

Deep patterns and design decisions for production multi-scenario demos.

Pattern 1: Three-Tier Synchronization

Tier 1: Free-Running (Baseline)

Each scenario runs independently, no blocking.

Time →
─────────────────────────────────────────────────┐
Simple   ███████████████ Complete at 1.2s        │
         └─────────────────────────────────────┘ │
                                                  │
Medium      ██████████████████████░░░░ In progress at 3.5s
            └──────────────────────────────────┘ │
                                                  │
Complex      ██████████░░░░░░░░░░░░░░░░ In progress at 25.7s
             └─────────────────────────────────────┘ │
─────────────────────────────────────────────────┘

Advantages:

• Realistic—shows natural skill behavior
• Tolerates slowness in one scenario
• Lower synchronization overhead

Implementation:

# Each scenario runs its own event loop
# No waiting between scenarios
# Checkpoints are independent

Tier 2: Milestone Synchronization

Scenarios pause at checkpoints (30%, 50%, 70%, 90%) to allow others to catch up.

Time →
─────────────────────────────────────────────────┐
Simple   ███ PAUSE ███ PAUSE ███ Complete        │
         └───┬────────┬────────┬────────────────┘│
                │      │      │                  │
Medium      ██ PAUSE ██ PAUSE ██████░░░░ In-prog │
            └────┬────────┬──────────────────────┘│
                 │      │                        │
Complex      █ PAUSE █ PAUSE ██░░░░░░░░░░░░░░░ │
             └──┬────────┬──────────────────────────┘│
─────────────────────────────────────────────────┘

Advantages:

• Synchronized checkpoints for state capture
• Better for demos (shows progression together)
• Easier to explain ("all at 30%")

Implementation:

async def synchronize_at_milestone(milestone_pct, timeout_seconds=60):
    while time.time() - start < timeout_seconds:
        if all_scenarios_at_milestone:
            return True
        await asyncio.sleep(0.5)

    # Timeout: proceed anyway (don't block forever)
    return False

Tier 3: Lock-Step (Strict Synchronization)

All scenarios advance together, slowest determines pace.

Time →
─────────────────────────────────────────────────┐
Step 1:  Simple   ███ | Medium   ███ | Complex █  │
         └────────────────────────────────────────┘│
Step 2:  Simple   ███ | Medium   ███ | Complex █  │
         └────────────────────────────────────────┘│
Step 3:  All complete together                    │
─────────────────────────────────────────────────┘

Advantages:

• Perfect synchronization for demos
• Easy to explain ("all scenarios complete together")

Disadvantages:

• Complex scenario blocks others (1-2 min delays)
• Unrealistic performance representation

Recommendation: Use Tier 1 (Free-Running) for production, Tier 2 (Milestone) for interactive demos.

---

Pattern 2: Input Scaling Strategies

Strategy A: Linear Scaling (Additive)

Simple:  100 items
Medium:  100 + 200 = 300 items (+200%)
Complex: 300 + 500 = 800 items (+267%)

Time complexity: O(n)
Expected medium time ≈ 3x simple
Expected complex time ≈ 8x simple

Best for: I/O-bound skills (API calls, database queries)

Strategy B: Exponential Scaling (Multiplicative)

Simple:  100 items
Medium:  100 × 3 = 300 items (3x)
Complex: 100 × 8 = 800 items (8x)

Time complexity: O(n) or O(n log n)
Expected medium time ≈ 3x simple (if linear)
Expected complex time ≈ 8x simple (if linear)

Best for: Batch processing, LLM calls

Strategy C: Quadratic Scaling

Simple:  100 items
Medium:  300 items (3x)
Complex: 800 items (8x)

But if algorithm is O(n²):
Expected medium time ≈ 9x simple
Expected complex time ≈ 64x simple

Detection:

# Calculate actual time complexity
simple_time = 1.2  # seconds
medium_time = 3.5
complex_time = 25.7

simple_size = 100
medium_size = 300
complex_size = 800

# Time per item
simple_tpi = simple_time / simple_size        # 0.012 s/item
medium_tpi = medium_time / medium_size        # 0.012 s/item
complex_tpi = complex_time / complex_size     # 0.032 s/item

# Ratio indicates scaling behavior
ratio = complex_tpi / simple_tpi  # 2.67 → O(n log n) or worse

Strategy D: Adaptive Scaling

Choose scaling based on skill characteristics:

SKILL_SCALING_PROFILES = {
    "performance-testing": {
        "scaling": "linear",
        "simple": 10,
        "medium": 30,
        "complex": 80
    },
    "security-scanning": {
        "scaling": "sublinear",  # Gets faster with caching
        "simple": 20,
        "medium": 100,
        "complex": 500
    },
    "data-transformation": {
        "scaling": "quadratic",  # O(n²) worst case
        "simple": 100,
        "medium": 200,  # Limit increase
        "complex": 300
    }
}

---

Pattern 3: Quality Metrics Framework

Metric Category 1: Functional Metrics

What the skill is designed to measure:

{
    "performance-testing": {
        "latency_p95_ms": {"target": "<500ms", "weight": 0.5},
        "error_rate": {"target": "<1%", "weight": 0.5},
    },
    "security-scanning": {
        "vulnerabilities_found": {"target": ">0", "weight": 0.3},
        "coverage_pct": {"target": "100%", "weight": 0.7},
    }
}

Metric Category 2: Comparative Metrics

How scenarios compare:

{
    "quality_scaling": {
        "formula": "complex_quality / simple_quality",
        "expected": 1.0,  # Expect no degradation
        "acceptable": ">0.8"
    },
    "time_efficiency": {
        "formula": "simple_tpi / complex_tpi",
        "expected": 1.0,  # Linear scaling
        "acceptable": ">0.5"
    },
    "resource_efficiency": {
        "formula": "quality_per_second_complex / quality_per_second_simple",
        "expected": 0.8,  # Complex less efficient (higher overhead)
        "acceptable": ">0.5"
    }
}

Metric Category 3: Stability Metrics

Consistency across scenarios:

{
    "quality_variance": {
        "formula": "stdev(simple_quality, medium_quality, complex_quality)",
        "expected": "<0.05",
        "interpretation": "Low variance = stable algorithm"
    },
    "error_consistency": {
        "formula": "all_scenarios_error_rate < threshold",
        "expected": True,
        "interpretation": "Same error rate across loads"
    }
}

---

Pattern 4: Failure Modes & Recovery

Failure Mode 1: One Scenario Fails (Independent)

Simple   ███████████████ Complete ✓
Medium   ██████████ FAILED ✗
Complex  ███ In progress...

Recovery:
• Medium stores checkpoint at failure point
• Can be restarted independently
• Simple/Complex continue
• Aggregator combines partial results

Implementation:

# Isolate failures
try:
    result = await invoke_skill(batch)
except Exception as e:
    progress.errors.append({"message": str(e), "batch_index": i})
    # Don't raise—let other scenarios continue

# Report but don't block
if progress.errors:
    print(f"⚠ {scenario_id} had {len(progress.errors)} errors")

Failure Mode 2: All Scenarios Fail (Systematic)

Simple   ███ FAILED ✗
Medium   ███ FAILED ✗
Complex  ███ FAILED ✗

Possible causes:
• Skill has a bug
• Resource limit exceeded
• Network/database unavailable

Recovery:

async def orchestrator_with_recovery(initial_state):
    """Attempt recovery if all scenarios fail."""

    result = await app.ainvoke(initial_state)

    all_failed = all(
        state[f"progress_{s}"].status == "failed"
        for s in ["simple", "medium", "complex"]
    )

    if all_failed:
        print("All scenarios failed—attempting recovery...")

        # 1. Reduce resource contention
        # 2. Retry with smaller batches
        # 3. Or abort with diagnostic info

        return retry_with_reduced_load(initial_state)

Failure Mode 3: Timeout (Skill Takes Too Long)

Simple   ███████████ Complete in 1.2s (budget: 30s) ✓
Medium   ██████░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ TIMEOUT ✗ (budget: 90s)
Complex  █░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ IN PROGRESS

Recovery:
• Medium: Cancel at 90s, return partial results
• Complex: Continue until 300s timeout

Implementation:

async def invoke_skill_with_timeout(skill, input_data, timeout_seconds):
    try:
        return await asyncio.wait_for(
            invoke_skill(skill, input_data),
            timeout=timeout_seconds
        )
    except asyncio.TimeoutError:
        print(f"Timeout after {timeout_seconds}s, returning partial results")
        return {
            "processed": len(input_data),
            "results": [],
            "error": "timeout",
            "quality_score": 0.0,
        }

---

Pattern 5: Observability & Monitoring

Monitoring Strategy 1: Real-Time Progress

# Stream progress from PostgreSQL checkpoints
async def monitor_real_time():
    while orchestration_running:
        progress = await db.query("""
            SELECT scenario_id, MAX(progress_pct), MAX(elapsed_ms)
            FROM scenario_checkpoints
            WHERE orchestration_id = $1
            GROUP BY scenario_id
        """, orchestration_id)

        for scenario_id, progress_pct, elapsed_ms in progress:
            bar = "█" * int(progress_pct / 5) + "░" * (20 - int(progress_pct / 5))
            print(f"{scenario_id}: │{bar}│ {progress_pct:.0f}%")

        await asyncio.sleep(2)

Monitoring Strategy 2: Comparative Timeline

                   0s      10s      20s      30s      40s
Simple:            |████████|                            (complete)
Medium:            |              |██████████|           (in progress)
Complex:           |                   |████|            (in progress)
                   |─────────────────────────────────────|

Milestones:
Simple:  ✓ 30% @ 0.4s   ✓ 50% @ 0.6s   ✓ 70% @ 0.9s   ✓ 100% @ 1.2s
Medium:  ✓ 30% @ 3.2s   ✓ 50% @ 5.1s   ⏳ 70% @ 8.3s  ⏳ In progress
Complex: ✓ 30% @ 9.1s   ⏳ 50% in progress

Monitoring Strategy 3: Quality Trend

Quality Score (0-1)
1.0 ├─────────────────
    │ Simple    ████░░░░░░
0.8 ├───────────────────
    │ Medium      ██████░░
0.6 ├───────────────────
    │ Complex       ███░░░░
0.4 ├───────────────────
    │
0.2 ├───────────────────
    └─────────────────────
    0%  30%  50%  70%  100%
         Progress

---

Pattern 6: Result Aggregation Strategies

Aggregation Type 1: Comparative (Default)

Compare metrics across all 3 scenarios:

{
  "quality_comparison": {
    "simple": {"latency_p95": 120, "score": 0.92},
    "medium": {"latency_p95": 145, "score": 0.88},
    "complex": {"latency_p95": 185, "score": 0.84}
  },
  "scaling_analysis": {
    "quality_degradation": "8% from simple to complex",
    "time_growth": "linear (as expected)",
    "recommendation": "Quality acceptable, can scale to complex"
  }
}

Aggregation Type 2: Pattern Extraction

Find common patterns across scenarios:

{
  "success_patterns": [
    "Caching strategy effective at all scales",
    "Batch size of 50+ preferred",
    "Memory usage stays below 512MB"
  ],
  "failure_patterns": [
    "Timeout at >5000 items per batch",
    "Quality drops with skewed data distribution"
  ]
}

Aggregation Type 3: Recommendation Engine

Suggest optimal difficulty for production:

{
  "recommended_difficulty": "medium",
  "reasoning": [
    "Simple: Insufficient load to detect bottlenecks",
    "Medium: Good balance of realism and speed",
    "Complex: Takes too long for frequent testing (300s)"
  ],
  "production_scaling": {
    "estimated_items_per_request": 50,
    "estimated_response_time_ms": 450,
    "required_concurrency_support": 10
  }
}

---

Pattern 7: Checkpointing Strategy

Checkpoint Type 1: Scenario-Level

Save progress at each scenario milestone:

INSERT INTO scenario_checkpoints (
    orchestration_id, scenario_id, milestone_pct, elapsed_ms, state_snapshot
) VALUES (
    'demo-001', 'medium', 30, 3200, {'items': 90, 'results': [...]}
);

Use case: Resume interrupted scenario

Checkpoint Type 2: Milestone-Level

Save synchronized state across all scenarios:

INSERT INTO orchestration_milestones (
    orchestration_id, milestone_pct, timestamp, simple_status, medium_status, complex_status
) VALUES (
    'demo-001', 30, NOW(), 'complete', 'paused', 'in_progress'
);

Use case: Track synchronization progress

Checkpoint Type 3: Full-State

Periodic snapshots for recovery:

async def checkpoint_full_state(state: ScenarioOrchestratorState):
    """Save complete state to disk."""

    checkpoint_data = {
        "orchestration_id": state["orchestration_id"],
        "timestamp": datetime.now().isoformat(),
        "progress_simple": state["progress_simple"].to_dict(),
        "progress_medium": state["progress_medium"].to_dict(),
        "progress_complex": state["progress_complex"].to_dict(),
    }

    await db.insert("full_state_checkpoints", checkpoint_data)

Use case: Complete recovery from any point

---

Pattern 8: Cost & Performance Analysis

Cost Analysis

def estimate_orchestration_cost(
    scenarios: dict[str, ScenarioDefinition]
) -> dict:
    """Estimate total execution cost."""

    # LLM cost (if using Claude)
    llm_cost_per_scenario = {
        "simple": estimate_tokens(100) * 0.001,   # ~$0.002
        "medium": estimate_tokens(300) * 0.001,   # ~$0.005
        "complex": estimate_tokens(800) * 0.001,  # ~$0.010
    }

    # Compute cost (if cloud)
    compute_cost = {
        "simple": 30 / 3600 * 0.10,   # 30s @ $0.10/hour
        "medium": 90 / 3600 * 0.10,   # 90s @ $0.10/hour
        "complex": 300 / 3600 * 0.10, # 300s @ $0.10/hour
    }

    # Database cost
    db_cost = 0.001  # Negligible for checkpointing

    return {
        "llm_cost": sum(llm_cost_per_scenario.values()),
        "compute_cost": sum(compute_cost.values()),
        "db_cost": db_cost,
        "total_cost": sum(llm_cost_per_scenario.values()) + sum(compute_cost.values()) + db_cost,
        "cost_per_scenario": llm_cost_per_scenario,
    }

Performance Analysis

def analyze_performance(
    results: dict
) -> dict:
    """Analyze orchestration performance."""

    return {
        "total_execution_time_minutes": (sum(r["elapsed_ms"] for r in results.values()) / 60000),
        "critical_path_seconds": max(r["elapsed_ms"] for r in results.values()) / 1000,
        "parallel_efficiency": (
            (sum(r["elapsed_ms"] for r in results.values()) / 1000) /
            (max(r["elapsed_ms"] for r in results.values()) / 1000)
        ),
        "cost_per_quality_point": estimate_orchestration_cost({}) / avg_quality_score,
    }

---

Key Architectural Decisions

---

References

• langgraph-implementation.md - Python implementation details
• claude-code-instance-management.md - Multi-terminal setup
• state-machine-design.md - Detailed state transitions
• skill-agnostic-template.md - Template for new skills

Claude Code Instance Management: Multi-Scenario Demos

Structure 3 parallel Claude Code terminal instances for simultaneous scenario execution with shared state synchronization.

Instance Architecture

┌─────────────────────────────────────────────────────────────────────┐
│                    COORDINATOR PROCESS (Python)                      │
│                   (Runs orchestrator graph)                          │
├─────────────────────────────────────────────────────────────────────┤
│  ┌─────────────┐      ┌─────────────┐      ┌─────────────┐         │
│  │  Terminal 1 │      │  Terminal 2 │      │  Terminal 3 │         │
│  │  (Simple)   │      │  (Medium)   │      │  (Complex)  │         │
│  │             │      │             │      │             │         │
│  │  Session:   │      │  Session:   │      │  Session:   │         │
│  │  simple-123 │      │  medium-123 │      │  complex-123│         │
│  └─────────────┘      └─────────────┘      └─────────────┘         │
│       │                    │                    │                   │
│  Claude Code instances     Claude Code instances     Claude Code     │
│  (3 parallel processes)    (3 parallel processes)    instance        │
│                                                                     │
│  ┌─────────────────────────────────────────────────────────┐       │
│  │   PostgreSQL Checkpoint Table                            │       │
│  │   (Shared state synchronization across instances)        │       │
│  └─────────────────────────────────────────────────────────┘       │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

Setup Instructions

Step 1: Prepare the Project

Ensure your project has the orchestrator graph and shared utilities:

# At project root
mkdir -p backend/app/workflows/multi_scenario
cp src/skills/multi-scenario-orchestration/references/langgraph-implementation.py \
   backend/app/workflows/multi_scenario/orchestrator.py

# Create coordinator script
cat > backend/app/workflows/multi_scenario/coordinator.py << 'EOF'
"""
Main coordinator that launches and monitors 3 Claude Code instances.
"""
import asyncio
import subprocess
import os
from pathlib import Path

PROJECT_ROOT = Path(__file__).parent.parent.parent.parent
SCENARIOS = ["simple", "medium", "complex"]
SKILL_NAME = "your-skill-name"  # Change this

async def launch_scenario_instance(scenario_id: str, orchestration_id: str):
    """Launch one Claude Code instance for a scenario."""

    env = os.environ.copy()
    env["SCENARIO_ID"] = scenario_id
    env["ORCHESTRATION_ID"] = orchestration_id
    env["PROJECT_ROOT"] = str(PROJECT_ROOT)

    # Launch Claude Code instance
    process = subprocess.Popen(
        [
            "claude", "code",
            str(PROJECT_ROOT),
            "--session", f"scenario-{scenario_id}-{orchestration_id}",
            "--skill", SKILL_NAME,
        ],
        env=env,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE
    )

    print(f"[COORDINATOR] Launched {scenario_id} instance (PID: {process.pid})")
    return process

async def monitor_instances(processes: dict):
    """Monitor all instances for completion."""

    while any(p.poll() is None for p in processes.values()):
        for scenario_id, process in processes.items():
            if process.poll() is not None:
                print(f"[COORDINATOR] {scenario_id} instance completed")

        await asyncio.sleep(1)

async def main():
    orchestration_id = "demo-001"

    print(f"[COORDINATOR] Starting orchestration {orchestration_id}")
    print(f"[COORDINATOR] Launching 3 parallel instances...")

    # Launch all instances
    processes = {}
    for scenario_id in SCENARIOS:
        process = await launch_scenario_instance(scenario_id, orchestration_id)
        processes[scenario_id] = process

    print(f"[COORDINATOR] All instances launched. Monitoring...")

    # Monitor
    await monitor_instances(processes)

    print(f"[COORDINATOR] All instances completed")

if __name__ == "__main__":
    asyncio.run(main())
EOF

Step 2: Create Scenario Runner Script

Create /backend/app/workflows/multi_scenario/run_scenario.py:

"""
Runner for single scenario. Invoked by Claude Code instance.
Sets up environment from SCENARIO_ID and ORCHESTRATION_ID env vars.
"""
import os
import asyncio
from orchestrator import (
    build_scenario_orchestrator,
    ScenarioOrchestratorState,
    ScenarioDefinition,
    ScenarioProgress,
)
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

async def run_scenario():
    # Read from environment
    scenario_id = os.getenv("SCENARIO_ID", "simple")
    orchestration_id = os.getenv("ORCHESTRATION_ID", "demo-001")
    project_root = os.getenv("PROJECT_ROOT", ".")

    print(f"[{scenario_id.upper()}] Starting scenario execution")
    print(f"[{scenario_id.upper()}] Orchestration ID: {orchestration_id}")

    # Setup checkpointer — from_conn_string is a @contextmanager, so for a
    # long-running orchestrator use an explicit pool with the PostgresSaver
    # constructor (a `with` block would close the pool mid-run).
    db_url = os.getenv("DATABASE_URL", "postgresql://localhost/orchestkit")
    pool = ConnectionPool(db_url, max_size=20, kwargs={"autocommit": True, "prepare_threshold": 0})
    checkpointer = PostgresSaver(pool)
    checkpointer.setup()  # first run creates the checkpoint tables

    # Build orchestrator
    app = build_scenario_orchestrator(checkpointer=checkpointer)

    # Prepare scenario definitions
    configs = {
        "simple": {
            "complexity_multiplier": 1.0,
            "input_size": 100,
            "time_budget_seconds": 30,
            "skill_params": {"batch_size": 10, "cache_enabled": True}
        },
        "medium": {
            "complexity_multiplier": 3.0,
            "input_size": 300,
            "time_budget_seconds": 90,
            "skill_params": {"batch_size": 50, "cache_enabled": True}
        },
        "complex": {
            "complexity_multiplier": 8.0,
            "input_size": 800,
            "time_budget_seconds": 300,
            "skill_params": {"batch_size": 100, "cache_enabled": True, "parallel_workers": 4}
        }
    }

    cfg = configs[scenario_id]

    # Build initial state
    initial_state: ScenarioOrchestratorState = {
        "orchestration_id": orchestration_id,
        "start_time_unix": int(time.time()),
        "skill_name": "your-skill-name",
        "skill_version": "1.0.0",

        # Current scenario only
        "scenario_simple": None,
        "scenario_medium": None,
        "scenario_complex": None,
        "progress_simple": None,
        "progress_medium": None,
        "progress_complex": None,
    }

    # Set only the relevant scenario
    initial_state[f"scenario_{scenario_id}"] = ScenarioDefinition(
        name=scenario_id,
        difficulty={"simple": "easy", "medium": "intermediate", "complex": "advanced"}[scenario_id],
        complexity_multiplier=cfg["complexity_multiplier"],
        input_size=cfg["input_size"],
        dataset_characteristics={"distribution": "uniform"},
        time_budget_seconds=cfg["time_budget_seconds"],
        memory_limit_mb={"simple": 256, "medium": 512, "complex": 1024}[scenario_id],
        error_tolerance={"simple": 0.0, "medium": 0.05, "complex": 0.1}[scenario_id],
        skill_params=cfg["skill_params"],
        expected_quality={"simple": "basic", "medium": "good", "complex": "excellent"}[scenario_id],
        quality_metrics=["accuracy", "coverage"]
    )

    initial_state[f"progress_{scenario_id}"] = ScenarioProgress(scenario_id=scenario_id)

    # Run orchestrator
    config = {"configurable": {"thread_id": f"orch-{orchestration_id}"}}

    print(f"[{scenario_id.upper()}] Invoking orchestrator...")

    try:
        # Stream progress
        async for update in app.astream(initial_state, config=config, stream_mode="updates"):
            if f"progress_{scenario_id}" in update:
                progress = update[f"progress_{scenario_id}"]
                print(f"[{scenario_id.upper()}] Progress: {progress.progress_pct:.1f}% "
                      f"({progress.items_processed} items, {progress.elapsed_ms}ms)")

        print(f"[{scenario_id.upper()}] Scenario complete")

    except Exception as e:
        print(f"[{scenario_id.upper()}] Error: {e}")
        raise

if __name__ == "__main__":
    import time
    asyncio.run(run_scenario())

Execution: Three-Terminal Mode

Terminal 1: Coordinator

cd /path/to/project
python backend/app/workflows/multi_scenario/coordinator.py

Output:

[COORDINATOR] Starting orchestration demo-001
[COORDINATOR] Launching 3 parallel instances...
[COORDINATOR] Launched simple instance (PID: 1234)
[COORDINATOR] Launched medium instance (PID: 1235)
[COORDINATOR] Launched complex instance (PID: 1236)
[COORDINATOR] All instances launched. Monitoring...

Terminal 2: Simple Scenario

cd /path/to/project
export SCENARIO_ID=simple
export ORCHESTRATION_ID=demo-001
python backend/app/workflows/multi_scenario/run_scenario.py

Output:

[SIMPLE] Starting scenario execution
[SIMPLE] Orchestration ID: demo-001
[SIMPLE] Invoking orchestrator...
[SIMPLE] Progress: 10.0% (10 items, 100ms)
[SIMPLE] Progress: 20.0% (20 items, 200ms)
...
[SIMPLE] Progress: 100.0% (100 items, 1050ms)
[SIMPLE] Scenario complete

Terminal 3: Medium Scenario

cd /path/to/project
export SCENARIO_ID=medium
export ORCHESTRATION_ID=demo-001
python backend/app/workflows/multi_scenario/run_scenario.py

Output:

[MEDIUM] Starting scenario execution
[MEDIUM] Orchestration ID: demo-001
[MEDIUM] Invoking orchestrator...
[MEDIUM] Progress: 3.3% (10 items, 100ms)
[MEDIUM] Progress: 6.7% (20 items, 200ms)
...
[MEDIUM] Progress: 100.0% (300 items, 3100ms)
[MEDIUM] Scenario complete

Terminal 4 (Optional): Complex Scenario

If you have 4 terminals, run complex in parallel:

export SCENARIO_ID=complex
export ORCHESTRATION_ID=demo-001
python backend/app/workflows/multi_scenario/run_scenario.py

Shared State Synchronization

PostgreSQL Checkpoint Schema

-- Create checkpoint table (run once)
CREATE TABLE IF NOT EXISTS scenario_orchestration_checkpoints (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    orchestration_id VARCHAR(255) NOT NULL,
    scenario_id VARCHAR(50) NOT NULL,
    milestone_name VARCHAR(100),
    progress_pct FLOAT,
    timestamp_unix BIGINT NOT NULL,
    state_snapshot JSONB,
    metrics JSONB,
    created_at TIMESTAMP DEFAULT NOW(),
    INDEX idx_orchestration_id (orchestration_id),
    INDEX idx_scenario_id (scenario_id),
    INDEX idx_timestamp (timestamp_unix)
);

-- View progress across all scenarios
SELECT
    orchestration_id,
    scenario_id,
    progress_pct,
    milestone_name,
    timestamp_unix,
    (timestamp_unix / 1000.0) as seconds_elapsed
FROM scenario_orchestration_checkpoints
WHERE orchestration_id = 'demo-001'
ORDER BY scenario_id, progress_pct;

-- Example output:
-- orchestration_id | scenario_id | progress_pct | milestone_name | seconds_elapsed
-- demo-001         | simple      | 30           | checkpoint_1   | 1.2
-- demo-001         | simple      | 70           | checkpoint_2   | 2.8
-- demo-001         | simple      | 100          | completion     | 3.1
-- demo-001         | medium      | 30           | checkpoint_1   | 3.5
-- demo-001         | medium      | 70           | checkpoint_2   | 8.2
-- demo-001         | medium      | 100          | completion     | 9.3
-- demo-001         | complex     | 30           | checkpoint_1   | 9.1
-- demo-001         | complex     | 70           | checkpoint_2   | 22.5
-- demo-001         | complex     | 100          | completion     | 25.7

Monitor Progress from Coordinator

"""Monitor script to watch progress across all instances."""
import asyncio
import time
from datetime import datetime
import psycopg2

async def monitor_orchestration(orchestration_id: str, interval: int = 2):
    """Watch progress of all scenarios."""

    conn = psycopg2.connect("dbname=orchestkit user=postgres")
    cursor = conn.cursor()

    print(f"Monitoring orchestration {orchestration_id}...\n")

    while True:
        cursor.execute("""
            SELECT
                scenario_id,
                MAX(progress_pct) as progress,
                MAX(timestamp_unix) as last_update
            FROM scenario_orchestration_checkpoints
            WHERE orchestration_id = %s
            GROUP BY scenario_id
            ORDER BY scenario_id
        """, (orchestration_id,))

        rows = cursor.fetchall()
        if not rows:
            print("No progress yet...")
            await asyncio.sleep(interval)
            continue

        # Clear screen and print progress
        print(f"\r{datetime.now().strftime('%H:%M:%S')}")
        print("-" * 50)

        all_complete = True
        for scenario_id, progress, timestamp in rows:
            bar_length = int(progress / 5)  # 20-char bar
            bar = "█" * bar_length + "░" * (20 - bar_length)

            print(f"{scenario_id:10} │{bar}│ {progress:3.0f}%")

            if progress < 100:
                all_complete = False

        if all_complete:
            print("\n✓ All scenarios complete!")
            break

        await asyncio.sleep(interval)

    conn.close()

if __name__ == "__main__":
    asyncio.run(monitor_orchestration("demo-001"))

Synchronization at Milestones

To enable forced synchronization at milestones (all scenarios pause and wait):

# In run_scenario.py
import asyncpg

async def wait_for_milestone_sync(
    orchestration_id: str,
    scenario_id: str,
    milestone_pct: int,
    timeout_seconds: int = 30
):
    """Wait for all scenarios to reach milestone."""

    # Poll the progress table directly with an asyncpg pool — the langgraph
    # checkpointer is a @contextmanager factory, not a general connection source.
    pool = await asyncpg.create_pool(DATABASE_URL)
    start = time.time()

    while time.time() - start < timeout_seconds:
        # Query checkpoint status
        async with pool.acquire() as conn:
            result = await conn.fetch("""
                SELECT DISTINCT scenario_id, MAX(progress_pct)
                FROM scenario_orchestration_checkpoints
                WHERE orchestration_id = $1
                GROUP BY scenario_id
            """, orchestration_id)

            scenarios_at_milestone = {
                row["scenario_id"]: row["max"] >= milestone_pct
                for row in result
            }

            if all(scenarios_at_milestone.values()):
                print(f"[{scenario_id.upper()}] All scenarios reached {milestone_pct}%")
                return True

        await asyncio.sleep(0.5)

    print(f"[{scenario_id.upper()}] Sync timeout at {milestone_pct}%")
    return False

Advanced: Multi-Host Execution

For even greater parallelism, run scenarios on different machines:

# Host 1: Coordinator + Simple
python backend/app/workflows/multi_scenario/coordinator.py

# Host 2: Medium (different machine, same DB)
export DATABASE_URL="postgresql://user:pass@coordinator-host/orchestkit"
export SCENARIO_ID=medium
export ORCHESTRATION_ID=demo-001
python backend/app/workflows/multi_scenario/run_scenario.py

# Host 3: Complex (different machine, same DB)
export DATABASE_URL="postgresql://user:pass@coordinator-host/orchestkit"
export SCENARIO_ID=complex
export ORCHESTRATION_ID=demo-001
python backend/app/workflows/multi_scenario/run_scenario.py

PostgreSQL checkpoints serve as the distributed state store.

Best Practices

1. Unique Orchestration IDs: Use timestamp or UUID for each demo run 2. Session Isolation: Each instance gets its own Claude Code session 3. Checkpointing: Always enable PostgreSQL persistence 4. Monitoring: Watch progress via checkpoint table queries 5. Timeout Handling: Allow asynchronous completion, don't force lock-step 6. Error Recovery: Failed instances can be restarted without resetting state

Troubleshooting

Instances get stuck at milestone: → Increase timeout_seconds in wait_for_milestone_sync()

Database connection errors: → Check DATABASE_URL environment variable, ensure PostgreSQL is running

One instance much slower than others: → This is expected! Use Mode A (free-running), not lock-step. Slower instance will eventually complete.

Memory usage grows over time: → Enable checkpointing to disk, reduce batch sizes for complex scenario

Agent Coordination Patterns

Patterns for coordinating multiple specialized agents in complex workflows.

Supervisor-Worker Pattern

from typing import Protocol, Any
import asyncio

class Agent(Protocol):
    async def run(self, task: str, context: dict) -> dict: ...

class SupervisorCoordinator:
    """Central supervisor that routes tasks to worker agents."""

    def __init__(self, workers: dict[str, Agent]):
        self.workers = workers
        self.execution_log: list[dict] = []

    async def route_and_execute(
        self,
        task: str,
        required_agents: list[str],
        parallel: bool = True
    ) -> dict[str, Any]:
        """Route task to specified agents."""
        context = {"task": task, "results": {}}

        if parallel:
            tasks = [
                self._run_worker(name, task, context)
                for name in required_agents
            ]
            results = await asyncio.gather(*tasks, return_exceptions=True)
            return dict(zip(required_agents, results))
        else:
            for name in required_agents:
                context["results"][name] = await self._run_worker(
                    name, task, context
                )
            return context["results"]

    async def _run_worker(
        self, name: str, task: str, context: dict
    ) -> dict:
        """Execute single worker with timeout."""
        try:
            result = await asyncio.wait_for(
                self.workers[name].run(task, context),
                timeout=30.0
            )
            self.execution_log.append({
                "agent": name, "status": "success", "result": result
            })
            return result
        except asyncio.TimeoutError:
            return {"error": f"{name} timed out"}

Conflict Resolution

async def resolve_agent_conflicts(
    findings: list[dict],
    llm: Any
) -> dict:
    """Resolve conflicts between agent outputs."""
    conflicts = []
    for i, f1 in enumerate(findings):
        for f2 in findings[i+1:]:
            if f1.get("recommendation") != f2.get("recommendation"):
                conflicts.append((f1, f2))

    if not conflicts:
        return {"status": "no_conflicts", "findings": findings}

    # LLM arbitration
    resolution = await llm.ainvoke(f"""
        Agents disagree. Determine best recommendation:
        Agent 1: {conflicts[0][0]}
        Agent 2: {conflicts[0][1]}
        Provide: winner, reasoning, confidence (0-1)
    """)
    return {"status": "resolved", "resolution": resolution}

Configuration

• Worker timeout: 30s default
• Max parallel agents: 8
• Retry failed agents: 1 attempt
• Log all executions for debugging

Cost Optimization

• Batch similar tasks to reduce overhead
• Cache agent results by task hash
• Use cheaper models for simple agents
• Parallelize independent agents always

CrewAI Patterns (v1.8+)

CrewAI patterns for role-based multi-agent collaboration with Flows architecture, hierarchical crews, MCP tools, and async execution.

Version: This document covers CrewAI 1.8.x - 1.9.x (2026). For earlier versions, patterns may differ.

Table of Contents

• Flows Architecture (1.8+)
• MCP Tool Support (1.8+)
• Hierarchical Process
• Agent Configuration (1.8+)
• Task Configuration (1.8+)
• Async Execution
• Streaming Output
• Knowledge Sources (1.8+)
• Memory Configuration
• Custom Tools
• Decorator-Based Crew Definition
• Human-in-the-Loop (Flows)
• Configuration Summary
• Best Practices
• Migration from 0.x

---

Flows Architecture (1.8+)

Flows provide event-driven orchestration with state management. This is the major 1.x feature for complex multi-step workflows.

Basic Flow

from crewai.flow.flow import Flow, listen, start

class ResearchFlow(Flow):
    @start()
    def generate_topic(self):
        """Entry point - marked with @start()"""
        return "AI Safety"

    @listen(generate_topic)
    def research_topic(self, topic):
        """Triggered when generate_topic completes"""
        return f"Research findings on {topic}"

    @listen(research_topic)
    def summarize(self, findings):
        """Chain multiple listeners"""
        return f"Summary: {findings[:100]}..."

# Execute flow
flow = ResearchFlow()
result = flow.kickoff()

Structured State (Pydantic)

from pydantic import BaseModel
from crewai.flow.flow import Flow, listen, start

class WorkflowState(BaseModel):
    topic: str = ""
    research: str = ""
    summary: str = ""
    iteration: int = 0

class StatefulFlow(Flow[WorkflowState]):
    @start()
    def initialize(self):
        self.state.topic = "Machine Learning"
        self.state.iteration = 1

    @listen(initialize)
    def process(self):
        self.state.research = f"Research on {self.state.topic}"
        self.state.iteration += 1
        return self.state.research

Router for Conditional Branching

from crewai.flow.flow import Flow, listen, start, router

class ConditionalFlow(Flow):
    @start()
    def evaluate(self):
        # Returns condition result
        return {"score": 85, "passed": True}

    @router(evaluate)
    def route_result(self, result):
        """Route based on evaluation"""
        if result["passed"]:
            return "success"
        return "retry"

    @listen("success")
    def handle_success(self):
        return "Workflow completed successfully"

    @listen("retry")
    def handle_retry(self):
        return "Retrying workflow..."

Parallel Execution with and_/or_

from crewai.flow.flow import Flow, listen, start, and_, or_

class ParallelFlow(Flow):
    @start()
    def task_a(self):
        return "Result A"

    @start()
    def task_b(self):
        return "Result B"

    @listen(and_(task_a, task_b))
    def combine_results(self):
        """Triggers when BOTH complete"""
        return "Combined results"

    @listen(or_(task_a, task_b))
    def first_result(self):
        """Triggers when EITHER completes"""
        return "First result received"

Integrating Crews with Flows

from crewai.flow.flow import Flow, listen, start
from crewai import Crew, Agent, Task

class CrewFlow(Flow):
    @start()
    def prepare_inputs(self):
        return {"topic": "AI Agents", "depth": "detailed"}

    @listen(prepare_inputs)
    def run_research_crew(self, inputs):
        researcher = Agent(
            role="Researcher",
            goal="Research the given topic thoroughly",
            backstory="Expert researcher with domain knowledge"
        )

        task = Agent(
            description=f"Research {inputs['topic']} at {inputs['depth']} level",
            expected_output="Comprehensive research report",
            agent=researcher
        )

        crew = Crew(agents=[researcher], tasks=[task])
        result = crew.kickoff()
        return result.raw

---

MCP Tool Support (1.8+)

CrewAI supports Model Context Protocol (MCP) for external tool integration.

Simple DSL (Recommended)

from crewai import Agent

# URL-based MCP server
agent = Agent(
    role="Research Analyst",
    goal="Research and analyze information",
    backstory="Expert analyst",
    mcps=[
        "https://mcp.example.com/mcp?api_key=your_key",
        "crewai-amp:financial-data",  # CrewAI marketplace
        "crewai-amp:research-tools#pubmed_search"  # Specific tool
    ]
)

Transport-Specific Configuration

from crewai import Agent
from crewai.mcp import MCPServerStdio, MCPServerHTTP, MCPServerSSE
from crewai.mcp.tool_filter import create_static_tool_filter

# Local server via stdio
agent = Agent(
    role="File Analyst",
    goal="Analyze local files",
    backstory="File processing expert",
    mcps=[
        MCPServerStdio(
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem"],
            tool_filter=create_static_tool_filter(
                allowed_tool_names=["read_file", "list_directory"]
            )
        )
    ]
)

# Remote HTTP server
agent = Agent(
    role="API Analyst",
    goal="Query external APIs",
    backstory="Integration specialist",
    mcps=[
        MCPServerHTTP(
            url="https://api.example.com/mcp",
            headers={"Authorization": "Bearer token"},
            connect_timeout=60
        )
    ]
)

# Server-Sent Events (streaming)
agent = Agent(
    role="Real-time Analyst",
    goal="Monitor streaming data",
    backstory="Real-time data specialist",
    mcps=[
        MCPServerSSE(
            url="https://stream.example.com/mcp",
            headers={"Authorization": "Bearer token"}
        )
    ]
)

MCPServerAdapter (Advanced)

from crewai import Agent, Crew, Task
from crewai_tools import MCPServerAdapter

# Context manager for manual connection management
with MCPServerAdapter(server_params, connect_timeout=60) as mcp_tools:
    agent = Agent(
        role="MCP Tool User",
        goal="Use MCP tools effectively",
        backstory="Tool specialist",
        tools=mcp_tools,
        verbose=True
    )

    # Or filter specific tools
    filtered_tools = mcp_tools["specific_tool_name"]

---

Hierarchical Process

from crewai import Agent, Crew, Task, Process

manager = Agent(
    role="Project Manager",
    goal="Coordinate team and ensure deliverables",
    backstory="Senior PM with 10 years experience",
    allow_delegation=True,
    verbose=True
)

researcher = Agent(
    role="Researcher",
    goal="Find accurate information",
    backstory="Expert researcher",
    allow_delegation=False
)

writer = Agent(
    role="Content Writer",
    goal="Create compelling content",
    backstory="Professional writer"
)

crew = Crew(
    agents=[manager, researcher, writer],
    tasks=[research_task, write_task, review_task],
    process=Process.hierarchical,
    manager_llm="gpt-5.5",  # Required for hierarchical
    memory=True,
    verbose=True
)

---

Agent Configuration (1.8+)

from crewai import Agent

agent = Agent(
    # Core identity
    role="Senior Data Scientist",
    goal="Analyze data and provide insights",
    backstory="Expert with 10 years experience",

    # LLM configuration
    llm="gpt-5.5",
    function_calling_llm="gpt-5-mini",  # Cheaper model for tools
    use_system_prompt=True,

    # Execution control
    max_iter=20,
    max_rpm=100,
    max_execution_time=300,  # seconds
    max_retry_limit=2,

    # Advanced features (1.8+)
    reasoning=True,  # Enable reflection before tasks
    max_reasoning_attempts=3,
    multimodal=True,  # Text and visual processing
    inject_date=True,
    date_format="%Y-%m-%d",

    # Memory and context
    memory=True,
    respect_context_window=True,  # Auto-summarize on limit

    # Tools
    tools=[tool1, tool2],
    cache=True,

    # Delegation
    allow_delegation=True,
    verbose=True
)

---

Task Configuration (1.8+)

Structured Output

from pydantic import BaseModel
from crewai import Task

class ReportOutput(BaseModel):
    title: str
    summary: str
    findings: list[str]
    confidence: float

task = Agent(
    description="Analyze market trends and create report",
    expected_output="Structured market analysis report",
    agent=analyst,
    output_pydantic=ReportOutput  # Structured output
)

# Access structured result
result = crew.kickoff()
report = result.pydantic
print(report.title, report.confidence)

Async Task Execution

from crewai import Task

# Parallel research tasks
research_task1 = Agent(
    description="Research topic A",
    expected_output="Research findings",
    agent=researcher,
    async_execution=True  # Non-blocking
)

research_task2 = Agent(
    description="Research topic B",
    expected_output="Research findings",
    agent=researcher,
    async_execution=True
)

# Dependent task waits for async tasks
synthesis_task = Agent(
    description="Synthesize all research",
    expected_output="Integrated analysis",
    agent=analyst,
    context=[research_task1, research_task2]  # Waits for completion
)

Task Guardrails (Validation)

from crewai import Task
from crewai.tasks import TaskOutput

def validate_length(result: TaskOutput) -> tuple[bool, any]:
    """Validate output meets requirements"""
    if len(result.raw.split()) < 100:
        return (False, "Content too brief, expand analysis")
    return (True, result.raw)

task = Agent(
    description="Write comprehensive analysis",
    expected_output="Detailed analysis (100+ words)",
    agent=writer,
    guardrail=validate_length,
    guardrail_max_retries=3
)

# Multiple guardrails
task = Agent(
    description="Generate report",
    expected_output="Validated report",
    agent=analyst,
    guardrails=[
        validate_length,
        validate_sources,
        "Content must be objective and data-driven"  # LLM-based
    ]
)

Human Input Tasks

task = Agent(
    description="Review and approve recommendations",
    expected_output="Approved recommendations",
    agent=reviewer,
    human_input=True  # Pauses for human verification
)

Task Callbacks

from crewai.tasks import TaskOutput

def task_callback(output: TaskOutput):
    print(f"Task completed: {output.description}")
    print(f"Result: {output.raw[:100]}...")
    # Send notifications, log metrics, etc.

task = Agent(
    description="Analyze data",
    expected_output="Analysis results",
    agent=analyst,
    callback=task_callback
)

---

Async Execution

Async Crew Kickoff

import asyncio
from crewai import Crew

async def run_crews_parallel():
    crew1 = Crew(agents=[agent1], tasks=[task1])
    crew2 = Crew(agents=[agent2], tasks=[task2])

    # Run multiple crews in parallel
    results = await asyncio.gather(
        crew1.kickoff_async(),
        crew2.kickoff_async()
    )
    return results

# Execute
results = asyncio.run(run_crews_parallel())

Async Flow Kickoff

from crewai.flow.flow import Flow, start, listen

class AsyncFlow(Flow):
    @start()
    async def fetch_data(self):
        # Async operations supported
        data = await external_api.fetch()
        return data

    @listen(fetch_data)
    async def process_data(self, data):
        result = await process_async(data)
        return result

# Async execution
async def main():
    flow = AsyncFlow()
    result = await flow.kickoff_async()
    return result

asyncio.run(main())

---

Streaming Output

from crewai import Crew

# Enable streaming on crew
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    stream=True  # Enable real-time output
)

# Stream results
result = crew.kickoff()

# Flow streaming
flow = ExampleFlow()
flow.stream = True
streaming = flow.kickoff()

for chunk in streaming:
    print(chunk.content, end="", flush=True)

final_result = streaming.result

---

Knowledge Sources (1.8+)

from crewai import Agent, Crew
from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
from crewai.knowledge.source.pdf_knowledge_source import PDFKnowledgeSource
from crewai.knowledge.source.crew_docling_source import CrewDoclingSource
from crewai.knowledge.knowledge_config import KnowledgeConfig

# String knowledge
company_info = StringKnowledgeSource(
    content="Company policies and guidelines..."
)

# PDF knowledge
docs = PDFKnowledgeSource(file_paths=["manual.pdf", "guide.pdf"])

# Web knowledge
web_source = CrewDoclingSource(
    file_paths=["https://example.com/docs"]
)

# Configure retrieval
config = KnowledgeConfig(
    results_limit=10,      # Documents returned (default: 3)
    score_threshold=0.5    # Relevance minimum (default: 0.35)
)

# Agent-level knowledge
agent = Agent(
    role="Support Agent",
    goal="Answer questions using company knowledge",
    backstory="Expert support representative",
    knowledge_sources=[company_info]
)

# Crew-level knowledge (all agents)
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    knowledge_sources=[docs, web_source]
)

---

Memory Configuration

from crewai import Crew
from crewai.memory import ShortTermMemory, LongTermMemory, EntityMemory

# Simple memory
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    memory=True  # Enable all memory types
)

# Custom memory configuration
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    short_term_memory=ShortTermMemory(),
    long_term_memory=LongTermMemory(
        storage=ChromaStorage(collection_name="crew_memory")
    ),
    entity_memory=EntityMemory()
)

---

Custom Tools

from crewai.tools import tool

@tool("Search Database")
def search_database(query: str) -> str:
    """Search the internal database for relevant information.

    Args:
        query: The search query string
    """
    results = db.search(query)
    return json.dumps(results)

# Async tool
@tool("Fetch API Data")
async def fetch_api_data(endpoint: str) -> str:
    """Fetch data from external API asynchronously.

    Args:
        endpoint: API endpoint to query
    """
    async with aiohttp.ClientSession() as session:
        async with session.get(endpoint) as response:
            return await response.text()

# Assign tools to agent
researcher = Agent(
    role="Researcher",
    goal="Find accurate information",
    backstory="Expert researcher",
    tools=[search_database, fetch_api_data],
    verbose=True
)

---

Decorator-Based Crew Definition (Recommended)

from crewai import Agent, Crew, Task, CrewBase, agent, task, crew

@CrewBase
class ResearchCrew:
    agents_config = 'config/agents.yaml'
    tasks_config = 'config/tasks.yaml'

    @agent
    def researcher(self) -> Agent:
        return Agent(
            config=self.agents_config['researcher'],
            tools=[search_tool]
        )

    @agent
    def analyst(self) -> Agent:
        return Agent(config=self.agents_config['analyst'])

    @task
    def research_task(self) -> Task:
        return Agent(config=self.tasks_config['research'])

    @task
    def analysis_task(self) -> Task:
        return Agent(
            config=self.tasks_config['analysis'],
            context=[self.research_task()]
        )

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,  # Auto-collected
            tasks=self.tasks,    # Auto-collected
            process=Process.sequential
        )

# Execute
result = ResearchCrew().crew().kickoff(inputs={"topic": "AI Safety"})

---

Human-in-the-Loop (Flows)

from crewai.flow.flow import Flow, listen, start
from crewai.flow.human_feedback import human_feedback, HumanFeedbackResult

class ReviewFlow(Flow):
    @start()
    @human_feedback(
        message="Do you approve this content?",
        emit=["approved", "rejected"],
        llm="gpt-5-mini"
    )
    def generate_content(self):
        return "Content for human review..."

    @listen("approved")
    def handle_approval(self, result: HumanFeedbackResult):
        print(f"Approved with feedback: {result.feedback}")
        return "Processing approved content"

    @listen("rejected")
    def handle_rejection(self, result: HumanFeedbackResult):
        print(f"Rejected: {result.feedback}")
        return "Revising content"

---

Configuration Summary

---

Best Practices

1. Use Flows for complex workflows: Multi-step processes benefit from Flows architecture 2. Prefer decorator-based definition: Use @CrewBase for maintainable crew definitions 3. Leverage MCP for external tools: Use the simple DSL for quick MCP integration 4. Enable structured outputs: Use output_pydantic for type-safe results 5. Add guardrails: Validate outputs with function or LLM-based guardrails 6. Use async for parallel work: async_execution=True for independent tasks 7. Configure knowledge sources: Add crew/agent-level knowledge for context 8. Role clarity: Each agent has distinct, non-overlapping role 9. Task granularity: One clear deliverable per task 10. Memory scope: Use short-term for session, long-term for persistent knowledge

---

Migration from 0.x

Framework Comparison

Decision matrix for choosing between multi-agent frameworks.

Feature Comparison

Use Case Matrix

Decision Tree

Start
  |
  +-- Need complex state machines?
  |     |
  |     +-- Yes --> LangGraph
  |     |
  |     +-- No
  |           |
  +-- Role-based collaboration?
  |     |
  |     +-- Yes --> CrewAI
  |     |
  |     +-- No
  |           |
  +-- OpenAI ecosystem only?
  |     |
  |     +-- Yes --> OpenAI Agents SDK
  |     |
  |     +-- No
  |           |
  +-- Enterprise requirements?
  |     |
  |     +-- Yes --> Microsoft Agent Framework
  |     |
  |     +-- No
  |           |
  +-- Open-source priority?
        |
        +-- Yes --> AG2
        |
        +-- No --> LangGraph (default)

Migration Paths

From AutoGen to MS Agent Framework

# AutoGen 0.2 (old)
from autogen import AssistantAgent, UserProxyAgent
agent = AssistantAgent(name="assistant", llm_config=config)

# MS Agent Framework (new)
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
model_client = OpenAIChatCompletionClient(model="gpt-5.5")
agent = AssistantAgent(name="assistant", model_client=model_client)

From Custom to LangGraph

# Custom orchestration (old)
async def workflow(task):
    step1 = await agent1.run(task)
    step2 = await agent2.run(step1)
    return step2

# LangGraph (new)
from langgraph.graph import StateGraph
workflow = StateGraph(State)
workflow.add_node("agent1", agent1_node)
workflow.add_node("agent2", agent2_node)
workflow.add_edge("agent1", "agent2")

Cost Considerations

Performance Characteristics

Team Expertise Requirements

Recommendation Summary

1. Default choice: LangGraph (most capable, production-proven) 2. Fastest to prototype: CrewAI (minimal code, intuitive) 3. OpenAI shops: OpenAI Agents SDK (native integration) 4. Enterprise: Microsoft Agent Framework (compliance, Azure) 5. Research: AG2 (open community, experimental features)

GPT-5.2-Codex

OpenAI's specialized agentic coding model (January 2026) optimized for long-horizon software engineering tasks.

Overview

GPT-5.2-Codex is a specialized variant of GPT-5.2 purpose-built for agentic coding workflows. Unlike the general-purpose GPT-5.2, Codex is optimized for:

• Extended autonomous operation: Hours-long coding sessions without degradation
• Context compaction: Intelligent summarization for long-running tasks
• Project-scale understanding: Full codebase comprehension and refactoring
• Tool reliability: Deterministic file operations and terminal commands

Key Differences from GPT-5.2

Key Capabilities

Long-Horizon Work Through Context Compaction

Codex automatically compacts context during extended sessions, preserving critical information while discarding ephemeral details.

from openai import OpenAI

client = OpenAI()

# Codex maintains context across many tool calls
response = client.chat.completions.create(
    model="gpt-5.2-codex",
    messages=[
        {"role": "system", "content": "You are a senior software engineer."},
        {"role": "user", "content": "Refactor the authentication module to use JWT."}
    ],
    # Codex-specific parameters
    extra_body={
        "codex_config": {
            "compaction_strategy": "semantic",  # semantic, aggressive, minimal
            "preserve_file_state": True,
            "max_session_hours": 8
        }
    }
)

Compaction Strategies:

Project-Scale Tasks

Codex excels at large-scale operations that span entire codebases:

from agents import Agent, tool

# Define codebase navigation tools
@tool
def search_codebase(query: str, file_types: list[str] = None) -> str:
    """Search across the entire codebase for patterns or definitions."""
    # Implementation
    pass

@tool
def apply_refactor(pattern: str, replacement: str, scope: str = "project") -> dict:
    """Apply a refactoring pattern across multiple files with preview."""
    # Returns affected files and changes for approval
    pass

codex_agent = Agent(
    name="refactor-engineer",
    model="gpt-5.2-codex",
    instructions="""You are a senior engineer performing large-scale refactors.

    Guidelines:
    1. Analyze impact before changes
    2. Create rollback points
    3. Run tests after each file change
    4. Document breaking changes""",
    tools=[search_codebase, apply_refactor]
)

Supported Project Tasks:

• Full codebase migrations (Python 2 to 3, React class to hooks)
• Dependency upgrades with breaking change resolution
• Architecture refactors (monolith to microservices)
• Test coverage expansion across modules
• Security vulnerability remediation

Enhanced Cybersecurity Capabilities

Codex includes specialized training for security-aware coding:

# Security-focused agent configuration
security_agent = Agent(
    name="security-engineer",
    model="gpt-5.2-codex",
    instructions="""You are a security engineer. When writing or reviewing code:

    1. Identify OWASP Top 10 vulnerabilities
    2. Check for secrets/credentials in code
    3. Validate input sanitization
    4. Review authentication/authorization flows
    5. Check dependency vulnerabilities via CVE databases""",
    extra_config={
        "security_mode": True,  # Enables security-focused reasoning
        "cve_lookup": True      # Real-time CVE database access
    }
)

Security Capabilities:

Vision for Code Artifacts

Codex processes visual inputs with code-aware understanding:

import base64

# Process architecture diagram
with open("architecture.png", "rb") as f:
    image_data = base64.standard_b64encode(f.read()).decode("utf-8")

response = client.chat.completions.create(
    model="gpt-5.2-codex",
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": "Implement the microservices shown in this architecture diagram."
                },
                {
                    "type": "image_url",
                    "image_url": {"url": f"data:image/png;base64,{image_data}"}
                }
            ]
        }
    ]
)

Vision Use Cases:

• Architecture diagrams to code scaffolding
• UI mockups to component implementation
• Error screenshots to debugging steps
• Whiteboard sketches to technical specs
• Database schemas (ERD) to migrations

Benchmark Performance

GPT-5.2-Codex achieves state-of-the-art results on coding benchmarks:

SWE-Bench Pro Notes:

• Tests real GitHub issues requiring multi-file changes
• Codex excels at test-writing and edge case handling
• Strong performance on legacy codebase understanding

Terminal-Bench 2.0 Notes:

• Tests long-horizon terminal tasks (setup, deploy, debug)
• Codex maintains coherent state across 50+ commands
• Superior error recovery and alternative path exploration

API Usage Patterns

Basic Completion

from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-5.2-codex",
    messages=[
        {
            "role": "system",
            "content": "You are an expert Python developer."
        },
        {
            "role": "user",
            "content": "Write a connection pool manager with health checks."
        }
    ],
    temperature=0.2,  # Lower temperature for code generation
    max_tokens=4096
)

print(response.choices[0].message.content)

Streaming with Tool Use

import json

# Define tools
tools = [
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "Read contents of a file",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "File path"}
                },
                "required": ["path"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "write_file",
            "description": "Write contents to a file",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "File path"},
                    "content": {"type": "string", "description": "File content"}
                },
                "required": ["path", "content"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "run_command",
            "description": "Execute a shell command",
            "parameters": {
                "type": "object",
                "properties": {
                    "command": {"type": "string", "description": "Command to run"},
                    "timeout": {"type": "integer", "description": "Timeout in seconds"}
                },
                "required": ["command"]
            }
        }
    }
]

# Streaming with tool calls
stream = client.chat.completions.create(
    model="gpt-5.2-codex",
    messages=[
        {"role": "user", "content": "Set up a new FastAPI project with tests"}
    ],
    tools=tools,
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.tool_calls:
        tool_call = chunk.choices[0].delta.tool_calls[0]
        print(f"Tool: {tool_call.function.name}")
    elif chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="")

Async Operations

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

async def refactor_module(module_path: str) -> str:
    response = await async_client.chat.completions.create(
        model="gpt-5.2-codex",
        messages=[
            {
                "role": "system",
                "content": "Refactor code for readability and performance."
            },
            {
                "role": "user",
                "content": f"Refactor the module at {module_path}"
            }
        ],
        extra_body={
            "codex_config": {
                "preserve_behavior": True,
                "add_type_hints": True
            }
        }
    )
    return response.choices[0].message.content

# Parallel refactoring
async def refactor_project(modules: list[str]):
    tasks = [refactor_module(m) for m in modules]
    results = await asyncio.gather(*tasks)
    return dict(zip(modules, results))

Integration with Agents SDK

GPT-5.2-Codex integrates seamlessly with OpenAI Agents SDK 0.12+:

from agents import Agent, Runner, handoff, tool
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX

# File operation tools
@tool
def read_file(path: str) -> str:
    """Read file contents."""
    with open(path) as f:
        return f.read()

@tool
def write_file(path: str, content: str) -> str:
    """Write content to file."""
    with open(path, "w") as f:
        f.write(content)
    return f"Wrote {len(content)} bytes to {path}"

@tool
def run_tests(path: str = ".") -> str:
    """Run pytest on the specified path."""
    import subprocess
    result = subprocess.run(
        ["pytest", path, "-v", "--tb=short"],
        capture_output=True,
        text=True
    )
    return f"Exit code: {result.returncode}\n{result.stdout}\n{result.stderr}"

# Specialized agents using Codex
architect_agent = Agent(
    name="architect",
    model="gpt-5.2-codex",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
You are a software architect. Analyze requirements and design solutions.
Hand off to implementer for coding tasks.
Hand off to reviewer for code review.""",
    tools=[read_file]
)

implementer_agent = Agent(
    name="implementer",
    model="gpt-5.2-codex",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
You are a senior developer. Implement designs with clean, tested code.
Hand off to reviewer when implementation is complete.""",
    tools=[read_file, write_file, run_tests]
)

reviewer_agent = Agent(
    name="reviewer",
    model="gpt-5.2-codex",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
You are a code reviewer. Check for bugs, security issues, and style.
Request changes from implementer if needed.
Approve and hand back to architect when satisfied.""",
    tools=[read_file]
)

# Wire up handoffs
architect_agent.handoffs = [
    handoff(agent=implementer_agent),
    handoff(agent=reviewer_agent)
]
implementer_agent.handoffs = [
    handoff(agent=reviewer_agent),
    handoff(agent=architect_agent)
]
reviewer_agent.handoffs = [
    handoff(agent=implementer_agent),
    handoff(agent=architect_agent)
]

# Run development workflow
async def develop_feature(requirement: str):
    runner = Runner()
    result = await runner.run(
        architect_agent,
        f"Design and implement: {requirement}"
    )
    return result.final_output

RealtimeRunner for Interactive Sessions

from agents import Agent
from agents.realtime import RealtimeRunner

# Interactive coding session
codex_agent = Agent(
    name="pair-programmer",
    model="gpt-5.2-codex",
    instructions="You are a pair programming partner. Help write and debug code."
)

async def interactive_session():
    async with RealtimeRunner(codex_agent) as runner:
        # Continuous conversation with context preservation
        while True:
            user_input = input("> ")
            if user_input == "exit":
                break

            async for chunk in runner.stream(user_input):
                print(chunk.content, end="", flush=True)
            print()

IDE Integrations

GPT-5.2-Codex powers several IDE integrations:

Cursor

// .cursor/settings.json
{
  "ai.model": "gpt-5.2-codex",
  "ai.features": {
    "composer": true,
    "agent": true,
    "codebaseIndexing": true
  },
  "ai.codex": {
    "sessionDuration": "8h",
    "compactionStrategy": "semantic"
  }
}

Cursor Features with Codex:

• Composer for multi-file generation
• Agent mode for autonomous tasks
• Background indexing for codebase awareness
• Inline completions with project context

Windsurf (Codeium)

# .windsurf/config.yaml
model:
  provider: openai
  name: gpt-5.2-codex

cascade:
  enabled: true
  max_depth: 10
  auto_apply: false  # Review changes before applying

features:
  flows: true        # Multi-step guided workflows
  supercomplete: true
  terminal_agent: true

Windsurf Features:

• Cascade for chained operations
• Flows for guided development
• Terminal integration for full-stack tasks

GitHub Copilot Workspace

# .github/copilot-workspace.yml
model: gpt-5.2-codex

workspace:
  scope: repository
  features:
    - issue-to-pr
    - multi-file-edit
    - test-generation

review:
  auto_suggest: true
  security_scan: true

Copilot Workspace Features:

• Issue-to-PR automation
• Multi-repository awareness
• Integrated CI feedback

Factory (VSCode Extension)

// .vscode/settings.json
{
  "factory.model": "gpt-5.2-codex",
  "factory.drafter": {
    "enabled": true,
    "autoContext": true
  },
  "factory.pilot": {
    "enabled": true,
    "approvalRequired": true
  }
}

When to Use Codex vs Standard GPT-5.2

Use GPT-5.2-Codex When:

Use Standard GPT-5.2 When:

Decision Matrix

Task Duration > 1 hour?
  |
  +-- Yes --> GPT-5.2-Codex
  |
  +-- No
        |
        +-- Multiple files affected?
        |     |
        |     +-- Yes --> GPT-5.2-Codex
        |     |
        |     +-- No
        |           |
        |           +-- Security review needed?
        |           |     |
        |           |     +-- Yes --> GPT-5.2-Codex
        |           |     |
        |           |     +-- No --> GPT-5.2 (standard)

Pricing Considerations

Token Pricing (January 2026)

Cost Optimization Strategies

# 1. Use caching for repeated context
response = client.chat.completions.create(
    model="gpt-5.2-codex",
    messages=messages,
    extra_body={
        "cache_control": {
            "system_prompt": "ephemeral",  # Cache system prompt
            "file_contents": "persistent"   # Cache file reads
        }
    }
)

# 2. Batch similar operations
# Instead of separate calls per file:
files_to_refactor = ["auth.py", "users.py", "api.py"]
response = client.chat.completions.create(
    model="gpt-5.2-codex",
    messages=[
        {"role": "user", "content": f"Refactor these files: {files_to_refactor}"}
    ]
)

# 3. Use gpt-5.2-mini for preprocessing
# Filter/classify tasks before sending to Codex
classification = client.chat.completions.create(
    model="gpt-5.2-mini",
    messages=[{"role": "user", "content": f"Is this task complex? {task}"}]
)
if "complex" in classification.choices[0].message.content.lower():
    # Use Codex for complex tasks
    use_model = "gpt-5.2-codex"
else:
    # Use standard for simple tasks
    use_model = "gpt-5.2"

Estimated Costs by Task Type

Note: Codex typically requires fewer iterations for complex tasks, often making total cost comparable to standard GPT-5.2.

Configuration Reference

Codex-Specific Parameters

response = client.chat.completions.create(
    model="gpt-5.2-codex",
    messages=messages,
    extra_body={
        "codex_config": {
            # Context management
            "compaction_strategy": "semantic",  # semantic, aggressive, minimal
            "preserve_file_state": True,        # Remember file contents
            "max_session_hours": 8,             # Session duration limit

            # Code behavior
            "preserve_behavior": True,          # Ensure refactors don't change behavior
            "add_type_hints": True,             # Add type hints when refactoring
            "follow_style_guide": "project",    # project, google, pep8

            # Safety
            "security_mode": True,              # Enable security scanning
            "dry_run": False,                   # Preview changes without applying
            "require_tests": True,              # Require test coverage for changes

            # Tools
            "shell_timeout": 300,               # Max seconds for shell commands
            "file_size_limit": 1048576          # Max file size to read (1MB)
        }
    }
)

Best Practices

1. Start with clear goals: Define what "done" looks like upfront 2. Provide project context: Include README, architecture docs, coding standards 3. Use semantic compaction: Best balance of context and performance 4. Enable security mode: Catch vulnerabilities during development 5. Set session limits: Prevent runaway costs with max_session_hours 6. Review before applying: Use dry_run for large refactors 7. Batch related operations: Reduce API calls by grouping similar tasks 8. Cache file contents: Use persistent caching for frequently read files

Related Resources

• OpenAI Agents SDK Reference
• Framework Comparison
• Multi-Agent Orchestration

LangGraph Implementation: Multi-Scenario Orchestration

Complete Python implementation of the multi-scenario orchestration pattern using LangGraph 1.0.6+.

1. State Definition

from typing import TypedDict, Annotated, Literal
from dataclasses import dataclass, field, asdict
from operator import add
import time
from datetime import datetime

@dataclass
class ScenarioProgress:
    """Track execution state for one scenario."""
    scenario_id: str
    status: Literal["pending", "running", "paused", "complete", "failed"]
    progress_pct: float = 0.0

    # Milestones
    milestones_reached: list[str] = field(default_factory=list)
    current_milestone: str = "start"

    # Timing
    start_time_ms: int = 0
    elapsed_ms: int = 0
    elapsed_checkpoints: dict = field(default_factory=dict)  # {milestone: time_ms}

    # Metrics
    memory_used_mb: int = 0
    items_processed: int = 0
    batch_count: int = 0

    # Results
    partial_results: list[dict] = field(default_factory=list)
    quality_scores: dict = field(default_factory=dict)

    # Errors
    errors: list[dict] = field(default_factory=list)

    def to_dict(self):
        return asdict(self)

@dataclass
class ScenarioDefinition:
    """Configuration for one scenario."""
    name: str  # "simple", "medium", "complex"
    difficulty: Literal["easy", "intermediate", "advanced"]
    complexity_multiplier: float  # 1.0, 3.0, 8.0

    # Inputs
    input_size: int
    dataset_characteristics: dict  # {"distribution": "uniform"}

    # Constraints
    time_budget_seconds: int
    memory_limit_mb: int
    error_tolerance: float  # 0-1

    # Skill params
    skill_params: dict

    # Expectations
    expected_quality: Literal["basic", "good", "excellent"]
    quality_metrics: list[str]

    def to_dict(self):
        return asdict(self)

class ScenarioOrchestratorState(TypedDict, total=False):
    """State for the entire orchestration."""

    # Orchestration metadata
    orchestration_id: str
    start_time_unix: int
    skill_name: str
    skill_version: str

    # Scenario definitions
    scenario_simple: ScenarioDefinition
    scenario_medium: ScenarioDefinition
    scenario_complex: ScenarioDefinition

    # Progress tracking
    progress_simple: ScenarioProgress
    progress_medium: ScenarioProgress
    progress_complex: ScenarioProgress

    # Synchronization
    sync_points: dict  # {milestone: bool}
    last_sync_time: int

    # Aggregated results
    final_results: dict

2. Node Implementations

Supervisor Node

from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, Send

async def scenario_supervisor(state: ScenarioOrchestratorState) -> list[Command]:
    """
    Route to all 3 scenarios in parallel.

    Returns Send commands that trigger parallel execution.
    """
    print(f"[SUPERVISOR] Starting orchestration {state['orchestration_id']}")

    # Initialize progress for each scenario
    for scenario_id in ["simple", "medium", "complex"]:
        progress = ScenarioProgress(
            scenario_id=scenario_id,
            status="pending",
            start_time_ms=int(time.time() * 1000)
        )
        state[f"progress_{scenario_id}"] = progress

    # Return Send commands for parallel execution
    return [
        Send("scenario_worker", {"scenario_id": "simple", **state}),
        Send("scenario_worker", {"scenario_id": "medium", **state}),
        Send("scenario_worker", {"scenario_id": "complex", **state}),
    ]

async def scenario_worker(state: ScenarioOrchestratorState) -> dict:
    """
    Execute one scenario (simple, medium, or complex).

    Receives scenario_id from supervisor via Send.
    """
    scenario_id = state.get("scenario_id")
    progress = state[f"progress_{scenario_id}"]
    scenario_def = state[f"scenario_{scenario_id}"]

    print(f"[SCENARIO {scenario_id.upper()}] Starting ({scenario_def.complexity_multiplier}x complexity)")

    progress.status = "running"
    progress.start_time_ms = int(time.time() * 1000)

    try:
        # Execute skill for this scenario
        result = await execute_skill_with_milestones(
            skill_name=state["skill_name"],
            scenario_def=scenario_def,
            progress=progress,
            state=state
        )

        progress.status = "complete"
        progress.elapsed_ms = int(time.time() * 1000) - progress.start_time_ms
        progress.partial_results.append(result)

        print(f"[SCENARIO {scenario_id.upper()}] Complete in {progress.elapsed_ms}ms")

        return {f"progress_{scenario_id}": progress}

    except Exception as e:
        progress.status = "failed"
        progress.errors.append({
            "timestamp": datetime.now().isoformat(),
            "message": str(e),
            "severity": "error"
        })
        print(f"[SCENARIO {scenario_id.upper()}] Failed: {e}")

        return {f"progress_{scenario_id}": progress}


async def execute_skill_with_milestones(
    skill_name: str,
    scenario_def: ScenarioDefinition,
    progress: ScenarioProgress,
    state: ScenarioOrchestratorState
) -> dict:
    """
    Execute skill, recording milestones and checkpoints.

    This is where you call YOUR SKILL.
    """

    milestones = [0, 30, 50, 70, 90, 100]  # Percentage checkpoints
    results = {"batches": [], "quality": {}}

    input_items = generate_test_data(
        size=scenario_def.input_size,
        characteristics=scenario_def.dataset_characteristics
    )

    batch_size = scenario_def.skill_params.get("batch_size", 10)

    for batch_idx, batch in enumerate(chunks(input_items, batch_size)):
        # Execute skill on this batch
        # Replace this with your actual skill invocation
        batch_result = await invoke_skill(
            skill_name=skill_name,
            input_data=batch,
            params=scenario_def.skill_params
        )

        results["batches"].append(batch_result)
        progress.batch_count += 1
        progress.items_processed += len(batch)

        # Update progress percentage
        progress.progress_pct = (progress.items_processed / scenario_def.input_size) * 100

        # Check if we've reached a milestone
        reached_milestones = [m for m in milestones if m <= progress.progress_pct]
        new_milestones = [m for m in reached_milestones if m not in progress.milestones_reached]

        for milestone in new_milestones:
            progress.milestones_reached.append(milestone)
            elapsed = int(time.time() * 1000) - progress.start_time_ms
            progress.elapsed_checkpoints[f"milestone_{milestone}"] = elapsed

            print(f"  [{progress.scenario_id}] Reached {milestone}% at {elapsed}ms")

            # Optional: Wait for other scenarios at major milestones
            if milestone in [30, 70]:
                await synchronize_at_milestone(milestone, state)

    # Score results
    results["quality"] = calculate_quality_metrics(results["batches"], scenario_def.quality_metrics)
    progress.quality_scores = results["quality"]

    return results

Synchronization Node

async def synchronize_at_milestone(
    milestone_pct: int,
    state: ScenarioOrchestratorState,
    timeout_seconds: int = 30
) -> bool:
    """
    Optional: Wait for other scenarios at major milestones.

    Returns True if all scenarios reached milestone, False if timeout.
    """

    start = time.time()
    milestone_key = f"checkpoint_{milestone_pct}"

    while time.time() - start < timeout_seconds:
        simple_at_milestone = milestone_pct in state["progress_simple"].milestones_reached
        medium_at_milestone = milestone_pct in state["progress_medium"].milestones_reached
        complex_at_milestone = milestone_pct in state["progress_complex"].milestones_reached

        all_reached = simple_at_milestone and medium_at_milestone and complex_at_milestone

        if all_reached:
            state["sync_points"][milestone_key] = True
            print(f"[SYNC] All scenarios reached {milestone_pct}%")
            return True

        # Check if any scenario failed
        if any(state[f"progress_{s}"].status == "failed" for s in ["simple", "medium", "complex"]):
            print(f"[SYNC] A scenario failed, proceeding without sync")
            return False

        await asyncio.sleep(0.5)

    print(f"[SYNC] Timeout at {milestone_pct}%, proceeding")
    return False

Aggregator Node

async def scenario_aggregator(state: ScenarioOrchestratorState) -> dict:
    """
    Collect all scenario results and synthesize findings.
    """

    print("[AGGREGATOR] Combining results from all scenarios")

    aggregated = {
        "orchestration_id": state["orchestration_id"],
        "skill": state["skill_name"],
        "timestamp": datetime.now().isoformat(),

        # Raw results
        "results_by_scenario": {
            "simple": state["progress_simple"].partial_results[-1] if state["progress_simple"].partial_results else {},
            "medium": state["progress_medium"].partial_results[-1] if state["progress_medium"].partial_results else {},
            "complex": state["progress_complex"].partial_results[-1] if state["progress_complex"].partial_results else {},
        },

        # Metrics
        "metrics": {},

        # Comparison
        "comparison": {},

        # Recommendations
        "recommendations": []
    }

    # Calculate comparative metrics
    for scenario_id in ["simple", "medium", "complex"]:
        progress = state[f"progress_{scenario_id}"]

        aggregated["metrics"][scenario_id] = {
            "elapsed_ms": progress.elapsed_ms,
            "items_processed": progress.items_processed,
            "quality_scores": progress.quality_scores,
            "errors": len(progress.errors)
        }

    # Compare quality vs. complexity
    simple_quality = state["progress_simple"].quality_scores.get("overall", 0)
    medium_quality = state["progress_medium"].quality_scores.get("overall", 0)
    complex_quality = state["progress_complex"].quality_scores.get("overall", 0)

    aggregated["comparison"]["quality_ranking"] = {
        "best": max(
            ("simple", simple_quality),
            ("medium", medium_quality),
            ("complex", complex_quality),
            key=lambda x: x[1]
        )[0],
        "scores": {
            "simple": simple_quality,
            "medium": medium_quality,
            "complex": complex_quality
        }
    }

    # Time complexity analysis
    simple_time = state["progress_simple"].elapsed_ms
    medium_time = state["progress_medium"].elapsed_ms
    complex_time = state["progress_complex"].elapsed_ms

    simple_size = 100 * 1.0
    medium_size = 100 * 3.0
    complex_size = 100 * 8.0

    aggregated["comparison"]["time_per_item_ms"] = {
        "simple": simple_time / simple_size,
        "medium": medium_time / medium_size,
        "complex": complex_time / complex_size,
    }

    # Identify scaling issues
    if complex_time / complex_size > simple_time / simple_size * 2:
        aggregated["recommendations"].append("Sublinear scaling—excellent performance with increased load")
    elif complex_time / complex_size < simple_time / simple_size * 0.8:
        aggregated["recommendations"].append("Superlinear scaling—overhead increases with load")

    # Success patterns
    success_patterns = []
    for scenario_id in ["simple", "medium", "complex"]:
        if state[f"progress_{scenario_id}"].status == "complete" and state[f"progress_{scenario_id}"].errors == []:
            success_patterns.append(scenario_id)

    aggregated["recommendations"].append(f"Successful in all scenarios: {', '.join(success_patterns)}")

    return {"final_results": aggregated}

3. Graph Construction

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import Command

def build_scenario_orchestrator(
    checkpointer: PostgresSaver | None = None
) -> Any:
    """
    Build the complete orchestration graph.
    """

    graph = StateGraph(ScenarioOrchestratorState)

    # Nodes
    graph.add_node("supervisor", scenario_supervisor)
    graph.add_node("scenario_worker", scenario_worker)
    graph.add_node("aggregator", scenario_aggregator)

    # Edges
    graph.add_edge(START, "supervisor")

    # Fan-out: supervisor sends to 3 parallel workers
    graph.add_conditional_edges(
        "supervisor",
        lambda _: ["scenario_worker", "scenario_worker", "scenario_worker"]
    )

    # Workers converge at aggregator
    graph.add_edge("scenario_worker", "aggregator")
    graph.add_edge("aggregator", END)

    # Compile with checkpointing
    return graph.compile(checkpointer=checkpointer)

4. Invocation Example

import asyncio
import uuid
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

async def main():
    # Setup checkpointing — from_conn_string is a @contextmanager; use an
    # explicit pool + PostgresSaver constructor for a long-running orchestrator
    pool = ConnectionPool(
        "postgresql://user:password@localhost/orchestkit",
        max_size=20,
        kwargs={"autocommit": True, "prepare_threshold": 0},
    )
    checkpointer = PostgresSaver(pool)
    checkpointer.setup()  # first run creates the checkpoint tables

    # Build orchestrator
    app = build_scenario_orchestrator(checkpointer=checkpointer)

    # Prepare initial state
    initial_state: ScenarioOrchestratorState = {
        "orchestration_id": f"demo-{uuid.uuid4().hex[:8]}",
        "start_time_unix": int(time.time()),
        "skill_name": "your-skill-name",
        "skill_version": "1.0.0",

        # Scenarios
        "scenario_simple": ScenarioDefinition(
            name="simple",
            difficulty="easy",
            complexity_multiplier=1.0,
            input_size=100,
            dataset_characteristics={"distribution": "uniform"},
            time_budget_seconds=30,
            memory_limit_mb=256,
            error_tolerance=0.0,
            skill_params={"batch_size": 10, "cache_enabled": True},
            expected_quality="basic",
            quality_metrics=["accuracy", "coverage"]
        ),
        "scenario_medium": ScenarioDefinition(
            name="medium",
            difficulty="intermediate",
            complexity_multiplier=3.0,
            input_size=300,
            dataset_characteristics={"distribution": "uniform"},
            time_budget_seconds=90,
            memory_limit_mb=512,
            error_tolerance=0.05,
            skill_params={"batch_size": 50, "cache_enabled": True},
            expected_quality="good",
            quality_metrics=["accuracy", "coverage"]
        ),
        "scenario_complex": ScenarioDefinition(
            name="complex",
            difficulty="advanced",
            complexity_multiplier=8.0,
            input_size=800,
            dataset_characteristics={"distribution": "skewed"},
            time_budget_seconds=300,
            memory_limit_mb=1024,
            error_tolerance=0.1,
            skill_params={"batch_size": 100, "cache_enabled": True, "parallel_workers": 4},
            expected_quality="excellent",
            quality_metrics=["accuracy", "coverage", "latency"]
        ),

        # Progress tracking
        "progress_simple": ScenarioProgress(scenario_id="simple"),
        "progress_medium": ScenarioProgress(scenario_id="medium"),
        "progress_complex": ScenarioProgress(scenario_id="complex"),

        # Synchronization
        "sync_points": {},
        "last_sync_time": 0,
    }

    # Run with thread_id for checkpointing
    config = {"configurable": {"thread_id": f"orch-{initial_state['orchestration_id']}"}}

    print("Starting multi-scenario orchestration...")
    result = await app.ainvoke(initial_state, config=config)

    # Print results
    final = result["final_results"]
    print("\n" + "="*60)
    print("ORCHESTRATION RESULTS")
    print("="*60)
    print(f"Orchestration ID: {final['orchestration_id']}")
    print(f"Skill: {final['skill']}")
    print("\nQuality Comparison:")
    for scenario, score in final["comparison"]["quality_ranking"]["scores"].items():
        print(f"  {scenario}: {score:.2f}")
    print("\nTime per Item (ms):")
    for scenario, time in final["comparison"]["time_per_item_ms"].items():
        print(f"  {scenario}: {time:.2f}ms")
    print("\nRecommendations:")
    for rec in final["recommendations"]:
        print(f"  • {rec}")

if __name__ == "__main__":
    asyncio.run(main())

5. Helper Functions

def chunks(items: list, size: int):
    """Split items into chunks."""
    for i in range(0, len(items), size):
        yield items[i:i + size]

def generate_test_data(size: int, characteristics: dict) -> list:
    """Generate test data based on scenario characteristics."""
    import random

    distribution = characteristics.get("distribution", "uniform")

    if distribution == "uniform":
        return [{"id": i, "value": random.random()} for i in range(size)]
    elif distribution == "skewed":
        # Zipfian distribution
        return [
            {"id": i, "value": random.random() ** 2}
            for i in range(size)
        ]
    else:
        return [{"id": i, "value": random.random()} for i in range(size)]

async def invoke_skill(
    skill_name: str,
    input_data: list,
    params: dict
) -> dict:
    """
    Invoke your skill here.

    Replace with actual skill invocation.
    """
    # Simulate processing
    await asyncio.sleep(0.1)  # 100ms per batch

    return {
        "processed": len(input_data),
        "quality_score": 0.85 + (random.random() * 0.15),
        "timestamp": datetime.now().isoformat()
    }

def calculate_quality_metrics(batches: list, metrics: list[str]) -> dict:
    """Calculate quality metrics across batches."""
    if not batches:
        return {metric: 0.0 for metric in metrics}

    scores = {
        "accuracy": sum(b.get("quality_score", 0) for b in batches) / len(batches),
        "coverage": 1.0,
    }

    return {metric: scores.get(metric, 0.0) for metric in metrics}

6. Streaming Results (Real-time Progress)

async def stream_orchestration_progress(
    app,
    initial_state: ScenarioOrchestratorState,
    config: dict
):
    """
    Stream progress updates as scenarios execute.
    """

    async for step in app.astream(initial_state, config=config, stream_mode="updates"):
        print(f"\n[UPDATE] {step}")

        # Extract progress from step
        if "progress_simple" in step:
            p = step["progress_simple"]
            print(f"  Simple: {p.progress_pct:.1f}% ({p.items_processed} items)")

        if "progress_medium" in step:
            p = step["progress_medium"]
            print(f"  Medium: {p.progress_pct:.1f}% ({p.items_processed} items)")

        if "progress_complex" in step:
            p = step["progress_complex"]
            print(f"  Complex: {p.progress_pct:.1f}% ({p.items_processed} items)")

Key Features

1. Fan-Out/Fan-In: All 3 scenarios execute in parallel 2. Milestone Tracking: Progress recorded at key checkpoints 3. Synchronization: Optional wait points at 30% and 70% 4. Error Isolation: One scenario's failure doesn't block others 5. Checkpointing: State saved to PostgreSQL for recovery 6. Aggregation: Cross-scenario analysis and recommendations 7. Streaming: Real-time progress updates

Testing

@pytest.mark.asyncio
async def test_multi_scenario_orchestration():
    # Mock checkpointer
    from langgraph.checkpoint.memory import MemorySaver

    app = build_scenario_orchestrator(checkpointer=MemorySaver())

    initial_state = {...}  # Setup
    config = {"configurable": {"thread_id": "test-123"}}

    result = await app.ainvoke(initial_state, config=config)

    assert result["final_results"]["orchestration_id"]
    assert "simple" in result["final_results"]["metrics"]
    assert "medium" in result["final_results"]["metrics"]
    assert "complex" in result["final_results"]["metrics"]

Microsoft Agent Framework

Microsoft Agent Framework (AutoGen + Semantic Kernel merger) patterns for enterprise multi-agent systems.

AssistantAgent Setup

from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Create model client
model_client = OpenAIChatCompletionClient(
    model="gpt-5.5",
    api_key=os.environ["OPENAI_API_KEY"]
)

# Define assistant agent
assistant = AssistantAgent(
    name="assistant",
    description="A helpful AI assistant",
    model_client=model_client,
    system_message="You are a helpful assistant. Answer questions concisely."
)

Team Patterns

Round Robin Chat

from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination

# Define team members
planner = AssistantAgent(
    name="planner",
    description="Plans tasks",
    model_client=model_client,
    system_message="You plan tasks. When done, say 'PLAN_COMPLETE'."
)

executor = AssistantAgent(
    name="executor",
    description="Executes tasks",
    model_client=model_client,
    system_message="You execute the plan. When done, say 'EXECUTION_COMPLETE'."
)

reviewer = AssistantAgent(
    name="reviewer",
    description="Reviews work",
    model_client=model_client,
    system_message="Review the work. Say 'APPROVED' if satisfactory."
)

# Create team with termination
termination = TextMentionTermination("APPROVED")
team = RoundRobinGroupChat(
    participants=[planner, executor, reviewer],
    termination_condition=termination
)

# Run team
result = await team.run(task="Create a marketing strategy")

Selector Group Chat

from autogen_agentchat.teams import SelectorGroupChat

# Selector chooses next speaker based on context
team = SelectorGroupChat(
    participants=[analyst, writer, reviewer],
    model_client=model_client,  # For selection decisions
    termination_condition=termination
)

Tool Integration

from autogen_core.tools import FunctionTool

# Define tool function
def search_database(query: str) -> str:
    """Search the database for information."""
    results = db.search(query)
    return json.dumps(results)

# Create tool
search_tool = FunctionTool(search_database, description="Search the database")

# Agent with tools
researcher = AssistantAgent(
    name="researcher",
    description="Researches information",
    model_client=model_client,
    tools=[search_tool],
    system_message="Use the search tool to find information."
)

Termination Conditions

from autogen_agentchat.conditions import (
    TextMentionTermination,
    MaxMessageTermination,
    TokenUsageTermination,
    TimeoutTermination
)

# Combine termination conditions
from autogen_agentchat.conditions import OrTerminationCondition

termination = OrTerminationCondition(
    TextMentionTermination("DONE"),
    MaxMessageTermination(max_messages=20),
    TimeoutTermination(timeout_seconds=300)
)

Streaming

# Stream team responses
async for message in team.run_stream(task="Analyze this data"):
    print(f"{message.source}: {message.content}")

State Management

from autogen_agentchat.state import TeamState

# Save state
state = await team.save_state()

# Restore state
await team.load_state(state)

# Resume conversation
result = await team.run(task="Continue from where we left off")

Agent-to-Agent Protocol (A2A)

from autogen_agentchat.protocols import A2AProtocol

# Enable A2A for cross-organization agent communication
protocol = A2AProtocol(
    agent=my_agent,
    endpoint="https://api.example.com/agent",
    auth_token=os.environ["A2A_TOKEN"]
)

# Send message to external agent
response = await protocol.send(
    to="external-agent-id",
    message="Process this request"
)

Migration from AutoGen 0.2

# Old AutoGen 0.2 pattern
# from autogen import AssistantAgent, UserProxyAgent

# New AutoGen 0.4+ pattern
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat

# Key differences:
# - No UserProxyAgent needed for simple tasks
# - Teams replace GroupChat
# - Explicit termination conditions required
# - Model client separate from agent

Configuration

• Model clients: OpenAI, Azure OpenAI, Anthropic supported
• Teams: RoundRobin, Selector, Custom
• Termination: Text mention, max messages, timeout, token usage
• Tools: FunctionTool wrapper for Python functions
• State: Full state serialization for persistence

Best Practices

1. Termination conditions: Always set explicit termination 2. Team size: 3-5 agents optimal for most workflows 3. System messages: Clear role definitions in system_message 4. Tool design: One function per tool, clear descriptions 5. Error handling: Use try/except around team.run() 6. Streaming: Use run_stream() for real-time feedback

Rule Categories

1. Agent Loops (loops) — HIGH — 2 rules

Autonomous LLM reasoning patterns for single-agent workflows.

• loops-react.md — ReAct (Reasoning + Acting) loop with tool execution, self-correction, and memory management
• loops-plan-execute.md — Plan-and-Execute with replanning, step validation, and goal-oriented synthesis

2. Multi-Agent Coordination (multi) — CRITICAL — 3 rules

Patterns for coordinating multiple specialized agents.

• multi-supervisor.md — Supervisor routing with fan-out/fan-in, dependency ordering, and CC Agent Teams
• multi-debate.md — Conflict resolution via confidence scoring and LLM arbitration, agent communication bus
• multi-synthesis.md — Result synthesis with category grouping, executive summaries, and confidence scoring

3. Alternative Frameworks (frameworks) — HIGH — 3 rules

Multi-agent frameworks beyond LangGraph for specialized use cases.

• frameworks-crewai.md — CrewAI hierarchical crews, Flows architecture (1.8+), MCP tools, structured output
• frameworks-autogen.md — Microsoft Agent Framework (AutoGen + SK), RoundRobin/Selector teams, A2A protocol
• frameworks-comparison.md — Decision matrix, feature comparison, migration paths, cost and performance analysis

4. Multi-Scenario (scenario) — MEDIUM — 2 rules

Orchestrate skills across parallel difficulty scenarios with synchronized state.

• scenario-orchestrator.md — Parallel fan-out to 3 difficulty tiers (1x/3x/8x), LangGraph state machine, result aggregation
• scenario-routing.md — Milestone synchronization modes, difficulty scaling strategies, checkpointing and failure recovery

[Rule Name]

[Brief description — 1-2 sentences.]

Incorrect:

// Bad pattern

Correct:

// Good pattern

Key rules:

• [Rule 1]
• [Rule 2]
• [Rule 3]

Reference: [link]

Microsoft Agent Framework (AutoGen + Semantic Kernel)

Enterprise multi-agent systems with RoundRobin/Selector teams, termination conditions, A2A protocol, and tool integration.

Team Setup

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Create model client
model_client = OpenAIChatCompletionClient(model="gpt-5.5")

# Define agents
planner = AssistantAgent(
    name="planner",
    description="Plans complex tasks and breaks them into steps",
    model_client=model_client,
    system_message="You are a planning expert. Break tasks into actionable steps."
)

executor = AssistantAgent(
    name="executor",
    description="Executes planned tasks",
    model_client=model_client,
    system_message="You execute tasks according to the plan."
)

reviewer = AssistantAgent(
    name="reviewer",
    description="Reviews work and provides feedback",
    model_client=model_client,
    system_message="You review work. Say 'APPROVED' if satisfactory."
)

# Create team with termination condition
termination = TextMentionTermination("APPROVED")
team = RoundRobinGroupChat(
    participants=[planner, executor, reviewer],
    termination_condition=termination
)

# Run team
result = await team.run(task="Create a marketing strategy")

Selector Group Chat

from autogen_agentchat.teams import SelectorGroupChat

# Selector chooses next speaker based on context
team = SelectorGroupChat(
    participants=[analyst, writer, reviewer],
    model_client=model_client,
    termination_condition=termination
)

Termination Conditions

from autogen_agentchat.conditions import (
    TextMentionTermination,
    MaxMessageTermination,
    TokenUsageTermination,
    TimeoutTermination,
    OrTerminationCondition,
)

termination = OrTerminationCondition(
    TextMentionTermination("DONE"),
    MaxMessageTermination(max_messages=20),
    TimeoutTermination(timeout_seconds=300)
)

Tool Integration

from autogen_core.tools import FunctionTool

def search_database(query: str) -> str:
    """Search the database for information."""
    results = db.search(query)
    return json.dumps(results)

search_tool = FunctionTool(search_database, description="Search the database")

researcher = AssistantAgent(
    name="researcher",
    description="Researches information",
    model_client=model_client,
    tools=[search_tool],
    system_message="Use the search tool to find information."
)

State Management

# Save state
state = await team.save_state()

# Restore state
await team.load_state(state)

# Resume conversation
result = await team.run(task="Continue from where we left off")

Agent-to-Agent Protocol (A2A)

from autogen_agentchat.protocols import A2AProtocol

protocol = A2AProtocol(
    agent=my_agent,
    endpoint="https://api.example.com/agent",
    auth_token=os.environ["A2A_TOKEN"]
)

response = await protocol.send(
    to="external-agent-id",
    message="Process this request"
)

Streaming

async for message in team.run_stream(task="Analyze this data"):
    print(f"{message.source}: {message.content}")

OpenAI Agents SDK (0.12+)

Alternative for OpenAI-native ecosystems:

from agents import Agent, Runner, handoff, RunConfig
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX

researcher_agent = Agent(
    name="researcher",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
You are a research specialist. Gather information and facts.
When research is complete, hand off to the writer.""",
    model="gpt-5.5"
)

writer_agent = Agent(
    name="writer",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
You are a content writer. Create compelling content from research.""",
    model="gpt-5.5"
)

orchestrator = Agent(
    name="orchestrator",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
You coordinate research and writing tasks.""",
    model="gpt-5.5",
    handoffs=[handoff(agent=researcher_agent), handoff(agent=writer_agent)]
)

async def run_workflow(task: str):
    runner = Runner()
    config = RunConfig(nest_handoff_history=True)
    result = await runner.run(orchestrator, task, run_config=config)
    return result.final_output

Migration from AutoGen 0.2

# Old AutoGen 0.2
# from autogen import AssistantAgent, UserProxyAgent

# New AutoGen 0.4+
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat

# Key differences:
# - No UserProxyAgent needed for simple tasks
# - Teams replace GroupChat
# - Explicit termination conditions required
# - Model client separate from agent

Best Practices

1. Always set explicit termination conditions 2. Team size: 3-5 agents optimal 3. Clear role definitions in system_message 4. One function per tool with clear descriptions 5. Use try/except around team.run() 6. Use run_stream() for real-time feedback

Incorrect — team without termination condition runs indefinitely:

team = RoundRobinGroupChat(
    participants=[planner, executor, reviewer]
    # No termination condition - infinite loop risk
)
result = await team.run(task="Complete task")

Correct — explicit termination prevents infinite loops:

termination = OrTerminationCondition(
    TextMentionTermination("APPROVED"),
    MaxMessageTermination(max_messages=20)
)
team = RoundRobinGroupChat(
    participants=[planner, executor, reviewer],
    termination_condition=termination
)
result = await team.run(task="Complete task")