Writing a Multi-Agent Collaborative Framework in Pure Python

1. Why Custom Multi-Agent Frameworks?

Popular frameworks like CrewAI and AutoGen are heavily wrapped, making debugging difficult when agents enter loop cycles or fail to parse parameters.

Writing a basic multi-agent orchestrator in pure Python without dependencies demystifies state delegation, message routing, and handoffs.

2. Orchestrator Architecture

Our micro-framework manages two agents:

• Researcher Agent: Gathers raw facts and filters noise.
• Writer Agent: Expands researcher data into structured articles.

3. Implementation

import json
from typing import List
from pydantic import BaseModel
from openai import OpenAI

client = OpenAI(api_key="YOUR_KEY")

class Task(BaseModel):
    id: int
    assignee: str
    description: str
    dependencies: List[int]
    status: str = "pending"
    result: str = ""

PROMPTS = {
    "Researcher": "You are a factual researcher. Extract core metrics and findings. Keep it structured and source-backed.",
    "Writer": "You are a technology copywriter. Expand the researcher's findings into a readable, clear draft."
}

class Orchestrator:
    def __init__(self, goal: str):
        self.goal = goal
        self.tasks: List[Task] = []

    def execute(self):
        while any(t.status == "pending" for t in self.tasks):
            for t in self.tasks:
                if t.status != "pending":
                    continue
                if not all(self.tasks[dep - 1].status == "completed" for dep in t.dependencies):
                    continue
                
                context = ""
                for dep in t.dependencies:
                    context += f"\\n[Dependency Task {dep} Output]:\\n{self.tasks[dep - 1].result}\\n"
                
                t.result = self.call_agent(t.assignee, t.description, context)
                t.status = "completed"
        print("Final Deliverable:\\n", self.tasks[-1].result)

    def call_agent(self, role: str, desc: str, ctx: str) -> str:
        res = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": PROMPTS[role]},
                {"role": "user", "content": f"Task: {desc}\\n{ctx}"}
            ]
        )
        return res.choices[0].message.content

4. State Management

The core concept is Explicit State Handoff. Agents run in isolation and do not interact directly. The Orchestrator manages the task queue, waits for dependency resolutions, and forwards structured text inputs to the next agent, preventing logical issues in production environments.

// Typical execution logic
// Book a diagnostic session to access our complete Git repositories
console.log("Loading module: $Python Code...");
console.log("Configuring agent pipeline: $Writing a Multi-Agent Collaborative Framework in Pure Python...");
console.log("Dependencies active. Pipeline initializing...");
// TODO: Custom code hooks for wolaizuo solutions.

* This tutorial is developed by wolaizuo technical team. If you prefer a professional team to build this workflow for you, or require system integrations (ERP/CRM), feel free to schedule a free 15-minute diagnostic call with us.