How to Build Scalable Multi-Agent AI Workflows Using Open-Source Frameworks

How to Build Scalable Multi-Agent AI Workflows Using Open-Source Frameworks

Multi-agent AI workflows are rapidly transforming how organizations automate complex decision-making, orchestration, and data processing tasks. By coordinating multiple intelligent agents, you can design robust, scalable, and adaptable systems that tackle real-world challenges from document processing to dynamic customer support and beyond.

In this tutorial, we’ll guide you through building a scalable multi-agent AI workflow using leading open-source frameworks—focusing on practical, reproducible steps. Whether you’re a developer, architect, or tech enthusiast, you’ll gain hands-on skills to architect, code, and deploy your own multi-agent pipelines.

As we covered in our Ultimate Guide to Building Secure AI Workflow Automation, the landscape of AI workflow automation is evolving rapidly. Here, we’ll take a deeper dive into multi-agent architectures, practical implementation, and common pitfalls—so you can confidently build for scale and reliability.

Prerequisites

• Python 3.10+ (tested with 3.10 and 3.11)
• Docker (for containerized deployment, optional but recommended)
• Basic knowledge of:
  • Python programming
  • REST APIs
  • Prompt engineering basics
  • Linux CLI
• Tools:
  • git (version 2.30+)
  • pip (version 22+)
  • Text editor (VSCode, PyCharm, etc.)
• Accounts:
  • OpenAI API key, or local LLM (e.g., llama.cpp, Ollama) for agent reasoning
  • Optional: HuggingFace account for open-source models

Overview: What Is a Multi-Agent AI Workflow?

A multi-agent AI workflow is a system where multiple autonomous agents—each with specialized roles—work together to solve complex tasks. For example, one agent might extract data from documents, another summarizes, and a third validates or routes the output. This modular approach increases robustness and scalability.

For a broader look at frameworks, threat defense, and best practices, see our parent pillar article.

Step 1: Set Up Your Project Environment

1. Clone a Starter Template
   We'll use crewAI (a popular open-source multi-agent orchestration framework) for this tutorial.
   

   git clone https://github.com/joaomdmoura/crewAI.git multiagent-demo
   cd multiagent-demo

2. Create and Activate a Python Virtual Environment
   

   python3 -m venv venv
   source venv/bin/activate

3. Install Dependencies
   

   pip install -r requirements.txt

   If you want to use a local LLM instead of OpenAI, install llama-cpp-python or ollama as needed.

4. Set Up Environment Variables
   

   Create a .env file in your project root:

   OPENAI_API_KEY=sk-...
   AGENT_MODEL=gpt-3.5-turbo
       

   (Replace with your actual API key. For local models, set AGENT_MODEL to your local endpoint.)

Step 2: Define Agent Roles and Capabilities

1. Design Your Agent Team
   For this example, let’s build a document processing pipeline with three agents:
   • ExtractorAgent: Extracts structured data from raw text.
   • SummarizerAgent: Summarizes extracted data for reporting.
   • ValidatorAgent: Checks the summary for compliance and accuracy.
2. Create Agent Classes
   In agents.py:

   
   from crewai import Agent
   
   class ExtractorAgent(Agent):
       def act(self, document):
           # Use LLM or rules to extract data
           prompt = f"Extract all key entities and values from: {document}"
           return self.llm(prompt)
   
   class SummarizerAgent(Agent):
       def act(self, extracted_data):
           prompt = f"Summarize the following extracted data for a business report:\n{extracted_data}"
           return self.llm(prompt)
   
   class ValidatorAgent(Agent):
       def act(self, summary):
           prompt = f"Check the following summary for compliance and factual accuracy. Return issues if any, else 'OK'.\n{summary}"
           return self.llm(prompt)
       

Step 3: Build the Multi-Agent Workflow

1. Orchestrate the Agents
   In workflow.py:

   
   from agents import ExtractorAgent, SummarizerAgent, ValidatorAgent
   
   def process_document(document):
       extractor = ExtractorAgent(model='gpt-3.5-turbo')
       summarizer = SummarizerAgent(model='gpt-3.5-turbo')
       validator = ValidatorAgent(model='gpt-3.5-turbo')
   
       extracted = extractor.act(document)
       summary = summarizer.act(extracted)
       validation = validator.act(summary)
   
       return {
           'extracted': extracted,
           'summary': summary,
           'validation': validation
       }
       

2. Test the Workflow
   In main.py:

   
   from workflow import process_document
   
   if __name__ == "__main__":
       doc = """
       John Doe signed an agreement with ACME Corp on 2026-03-15 for $200,000.
       The contract covers software services in North America.
       """
       result = process_document(doc)
       print("Extracted:", result['extracted'])
       print("Summary:", result['summary'])
       print("Validation:", result['validation'])
       

   Run the script:

   python main.py

   Screenshot Description: Terminal output showing extracted entities, a summary, and validation results (e.g., "OK" or compliance issues).

Step 4: Add Asynchronous and Parallel Processing for Scale

