多智能体协作模式：从层级到对等网络

         [编排者智能体]
        /      |      \
[研究智能体] [编码智能体] [测试智能体]

import autogen

# 配置模型
config_list = [{
    "model": "gpt-4",
    "api_key": "your-api-key"
}]

# 创建编排者
orchestrator = autogen.AssistantAgent(
    name="编排者",
    system_message="""你是项目编排者。职责：
    1. 理解用户需求
    2. 将需求分解为子任务
    3. 分配给合适的专家智能体
    4. 汇总结果并确保质量
    
    可用的专家：
    - 研究员：负责信息搜集和调研
    - 工程师：负责代码编写
    - 测试员：负责代码测试
    
    请用以下格式分配任务：
    [TO: 专家名] 任务描述""",
    llm_config={"config_list": config_list}
)

# 创建专家智能体
researcher = autogen.AssistantAgent(
    name="研究员",
    system_message="你是研究专家。接收编排者的任务，进行信息搜集和分析。",
    llm_config={"config_list": config_list}
)

engineer = autogen.AssistantAgent(
    name="工程师",
    system_message="你是资深工程师。接收编排者的任务，编写高质量代码。",
    llm_config={"config_list": config_list}
)

tester = autogen.AssistantAgent(
    name="测试员",
    system_message="你是测试工程师。接收编排者的任务，编写和执行测试。",
    llm_config={"config_list": config_list}
)

# 用户代理
user_proxy = autogen.UserProxyAgent(
    name="用户",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10
)

# 建立群聊
groupchat = autogen.GroupChat(
    agents=[user_proxy, orchestrator, researcher, engineer, tester],
    messages=[],
    max_round=30
)

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

# 启动协作
user_proxy.initiate_chat(
    manager,
    message="开发一个 Python 命令行工具，可以统计指定目录下各类文件的数量。"
)

[智能体A] ←→ [智能体B]
    ↕           ↕
[智能体C] ←→ [智能体D]

from crewai import Agent, Task, Crew, Process

# 定义智能体——每个都有不同的专业视角
product_manager = Agent(
    role='产品经理',
    goal='确保产品方向正确，满足用户需求',
    backstory='10年产品管理经验，擅长用户研究和需求分析',
    verbose=True
)

designer = Agent(
    role='设计师',
    goal='确保用户体验优秀，界面美观',
    backstory='资深 UI/UX 设计师，曾设计过多款百万用户产品',
    verbose=True
)

engineer = Agent(
    role='工程师',
    goal='确保技术方案可行，代码质量高',
    backstory='全栈工程师，精通前后端开发和系统架构',
    verbose=True
)

# 定义任务——每个智能体从自己的角度贡献
pm_task = Task(
    description='分析需求可行性，输出需求文档',
    agent=product_manager,
    expected_output='需求分析文档，包含用户故事和验收标准'
)

design_task = Task(
    description='基于需求文档设计界面和交互方案',
    agent=designer,
    expected_output='设计稿和交互说明文档',
    context=[pm_task]  # 依赖产品经理的输出
)

tech_task = Task(
    description='评估技术方案，编写核心代码',
    agent=engineer,
    expected_output='技术方案文档和可运行代码',
    context=[pm_task, design_task]  # 依赖前两者的输出
)

