Multi-Agent 系统设计与协作

# 单 Agent vs Multi-Agent 的代码结构对比

# ====== 单 Agent 模式 ======
class SingleAgent:
    """一个 Agent 干所有事"""
    def __init__(self, llm):
        self.llm = llm
    
    def handle_task(self, task: str) -> str:
        # 同一个 LLM 处理搜索、编码、审查...
        # prompt 里要包含所有领域的指令
        return self.llm(f"你是全能助手，请完成：{task}")

# ====== Multi-Agent 模式 ======
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class AgentMessage:
    """Agent 之间传递的消息"""
    sender: str
    content: str
    message_type: str  # "request", "response", "feedback"
    metadata: dict = None

class AgentRole:
    """定义 Agent 的角色"""
    def __init__(self, name: str, system_prompt: str, llm):
        self.name = name
        self.system_prompt = system_prompt
        self.llm = llm
        self.message_history: List[AgentMessage] = []
    
    def process(self, message: AgentMessage) -> AgentMessage:
        prompt = f"{self.system_prompt}\n\n收到消息：{message.content}"
        response = self.llm(prompt)
        self.message_history.append(message)
        return AgentMessage(
            sender=self.name, content=response,
            message_type="response", metadata={"source_role": self.name}
        )

# 创建专业 Agent 团队
researcher = AgentRole("Researcher", 
    "你是一个专业研究员，负责收集和分析信息。", llm)
coder = AgentRole("Coder",
    "你是一个资深工程师，负责编写高质量代码。", llm)
reviewer = AgentRole("Reviewer",
    "你是一个严格的代码审查员，负责发现潜在问题。", llm)

# Multi-Agent 协作管道（Pipeline）
from typing import Callable, List

class AgentPipeline:
    """将多个 Agent 串联成处理管道"""
    def __init__(self):
        self.stages: List[AgentRole] = []
    
    def add_stage(self, agent: AgentRole):
        self.stages.append(agent)
        return self
    
    def execute(self, initial_input: str) -> str:
        current_input = initial_input
        for stage in self.stages:
            msg = AgentMessage(
                sender="Pipeline", content=current_input,
                message_type="request"
            )
            response = stage.process(msg)
            current_input = response.content
            print(f"[{stage.name}] -> {current_input[:80]}...")
        return current_input

# 构建处理管道
pipeline = (AgentPipeline()
    .add_stage(researcher)    # 先研究
    .add_stage(coder)         # 再编码
    .add_stage(reviewer)      # 最后审查
)

result = pipeline.execute("实现一个支持并发请求的 HTTP 客户端")

# 基于发布-订阅模式的 Agent 消息总线
import asyncio
from typing import Callable, Dict, List, Any
import json
import time

class MessageBus:
    """异步消息总线，支持发布-订阅模式"""
    def __init__(self):
        self._subscribers: Dict[str, List[Callable]] = {}
        self._message_log: List[dict] = []
    
    def subscribe(self, topic: str, callback: Callable):
        """订阅某个话题"""
        if topic not in self._subscribers:
            self._subscribers[topic] = []
        self._subscribers[topic].append(callback)
    
    async def publish(self, topic: str, data: Any, sender: str = "system"):
        """发布消息到话题"""
        message = {
            "topic": topic, "data": data,
            "sender": sender, "timestamp": time.time(),
            "msg_id": f"{topic}_{len(self._message_log)}",
        }
        self._message_log.append(message)
        print(f"[BUS] 发布 [{topic}] 来自 {sender}")
        tasks = []
        for callback in self._subscribers.get(topic, []):
            tasks.append(asyncio.create_task(callback(message)))
        if tasks:
            results = await asyncio.gather(*tasks, return_exceptions=True)
            for r in results:
                if isinstance(r, Exception):
                    print(f"  [BUS] 回调异常: {r}")
        return message
    
    def get_history(self, topic: str = None) -> List[dict]:
        if topic:
            return [m for m in self._message_log if m["topic"] == topic]
        return self._message_log

# 使用示例
bus = MessageBus()

async def on_research_result(msg):
    print(f"  [Coder] 收到研究结果: {msg['data'][:50]}...")

async def on_code_complete(msg):
    print(f"  [Reviewer] 收到代码: {msg['data'][:50]}...")

bus.subscribe("research.done", on_research_result)
bus.subscribe("code.done", on_code_complete)

# Agent 之间的结构化消息协议
from enum import Enum
from typing import Optional, Dict, Any
from dataclasses import dataclass, asdict
import json

