Competitive Agent Ensembles: A High‑Reliability Pattern for Agentic AI

Competitive Agent Ensemble Pattern

The competitive ensemble mitigates single‑point failures by letting multiple agents generate answers and selecting the best one via a judge agent.

Reliability patterns overview

Parallel tool use : agents issue independent API calls in parallel to hide I/O latency.

Hierarchical agents : a manager splits tasks into smaller steps for execution agents.

Competitive agent ensembles : multiple agents generate answers and the system picks the optimal one.

Redundant execution : two or more agents solve the same task to detect errors.

Parallel and mixed retrieval : multiple retrieval strategies run together to improve context quality.

Multi‑hop retrieval : agents iteratively retrieve deeper, more relevant information.

Model setup

from langchain_huggingface import HuggingFacePipeline
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
from langchain_google_vertexai import ChatVertexAI
import torch

# LLM 1: Meta‑Llama‑3‑8B‑Instruct (open‑source, locally deployed)
model_id = "meta-llama/Meta-Llama-3-8B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(model_id)
hf_model = AutoModelForCausalLM.from_pretrained(
    model_id,
    torch_dtype=torch.bfloat16,
    device_map="auto",
    load_in_4bit=True,
)
pipe = pipeline(
    "text-generation",
    model=hf_model,
    tokenizer=tokenizer,
    max_new_tokens=1024,
    do_sample=True,
    temperature=0.7,
    top_p=0.9,
)
llama3_llm = HuggingFacePipeline(pipeline=pipe)

# LLM 2: Claude‑3‑5‑Sonnet on Vertex AI (proprietary, cloud service)
claude_sonnet_llm = ChatVertexAI(model_name="claude-3-5-sonnet@001", temperature=0.7)
print("LLMs Initialized: Llama 3 and Claude 3.5 Sonnet are ready to compete.")

Structured output schemas

from langchain_core.pydantic_v1 import BaseModel, Field
from typing import List

class ProductDescription(BaseModel):
    """A structured product description with a headline and body."""
    headline: str = Field(description="A catchy, attention‑grabbing headline for the product.")
    body: str = Field(description="A short paragraph (2‑3 sentences) detailing the product's benefits and features.")

class FinalEvaluation(BaseModel):
    """Output of the judge agent, containing the winning description and a critique."""
    best_description: ProductDescription = Field(description="The winning product description chosen by the judge.")
    critique: str = Field(description="A point‑by‑point critique explaining why the winner was chosen.")
    winning_agent: str = Field(description="Name of the agent that produced the winning description.")

Graph state and competitor node factory

from typing import TypedDict, Annotated, Dict
import operator, time

class GraphState(TypedDict):
    product_name: str
    product_category: str
    features: str
    competitor_results: Annotated[Dict[str, ProductDescription], operator.update]
    final_evaluation: FinalEvaluation
    performance_log: Annotated[List[str], operator.add]

def create_competitor_node(agent_name: str, llm, prompt):
    chain = prompt | llm.with_structured_output(ProductDescription)
    def competitor_node(state: GraphState):
        print(f"--- [COMPETITOR: {agent_name}] Starting generation... ---")
        start_time = time.time()
        result = chain.invoke({
            "product_name": state['product_name'],
            "product_category": state['product_category'],
            "features": state['features'],
        })
        execution_time = time.time() - start_time
        log = f"[{agent_name}] Completed in {execution_time:.2f}s."
        print(log)
        return {"competitor_results": {agent_name: result}, "performance_log": [log]}
    return competitor_node

Competitor and judge nodes

# Competitor nodes (three distinct agents)
claude_creative_node = create_competitor_node(
    "Claude_Sonnet_Creative", claude_sonnet_llm, claude_creative_prompt)
llama3_direct_node = create_competitor_node(
    "Llama3_Direct", llama3_llm, llama3_direct_prompt)
llama3_luxury_node = create_competitor_node(
    "Llama3_Luxury", llama3_llm, llama3_luxury_prompt)

# Judge node aggregates and selects the best description
def judge_node(state: GraphState):
    """Evaluate all competitor results and select the winner."""
    print("--- [JUDGE] Evaluating competing descriptions... ---")
    start_time = time.time()
    descriptions_to_evaluate = ""
    for name, desc in state['competitor_results'].items():
        descriptions_to_evaluate += (
            f"--- Option from {name} ---
"
            f"Headline: {desc.headline}
"
            f"Body: {desc.body}

"
        )
    judge_chain = judge_prompt | llm.with_structured_output(FinalEvaluation)
    evaluation = judge_chain.invoke({
        "product_name": state['product_name'],
        "descriptions_to_evaluate": descriptions_to_evaluate,
    })
    execution_time = time.time() - start_time
    log = f"[Judge] Completed evaluation in {execution_time:.2f}s."
    print(log)
    return {"final_evaluation": evaluation, "performance_log": [log]}

Workflow assembly (fan‑out / fan‑in)

from langgraph.graph import StateGraph, END

workflow = StateGraph(GraphState)
workflow.add_node("claude_creative", claude_creative_node)
workflow.add_node("llama3_direct", llama3_direct_node)
workflow.add_node("llama3_luxury", llama3_luxury_node)
workflow.add_node("judge", judge_node)

# Entry points run in parallel
workflow.set_entry_point(["claude_creative", "llama3_direct", "llama3_luxury"])
# Fan‑in to the judge after all competitors finish
workflow.add_edge(["claude_creative", "llama3_direct", "llama3_luxury"], "judge")
workflow.add_edge("judge", END)

app = workflow.compile()

Execution and performance results

# Sample performance numbers (seconds)
competitor_times = [7.33, 6.12, 6.45]
judge_time = 8.91
parallel_time = max(competitor_times)
sequential_time = sum(competitor_times)
total_time = parallel_time + judge_time
print(f"Total Execution Time: {total_time:.2f} seconds")

Running the workflow produces three distinct product descriptions and a final verdict. The judge selects the Claude Sonnet‑Creative description as the winner, providing a headline, body, and a detailed critique.

Analysis of benefits

Higher quality through diversity and evaluation : the three agents generate markedly different outputs, giving the judge a richer set of options and enabling transparent, reasoned selection.

Higher performance through parallelism : parallel execution reduces overall latency by about 63 % compared with sequential execution, incurring only the slowest agent’s runtime overhead.

Full code, data models, and the repository are available at https://github.com/FareedKhan-dev/agentic-parallelism