# 创建协作团队——层级流程
crew = Crew(
    agents=[product_manager, designer, engineer],
    tasks=[pm_task, design_task, tech_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff()

class SharedBlackboard:
    def __init__(self):
        self.state = {
            "facts": {},        # 已确认的事实
            "hypotheses": {},   # 假设
            "decisions": {},    # 决策记录
            "artifacts": {},    # 产出物
            "messages": []      # 消息历史
        }
        self.lock = threading.Lock()
    
    def write(self, key: str, value: str, agent: str):
        with self.lock:
            self.state[key][agent] = {
                "value": value,
                "timestamp": time.time()
            }
            self.state["messages"].append({
                "from": agent,
                "key": key,
                "value": value,
                "timestamp": time.time()
            })
    
    def read(self, key: str) -> dict:
        with self.lock:
            return self.state.get(key, {}).copy()
    
    def get_updates_since(self, timestamp: float) -> list:
        with self.lock:
            return [
                m for m in self.state["messages"]
                if m["timestamp"] > timestamp
            ]


class PeerAgent:
    def __init__(self, name: str, expertise: str, 
                 blackboard: SharedBlackboard):
        self.name = name
        self.expertise = expertise
        self.bb = blackboard
        self.last_read = time.time()
    
    def observe(self) -> list:
        """观察黑板上的新信息"""
        updates = self.bb.get_updates_since(self.last_read)
        self.last_read = time.time()
        return updates
    
    def contribute(self, key: str, value: str):
        """向黑板贡献信息"""
        self.bb.write(key, value, self.name)
    
    def deliberate(self) -> dict:
        """基于黑板信息进行推理"""
        all_facts = self.bb.read("facts")
        all_hypotheses = self.bb.read("hypotheses")
        
        # 基于自己的专业视角分析
        analysis = self._analyze(all_facts, all_hypotheses)
        return analysis

[需求分析] → [方案设计] → [编码实现] → [代码审查] → [测试验证]

from dataclasses import dataclass
from typing import Any, Optional

@dataclass
class PipelineContext:
    """在流水线中传递的上下文"""
    original_request: str
    requirements_doc: Optional[str] = None
    design_doc: Optional[str] = None
    code: Optional[str] = None
    review_feedback: Optional[str] = None
    test_results: Optional[dict] = None
    errors: list = None
    
    def __post_init__(self):
        if self.errors is None:
            self.errors = []


class PipelineStage:
    def __init__(self, name: str, agent, 
                 input_key: str, output_key: str):
        self.name = name
        self.agent = agent
        self.input_key = input_key
        self.output_key = output_key
    
    def execute(self, ctx: PipelineContext) -> PipelineContext:
        input_value = getattr(ctx, self.input_key)
        
        prompt = f"""
        阶段：{self.name}
        输入：{input_value}
        
        请完成你的工作并输出结果。
        """
        
        result = self.agent.generate(prompt)
        setattr(ctx, self.output_key, result)
        
        print(f"[{self.name}] 完成")
        return ctx


class AgentPipeline:
    def __init__(self, stages: list[PipelineStage]):
        self.stages = stages
    
    def run(self, request: str) -> PipelineContext:
        ctx = PipelineContext(original_request=request)
        
        for stage in self.stages:
            try:
                ctx = stage.execute(ctx)
            except Exception as e:
                ctx.errors.append({
                    "stage": stage.name,
                    "error": str(e)
                })
                # 可以选择中断或继续
                break
        
        return ctx


# 构建开发流水线
pipeline = AgentPipeline([
    PipelineStage("需求分析", analyst_agent, 
                  "original_request", "requirements_doc"),
    PipelineStage("方案设计", designer_agent,
                  "requirements_doc", "design_doc"),
    PipelineStage("编码实现", coder_agent,
                  "design_doc", "code"),
    PipelineStage("代码审查", reviewer_agent,
                  "code", "review_feedback"),
    PipelineStage("测试验证", tester_agent,
                  "code", "test_results"),
])

result = pipeline.run("开发一个待办事项管理 CLI 工具")

       [问题]
      /   |   \
[A方案] [B方案] [C方案]
      \   |   /
    [评审/融合]
       [最终方案]

class CompetitiveCollaboration:
    def __init__(self, solver_agents: list, judge_agent):
        self.solvers = solver_agents
        self.judge = judge_agent
    
    def solve(self, problem: str) -> dict:
        # 阶段1：并行求解
        solutions = {}
        for agent in self.solvers:
            solution = agent.generate(
                f"请解决以下问题，给出你的最佳方案：\n{problem}"
            )
            solutions[agent.name] = solution
        
        # 阶段2：评审打分
        review_prompt = f"""
        问题：{problem}
        
        以下是多个方案的解决方案：
        {self._format_solutions(solutions)}
        
        请作为评审专家：
        1. 对每个方案打分（1-10）
        2. 指出各方案的优缺点
        3. 选择最佳方案，或综合各方案优点给出最终方案
        """
        
        review = self.judge.generate(review_prompt)
        
        return {
            "solutions": solutions,
            "review": review,
            "final": extract_final_solution(review)
        }

         [编排者]
        /    |    \
   [开发流水线] [调研流水线] [运维流水线]

[智能体A] ←→ [智能体B] ←→ [智能体C]
                 ↓
          [竞争：独立提出方案]
                 ↓
          [投票/评审：选择最佳]

from dataclasses import dataclass, asdict
from enum import Enum
import json

class MessageType(Enum):
    TASK_ASSIGNMENT = "task_assignment"
    RESULT_REPORT = "result_report"
    QUESTION = "question"
    FEEDBACK = "feedback"
    HANDOFF = "handoff"

@dataclass
class AgentMessage:
    sender: str
    receiver: str  # "broadcast" for all
    type: MessageType
    content: str
    metadata: dict  # 附件、引用等
    timestamp: float
    reply_to: str = None  # 回复的消息ID
    
    def to_json(self) -> str:
        return json.dumps({
            **asdict(self),
            "type": self.type.value
        }, ensure_ascii=False, indent=2)

devils_advocate = Agent(
    role='魔鬼代言人',
    goal='挑战和质疑其他智能体的方案',
    backstory='你天生喜欢挑毛病。你的存在是为了防止群体思维。'
              '请尖锐但建设性地质疑每个方案。',
)

class ConversationManager:
    def __init__(self, max_rounds: int = 20):
        self.max_rounds = max_rounds
        self.round = 0
        self.consensus_threshold = 0.8
    
    def should_continue(self, messages: list) -> bool:
        self.round += 1
        if self.round >= self.max_rounds:
            return False
        
        # 检查共识度
        agreement = self._measure_agreement(messages[-5:])
        if agreement > self.consensus_threshold:
            return False
        
        return True

def maintain_shared_context(blackboard: SharedBlackboard):
    """每N轮总结一次共享上下文"""
    recent_messages = blackboard.get_recent(n=10)
    summary = summarizer_agent.generate(
        f"总结以下讨论的关键决策和待解决问题：\n{recent_messages}"
    )
    blackboard.write("context_summary", summary, "system")