class MessageType(Enum):
    TASK_ASSIGN = "task_assign"       # 分配任务
    TASK_COMPLETE = "task_complete"   # 任务完成
    REQUEST_INFO = "request_info"     # 请求信息
    PROVIDE_INFO = "provide_info"     # 提供信息
    ERROR = "error"                   # 错误报告
    FEEDBACK = "feedback"             # 反馈/评价
    VOTE = "vote"                     # 投票

@dataclass
class ProtocolMessage:
    """标准化的 Agent 间通信协议"""
    msg_type: MessageType
    sender: str
    receiver: str  # "*" 表示广播
    content: str
    payload: Dict[str, Any] = None
    priority: int = 5  # 1-10，10 最高
    thread_id: Optional[str] = None  # 关联同一对话线程
    reply_to: Optional[str] = None   # 回复哪条消息
    
    def to_json(self) -> str:
        d = asdict(self)
        d["msg_type"] = self.msg_type.value
        return json.dumps(d, ensure_ascii=False)
    
    @classmethod
    def from_json(cls, json_str: str) -> "ProtocolMessage":
        d = json.loads(json_str)
        d["msg_type"] = MessageType(d["msg_type"])
        return cls(d)

# 示例：Task Manager 分配任务
assign_msg = ProtocolMessage(
    msg_type=MessageType.TASK_ASSIGN,
    sender="Manager",
    receiver="Researcher",
    content="请调研 Python 异步 HTTP 库的优缺点",
    payload={
        "task_id": "T-001",
        "deadline": "5min",
        "expected_format": "对比表格",
    },
    priority=7,
)
print(assign_msg.to_json())

# 黑板（Blackboard）模式的实现
class Blackboard:
    """共享黑板：Agent 们通过读写黑板进行间接通信"""
    def __init__(self):
        self._board: Dict[str, Any] = {}
        self._lock = asyncio.Lock()
        self._observers: Dict[str, List[Callable]] = {}
    
    async def write(self, key: str, value: Any, author: str):
        async with self._lock:
            old_value = self._board.get(key)
            self._board[key] = {
                "value": value, "author": author,
                "updated_at": time.time(), "version": 1
            }
            if old_value:
                self._board[key]["version"] = old_value.get("version", 0) + 1
            for callback in self._observers.get(key, []):
                await callback(key, value, author)
    
    def read(self, key: str) -> Optional[dict]:
        return self._board.get(key)
    
    def observe(self, key: str, callback: Callable):
        if key not in self._observers:
            self._observers[key] = []
        self._observers[key].append(callback)
    
    def snapshot(self) -> Dict[str, dict]:
        return {k: v for k, v in self._board.items()}

# Agent 通过黑板协作的示例
blackboard = Blackboard()

async def researcher_work():
    await blackboard.write("research_result", 
        {"frameworks": ["aiohttp", "httpx", "FastAPI"], "recommendation": "httpx"},
        "Researcher")

async def coder_work():
    result = blackboard.read("research_result")
    if result:
        print(f"基于 {result['value']['recommendation']} 进行编码...")
        await blackboard.write("code_draft", 
            "import httpx; async def fetch(url): ...", "Coder")

# 角色工厂：动态创建不同类型的 Agent
from enum import Enum
from typing import Dict, Callable, Optional

class RoleType(Enum):
    MANAGER = "manager"
    RESEARCHER = "researcher"
    CODER = "coder"
    REVIEWER = "reviewer"
    TESTER = "tester"

# 角色 Prompt 模板库
ROLE_PROMPTS: Dict[RoleType, str] = {
    RoleType.MANAGER: """你是项目 Manager，负责分解复杂任务、分配给合适的专家、协调进度并整合结果。
你的核心职责：
1. 分析用户需求的复杂度和涉及的领域
2. 将任务拆分为可独立执行的子任务
3. 根据子任务类型选择合适的专家角色
4. 监控各子任务的进度，处理阻塞和冲突
5. 整合所有子任务的结果，生成最终输出

决策原则：宁可多分配专家，也不让一个角色承担过多职责。""",

    RoleType.RESEARCHER: """你是 Researcher，负责信息收集、技术调研和分析对比。
你的工作方式：
1. 明确调研的目标和范围
2. 从多个来源收集信息（文档、最佳实践、案例）
3. 对比分析不同方案的优缺点
4. 给出基于证据的推荐意见

输出格式：使用结构化的对比表格，列出方案、优势、劣势、适用场景。""",

    RoleType.CODER: """你是 Coder，一位拥有 10 年经验的资深软件工程师。
你的编码原则：
1. 代码必须清晰、可读、可维护
2. 遵循 SOLID 原则和设计模式
3. 包含必要的类型注解和文档字符串
4. 考虑边界条件和错误处理
5. 优先使用标准库和成熟框架

输出要求：完整的可运行代码，包含注释和示例用法。""",

    RoleType.REVIEWER: """你是 Reviewer，一位经验丰富的技术负责人，负责代码审查。
审查维度：
1. 安全性：是否存在注入、溢出、权限绕过等漏洞
2. 性能：是否有不必要的计算、内存泄漏、N+1 查询
3. 可读性：命名是否清晰、结构是否合理、注释是否充分
4. 可维护性：是否遵循 DRY 原则、是否易于扩展和修改
5. 测试覆盖：关键逻辑是否有对应的测试

输出格式：按严重程度列出问题（Critical/Major/Minor），每个问题附带修复建议。""",
}

