多智能体系统：何时构建，以及如何构建

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

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) -> Optional[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):
        raise NotImplementedError

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

# 模型标识符是易腐信息，请通过环境变量设置当前值。
llm = ChatOpenAI(model=os.environ.get("OPENAI_MODEL", "gpt-4o"))

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="研究当前多 Agent 协调的主流做法。",
    expected_output="一份带来源和出处的研究报告，含关键发现",
    agent=researcher,
)
analysis_task = Task(
    description="识别研究中最强和最弱的论断。",
    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 os
import autogen

config_list = [{
    "model": os.environ.get("OPENAI_MODEL", "gpt-4o"),
    "api_key": os.environ["OPENAI_API_KEY"],
}]

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

architect = autogen.AssistantAgent(
    name="architect",
    system_message="你负责设计整体结构，并为每个选择给出理由。",
    llm_config={"config_list": config_list},
)

developer = autogen.AssistantAgent(
    name="developer",
    system_message="你实现架构师的设计，并说明你的假设。",
    llm_config={"config_list": config_list},
)

reviewer = autogen.AssistantAgent(
    name="reviewer",
    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": ["researcher: 信息已收集"],
        "current_agent": "analyst",
        "task_status": {**state["task_status"], "research": "done"},
    }

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

def writing_node(state: MultiAgentState) -> dict:
    return {
        "messages": ["writer: 初稿已完成"],
        "current_agent": "reviewer",
        "task_status": {**state["task_status"], "writing": "done"},
    }

def review_node(state: MultiAgentState) -> dict:
    if needs_revision(state):
        return {"messages": ["reviewer: 需要修改"], "current_agent": "writer"}
    return {
        "messages": ["reviewer: 已通过"],
        "final_result": "任务完成",
        "current_agent": "end",
    }

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

def needs_revision(state) -> bool:
    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": "",
})