Agent Assembly Line: A High‑Reliability Design Pattern for Scalable AI Agents

The series explains design patterns that improve the reliability of modern AI agent systems. This seventh article focuses on the Agent Assembly Line pattern, which shifts the goal from minimizing latency to maximizing throughput by decomposing a workflow into parallel workstations.

Parallel Tools : agents make independent API calls concurrently to hide I/O latency.

Hierarchical Agents : a manager splits tasks into smaller steps handled by execution agents.

Competitive Agent Ensembles : multiple agents propose answers and the system selects the best.

Redundant Execution : two or more agents solve the same task to detect errors and increase reliability.

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

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

Agent Assembly Line Architecture

The assembly line breaks a batch of product reviews into three stations—triage, summarization, and data extraction—each running in parallel. A review moves through the line, being enriched at each stage, while all stations process different reviews simultaneously.

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

class TriageResult(BaseModel):
    """Initial triage output"""
    category: Literal["Feedback", "Bug Report", "Support Request", "Irrelevant"] = Field(description="The category of the review.")

class Summary(BaseModel):
    """Summary output"""
    summary: str = Field(description="A one-sentence summary of the key feedback in the review.")

class ExtractedData(BaseModel):
    """Data extraction output"""
    product_mentioned: str = Field(description="The specific product the review is about.")
    sentiment: Literal["Positive", "Negative", "Neutral"] = Field(description="The overall sentiment of the review.")
    key_feature: str = Field(description="The main feature or aspect discussed in the review.")

class ProcessedReview(BaseModel):
    """Aggregated review after all stations"""
    original_review: str
    category: str
    summary: Optional[str] = None
    extracted_data: Optional[ExtractedData] = None

These Pydantic models act as standardized containers that travel through the pipeline. The ProcessedReview object is created at the triage station and enriched by the subsequent stations, ensuring consistent data contracts.

from typing import TypedDict, Annotated, List
import operator

class PipelineState(TypedDict):
    # List of raw review strings
    initial_reviews: List[str]
    # List of ProcessedReview objects built as they move through the line
    processed_reviews: List[ProcessedReview]
    performance_log: Annotated[List[str], operator.add]

The pipeline is driven by three node functions, each using ThreadPoolExecutor to parallelise work on all items assigned to that stage.

# Station 1: Triage
MAX_WORKERS = 4

def triage_node(state: PipelineState):
    """Parallel triage of all initial reviews"""
    print(f"--- [Station 1: Triage] Processing {len(state['initial_reviews'])} reviews... ---")
    start_time = time.time()
    triaged_reviews = []
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        future_to_review = {executor.submit(triage_chain.invoke, {"review_text": r}): r for r in state['initial_reviews']}
        for future in tqdm(as_completed(future_to_review), total=len(state['initial_reviews']), desc="Triage Progress"):
            original_review = future_to_review[future]
            try:
                result = future.result()
                triaged_reviews.append(ProcessedReview(original_review=original_review, category=result.category))
            except Exception as exc:
                print(f"Review generated an exception: {exc}")
    execution_time = time.time() - start_time
    log = f"[Triage] Processed {len(state['initial_reviews'])} reviews in {execution_time:.2f}s."
    print(log)
    return {"processed_reviews": triaged_reviews, "performance_log": [log]}

# Station 2: Summarizer
def summarize_node(state: PipelineState):
    """Parallel summarisation of feedback‑type reviews"""
    feedback_reviews = [r for r in state['processed_reviews'] if r.category == "Feedback"]
    if not feedback_reviews:
        print("--- [Station 2: Summarizer] No feedback reviews to process. Skipping. ---")
        return {}
    print(f"--- [Station 2: Summarizer] Processing {len(feedback_reviews)} feedback reviews... ---")
    start_time = time.time()
    review_map = {r.original_review: r for r in state['processed_reviews']}
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        future_to_review = {executor.submit(summarizer_chain.invoke, {"review_text": r.original_review}): r for r in feedback_reviews}
        for future in tqdm(as_completed(future_to_review), total=len(feedback_reviews), desc="Summarizer Progress"):
            original_review_obj = future_to_review[future]
            try:
                result = future.result()
                review_map[original_review_obj.original_review].summary = result.summary
            except Exception as exc:
                print(f"Review generated an exception: {exc}")
    execution_time = time.time() - start_time
    log = f"[Summarizer] Processed {len(feedback_reviews)} reviews in {execution_time:.2f}s."
    print(log)
    return {"processed_reviews": list(review_map.values()), "performance_log": [log]}