class RoleFactory:
    """角色工厂：根据角色类型创建配置好的 Agent"""
    @staticmethod
    def create(role_type: RoleType, llm: Callable, kwargs) -> AgentRole:
        prompt = ROLE_PROMPTS.get(role_type)
        if not prompt:
            raise ValueError(f"未知角色类型: {role_type}")
        custom_prompt = kwargs.get("custom_prompt", "")
        if custom_prompt:
            prompt = f"{prompt}\n\n额外要求：{custom_prompt}"
        return AgentRole(role_type.value, prompt, llm)

# 动态角色分配算法
from collections import defaultdict

class DynamicRoleAssigner:
    """根据任务特征动态分配 Agent 角色"""
    
    def __init__(self):
        # 角色能力矩阵：角色 -> 擅长的任务类型
        self.role_capabilities = {
            RoleType.RESEARCHER: ["调研", "分析", "对比", "搜索", "总结"],
            RoleType.CODER: ["编码", "实现", "重构", "调试", "API"],
            RoleType.REVIEWER: ["审查", "评估", "安全", "性能", "优化"],
            RoleType.TESTER: ["测试", "验证", "覆盖率", "边界", "集成测试"],
            RoleType.MANAGER: ["协调", "规划", "整合", "分解"],
        }
    
    def assign_roles(self, task_description: str) -> Dict[str, list]:
        """分析任务描述，分配需要的角色"""
        words = set(task_description)
        
        # 计算每个角色与任务的匹配度
        role_scores = {}
        for role, keywords in self.role_capabilities.items():
            score = sum(1 for kw in keywords if any(kw in w for w in words))
            if score > 0:
                role_scores[role.value] = score
        
        # 总是需要 Manager
        role_scores.setdefault("manager", 1)
        
        # 按匹配度排序，选择匹配度 > 0 的角色
        assigned = {role: [] for role, score in role_scores.items() if score > 0}
        return assigned

# 使用示例
assigner = DynamicRoleAssigner()
task = "调研 Python 异步编程框架，对比 aiohttp 和 httpx，编写封装库并编写单元测试"
roles = assigner.assign_roles(task)
print(f"需要 {len(roles)} 个角色: {list(roles.keys())}")

# 基于 DAG 的任务调度器
from collections import defaultdict, deque
from typing import Dict, List, Set, Callable, Any
import asyncio

class TaskNode:
    """DAG 中的任务节点"""
    def __init__(self, task_id: str, description: str, 
                 executor: Callable, depends_on: List[str] = None):
        self.task_id = task_id
        self.description = description
        self.executor = executor
        self.depends_on = depends_on or []
        self.result: Any = None
        self.status = "pending"  # pending, running, done, failed

class DAGScheduler:
    """基于有向无环图的任务调度器"""
    def __init__(self):
        self.tasks: Dict[str, TaskNode] = {}
        self._adjacency: Dict[str, List[str]] = defaultdict(list)
        self._in_degree: Dict[str, int] = defaultdict(int)
    
    def add_task(self, task: TaskNode):
        self.tasks[task.task_id] = task
        for dep in task.depends_on:
            self._adjacency[dep].append(task.task_id)
            self._in_degree[task.task_id] += 1
        self._in_degree.setdefault(task.task_id, 0)
    
    def get_ready_tasks(self) -> List[str]:
        """获取所有可以执行的任务（依赖已满足）"""
        return [tid for tid, node in self.tasks.items()
                if node.status == "pending" and self._in_degree.get(tid, 0) == 0]
    
    async def execute(self) -> Dict[str, Any]:
        """执行所有任务（BFS 拓扑排序）"""
        ready = self.get_ready_tasks()
        while ready:
            batch_tasks = []
            for tid in ready:
                task = self.tasks[tid]
                dep_results = {dep: self.tasks[dep].result 
                              for dep in task.depends_on}
                batch_tasks.append(self._run_task(task, dep_results))
            
            results = await asyncio.gather(*batch_tasks, return_exceptions=True)
            
            for tid, result in zip(ready, results):
                if isinstance(result, Exception):
                    self.tasks[tid].status = "failed"
                    self.tasks[tid].result = f"Error: {result}"
                    print(f"  任务 {tid} 失败: {result}")
                else:
                    self.tasks[tid].status = "done"
                    self.tasks[tid].result = result
                    for successor in self._adjacency.get(tid, []):
                        self._in_degree[successor] -= 1
            
            ready = self.get_ready_tasks()
        
        pending = [t for t in self.tasks.values() if t.status == "pending"]
        if pending:
            print(f"  检测到循环依赖: {[t.task_id for t in pending]}")
        
        return {tid: t.result for tid, t in self.tasks.items()}
    
    async def _run_task(self, task: TaskNode, dep_results: dict) -> Any:
        task.status = "running"
        print(f"  执行任务: {task.description}")
        return await task.executor(dep_results) if asyncio.iscoroutinefunction(task.executor) else task.executor(dep_results)