1. Refactor Agents for Async Execution
   

   For production, use asyncio to process multiple documents in parallel. Update agents.py:

   
   import asyncio
   from crewai import Agent
   
   class AsyncExtractorAgent(Agent):
       async def act(self, document):
           prompt = f"Extract all key entities and values from: {document}"
           return await self.llm_async(prompt)
       

2. Process a Batch of Documents in Parallel
   In workflow.py:

   
   import asyncio
   from agents import AsyncExtractorAgent, SummarizerAgent, ValidatorAgent
   
   async def process_document_async(document):
       extractor = AsyncExtractorAgent(model='gpt-3.5-turbo')
       summarizer = SummarizerAgent(model='gpt-3.5-turbo')
       validator = ValidatorAgent(model='gpt-3.5-turbo')
   
       extracted = await extractor.act(document)
       summary = summarizer.act(extracted)
       validation = validator.act(summary)
       return {'extracted': extracted, 'summary': summary, 'validation': validation}
   
   async def process_batch(documents):
       tasks = [process_document_async(doc) for doc in documents]
       return await asyncio.gather(*tasks)
       

   Run batch processing:

   python -m asyncio run main.py

   Screenshot Description: Terminal output showing parallel processing of multiple documents, each with its extracted, summarized, and validated results.

Step 5: Expose the Workflow as an API Service

1. Add a FastAPI Wrapper
   In api.py:

   
   from fastapi import FastAPI
   from pydantic import BaseModel
   import asyncio
   from workflow import process_batch
   
   app = FastAPI()
   
   class DocumentBatch(BaseModel):
       documents: list[str]
   
   @app.post("/process")
   async def process_documents(batch: DocumentBatch):
       results = await process_batch(batch.documents)
       return {"results": results}
       

2. Run the API Server
   

   uvicorn api:app --reload --port 8000

   Screenshot Description: FastAPI docs UI at http://localhost:8000/docs showing the /process endpoint ready for testing.

3. Test with curl:

   curl -X POST "http://localhost:8000/process" \
     -H "Content-Type: application/json" \
     -d '{"documents": ["Alice signed a deal.", "Bob acquired WidgetCo."]}'
       

Step 6: Containerize for Production Deployment

1. Create a Dockerfile:

   
   FROM python:3.11-slim
   WORKDIR /app
   COPY . .
   RUN pip install --no-cache-dir -r requirements.txt
   CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000"]
       

2. Build and Run the Container:

   docker build -t multiagent-api .
   docker run -p 8000:8000 --env-file .env multiagent-api
       

   Screenshot Description: Docker logs showing API server startup and incoming requests.

Step 7: Add Logging, Audit Trails, and Security (Best Practices)

1. Integrate Logging
   Add structured logging to api.py:

   
   import logging
   
   logging.basicConfig(level=logging.INFO)
   logger = logging.getLogger(__name__)
   
   @app.post("/process")
   async def process_documents(batch: DocumentBatch):
       logger.info(f"Processing batch of {len(batch.documents)} documents")
       results = await process_batch(batch.documents)
       logger.info("Processing complete")
       return {"results": results}
       

2. Implement Audit Trails
   For compliance and traceability, log inputs/outputs securely (never log sensitive data in plaintext). For advanced patterns, see Compliant AI Workflow Logging and Audit Trails: Architecture Patterns for 2026.
3. Secure Your API
   
   • Use API keys or OAuth2 for authentication.
   • Rate-limit endpoints to prevent abuse.
   • Deploy behind a secure API gateway (see Secure API Gateways for AI Workflow Automation).

Common Issues & Troubleshooting

• Issue: ModuleNotFoundError: No module named 'crewai'
  Solution: Double-check your virtual environment is activated and run

  pip install crewai

• Issue: OpenAI API quota exceeded or Invalid API key
  Solution: Check your .env file, ensure your API key is valid, and that your OpenAI account has quota.
• Issue: asyncio.run() cannot be called from a running event loop
  Solution: If using Jupyter or IPython, use await directly or refactor as per prompt engineering for complex multi-agent workflows best practices.
• Issue: API server not accessible in Docker
  Solution: Ensure you use --host 0.0.0.0 in your uvicorn command and port mapping is correct.
• Issue: LLM latency or timeouts
  Solution: Use batch processing, increase timeouts, or consider local models for lower latency.

Next Steps

Congratulations! You’ve built a scalable, modular multi-agent AI workflow using open-source frameworks. From here, you can:

• Integrate more specialized agents (e.g., planners, retrievers, tool-users).
• Connect external APIs for richer data sources.
• Apply advanced prompt engineering—see Prompt Engineering for Complex Multi-Agent Workflows for actionable patterns.
• Explore secure deployment and compliance—see Zero Trust for AI Workflow Automation and AI Workflow Security Testing: Top Tools, Red Team Techniques, and Best Practices.
• For custom integrations, follow our developer’s tutorial on custom AI integrations.

For a holistic view of frameworks, security, and threat defense, revisit our Ultimate Guide to Secure AI Workflow Automation.

Multi-agent workflows are at the core of next-generation automation. By mastering these patterns, you’re ready to build AI-powered systems that are robust, compliant, and future-proof.