Multi-Agent System Complete Guide [2026] - Core Principles and Practice of Agent Collaboration

from enum import Enum
from dataclasses import dataclass
from typing import List, Dict, Any

class CoordinationType(Enum):
    CONTRACT_NET = "contract_net"
    VOTING = "voting"
    AUCTION = "auction"
    NEGOTIATION = "negotiation"

@dataclass
class Task:
    id: str
    description: str
    requirements: List[str]
    priority: int

@dataclass
class Bid:
    agent_id: str
    task_id: str
    capability_score: float
    estimated_time: float

class ContractNetProtocol:
    def __init__(self, manager_agent):
        self.manager = manager_agent
        self.bids: Dict[str, List[Bid]] = {}
    
    def announce_task(self, task: Task) -> None:
        self.bids[task.id] = []
        for agent in self.get_available_agents():
            agent.receive_announcement(task)
    
    def collect_bid(self, bid: Bid) -> None:
        if bid.task_id in self.bids:
            self.bids[bid.task_id].append(bid)
    
    def award_contract(self, task_id: str) -> str:
        bids = self.bids.get(task_id, [])
        if not bids:
            return None
        best_bid = max(bids, key=lambda b: b.capability_score / b.estimated_time)
        return best_bid.agent_id
    
    def get_available_agents(self):
        pass

from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4-turbo", temperature=0.7)

researcher = Agent(
    role='Senior Researcher',
    goal='Conduct in-depth research on specified topics, collecting comprehensive and accurate information',
    backstory='''You are an experienced research expert skilled at collecting
    and verifying information from multiple sources. You prioritize data accuracy and timeliness.''',
    verbose=True,
    allow_delegation=False,
    llm=llm
)

analyst = Agent(
    role='Data Analyst',
    goal='Analyze research data, extract key insights and trends',
    backstory='''You are a senior data analyst skilled at discovering patterns
    and trends in data. You can transform complex data into actionable insights.''',
    verbose=True,
    allow_delegation=False,
    llm=llm
)

writer = Agent(
    role='Content Writer',
    goal='Transform analysis results into clear and understandable reports',
    backstory='''You are a professional technical writing expert skilled at
    transforming complex technical content into easy-to-understand articles.''',
    verbose=True,
    allow_delegation=False,
    llm=llm
)

research_task = Task(
    description='''Research the latest development trends in multi-agent systems for 2026,
    including mainstream frameworks, application scenarios, and technological breakthroughs.''',
    expected_output='Detailed research report with data sources and key findings',
    agent=researcher
)

analysis_task = Task(
    description='''Based on research data, analyze the development direction of multi-agent systems,
    identify major trends and potential opportunities.''',
    expected_output='Trend analysis report with data visualization suggestions',
    agent=analyst
)

writing_task = Task(
    description='''Integrate research and analysis results to write a complete
    multi-agent system development report.''',
    expected_output='Well-structured, detailed final report',
    agent=writer
)

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff()
print(result)

import autogen

config_list = [{"model": "gpt-4-turbo", "api_key": "your-api-key"}]

user_proxy = autogen.UserProxyAgent(
    name="User Proxy",
    human_input_mode="TERMINATE",
    max_consecutive_auto_reply=10,
    code_execution_config={"work_dir": "workspace"}
)

architect = autogen.AssistantAgent(
    name="System Architect",
    system_message='''You are a senior system architect responsible for designing
    the overall architecture and technology selection for multi-agent systems.''',
    llm_config={"config_list": config_list}
)

developer = autogen.AssistantAgent(
    name="Development Engineer",
    system_message='''You are an experienced development engineer responsible
    for implementing system components designed by the architect.''',
    llm_config={"config_list": config_list}
)

reviewer = autogen.AssistantAgent(
    name="Code Reviewer",
    system_message='''You are a strict code review expert responsible
    for reviewing code quality and security.''',
    llm_config={"config_list": config_list}
)

groupchat = autogen.GroupChat(
    agents=[user_proxy, architect, developer, reviewer],
    messages=[],
    max_round=20
)

manager = autogen.GroupChatManager(
    groupchat=groupchat,
    llm_config={"config_list": config_list}
)

user_proxy.initiate_chat(
    manager,
    message="Design and implement a simple multi-agent task scheduling system"
)

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

class MultiAgentState(TypedDict):
    messages: Annotated[List[str], operator.add]
    current_agent: str
    task_status: dict
    final_result: str

def research_node(state: MultiAgentState) -> dict:
    return {
        "messages": ["Research Agent: Information collection complete"],
        "current_agent": "analyst",
        "task_status": {**state["task_status"], "research": "done"}
    }

def analysis_node(state: MultiAgentState) -> dict:
    return {
        "messages": ["Analysis Agent: Data analysis complete"],
        "current_agent": "writer",
        "task_status": {**state["task_status"], "analysis": "done"}
    }

def writing_node(state: MultiAgentState) -> dict:
    return {
        "messages": ["Writing Agent: Report writing complete"],
        "current_agent": "reviewer",
        "task_status": {**state["task_status"], "writing": "done"}
    }

def review_node(state: MultiAgentState) -> dict:
    if needs_revision(state):
        return {
            "messages": ["Review Agent: Revision needed"],
            "current_agent": "writer"
        }
    return {
        "messages": ["Review Agent: Approved"],
        "final_result": "Task complete",
        "current_agent": "end"
    }

def route_after_review(state: MultiAgentState) -> str:
    if state["current_agent"] == "end":
        return END
    return state["current_agent"]

def needs_revision(state):
    return False

workflow = StateGraph(MultiAgentState)

workflow.add_node("researcher", research_node)
workflow.add_node("analyst", analysis_node)
workflow.add_node("writer", writing_node)
workflow.add_node("reviewer", review_node)

workflow.set_entry_point("researcher")
workflow.add_edge("researcher", "analyst")
workflow.add_edge("analyst", "writer")
workflow.add_edge("writer", "reviewer")
workflow.add_conditional_edges("reviewer", route_after_review)

app = workflow.compile()

result = app.invoke({
    "messages": [],
    "current_agent": "researcher",
    "task_status": {},
    "final_result": ""
})

from dataclasses import dataclass
from typing import Any, Optional
from datetime import datetime
from enum import Enum

class MessageType(Enum):
    REQUEST = "request"
    RESPONSE = "response"
    BROADCAST = "broadcast"
    ACK = "acknowledgment"

@dataclass
class AgentMessage:
    sender: str
    receiver: str
    msg_type: MessageType
    content: Any
    timestamp: datetime
    correlation_id: Optional[str] = None
    
    def to_dict(self) -> dict:
        return {
            "sender": self.sender,
            "receiver": self.receiver,
            "type": self.msg_type.value,
            "content": self.content,
            "timestamp": self.timestamp.isoformat(),
            "correlation_id": self.correlation_id
        }