# 使用 DAG 调度器构建 Multi-Agent 任务流
async def build_article_pipeline(llm):
    """构建一个文章生成的 DAG 任务流"""
    scheduler = DAGScheduler()
    
    # T1: 研究（无依赖）
    async def research_task(deps):
        researcher = AgentRole("Researcher", ROLE_PROMPTS[RoleType.RESEARCHER], llm)
        msg = AgentMessage("Scheduler", "调研 2026 年 AI Agent 最新进展", "request")
        return researcher.process(msg).content
    scheduler.add_task(TaskNode("research", "调研", research_task))
    
    # T2: 大纲（依赖研究）
    async def outline_task(deps):
        result = deps.get("research", "")
        msg = AgentMessage("Scheduler", f"基于以下研究写文章大纲:\n{result[:200]}", "request")
        manager = AgentRole("Manager", ROLE_PROMPTS[RoleType.MANAGER], llm)
        return manager.process(msg).content
    scheduler.add_task(TaskNode("outline", "大纲", outline_task, ["research"]))
    
    # T3: 正文（依赖大纲）
    async def body_task(deps):
        outline = deps.get("outline", "")
        msg = AgentMessage("Scheduler", f"根据以下大纲撰写正文:\n{outline[:200]}", "request")
        coder = AgentRole("Coder", ROLE_PROMPTS[RoleType.CODER], llm)
        return coder.process(msg).content
    scheduler.add_task(TaskNode("body", "正文", body_task, ["outline"]))
    
    # T4: 审查（依赖正文）
    async def review_task(deps):
        body = deps.get("body", "")
        msg = AgentMessage("Scheduler", f"审查以下文章:\n{body[:300]}", "request")
        reviewer = AgentRole("Reviewer", ROLE_PROMPTS[RoleType.REVIEWER], llm)
        return reviewer.process(msg).content
    scheduler.add_task(TaskNode("review", "审查", review_task, ["body"]))
    
    results = await scheduler.execute()
    return results

# 执行
# results = asyncio.run(build_article_pipeline(some_llm))
# print(results)  # {'research': ..., 'outline': ..., 'body': ..., 'review': ...}

# 加权投票冲突解决机制
from typing import List, Dict, Any
from collections import Counter

class WeightedVoter:
    """基于权重的多 Agent 投票系统"""
    def __init__(self):
        self._weights: Dict[str, float] = {}
        self._votes: Dict[str, Any] = {}
    
    def set_weight(self, agent_name: str, weight: float):
        """设置 Agent 的投票权重"""
        self._weights[agent_name] = max(0.0, weight)
    
    def cast_vote(self, agent_name: str, vote: Any, confidence: float = 1.0):
        """Agent 投票"""
        weight = self._weights.get(agent_name, 1.0)
        effective_weight = weight * confidence
        self._votes[agent_name] = {"vote": vote, "weight": effective_weight}
    
    def tally(self) -> Dict[str, Any]:
        """统计投票结果"""
        score_map: Dict[str, float] = {}
        vote_map: Dict[str, str] = {}
        for agent, data in self._votes.items():
            vote_key = str(data["vote"])
            vote_map[vote_key] = data["vote"]
            score_map[vote_key] = score_map.get(vote_key, 0) + data["weight"]
        
        if not score_map:
            return {"winner": None, "scores": {}, "total_votes": 0}
        
        winner_key = max(score_map, key=score_map.get)
        total_weight = sum(score_map.values())
        
        return {
            "winner": vote_map[winner_key],
            "scores": {k: round(v / total_weight, 3) for k, v in score_map.items()},
            "total_votes": len(self._votes),
            "consensus": max(score_map.values()) / total_weight > 0.7,
        }

# 使用示例：三个 Agent 对技术方案投票
voter = WeightedVoter()
voter.set_weight("Researcher", 0.3)
voter.set_weight("Coder", 0.5)
voter.set_weight("Reviewer", 0.4)

