Pipeline Multi-Agent Pattern

What Is the Pipeline Pattern?

In the pipeline pattern, agents are arranged in a linear sequence. Each agent receives the output of the previous agent, transforms it, and passes it to the next. The data flows in one direction: no cycles, no branching.

Input → [Agent 1] → [Agent 2] → [Agent 3] → Output
          Research    Analyze     Write

This is the simplest multi-agent topology. It maps naturally to workflows that have clear, ordered stages.

When to Use Pipelines

Use when:

The task has clear sequential stages (research → outline → write → review)
Each stage requires different specialization or a different context
You want to validate output at each step before continuing
Stages can be independently tested and replaced

Avoid when:

Stages can be parallelized (use supervisor with parallel workers)
Steps depend on each other in complex ways (use graph-based control flow like LangGraph)

Typed Interfaces Between Stages

The key to a robust pipeline is typed boundaries. Each stage has a defined input and output schema. If a stage fails to produce the expected output, the pipeline stops with a clear error — not silent garbage propagating downstream.

Python

# pharmabot/pipeline/schemas.py
from pydantic import BaseModel, Field

class ResearchOutput(BaseModel):
    topic: str
    key_facts: list[str]
    sources: list[str]
    gaps: list[str] = Field(default_factory=list)

class AnalysisOutput(BaseModel):
    topic: str
    risk_level: str  # "low" | "medium" | "high"
    key_findings: list[str]
    contraindications: list[str]
    confidence: float

class ContentOutput(BaseModel):
    title: str
    summary: str
    full_content: str
    citations: list[str]

Building the Pipeline

Python

# pharmabot/pipeline/drug_content_pipeline.py
from openai import AsyncAzureOpenAI
from pharmabot.pipeline.schemas import ResearchOutput, AnalysisOutput, ContentOutput
import structlog

log = structlog.get_logger()

class DrugContentPipeline:
    """
    3-stage pipeline for generating drug information content:
    1. Research: gather facts about the drug
    2. Analyze: assess clinical significance
    3. Write: produce patient-facing content
    """

    def __init__(self, client: AsyncAzureOpenAI):
        self.client = client

    async def _call_stage(
        self,
        stage_name: str,
        system_prompt: str,
        user_content: str,
        output_schema: type,
    ):
        """Generic stage runner with structured output."""
        log.info("pipeline_stage_start", stage=stage_name)

        response = await self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_content},
            ],
            response_format={"type": "json_object"},
            temperature=0.2,
        )

        raw = response.choices[0].message.content
        result = output_schema.model_validate_json(raw)

        log.info("pipeline_stage_complete", stage=stage_name)
        return result

    async def research(self, drug_name: str) -> ResearchOutput:
        return await self._call_stage(
            stage_name="research",
            system_prompt="""You are a pharmaceutical researcher.
            Extract factual information about the given drug.
            Return JSON matching ResearchOutput schema:
            {topic, key_facts (list), sources (list), gaps (list)}""",
            user_content=f"Research the drug: {drug_name}",
            output_schema=ResearchOutput,
        )

    async def analyze(self, research: ResearchOutput) -> AnalysisOutput:
        return await self._call_stage(
            stage_name="analyze",
            system_prompt="""You are a clinical safety analyst.
            Analyze the research findings for clinical significance.
            Return JSON matching AnalysisOutput schema:
            {topic, risk_level, key_findings (list), contraindications (list), confidence (0-1)}""",
            user_content=(
                f"Analyze this research:\n"
                f"Topic: {research.topic}\n"
                f"Facts: {research.key_facts}\n"
                f"Gaps: {research.gaps}"
            ),
            output_schema=AnalysisOutput,
        )

    async def write(
        self,
        research: ResearchOutput,
        analysis: AnalysisOutput,
    ) -> ContentOutput:
        return await self._call_stage(
            stage_name="write",
            system_prompt="""You are a medical content writer for a patient-facing app.
            Write clear, accurate, safe content about the drug.
            Always include safety warnings. Never give dosage without mentioning to consult a pharmacist.
            Return JSON matching ContentOutput schema:
            {title, summary, full_content, citations (list)}""",
            user_content=(
                f"Write patient-facing content about: {research.topic}\n\n"
                f"Key facts: {research.key_facts}\n"
                f"Risk level: {analysis.risk_level}\n"
                f"Key findings: {analysis.key_findings}\n"
                f"Contraindications: {analysis.contraindications}"
            ),
            output_schema=ContentOutput,
        )

    async def run(self, drug_name: str) -> ContentOutput:
        """Execute the full pipeline."""
        log.info("pipeline_start", drug=drug_name)

        # Stage 1
        research_result = await self.research(drug_name)

        # Stage 2 — receives Stage 1 output
        analysis_result = await self.analyze(research_result)

        # Stage 3 — receives Stage 1 and 2 outputs
        content_result = await self.write(research_result, analysis_result)

        log.info("pipeline_complete", drug=drug_name, title=content_result.title)
        return content_result

Error Propagation

When a stage fails, the error should be caught and handled — not silently pass garbage:

Python

async def run_safe(self, drug_name: str) -> ContentOutput | None:
    """Pipeline with explicit error handling at each stage."""
    try:
        research_result = await self.research(drug_name)
    except Exception as e:
        log.error("pipeline_stage_failed", stage="research", error=str(e))
        return None  # or raise a PipelineError

    try:
        analysis_result = await self.analyze(research_result)
    except Exception as e:
        log.error("pipeline_stage_failed", stage="analyze", error=str(e))
        # Fallback: skip analysis, write from research only
        return await self.write_from_research_only(research_result)

    try:
        return await self.write(research_result, analysis_result)
    except Exception as e:
        log.error("pipeline_stage_failed", stage="write", error=str(e))
        return None

Testing Pipeline Stages Independently

Because stages have typed interfaces, each can be unit-tested in isolation:

Python

# tests/test_pipeline.py
import pytest
from pharmabot.pipeline.drug_content_pipeline import DrugContentPipeline
from pharmabot.pipeline.schemas import ResearchOutput

@pytest.mark.asyncio
async def test_analyze_stage(mock_client):
    """Test analyze stage in isolation with mock research data."""
    pipeline = DrugContentPipeline(client=mock_client)

    mock_research = ResearchOutput(
        topic="Ibuprofen",
        key_facts=["NSAID", "inhibits COX-1 and COX-2", "reduces inflammation"],
        sources=["FDA label"],
        gaps=["long-term safety in elderly"],
    )

    result = await pipeline.analyze(mock_research)

    assert result.risk_level in ("low", "medium", "high")
    assert isinstance(result.confidence, float)
    assert 0.0 <= result.confidence <= 1.0
    assert isinstance(result.key_findings, list)

Pipeline vs Single Long Prompt

Why not just put all three stages in one giant prompt?

| | Single Prompt | Pipeline | |---|---|---| | Context usage | One call, but huge context | Three focused calls | | Intermediate validation | No checkpoints | Validate at each stage | | Debugging | Hard to isolate failures | Clear which stage failed | | Stage replacement | Rewrite everything | Swap one stage | | Token efficiency | Lower (all context in one call) | Higher (each stage focused) | | Latency | Lower (one call) | Higher (three serial calls) |

For simple tasks: use a single prompt. For complex multi-stage workflows where quality and debuggability matter: use the pipeline pattern.

Pipeline Multi-Agent Pattern

What Is the Pipeline Pattern?

When to Use Pipelines

Typed Interfaces Between Stages

Building the Pipeline

Error Propagation

Testing Pipeline Stages Independently

Pipeline vs Single Long Prompt

Enjoyed this article?

Leave a comment