多智能体系统完全指南【2026】- 多Agent协作的核心原理与实战

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='高级研究员',
    goal='深入研究指定主题，收集全面准确的信息',
    backstory='''你是一位经验丰富的研究专家，擅长从多个来源
    收集和验证信息。你注重数据的准确性和时效性。''',
    verbose=True,
    allow_delegation=False,
    llm=llm
)

analyst = Agent(
    role='数据分析师',
    goal='分析研究数据，提取关键洞察和趋势',
    backstory='''你是一位资深数据分析师，擅长发现数据中的
    模式和趋势。你能够将复杂数据转化为可操作的洞察。''',
    verbose=True,
    allow_delegation=False,
    llm=llm
)

writer = Agent(
    role='内容撰稿人',
    goal='将分析结果转化为清晰易懂的报告',
    backstory='''你是一位专业的技术写作专家，擅长将复杂的
    技术内容转化为易于理解的文章。''',
    verbose=True,
    allow_delegation=False,
    llm=llm
)

research_task = Task(
    description='''研究2026年多智能体系统的最新发展趋势，
    包括主流框架、应用场景和技术突破。''',
    expected_output='详细的研究报告，包含数据来源和关键发现',
    agent=researcher
)

analysis_task = Task(
    description='''基于研究数据，分析多Agent系统的发展方向，
    识别主要趋势和潜在机会。''',
    expected_output='趋势分析报告，包含数据可视化建议',
    agent=analyst
)

writing_task = Task(
    description='''整合研究和分析结果，撰写一份完整的
    多智能体系统发展报告。''',
    expected_output='结构清晰、内容详实的最终报告',
    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="用户代理",
    human_input_mode="TERMINATE",
    max_consecutive_auto_reply=10,
    code_execution_config={"work_dir": "workspace"}
)

architect = autogen.AssistantAgent(
    name="系统架构师",
    system_message='''你是一位资深系统架构师，负责设计
    多Agent系统的整体架构和技术选型。''',
    llm_config={"config_list": config_list}
)

developer = autogen.AssistantAgent(
    name="开发工程师",
    system_message='''你是一位经验丰富的开发工程师，
    负责实现架构师设计的系统组件。''',
    llm_config={"config_list": config_list}
)

reviewer = autogen.AssistantAgent(
    name="代码审查员",
    system_message='''你是一位严格的代码审查专家，
    负责审查代码质量和安全性。''',
    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="设计并实现一个简单的多Agent任务调度系统"
)

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": ["研究Agent: 完成信息收集"],
        "current_agent": "analyst",
        "task_status": {**state["task_status"], "research": "done"}
    }

def analysis_node(state: MultiAgentState) -> dict:
    return {
        "messages": ["分析Agent: 完成数据分析"],
        "current_agent": "writer",
        "task_status": {**state["task_status"], "analysis": "done"}
    }

def writing_node(state: MultiAgentState) -> dict:
    return {
        "messages": ["撰写Agent: 完成报告撰写"],
        "current_agent": "reviewer",
        "task_status": {**state["task_status"], "writing": "done"}
    }

def review_node(state: MultiAgentState) -> dict:
    if needs_revision(state):
        return {
            "messages": ["审核Agent: 需要修改"],
            "current_agent": "writer"
        }
    return {
        "messages": ["审核Agent: 审核通过"],
        "final_result": "任务完成",
        "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
        }