# Station 3: Data Extraction
def extract_data_node(state: PipelineState):
    """Parallel extraction of structured data from summarized reviews"""
    summarized_reviews = [r for r in state['processed_reviews'] if r.summary is not None]
    if not summarized_reviews:
        print("--- [Station 3: Extractor] No summarized reviews to process. Skipping. ---")
        return {}
    print(f"--- [Station 3: Extractor] Processing {len(summarized_reviews)} summarized reviews... ---")
    start_time = time.time()
    review_map = {r.original_review: r for r in state['processed_reviews']}
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        future_to_review = {executor.submit(extractor_chain.invoke, {"summary_text": r.summary}): r for r in summarized_reviews}
        for future in tqdm(as_completed(future_to_review), total=len(summarized_reviews), desc="Extractor Progress"):
            original_review_obj = future_to_review[future]
            try:
                result = future.result()
                review_map[original_review_obj.original_review].extracted_data = result
            except Exception as exc:
                print(f"Review generated an exception: {exc}")
    execution_time = time.time() - start_time
    log = f"[Extractor] Processed {len(summarized_reviews)} reviews in {execution_time:.2f}s."
    print(log)
    return {"processed_reviews": list(review_map.values()), "performance_log": [log]}

Each node filters the data it works on (e.g., the summarizer only processes reviews whose category is Feedback, and the extractor only handles reviews that already have a summary), which is the key specialization of the pattern.

from langgraph.graph import StateGraph, END

workflow = StateGraph(PipelineState)
workflow.add_node("triage", triage_node)
workflow.add_node("summarize", summarize_node)
workflow.add_node("extract_data", extract_data_node)
workflow.set_entry_point("triage")
workflow.add_edge("triage", "summarize")
workflow.add_edge("summarize", "extract_data")
workflow.add_edge("extract_data", END)
app = workflow.compile()

The final analysis compares the pipelined assembly line with a simulated monolithic sequential agent. Using ten reviews, the assembly line completes the batch in 20.40 seconds (≈0.49 reviews/s), while the sequential simulation needs 61.20 seconds (≈0.16 reviews/s), yielding a 206 % throughput increase.

# Performance calculations (simplified)
# pipelined_total_time = triage_time + summarize_time + extract_time
# pipelined_throughput = num_reviews / pipelined_total_time
# sequential_total_time = total_latency_per_review * num_reviews
# sequential_throughput = num_reviews / sequential_total_time
# throughput_increase = ((pipelined_throughput - sequential_throughput) / sequential_throughput) * 100

print("=== PERFORMANCE ANALYSIS ===")
print(f"Total Time to Process {num_reviews} Reviews: {pipelined_total_time:.2f} seconds")
print(f"Calculated Throughput: {pipelined_throughput:.2f} reviews/second
")
print("--- Monolithic (Sequential) Workflow (Simulated) ---")
print(f"Avg. Latency For One Review: {total_latency_per_review:.2f} seconds")
print(f"Simulated Total Time to Process {num_reviews} Reviews: {sequential_total_time:.2f} seconds")
print(f"Simulated Throughput: {sequential_throughput:.2f} reviews/second
")
print("=== CONCLUSION ===")
print(f"Throughput Increase: {throughput_increase:.0f}%")

The results confirm that the assembly‑line pattern dramatically improves throughput while keeping per‑review latency acceptable, making it a core technique for building high‑throughput AI‑agent data‑processing systems.