Multi-Agent AI System Development

Multi-Agent AI System Development

A multi-agent system (MAS) is an architecture where multiple specialized AI agents interact to complete complex tasks impossible for a single agent. Dividing responsibility reduces complexity of each agent, improves quality in specialized tasks, and enables horizontal scaling.

Multi-Agent System Architectures

Supervisor (Orchestrator): central agent distributes tasks among specialized agents and aggregates results. Well managed, but has a bottleneck in orchestrator.

Peer-to-peer (P2P): agents communicate directly without central coordinator. More resilient to failures, harder to debug.

Hierarchical: multi-level structure — upper-level agents manage lower-level agents. Used for very complex workflows.

Pipeline: each agent executes its part and passes result to next. Simple, predictable, limited by linearity.

Supervisor Pattern Implementation with LangGraph

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Literal
import operator
from langchain_core.messages import HumanMessage

class MultiAgentState(TypedDict):
    messages: list
    current_task: str
    task_result: str
    next_agent: str

# Specialized agents
def researcher_agent(state: MultiAgentState) -> MultiAgentState:
    """Agent for information retrieval"""
    llm = ChatOpenAI(model="gpt-4o")
    task = state["current_task"]

    # Retrieval + analysis
    docs = retriever.invoke(task)
    context = "\n".join([d.page_content for d in docs])

    result = llm.invoke([
        HumanMessage(content=f"Research task: {task}\n\nContext:\n{context}\n\nProvide key facts:")
    ]).content

    return {**state, "task_result": result, "next_agent": "writer"}

def writer_agent(state: MultiAgentState) -> MultiAgentState:
    """Agent for text writing"""
    llm = ChatOpenAI(model="gpt-4o")
    research = state["task_result"]
    original_task = state["current_task"]

    result = llm.invoke([
        HumanMessage(content=f"Write answer to task: {original_task}\n\nMaterials: {research}")
    ]).content

    return {**state, "task_result": result, "next_agent": "reviewer"}

def reviewer_agent(state: MultiAgentState) -> MultiAgentState:
    """Agent for quality checking"""
    llm = ChatOpenAI(model="gpt-4o")
    draft = state["task_result"]

    review = llm.invoke([
        HumanMessage(content=f"""Check the following text for:
1. Factual errors
2. Completeness
3. Structure and clarity

Text: {draft}

If OK, reply "APPROVED". Otherwise, specify edits.""")
    ]).content

    if "APPROVED" in review:
        return {**state, "next_agent": "complete"}
    else:
        return {**state, "task_result": review, "next_agent": "writer"}

def supervisor_agent(state: MultiAgentState) -> MultiAgentState:
    """Orchestrator: determines first agent for task"""
    return {**state, "next_agent": "researcher"}

def route_agent(state: MultiAgentState) -> str:
    return state["next_agent"]

# Build graph
graph = StateGraph(MultiAgentState)
graph.add_node("supervisor", supervisor_agent)
graph.add_node("researcher", researcher_agent)
graph.add_node("writer", writer_agent)
graph.add_node("reviewer", reviewer_agent)

graph.set_entry_point("supervisor")
graph.add_conditional_edges("supervisor", route_agent)
graph.add_conditional_edges("researcher", route_agent)
graph.add_conditional_edges("writer", route_agent)
graph.add_conditional_edges("reviewer", lambda s: END if s["next_agent"] == "complete" else s["next_agent"])

mas = graph.compile()

CrewAI: High-Level Framework

from crewai import Agent, Task, Crew, Process

# Define agents with roles
analyst = Agent(
    role="Financial Analyst",
    goal="Analyze financial data and identify trends",
    backstory="Experienced financial analyst with 10 years in investment banking",
    tools=[search_tool, calculator_tool, db_query_tool],
    llm=ChatOpenAI(model="gpt-4o"),
    verbose=True,
)

report_writer = Agent(
    role="Report Writer",
    goal="Create professional financial reports",
    backstory="Business communication specialist with finance experience",
    tools=[document_writer_tool],
    llm=ChatOpenAI(model="gpt-4o"),
)

fact_checker = Agent(
    role="Fact Checker",
    goal="Verify all numbers and statements in report",
    tools=[search_tool, calculator_tool],
    llm=ChatOpenAI(model="gpt-4o"),
)

# Tasks
analysis_task = Task(
    description="Analyze financial metrics for Company X in Q1 2026",
    expected_output="JSON with KPI: revenue, EBITDA, net_profit, growth_rates",
    agent=analyst,
)

report_task = Task(
    description="Create investment memorandum based on analysis",
    expected_output="PDF-ready investment memorandum text",
    agent=report_writer,
    context=[analysis_task],
)

# Crew
crew = Crew(
    agents=[analyst, report_writer, fact_checker],
    tasks=[analysis_task, report_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(inputs={"company": "Example Inc", "period": "Q1 2026"})

Practical Case: Due Diligence Automation System

Task: comprehensive company verification in M&A transaction. Includes legal, financial, operational and HR analysis.

Agent composition:

• Financial Analyst Agent: IFRS/GAAP reporting analysis
• Legal Agent: contract verification, litigation checks
• HR Agent: personnel structure analysis, turnover, key staff
• Risk Agent: consolidated risk analysis
• Report Agent: final Due Diligence report

Infrastructure: LangGraph, each agent with access to specialized RAG indexes.

Results:

• DD timeline reduction: 4 weeks to 3 days (standard cases)
• Coverage (aspect verification ratio): 78% → 94%
• Human-in-the-loop: final validation of each section
• Cost per DD: 67% reduction

Agent Communication

# Pattern: agents pass structured messages through shared state
class AgentMessage:
    source_agent: str
    target_agent: str
    message_type: str  # "request", "result", "error"
    content: dict
    priority: int

# Message queue for async communication
import asyncio
from asyncio import Queue

class AgentCommunicationBus:
    def __init__(self):
        self.queues: dict[str, Queue] = {}

    def register_agent(self, agent_id: str):
        self.queues[agent_id] = Queue()

    async def send(self, msg: AgentMessage):
        await self.queues[msg.target_agent].put(msg)

    async def receive(self, agent_id: str) -> AgentMessage:
        return await self.queues[agent_id].get()

Timeline

• MAS architecture design: 1–2 weeks
• Basic agent development (3–5): 3–5 weeks
• Communication integration and testing: 2–3 weeks
• Monitoring and production hardening: 1–2 weeks
• Total: 7–12 weeks