voter.cast_vote("Researcher", "方案A", 0.8)
voter.cast_vote("Coder", "方案B", 0.9)
voter.cast_vote("Reviewer", "方案A", 0.7)

result = voter.tally()
print(f"胜出: {result['winner']}, 共识: {result['consensus']}")

# 辩论式冲突解决（Debate-based Resolution）
class DebateModerator:
    """辩论主持人：引导 Agent 进行结构化辩论"""
    def __init__(self, llm, max_rounds: int = 3):
        self.llm = llm
        self.max_rounds = max_rounds
        self.transcript: List[Dict] = []
    
    def moderate(self, topic: str, positions: Dict[str, str]) -> str:
        """主持辩论
        topic: 争议话题
        positions: {agent_name: 立场}
        """
        agent_names = list(positions.keys())
        print(f"辩论开始：{topic}")
        print(f"   参与方: {', '.join(agent_names)}")
        
        for round_num in range(self.max_rounds):
            print(f"\n--- 第 {round_num + 1} 轮 ---")
            for agent_name in agent_names:
                history = "\n".join(
                    f"[{t['agent']}]: {t['argument'][:100]}" 
                    for t in self.transcript[-4:]
                )
                prompt = f"""辩论话题：{topic}
你的立场：{positions[agent_name]}
之前的辩论记录：
{history}

请用 3-5 句话陈述你的观点。如果是后续轮次，请回应其他参与方的论点。"""
                argument = self.llm(prompt)
                self.transcript.append({
                    "agent": agent_name, "round": round_num + 1,
                    "argument": argument
                })
                print(f"  [{agent_name}]: {argument[:80]}...")
        
        full_transcript = "\n".join(
            f"[{t['agent']} R{t['round']}]: {t['argument']}" 
            for t in self.transcript
        )
        verdict_prompt = f"""作为公正的裁判，请根据以下辩论内容做出裁决。
辩论话题：{topic}
{full_transcript}

请给出裁决结果，说明哪一方的论点更有说服力，为什么。"""
        return self.llm(verdict_prompt)

# 冲突日志与学习系统
import json
from datetime import datetime

class ConflictLogger:
    """记录和分析 Agent 之间的冲突"""
    def __init__(self, log_file: str = "conflicts.json"):
        self.log_file = log_file
        self.conflicts = self._load()
    
    def _load(self) -> List[dict]:
        try:
            with open(self.log_file) as f:
                return json.load(f)
        except FileNotFoundError:
            return []
    
    def log_conflict(self, conflict_type: str, agents: List[str],
                     descriptions: Dict[str, str], resolution: str,
                     winner: str = None):
        entry = {
            "timestamp": datetime.now().isoformat(),
            "type": conflict_type,
            "agents": agents,
            "positions": descriptions,
            "resolution": resolution,
            "winner": winner,
        }
        self.conflicts.append(entry)
        self._save()
    
    def _save(self):
        with open(self.log_file, 'w') as f:
            json.dump(self.conflicts, f, ensure_ascii=False, indent=2)
    
    def get_patterns(self) -> Dict[str, Any]:
        """分析冲突模式"""
        type_counts = Counter(c["type"] for c in self.conflicts)
        agent_conflict_count = Counter()
        for c in self.conflicts:
            for agent in c["agents"]:
                agent_conflict_count[agent] += 1
        
        return {
            "total_conflicts": len(self.conflicts),
            "by_type": dict(type_counts),
            "most_conflicted_agents": dict(agent_conflict_count.most_common(3)),
            "recent_trend": "increasing" if len(self.conflicts) > 5 
                           and self.conflicts[-3:] else "stable",
        }

# 使用
logger = ConflictLogger()
logger.log_conflict(
    conflict_type="content_disagreement",
    agents=["Researcher", "Coder"],
    descriptions={
        "Researcher": "推荐使用 httpx，性能更好",
        "Coder": "推荐使用 aiohttp，生态更成熟"
    },
    resolution="加权投票，Coder 权重更高（实际开发经验）",
    winner="Coder"
)
print(json.dumps(logger.get_patterns(), indent=2, ensure_ascii=False))

# CrewAI 风格的架构实现（简化版）
from typing import List, Optional
from dataclasses import dataclass, field

@dataclass
class Task:
    description: str
    agent: Optional[str] = None
    expected_output: str = ""
    is_complete: bool = False
    result: str = ""

