Combining Agent Patterns

Combine multiple advanced patterns to create sophisticated multi-stage reasoning systems with enhanced capabilities.

Overview

While individual patterns are powerful, combining them unlocks even greater potential. This guide shows proven combinations and best practices for building advanced multi-pattern systems.

Common Combinations

Self-Consistency + Agent Judge

Generate multiple solutions, then use judge to select the best.

from azcore import SelfConsistencyAgent, AgentJudge

# Generate multiple candidates
sc_agent = SelfConsistencyAgent(
    agent=base_agent,
    num_generations=5,
)

# Judge selects best
judge = AgentJudge(
    system_prompt="Evaluate solutions and select the best one."
)

# Combined flow
candidates = sc_agent.generate_all(task)
evaluations = [judge.evaluate(c) for c in candidates]
best = max(zip(candidates, evaluations), key=lambda x: x[1].score)

Benefits:

• Diverse solutions from self-consistency
• Objective selection from judge
• Higher quality than simple voting

Use cases: Complex problems where "best" is subjective

Reflexion + Agent Judge

Use judge for evaluation in reflexion loop.

from azcore import ReflexionAgent, AgentJudge

judge = AgentJudge(system_prompt="Score output quality 0-1")

reflexion = ReflexionAgent(
    agent=agent,
    evaluator=lambda output: judge.evaluate(output).score,
    max_iterations=5,
    target_score=0.9,
)

result = reflexion.run(task)

Benefits:

• Consistent evaluation criteria
• Clear improvement targets
• Objective stopping condition

Use cases: Quality-critical tasks with measurable criteria

Reasoning Duo + Reflexion

Use reasoning duo within reflexion iterations.

from azcore import ReasoningDuoAgent, ReflexionAgent

duo = ReasoningDuoAgent(proposer=proposer, critic=critic, max_rounds=2)

reflexion = ReflexionAgent(
    agent=duo,  # Duo generates each attempt
    max_iterations=3,
    self_reflect=True,
)

result = reflexion.run(task)

Benefits:

• Structured improvement at two levels
• External critique + self-reflection
• High-quality outputs

Use cases: Complex creative tasks, strategic planning

Self-Consistency + Reflexion

Generate diverse attempts, refine the best one.

# Generate multiple approaches
sc_agent = SelfConsistencyAgent(agent=agent, num_generations=5)
candidates = sc_agent.generate_all(task)

# Refine the best candidate
judge = AgentJudge()
best_candidate = max(candidates, key=lambda c: judge.evaluate(c).score)

reflexion = ReflexionAgent(agent=agent, max_iterations=3)
final = reflexion.run_with_seed(task, seed=best_candidate)

Benefits:

• Explore multiple approaches (SC)
• Deep refinement of best approach (Reflexion)
• Best of both breadth and depth

Use cases: Complex problems with multiple solution paths

Multi-Stage Pipelines

Three-Stage Quality Pipeline

# Stage 1: Generate diverse candidates
candidates = self_consistency.generate_all(task, num=5)

# Stage 2: Select most promising
top_candidates = sorted(
    candidates,
    key=lambda c: judge.evaluate(c).score,
    reverse=True
)[:2]

# Stage 3: Refine top candidates
refined = []
for candidate in top_candidates:
    improved = reasoning_duo.run_with_seed(task, seed=candidate)
    refined.append(improved)

# Final selection
final = judge.select_best(refined)

Pipeline: SC → Judge → Duo → Judge

Benefits: Maximum quality through multiple improvement stages

Iterative Ensemble

def iterative_ensemble(task, rounds=3):
    candidates = []

    for round_num in range(rounds):
        # Generate with self-consistency
        new_candidates = sc_agent.generate_all(task, num=3)

        # Refine with reflexion
        for candidate in new_candidates:
            refined = reflexion.improve(candidate, context=candidates)
            candidates.append(refined)

        # Keep only top candidates
        evaluations = [judge.evaluate(c) for c in candidates]
        candidates = [
            c for c, e in sorted(
                zip(candidates, evaluations),
                key=lambda x: x[1].score,
                reverse=True
            )[:5]  # Keep top 5
        ]

    return judge.select_best(candidates)

Benefits: Progressive improvement through iteration

Pattern Selection Matrix

Best Practices

1. Start Simple, Add Complexity

# Start: Single pattern
result = reflexion.run(task)

# If needed: Add judge for evaluation
result = reflexion_with_judge.run(task)

# If still needed: Add self-consistency
result = sc_reflexion_with_judge.run(task)

2. Balance Cost vs. Benefit

# Simple task: Single agent
if is_simple(task):
    return agent.run(task)

# Medium task: One pattern
elif is_medium(task):
    return reflexion.run(task)

# Complex task: Combined patterns
else:
    return sc_reflexion_duo.run(task)

3. Set Appropriate Limits

# Don't over-engineer
combined = CombinedPattern(
    self_consistency_generations=5,  # Not 20
    reflexion_iterations=3,  # Not 10
    duo_rounds=2,  # Not 5
)

4. Monitor Performance

result = combined_pattern.run(task)

metrics = {
    "total_latency": result.total_time,
    "total_cost": result.total_tokens * price_per_token,
    "quality_score": evaluate(result.output),
    "improvement": result.final_score - result.initial_score,
}

log_metrics(metrics)

5. Use Early Stopping

# Stop early if quality achieved
combined = CombinedPattern(
    early_stopping=True,
    target_quality=0.9,
)

