Building a Multi-Agent AI Workflow with Python, LangChain, and FastAPI

Why Single-Prompt AI is Failing in Production

Most tutorials show you a single chatbot responding to user prompts. But in real-world software engineering, complex business tasks need collaboration across specialized roles — just like a software team has researchers, developers, and QA engineers.

In this tutorial, we build an Autonomous Multi-Agent Workflow Engine using Python, Pydantic v2, and FastAPI with real-time Server-Sent Events (SSE) streaming.

🏗️ Architecture Overview

Our system coordinates three autonomous agents:

Researcher Agent: Gathers and synthesizes structured domain intelligence.
Writer Agent: Transforms research briefings into technical reports.
Reviewer Agent: Performs automated quality evaluation and refines outputs in an iterative loop.

📦 1. Installation & Environment

pip install fastapi uvicorn pydantic python-dotenv langchain

🧠 2. Implementing Specialized Agents

import asyncio
from pydantic import BaseModel, Field

class ReviewResult(BaseModel):
    score: int = Field(description="Score from 1 to 10 evaluating quality", ge=1, le=10)
    feedback: str = Field(description="Critique and suggestions")
    improved_version: str = Field(description="Enhanced output")

class ResearchAgent:
    async def execute(self, topic: str) -> str:
        await asyncio.sleep(1) # Simulating tool use
        return f"Key insights, trade-offs, and toolchain for: {topic}"

class WriterAgent:
    async def execute(self, topic: str, research_data: str) -> str:
        await asyncio.sleep(1)
        return f"# Technical Report: {topic}\n\n## Insights\n{research_data}"

class ReviewerAgent:
    async def execute(self, draft: str) -> ReviewResult:
        await asyncio.sleep(1)
        return ReviewResult(
            score=9,
            feedback="Strong coverage and clear structure.",
            improved_version=draft + "\n\n*Verified: Conforms to Production Standards.*"
        )

🔄 3. Building the Multi-Agent Orchestrator

class MultiAgentOrchestrator:
    def __init__(self):
        self.researcher = ResearchAgent()
        self.writer = WriterAgent()
        self.reviewer = ReviewerAgent()

    async def run(self, topic: str, max_iterations: int = 2):
        research = await self.researcher.execute(topic)
        draft = await self.writer.execute(topic, research)

        final_content = draft
        for _ in range(max_iterations):
            review = await self.reviewer.execute(final_content)
            final_content = review.improved_version
            if review.score >= 8:
                break

        return {"topic": topic, "report": final_content, "score": review.score}

⚡ 4. Exposing FastAPI with Streaming (SSE)

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

app = FastAPI(title="Multi-Agent AI Engine")
orchestrator = MultiAgentOrchestrator()

@app.post("/api/workflow/stream")
async def stream_workflow(topic: str):
    async def event_stream():
        yield f"data: {json.dumps({'stage': 'RESEARCH', 'message': 'Researching...' })}\n\n"
        result = await orchestrator.run(topic)
        yield f"data: {json.dumps({'stage': 'COMPLETE', 'result': result})}\n\n"

    return StreamingResponse(event_stream(), media_type="text/event-stream")