@dataclass
class Crew:
    """CrewAI 风格的团队编排"""
    agents: Dict[str, AgentRole] = field(default_factory=dict)
    tasks: List[Task] = field(default_factory=list)
    process: str = "sequential"  # sequential | hierarchical
    
    def add_agent(self, agent: AgentRole):
        self.agents[agent.name] = agent
    
    def add_task(self, task: Task):
        self.tasks.append(task)
    
    def kickoff(self) -> str:
        """启动团队执行任务"""
        if self.process == "sequential":
            return self._sequential_process()
        elif self.process == "hierarchical":
            return self._hierarchical_process()
        else:
            raise ValueError(f"未知流程: {self.process}")
    
    def _sequential_process(self) -> str:
        """顺序执行：按任务列表依次执行"""
        context = ""
        for task in self.tasks:
            if task.agent and task.agent in self.agents:
                agent = self.agents[task.agent]
                input_text = f"{task.description}\n\n上下文信息:\n{context}" if context else task.description
                msg = AgentMessage("Crew", input_text, "request")
                result = agent.process(msg).content
                task.result = result
                task.is_complete = True
                context += f"\n[{task.agent} 的结果]: {result[:200]}"
                print(f"  完成: {task.description[:40]}...")
        return context
    
    def _hierarchical_process(self) -> str:
        """层级执行：Manager Agent 协调所有任务"""
        manager = self.agents.get("manager")
        if not manager:
            return self._sequential_process()
        
        task_summary = "\n".join(
            f"- [{t.agent}] {t.description}" for t in self.tasks
        )
        msg = AgentMessage("Crew", f"请协调完成以下任务:\n{task_summary}", "request")
        return manager.process(msg).content

# AutoGen 风格的对话驱动架构
class ConversationalAgent:
    """AutoGen 风格的对话式 Agent"""
    def __init__(self, name: str, system_prompt: str, llm, 
                 can_terminate: bool = False, can_ask_human: bool = False):
        self.name = name
        self.system_prompt = system_prompt
        self.llm = llm
        self.can_terminate = can_terminate
        self.can_ask_human = can_ask_human
        self.chat_history: List[Dict] = []
    
    def receive(self, message: str, sender: str) -> str:
        """接收消息并回复"""
        self.chat_history.append({"role": "user", "content": message, "sender": sender})
        
        context = "\n".join(
            f"[{m['sender']}]: {m['content']}" for m in self.chat_history[-10:]
        )
        prompt = f"{self.system_prompt}\n\n对话历史:\n{context}\n\n请回复："
        
        reply = self.llm(prompt)
        self.chat_history.append({"role": "assistant", "content": reply, "sender": self.name})
        return reply
    
    def is_termination(self, message: str) -> bool:
        """判断对话是否应该终止"""
        if not self.can_terminate:
            return False
        termination_signals = ["TERMINATE", "任务完成", "没有问题了", "done"]
        return any(signal in message.upper() for signal in termination_signals)

class GroupChat:
    """AutoGen 风格的群聊"""
    def __init__(self, agents: List[ConversationalAgent], max_rounds: int = 10):
        self.agents = {a.name: a for a in agents}
        self.max_rounds = max_rounds
        self.messages: List[Dict] = []
    
    def run(self, initial_message: str, speaker_order: List[str] = None) -> str:
        """运行群聊"""
        self.messages.append({"sender": "User", "content": initial_message})
        current_speaker_idx = 0
        
        for round_num in range(self.max_rounds):
            if speaker_order:
                speaker_name = speaker_order[current_speaker_idx % len(speaker_order)]
                current_speaker_idx += 1
            else:
                speaker_name = list(self.agents.keys())[round_num % len(self.agents)]
            
            agent = self.agents[speaker_name]
            last_message = self.messages[-1]["content"]
            
            reply = agent.receive(last_message, self.messages[-1]["sender"])
            self.messages.append({"sender": speaker_name, "content": reply})
            print(f"  [{speaker_name}]: {reply[:60]}...")
            
            if agent.is_termination(reply):
                print(f"  {speaker_name} 发起终止")
                break
        
        return self.messages[-1]["content"]

# Multi-ReAct 架构：多个 Agent 共享观察空间
class SharedObservationSpace:
    """多个 ReAct Agent 共享的观察空间"""
    def __init__(self):
        self.observations: List[Dict] = []
        self.global_state: Dict[str, Any] = {}
    
    def publish(self, agent_name: str, thought: str, action: str, observation: str):
        """Agent 发布自己的 ReAct 步骤"""
        entry = {
            "agent": agent_name, "thought": thought,
            "action": action, "observation": observation,
            "timestamp": time.time(),
        }
        self.observations.append(entry)
        self.global_state[f"{agent_name}_last_observation"] = observation
    
    def get_shared_context(self, agent_name: str, last_n: int = 5) -> str:
        """获取共享上下文（不包括自己的最新步骤）"""
        others = [o for o in self.observations[-last_n*2:] if o["agent"] != agent_name]
        if not others:
            return "暂无其他 Agent 的观察结果。"
        return "\n".join(
            f"[{o['agent']}] 行动: {o['action']}\n  结果: {o['observation'][:100]}"
            for o in others
        )
    
    def get_global_state(self) -> Dict[str, Any]:
        return dict(self.global_state)