Advanced Architectures

Hierarchical Refinement

class HierarchicalRefinement:
    def __init__(self):
        # Level 1: Broad exploration
        self.sc_agent = SelfConsistencyAgent(num_generations=10)

        # Level 2: Focused refinement
        self.reflexion = ReflexionAgent(max_iterations=3)

        # Level 3: Final polish
        self.duo = ReasoningDuoAgent(max_rounds=2)

        # Evaluation
        self.judge = AgentJudge()

    def run(self, task):
        # Level 1: Generate candidates
        candidates = self.sc_agent.generate_all(task)

        # Select top 3
        top_3 = self.judge.rank(candidates)[:3]

        # Level 2: Refine each
        refined = [self.reflexion.run(c) for c in top_3]

        # Select best
        best = self.judge.select_best(refined)

        # Level 3: Final polish
        final = self.duo.run(best)

        return final

Adaptive Pattern Selection

class AdaptivePatternCombination:
    def run(self, task):
        # Analyze task complexity
        complexity = self.analyze_complexity(task)

        if complexity < 0.3:
            # Simple: Single agent
            return self.agent.run(task)

        elif complexity < 0.6:
            # Medium: Reflexion
            return self.reflexion.run(task)

        elif complexity < 0.8:
            # Complex: Reflexion + Judge
            return self.reflexion_with_judge.run(task)

        else:
            # Very complex: Full pipeline
            return self.full_pipeline.run(task)

Ensemble of Patterns

class PatternEnsemble:
    def run(self, task):
        # Run multiple pattern combinations
        results = []

        # Approach 1: Self-consistency
        results.append(self.sc.run(task))

        # Approach 2: Reflexion
        results.append(self.reflexion.run(task))

        # Approach 3: Reasoning Duo
        results.append(self.duo.run(task))

        # Meta-judge selects best
        return self.meta_judge.select_best(results)

Pattern Combination Cookbook

For Code Generation

# 1. Generate multiple implementations
codes = sc_agent.generate_all(task, num=5)

# 2. Test and score each
tested = [run_tests(c) for c in codes]
best = max(zip(codes, tested), key=lambda x: x[1].pass_rate)[0]

# 3. Refine with reflexion
improved = reflexion.improve(best, test_results=tested)

# 4. Final review
final = duo.polish(improved)

For Research and Analysis

# 1. Multiple research approaches
analyses = sc_agent.generate_all(task, num=3)

# 2. Deep refinement of each
refined = [reflexion.run(a) for a in analyses]

# 3. Comparative evaluation
ranked = judge.rank(refined, criteria="depth,accuracy,clarity")

# 4. Synthesize best elements
final = synthesizer.combine(ranked[:2])

For Decision Making

# 1. Generate options
options = sc_agent.generate_all(task, num=5)

# 2. Critique each option
critiqued = [duo.evaluate(o) for o in options]

# 3. Iterative refinement
for _ in range(2):
    improved = [reflexion.improve(c) for c in critiqued]
    critiqued = improved

# 4. Final selection
decision = judge.select_best_with_rationale(critiqued)

Performance Optimization

Parallel Execution

from concurrent.futures import ThreadPoolExecutor

def parallel_refinement(candidates):
    with ThreadPoolExecutor(max_workers=3) as executor:
        # Refine candidates in parallel
        refined = list(executor.map(reflexion.run, candidates))
    return refined

Caching

from functools import lru_cache

@lru_cache(maxsize=100)
def cached_evaluation(output_hash):
    return judge.evaluate(output)

Progressive Refinement

def progressive_pipeline(task):
    # Start with quick pass
    quick = sc_agent.run(task, num_generations=3)

    # Check if good enough
    if judge.evaluate(quick).score >= 0.8:
        return quick

    # If not, do deep refinement
    deep = reflexion.run(quick, max_iterations=5)
    return deep

Debugging Combined Patterns

Trace Execution

def trace_combined_pattern(task):
    trace = []

    # Stage 1
    candidates = sc_agent.generate_all(task)
    trace.append(("SC", candidates))

    # Stage 2
    scores = [judge.evaluate(c) for c in candidates]
    trace.append(("Judge", scores))

    # Stage 3
    best = candidates[scores.index(max(scores))]
    refined = reflexion.run(best)
    trace.append(("Reflexion", refined))

    return refined, trace

Measure Stage Impact

# Measure quality at each stage
quality_progression = []

# After SC
sc_result = sc_agent.run(task)
quality_progression.append(("SC", evaluate(sc_result)))

# After Reflexion
refl_result = reflexion.run(sc_result)
quality_progression.append(("Reflexion", evaluate(refl_result)))

# After Duo
final_result = duo.run(refl_result)
quality_progression.append(("Duo", evaluate(final_result)))

# Visualize improvement
plot_quality_progression(quality_progression)

When NOT to Combine

Overkill Scenarios

• Simple factual questions
• Quick lookups
• Straightforward tasks
• Time-critical operations

Diminishing Returns

# Check if additional patterns help
baseline = agent.run(task)
with_pattern1 = pattern1.run(task)
with_both = pattern2.run(pattern1.run(task))

improvement1 = quality(with_pattern1) - quality(baseline)
improvement2 = quality(with_both) - quality(with_pattern1)

if improvement2 < 0.1 * improvement1:
    print("Diminishing returns - don't use both patterns")