class MultiReactAgent(AgentRole):
    """执行 ReAct 循环的 Agent，可以感知其他 Agent 的观察"""
    def __init__(self, name: str, system_prompt: str, llm, shared_space: SharedObservationSpace):
        super().__init__(name, system_prompt, llm)
        self.shared_space = shared_space
    
    def react_step(self, goal: str, max_steps: int = 5) -> str:
        for step in range(max_steps):
            shared_context = self.shared_space.get_shared_context(self.name)
            
            prompt = f"""{self.system_prompt}

目标: {goal}
其他 Agent 的观察:
{shared_context}

请用 ReAct 格式回复:
Thought: <思考>
Action: <行动>
Observation: <你观察到的结果>"""
            
            response = self.llm(prompt)
            
            thought = self._extract(response, "Thought:")
            action = self._extract(response, "Action:")
            observation = self._extract(response, "Observation:")
            
            self.shared_space.publish(self.name, thought, action, observation)
            
            if "done" in observation.lower() or not action:
                return observation
        
        return "达到最大步数"
    
    def _extract(self, text: str, prefix: str) -> str:
        for line in text.split("\n"):
            if line.startswith(prefix):
                return line[len(prefix):].strip()
        return ""

# 静态分析 Agent（模拟 AST 级别的代码分析）
import ast
import re
from typing import List, Dict

class StaticAnalyzerAgent:
    """模拟静态分析的 Agent"""
    def __init__(self, llm):
        self.llm = llm
        self.name = "StaticAnalyzer"
    
    def analyze(self, code: str) -> Dict:
        """执行多维度代码分析"""
        issues = []
        
        # 规则 1: 检查函数长度（超过 50 行的大函数）
        try:
            tree = ast.parse(code)
            for node in ast.walk(tree):
                if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                    lines = node.end_lineno - node.lineno + 1 if hasattr(node, 'end_lineno') else 0
                    if lines > 50:
                        issues.append({
                            "line": node.lineno, "type": "complexity",
                            "severity": "warning",
                            "message": f"函数 '{node.name}' 有 {lines} 行，建议拆分为更小的函数"
                        })
        except SyntaxError as e:
            issues.append({"line": e.lineno, "type": "syntax", "severity": "error",
                          "message": f"语法错误: {e.msg}"})
        
        # 规则 2: 检查 TODO/FIXME/HACK 注释
        for i, line in enumerate(code.split("\n"), 1):
            if re.search(r'#.*\b(TODO|FIXME|HACK|XXX)\b', line):
                issues.append({
                    "line": i, "type": "code_smell",
                    "severity": "info",
                    "message": f"发现标记注释: {line.strip()}"
                })
        
        # 规则 3: LLM 辅助语义分析
        if issues:
            llm_prompt = f"""分析以下 Python 代码的质量问题：

代码片段:
{code[:2000]}

已发现的机械检查问题：
{issues}

请补充：命名是否清晰、是否有更好的设计模式、整体代码结构建议"""
            llm_feedback = self.llm(llm_prompt)
        else:
            llm_feedback = "代码结构良好，未发现明显质量问题。"
        
        return {
            "agent": self.name,
            "total_issues": len(issues),
            "issues": issues,
            "llm_feedback": llm_feedback,
            "status": "passed" if not any(i["severity"] == "error" for i in issues) else "failed"
        }

# 安全审查 Agent
class SecurityReviewerAgent:
    """安全漏洞扫描 Agent"""
    def __init__(self, llm):
        self.llm = llm
        self.name = "SecurityReviewer"
        # 常见的安全模式
        self.dangerous_patterns = [
            (r'eval\s*\(', "使用 eval() 可能导致代码注入"),
            (r'exec\s*\(', "使用 exec() 可能导致代码注入"),
            (r'os\.system\s*\(', "os.system() 可能导致命令注入"),
            (r'subprocess\.call\s*\(.*shell\s*=\s*True', "shell=True 可能导致命令注入"),
            (r'(?i)password\s*=\s*["\x27][^"\x27]+["\x27]', "硬编码密码"),
            (r'(?i)api[_-]?key\s*=\s*["\x27][^"\x27]+["\x27]', "硬编码 API 密钥"),
            (r'(?i)secret\s*=\s*["\x27][^"\x27]+["\x27]', "硬编码密钥"),
            (r'pickle\.loads?', "pickle 反序列化可能导致远程代码执行"),
            (r'marshal\.loads?', "marshal 反序列化可能不安全"),
        ]
    
    def review(self, code: str) -> Dict:
        vulnerabilities = []
        
        # 规则扫描
        for pattern, message in self.dangerous_patterns:
            for i, line in enumerate(code.split("\n"), 1):
                if re.search(pattern, line):
                    vulnerabilities.append({
                        "line": i, "severity": "critical" if "注入" in message else "high",
                        "type": "security", "description": message,
                        "code": line.strip()
                    })
        
        # LLM 深度安全分析
        llm_prompt = f"""作为安全专家，审查以下代码的安全漏洞：

代码片段:
{code[:2000]}

请检查：
1. 输入验证是否充分
2. 是否存在 SQL 注入、XSS、CSRF 风险
3. 认证和授权逻辑是否安全
4. 敏感数据处理是否合规
5. 依赖包是否有已知漏洞

列出发现的所有安全问题，按严重程度排序。"""
        llm_security_report = self.llm(llm_prompt)
        
        critical_count = sum(1 for v in vulnerabilities if v["severity"] == "critical")
        return {
            "agent": self.name,
            "vulnerabilities": vulnerabilities,
            "llm_security_report": llm_security_report,
            "critical_count": critical_count,
            "status": "blocked" if critical_count > 0 else "passed"
        }

# 技术负责人 Agent + 完整的代码评审流程
class TechLeadAgent:
    """技术负责人：综合分析并做出评审决策"""
    def __init__(self, llm):
        self.llm = llm
        self.name = "TechLead"
    
    def review(self, code: str, static_result: Dict, security_result: Dict) -> Dict:
        """综合评审"""
        all_issues = []
        all_issues.extend(static_result.get("issues", []))
        all_issues.extend(security_result.get("vulnerabilities", []))
        
        critical = [i for i in all_issues if i.get("severity") in ("critical", "error")]
        warnings = [i for i in all_issues if i.get("severity") == "warning"]
        infos = [i for i in all_issues if i.get("severity") in ("info", "minor")]
        
        # 自动生成决策
        if critical:
            decision = "REJECT"
            reason = f"发现 {len(critical)} 个严重问题，必须修复后才能合并"
        elif len(warnings) > 5:
            decision = "REQUEST_CHANGES"
            reason = f"发现 {len(warnings)} 个警告，建议修改后重新提交"
        elif len(all_issues) <= 3:
            decision = "APPROVE"
            reason = "代码质量良好，可以合并"
        else:
            decision = "APPROVE_WITH_COMMENTS"
            reason = f"可以合并，但有 {len(all_issues)} 个建议改进项"
        
        # LLM 生成详细评审报告
        report_prompt = f"""作为技术负责人，请生成代码评审报告。

代码片段:
{code[:1500]}

静态分析结果（{static_result['total_issues']} 个问题）：
{str(static_result.get('issues', []))[:500]}

安全审查结果（{security_result['critical_count']} 个严重漏洞）：
{str(security_result.get('vulnerabilities', []))[:500]}

安全分析报告：
{security_result.get('llm_security_report', 'N/A')[:500]}

请生成格式化的评审报告，包括：
1. 总体评价（1-2 句话）
2. 必须修复的问题（按优先级排序）
3. 建议改进的方面
4. 最终结论：APPROVE / REQUEST_CHANGES / REJECT"""
        report = self.llm(report_prompt)
        
        return {
            "agent": self.name,
            "decision": decision,
            "reason": reason,
            "summary": {
                "critical": len(critical),
                "warnings": len(warnings),
                "info": len(infos),
                "total": len(all_issues),
            },
            "report": report,
        }

# ====== 完整的代码评审 Pipeline ======
def code_review_pipeline(code: str, llm) -> Dict:
    """完整的 Multi-Agent 代码评审流程"""
    print("开始 Multi-Agent 代码评审...")
    
    # Step 1: 并行执行静态分析和安全审查
    static_analyzer = StaticAnalyzerAgent(llm)
    security_reviewer = SecurityReviewerAgent(llm)
    
    print("  静态分析中...")
    static_result = static_analyzer.analyze(code)
    print(f"    发现 {static_result['total_issues']} 个问题")
    
    print("  安全审查中...")
    security_result = security_reviewer.review(code)
    print(f"    发现 {security_result['critical_count']} 个严重漏洞")
    
    # Step 2: TechLead 综合评审
    print("  TechLead 评审中...")
    tech_lead = TechLeadAgent(llm)
    final_result = tech_lead.review(code, static_result, security_result)
    
    print(f"\n评审结果: {final_result['decision']}")
    print(f"   原因: {final_result['reason']}")
    print(f"   问题统计: {final_result['summary']}")
    
    return final_result