03_multi_agent.py

"""
多智能体系统案例：团队协作任务处理
==================================

本案例展示如何使用 LangGraph 构建一个多智能体协作系统，
模拟一个团队处理复杂任务的场景，包括：
1. 任务分解和分配
2. 智能体间通信和协作
3. 结果整合和验证
4. 冲突解决机制
"""

from typing import TypedDict, Annotated, List, Dict, Any, Optional
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
import operator
from datetime import datetime
from enum import Enum
import json
import random
from dataclasses import dataclass

# 定义智能体角色
class AgentRole(Enum):
    MANAGER = "manager"  # 经理：任务分解和分配
    RESEARCHER = "researcher"  # 研究员：信息收集
    ANALYST = "analyst"  # 分析师：数据分析
    WRITER = "writer"  # 写作者：报告撰写
    REVIEWER = "reviewer"  # 评审员：质量检查
    COORDINATOR = "coordinator"  # 协调员：冲突解决

# 定义任务状态
class TaskStatus(Enum):
    PENDING = "pending"  # 待处理
    ASSIGNED = "assigned"  # 已分配
    IN_PROGRESS = "in_progress"  # 进行中
    COMPLETED = "completed"  # 已完成
    BLOCKED = "blocked"  # 阻塞
    FAILED = "failed"  # 失败

# 定义消息类型
class MessageType(Enum):
    TASK_ASSIGNMENT = "task_assignment"  # 任务分配
    PROGRESS_UPDATE = "progress_update"  # 进度更新
    RESULT_SUBMISSION = "result_submission"  # 结果提交
    HELP_REQUEST = "help_request"  # 帮助请求
    CONFLICT_REPORT = "conflict_report"  # 冲突报告
    COORDINATION = "coordination"  # 协调消息

# 定义智能体基类
@dataclass
class Agent:
    """智能体基类"""
    id: str
    role: AgentRole
    name: str
    expertise: List[str]
    workload: int = 0
    is_available: bool = True
    
    def process_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
        """处理任务（基类方法，子类应重写）"""
        raise NotImplementedError("子类必须实现 process_task 方法")
    
    def send_message(self, recipient: 'Agent', message_type: MessageType, content: Any) -> Dict[str, Any]:
        """发送消息"""
        return {
            "sender": self.id,
            "recipient": recipient.id,
            "message_type": message_type.value,
            "content": content,
            "timestamp": datetime.now()
        }

# 具体智能体实现
class ManagerAgent(Agent):
    """经理智能体"""
    
    def __init__(self, agent_id: str, name: str):
        super().__init__(
            id=agent_id,
            role=AgentRole.MANAGER,
            name=name,
            expertise=["任务分解", "资源分配", "进度监控"]
        )
    
    def process_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
        """处理任务：分解主任务为子任务"""
        print(f"[Manager {self.name}] 分解任务: {task.get('title', '未知任务')}")
        
        main_task = task.get("description", "")
        
        # 简单的任务分解逻辑
        subtasks = []
        if "研究" in main_task or "调研" in main_task:
            subtasks.append({
                "id": f"subtask_{len(subtasks)+1}",
                "title": "信息收集",
                "description": "收集相关信息和数据",
                "assigned_to": AgentRole.RESEARCHER,
                "priority": "high"
            })
        
        if "分析" in main_task or "评估" in main_task:
            subtasks.append({
                "id": f"subtask_{len(subtasks)+1}",
                "title": "数据分析",
                "description": "分析收集到的数据",
                "assigned_to": AgentRole.ANALYST,
                "priority": "medium"
            })
        
        if "报告" in main_task or "总结" in main_task:
            subtasks.append({
                "id": f"subtask_{len(subtasks)+1}",
                "title": "报告撰写",
                "description": "撰写最终报告",
                "assigned_to": AgentRole.WRITER,
                "priority": "medium"
            })
        
        # 如果没有识别出特定任务，创建默认子任务
        if not subtasks:
            subtasks.append({
                "id": "subtask_1",
                "title": "任务执行",
                "description": main_task,
                "assigned_to": AgentRole.RESEARCHER,
                "priority": "medium"
            })
        
        return {
            "action": "task_decomposition",
            "original_task": task,
            "subtasks": subtasks,
            "recommendations": f"建议分配 {len(subtasks)} 个子任务"
        }

class ResearcherAgent(Agent):
    """研究员智能体"""
    
    def __init__(self, agent_id: str, name: str):
        super().__init__(
            id=agent_id,
            role=AgentRole.RESEARCHER,
            name=name,
            expertise=["信息收集", "数据整理", "文献调研"]
        )
    
    def process_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
        """处理任务：收集信息"""
        print(f"[Researcher {self.name}] 收集信息: {task.get('title', '未知任务')}")
        
        topic = task.get("description", "")
        
        # 模拟信息收集
        sources = [
            "学术数据库",
            "行业报告",
            "公开数据集",
            "专家访谈"
        ]
        
        collected_data = {
            "topic": topic,
            "sources_used": random.sample(sources, min(2, len(sources))),
            "key_findings": [
                f"关于'{topic}'的相关信息1",
                f"关于'{topic}'的相关信息2",
                f"关于'{topic}'的相关信息3"
            ],
            "data_points": random.randint(5, 15),
            "confidence_score": random.uniform(0.7, 0.95)
        }
        
        return {
            "action": "information_gathering",
            "task": task,
            "results": collected_data,
            "status": "completed",
            "notes": "信息收集完成，已整理关键发现"
        }

class AnalystAgent(Agent):
    """分析师智能体"""
    
    def __init__(self, agent_id: str, name: str):
        super().__init__(
            id=agent_id,
            role=AgentRole.ANALYST,
            name=name,
            expertise=["数据分析", "模式识别", "趋势预测"]
        )
    
    def process_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
        """处理任务：分析数据"""
        print(f"[Analyst {self.name}] 分析数据: {task.get('title', '未知任务')}")
        
        # 模拟数据分析
        analysis_results = {
            "patterns_identified": random.randint(2, 5),
            "trends_detected": random.randint(1, 3),
            "insights": [
                "发现主要趋势：...",
                "识别关键模式：...",
                "提出建议：..."
            ],
            "metrics": {
                "accuracy": random.uniform(0.8, 0.99),
                "completeness": random.uniform(0.7, 0.95),
                "relevance": random.uniform(0.75, 0.98)
            },
            "visualizations": ["图表1", "图表2"]
        }
        
        return {
            "action": "data_analysis",
            "task": task,
            "results": analysis_results,
            "status": "completed",
            "notes": "数据分析完成，已生成洞察和建议"
        }

class WriterAgent(Agent):
    """写作者智能体"""
    
    def __init__(self, agent_id: str, name: str):
        super().__init__(
            id=agent_id,
            role=AgentRole.WRITER,
            name=name,
            expertise=["报告撰写", "内容组织", "语言优化"]
        )
    
    def process_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
        """处理任务：撰写报告"""
        print(f"[Writer {self.name}] 撰写报告: {task.get('title', '未知任务')}")
        
        # 模拟报告撰写
        report = {
            "title": f"关于{task.get('description', '任务')}的最终报告",
            "sections": [
                {"title": "执行摘要", "content": "本报告总结了...", "length": 150},
                {"title": "背景介绍", "content": "任务背景是...", "length": 300},
                {"title": "方法", "content": "采用的方法是...", "length": 250},
                {"title": "结果", "content": "主要结果包括...", "length": 400},
                {"title": "结论", "content": "综上所述...", "length": 200},
                {"title": "建议", "content": "建议采取以下措施...", "length": 300}
            ],
            "total_length": 1600,
            "readability_score": random.uniform(0.7, 0.95),
            "key_points": ["要点1", "要点2", "要点3"]
        }
        
        return {
            "action": "report_writing",
            "task": task,
            "results": report,
            "status": "completed",
            "notes": "报告撰写完成，已优化语言和结构"
        }

class ReviewerAgent(Agent):
    """评审员智能体"""
    
    def __init__(self, agent_id: str, name: str):
        super().__init__(
            id=agent_id,
            role=AgentRole.REVIEWER,
            name=name,
            expertise=["质量检查", "错误检测", "改进建议"]
        )
    
    def process_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
        """处理任务：评审结果"""
        print(f"[Reviewer {self.name}] 评审结果: {task.get('title', '未知任务')}")
        
        content_to_review = task.get("content", {})
        
        # 模拟评审过程
        issues_found = random.randint(0, 3)
        suggestions = []
        
        if issues_found > 0:
            suggestions = [
                "建议改进语言表达",
                "建议增加数据支持",
                "建议优化结构"
            ][:issues_found]
        
        review_result = {
            "quality_score": random.uniform(0.6, 1.0),
            "issues_found": issues_found,
            "suggestions": suggestions,
            "approval_status": "approved" if random.random() > 0.3 else "needs_revision",
            "feedback": "整体质量良好，建议进行少量改进" if issues_found < 2 else "需要重大改进"
        }
        
        return {
            "action": "quality_review",
            "task": task,
            "results": review_result,
            "status": "completed",
            "notes": "评审完成，已提供反馈和建议"
        }

class CoordinatorAgent(Agent):
    """协调员智能体"""
    
    def __init__(self, agent_id: str, name: str):
        super().__init__(
            id=agent_id,
            role=AgentRole.COORDINATOR,
            name=name,
            expertise=["冲突解决", "资源协调", "流程优化"]
        )
    
    def process_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
        """处理任务：协调冲突"""
        print(f"[Coordinator {self.name}] 协调冲突: {task.get('title', '未知任务')}")
        
        conflict_details = task.get("conflict", {})
        
        # 模拟冲突解决
        resolution = {
            "conflict_type": conflict_details.get("type", "unknown"),
            "parties_involved": conflict_details.get("parties", []),
            "resolution_strategy": random.choice([
                "协商妥协",
                "权威决策",
                "第三方调解",
                "重新分配任务"
            ]),
            "agreement_reached": random.random() > 0.2,
            "follow_up_actions": [
                "更新任务分配",
                "调整时间表",
                "提供额外资源"
            ]
        }
        
        return {
            "action": "conflict_resolution",
            "task": task,
            "results": resolution,
            "status": "completed",
            "notes": "冲突协调完成，已提出解决方案"
        }

# 定义多智能体系统状态
class MultiAgentState(TypedDict):
    """多智能体系统状态"""
    task_id: str  # 任务ID
    original_task: Dict[str, Any]  # 原始任务
    subtasks: Annotated[List[Dict], operator.add]  # 子任务列表
    assigned_tasks: Dict[str, List[Dict]]  # 已分配任务（按智能体）
    completed_tasks: Annotated[List[Dict], operator.add]  # 已完成任务
    agent_results: Dict[str, Dict]  # 智能体结果
    messages: Annotated[List[Dict], operator.add]  # 消息记录
    conflicts: Annotated[List[Dict], operator.add]  # 冲突记录
    overall_status: TaskStatus  # 整体状态
    final_report: Optional[Dict[str, Any]]  # 最终报告
    timestamp: datetime  # 时间戳

# 智能体管理器
class AgentManager:
    """智能体管理器"""
    
    def __init__(self):
        self.agents = {}
        self.initialize_agents()
    
    def initialize_agents(self):
        """初始化智能体"""
        # 创建各种角色的智能体
        self.agents[AgentRole.MANAGER] = ManagerAgent("manager_001", "张经理")
        self.agents[AgentRole.RESEARCHER] = ResearcherAgent("researcher_001", "李研究员")
        self.agents[AgentRole.ANALYST] = AnalystAgent("analyst_001", "王分析师")
        self.agents[AgentRole.WRITER] = WriterAgent("writer_001", "赵作家")
        self.agents[AgentRole.REVIEWER] = ReviewerAgent("reviewer_001", "钱评审")
        self.agents[AgentRole.COORDINATOR] = CoordinatorAgent("coordinator_001", "孙协调员")
    
    def get_agent_by_role(self, role: AgentRole) -> Optional[Agent]:
        """根据角色获取智能体"""
        return self.agents.get(role)
    
    def assign_task(self, role: AgentRole, task: Dict[str, Any]) -> Dict[str, Any]:
        """分配任务给智能体"""
        agent = self.get_agent_by_role(role)
        if not agent:
            return {"error": f"没有找到角色为 {role.value} 的智能体"}
        
        if not agent.is_available:
            return {"error": f"智能体 {agent.name} 不可用"}
        
        # 更新智能体工作负载
        agent.workload += 1
        
        # 执行任务
        result = agent.process_task(task)
        
        # 更新可用性
        agent.workload -= 1
        if agent.workload <= 0:
            agent.is_available = True
        
        return result

# 节点函数定义
def receive_task(state: MultiAgentState) -> dict:
    """接收任务节点"""
    print(f"[{datetime.now()}] 接收新任务: {state['task_id']}")
    
    return {
        "overall_status": TaskStatus.PENDING,
        "timestamp": datetime.now(),
        "messages": [{
            "type": "system",
            "content": f"任务 {state['task_id']} 已接收",
            "timestamp": datetime.now()
        }]
    }

def decompose_task(state: MultiAgentState) -> dict:
    """分解任务节点"""
    print(f"[{datetime.now()}] 分解任务")
    
    # 获取经理智能体
    agent_manager = AgentManager()
    manager = agent_manager.get_agent_by_role(AgentRole.MANAGER)
    
    if not manager:
        return {
            "overall_status": TaskStatus.FAILED,
            "messages": [{
                "type": "error",
                "content": "没有可用的经理智能体",
                "timestamp": datetime.now()
            }]
        }
    
    # 经理分解任务
    decomposition_result = manager.process_task(state["original_task"])
    
    subtasks = decomposition_result.get("subtasks", [])
    
    return {
        "subtasks": subtasks,
        "overall_status": TaskStatus.ASSIGNED,
        "messages": [{
            "type": "task_assignment",
            "content": f"任务已分解为 {len(subtasks)} 个子任务",
            "timestamp": datetime.now()
        }]
    }

def assign_subtasks(state: MultiAgentState) -> dict:
    """分配子任务节点"""
    print(f"[{datetime.now()}] 分配子任务")
    
    agent_manager = AgentManager()
    assigned_tasks = {}
    
    for subtask in state.get("subtasks", []):
        role_str = subtask.get("assigned_to")
        try:
            role = AgentRole(role_str)
            
            # 分配任务
            result = agent_manager.assign_task(role, subtask)
            
            if "error" not in result:
                # 记录分配
                agent_key = role.value
                if agent_key not in assigned_tasks:
                    assigned_tasks[agent_key] = []
                assigned_tasks[agent_key].append({
                    "task": subtask,
                    "result": result
                })
                
                # 记录消息
                state["messages"].append({
                    "type": "task_assignment",
                    "content": f"任务 {subtask['id']} 已分配给 {role.value}",
                    "timestamp": datetime.now()
                })
            else:
                state["messages"].append({
                    "type": "error",
                    "content": f"分配任务失败: {result['error']}",
                    "timestamp": datetime.now()
                })
                
        except ValueError:
            state["messages"].append({
                "type": "error",
                "content": f"未知角色: {role_str}",
                "timestamp": datetime.now()
            })
    
    return {
        "assigned_tasks": assigned_tasks,
        "overall_status": TaskStatus.IN_PROGRESS
    }

def execute_tasks(state: MultiAgentState) -> dict:
    """执行任务节点"""
    print(f"[{datetime.now()}] 执行任务")
    
    agent_manager = AgentManager()
    agent_results = {}
    completed_tasks = []
    
    # 执行所有已分配的任务
    for agent_role_str, tasks in state.get("assigned_tasks", {}).items():
        try:
            role = AgentRole(agent_role_str)
            
            for task_info in tasks:
                task = task_info["task"]
                
                # 获取智能体并执行任务
                agent = agent_manager.get_agent_by_role(role)
                if agent:
                    result = agent.process_task(task)
                    
                    # 记录结果
                    agent_results[task["id"]] = {
                        "agent_role": role.value,
                        "agent_name": agent.name,
                        "task": task,
                        "result": result,
                        "timestamp": datetime.now()
                    }
                    
                    completed_tasks.append({
                        "task_id": task["id"],
                        "status": "completed",
                        "result": result
                    })
                    
                    # 记录消息
                    state["messages"].append({
                        "type": "progress_update",
                        "content": f"任务 {task['id']} 已完成 ({role.value})",
                        "timestamp": datetime.now()
                    })
                    
                    # 模拟偶尔的冲突
                    if random.random() < 0.2:  # 20% 概率发生冲突
                        conflict = {
                            "task_id": task["id"],
                            "type": random.choice(["资源冲突", "时间冲突", "优先级冲突"]),
                            "parties": [role.value],
                            "description": f"任务 {task['id']} 遇到冲突",
                            "timestamp": datetime.now()
                        }
                        state["conflicts"].append(conflict)
                        
                        state["messages"].append({
                            "type": "conflict_report",
                            "content": f"检测到冲突: {conflict['description']}",
                            "timestamp": datetime.now()
                        })
        
        except ValueError:
            continue
    
    return {
        "agent_results": agent_results,
        "completed_tasks": completed_tasks,
        "overall_status": TaskStatus.COMPLETED if completed_tasks else TaskStatus.BLOCKED
    }

def resolve_conflicts(state: MultiAgentState) -> dict:
    """解决冲突节点"""
    print(f"[{datetime.now()}] 解决冲突")
    
    if not state.get("conflicts"):
        return {"overall_status": state["overall_status"]}
    
    agent_manager = AgentManager()
    coordinator = agent_manager.get_agent_by_role(AgentRole.COORDINATOR)
    
    resolved_conflicts = []
    
    for conflict in state.get("conflicts", []):
        if coordinator:
            # 协调员解决冲突
            resolution_result = coordinator.process_task({
                "title": f"解决冲突: {conflict['type']}",
                "conflict": conflict
            })
            
            resolved_conflicts.append({
                "original_conflict": conflict,
                "resolution": resolution_result,
                "resolved_at": datetime.now()
            })
            
            state["messages"].append({
                "type": "coordination",
                "content": f"冲突已解决: {conflict['description']}",
                "timestamp": datetime.now()
            })
    
    # 更新状态
    new_status = TaskStatus.IN_PROGRESS if resolved_conflicts else state["overall_status"]
    
    return {
        "conflicts": resolved_conflicts,
        "overall_status": new_status
    }

def review_results(state: MultiAgentState) -> dict:
    """评审结果节点"""
    print(f"[{datetime.now()}] 评审结果")
    
    agent_manager = AgentManager()
    reviewer = agent_manager.get_agent_by_role(AgentRole.REVIEWER)
    
    if not reviewer:
        return {
            "overall_status": TaskStatus.COMPLETED,
            "messages": [{
                "type": "warning",
                "content": "没有可用的评审员，跳过评审",
                "timestamp": datetime.now()
            }]
        }
    
    # 收集所有结果进行评审
    all_results = list(state.get("agent_results", {}).values())
    
    review_tasks = []
    for result in all_results:
        review_tasks.append({
            "title": f"评审任务结果: {result['task'].get('title', '未知')}",
            "content": result
        })
    
    review_results = []
    for task in review_tasks:
        review_result = reviewer.process_task(task)
        review_results.append(review_result)
    
    # 计算整体质量分数
    quality_scores = [r["results"].get("quality_score", 0) for r in review_results]
    avg_quality = sum(quality_scores) / len(quality_scores) if quality_scores else 0
    
    # 生成最终报告
    final_report = {
        "task_id": state["task_id"],
        "original_task": state["original_task"],
        "subtasks_completed": len(state.get("completed_tasks", [])),
        "agents_involved": list(set([r["agent_role"] for r in all_results])),
        "overall_quality_score": avg_quality,
        "review_results": review_results,
        "completion_time": datetime.now(),
        "summary": f"任务 {state['task_id']} 已完成，整体质量评分: {avg_quality:.2f}/1.0"
    }
    
    return {
        "final_report": final_report,
        "overall_status": TaskStatus.COMPLETED,
        "messages": [{
            "type": "result_submission",
            "content": f"任务完成，最终报告已生成，质量评分: {avg_quality:.2f}",
            "timestamp": datetime.now()
        }]
    }

def handle_failure(state: MultiAgentState) -> dict:
    """处理失败节点"""
    print(f"[{datetime.now()}] 处理失败")
    
    return {
        "overall_status": TaskStatus.FAILED,
        "messages": [{
            "type": "error",
            "content": "任务处理失败",
            "timestamp": datetime.now()
        }]
    }

# 条件判断函数
def check_task_decomposition(state: MultiAgentState) -> str:
    """检查任务分解状态"""
    if state.get("subtasks"):
        return "assign_tasks"
    else:
        return "failure"

def check_conflicts(state: MultiAgentState) -> str:
    """检查是否有冲突"""
    if state.get("conflicts"):
        return "resolve_conflicts"
    else:
        return "execute_tasks"

def check_execution_complete(state: MultiAgentState) -> str:
    """检查执行是否完成"""
    if state["overall_status"] == TaskStatus.COMPLETED:
        return "review_results"
    elif state["overall_status"] == TaskStatus.BLOCKED:
        return "resolve_conflicts"
    else:
        return "failure"

def check_review_needed(state: MultiAgentState) -> str:
    """检查是否需要评审"""
    if state.get("completed_tasks"):
        return "review_results"
    else:
        return "failure"

# 构建多智能体系统图
def build_multi_agent_system():
    """构建多智能体系统图"""
    workflow = StateGraph(MultiAgentState)
    
    # 添加节点
    workflow.add_node("receive", receive_task)
    workflow.add_node("decompose", decompose_task)
    workflow.add_node("assign", assign_subtasks)
    workflow.add_node("execute", execute_tasks)
    workflow.add_node("resolve", resolve_conflicts)
    workflow.add_node("review", review_results)
    workflow.add_node("failure", handle_failure)
    
    # 设置入口点
    workflow.set_entry_point("receive")
    
    # 添加边
    workflow.add_edge("receive", "decompose")
    
    # 添加条件边
    workflow.add_conditional_edges(
        "decompose",
        check_task_decomposition,
        {
            "assign_tasks": "assign",
            "failure": "failure"
        }
    )
    
    workflow.add_edge("assign", "execute")
    
    workflow.add_conditional_edges(
        "execute",
        check_execution_complete,
        {
            "review_results": "review",
            "resolve_conflicts": "resolve",
            "failure": "failure"
        }
    )
    
    workflow.add_conditional_edges(
        "resolve",
        check_review_needed,
        {
            "review_results": "review",
            "failure": "failure"
        }
    )
    
    # 添加结束边
    workflow.add_edge("review", END)
    workflow.add_edge("failure", END)
    
    # 编译图
    return workflow.compile()

# 多智能体系统包装类
class MultiAgentSystem:
    """多智能体系统"""
    
    def __init__(self):
        self.system = build_multi_agent_system()
    
    def process_complex_task(self, task_description: str, task_title: str = "复杂任务") -> Dict[str, Any]:
        """处理复杂任务"""
        task_id = f"task_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        
        initial_state = {
            "task_id": task_id,
            "original_task": {
                "id": task_id,
                "title": task_title,
                "description": task_description,
                "priority": "high",
                "created_at": datetime.now()
            },
            "subtasks": [],
            "assigned_tasks": {},
            "completed_tasks": [],
            "agent_results": {},
            "messages": [],
            "conflicts": [],
            "overall_status": TaskStatus.PENDING,
            "final_report": None,
            "timestamp": datetime.now()
        }
        
        # 执行系统
        result = self.system.invoke(initial_state)
        
        return result

# 运行示例
def run_multi_agent_example():
    """运行多智能体系统示例"""
    print("=" * 70)
    print("多智能体协作系统示例")
    print("=" * 70)
    
    # 创建多智能体系统
    mas = MultiAgentSystem()
    
    # 测试用例1：市场调研任务
    print("\n测试用例1：市场调研任务")
    print("-" * 50)
    
    task1 = "进行人工智能市场调研，分析当前趋势，并撰写详细报告"
    result1 = mas.process_complex_task(task1, "AI市场调研")
    
    print(f"任务ID: {result1['task_id']}")
    print(f"任务状态: {result1['overall_status'].value}")
    print(f"完成子任务数: {len(result1['completed_tasks'])}")
    print(f"消息数量: {len(result1['messages'])}")
    print(f"冲突数量: {len(result1['conflicts'])}")
    
    if result1.get("final_report"):
        report = result1["final_report"]
        print(f"最终报告质量评分: {report.get('overall_quality_score', 0):.2f}")
        print(f"报告摘要: {report.get('summary', '无摘要')}")
    
    # 测试用例2：产品分析任务
    print("\n测试用例2：产品分析任务")
    print("-" * 50)
    
    task2 = "分析竞争对手产品，收集用户反馈，提出改进建议"
    result2 = mas.process_complex_task(task2, "竞品分析")
    
    print(f"任务ID: {result2['task_id']}")
    print(f"任务状态: {result2['overall_status'].value}")
    print(f"完成子任务数: {len(result2['completed_tasks'])}")
    
    # 显示消息日志
    print("\n消息日志（最近5条）:")
    for msg in result2["messages"][-5:]:
        print(f"  [{msg['timestamp']}] {msg['type']}: {msg['content']}")
    
    return mas

# 系统监控和分析工具
class SystemMonitor:
    """系统监控工具"""
    
    @staticmethod
    def analyze_system_performance(results: List[Dict[str, Any]]) -> Dict[str, Any]:
        """分析系统性能"""
        analysis = {
            "total_tasks": len(results),
            "successful_tasks": 0,
            "failed_tasks": 0,
            "avg_subtasks_per_task": 0,
            "avg_conflicts_per_task": 0,
            "avg_quality_score": 0,
            "agent_utilization": {}
        }
        
        total_subtasks = 0
        total_conflicts = 0
        total_quality = 0
        quality_count = 0
        
        agent_workload = {}
        
        for result in results:
            if result["overall_status"] == TaskStatus.COMPLETED:
                analysis["successful_tasks"] += 1
            else:
                analysis["failed_tasks"] += 1
            
            total_subtasks += len(result.get("completed_tasks", []))
            total_conflicts += len(result.get("conflicts", []))
            
            if result.get("final_report"):
                quality = result["final_report"].get("overall_quality_score", 0)
                total_quality += quality
                quality_count += 1
            
            # 统计智能体工作量
            for agent_result in result.get("agent_results", {}).values():
                agent_role = agent_result.get("agent_role")
                if agent_role:
                    agent_workload[agent_role] = agent_workload.get(agent_role, 0) + 1
        
        analysis["avg_subtasks_per_task"] = total_subtasks / len(results) if results else 0
        analysis["avg_conflicts_per_task"] = total_conflicts / len(results) if results else 0
        analysis["avg_quality_score"] = total_quality / quality_count if quality_count else 0
        analysis["agent_utilization"] = agent_workload
        
        return analysis
    
    @staticmethod
    def print_performance_report(analysis: Dict[str, Any]):
        """打印性能报告"""
        print("\n" + "=" * 60)
        print("多智能体系统性能报告")
        print("=" * 60)
        
        print(f"总任务数: {analysis['total_tasks']}")
        print(f"成功任务数: {analysis['successful_tasks']}")
        print(f"失败任务数: {analysis['failed_tasks']}")
        success_rate = analysis['successful_tasks'] / analysis['total_tasks'] if analysis['total_tasks'] > 0 else 0
        print(f"成功率: {success_rate:.1%}")
        
        print(f"\n平均子任务数: {analysis['avg_subtasks_per_task']:.1f}")
        print(f"平均冲突数: {analysis['avg_conflicts_per_task']:.1f}")
        print(f"平均质量评分: {analysis['avg_quality_score']:.2f}/1.0")
        
        print("\n智能体工作量分布:")
        for agent, workload in analysis['agent_utilization'].items():
            print(f"  {agent}: {workload} 个任务")

if __name__ == "__main__":
    # 运行示例
    mas = run_multi_agent_example()
    
    print("\n" + "=" * 70)
    print("多智能体系统案例总结:")
    print("1. 角色化智能体设计")
    print("2. 任务分解和分配机制")
    print("3. 智能体间通信和协作")
    print("4. 冲突检测和解决")
    print("5. 质量评审和报告生成")
    print("=" * 70)
    
    # 演示系统监控
    print("\n演示系统监控和分析:")
    
    # 运行多个任务进行性能分析
    test_tasks = [
        "研究区块链技术发展趋势",
        "分析新能源汽车市场",
        "评估远程办公工具效果",
        "调研健康科技应用场景"
    ]
    
    all_results = []
    for i, task_desc in enumerate(test_tasks):
        print(f"\n执行任务 {i+1}: {task_desc[:30]}...")
        result = mas.process_complex_task(task_desc, f"测试任务{i+1}")
        all_results.append(result)
    
    # 分析性能
    monitor = SystemMonitor()
    performance_analysis = monitor.analyze_system_performance(all_results)
    monitor.print_performance_report(performance_analysis)
    
    # 演示智能体协作场景
    print("\n演示智能体协作场景:")
    print("模拟一个需要多个智能体协作的复杂任务...")
    
    complex_task = """
    我们需要完成一个全面的数字化转型评估项目：
    1. 收集行业最佳实践和案例研究
    2. 分析当前技术架构和流程
    3. 评估数字化转型的潜在影响
    4. 制定实施路线图和风险缓解策略
    5. 撰写详细的评估报告和执行建议
    """
    
    final_result = mas.process_complex_task(complex_task, "数字化转型评估")
    
    print(f"\n复杂任务执行结果:")
    print(f"任务状态: {final_result['overall_status'].value}")
    print(f"涉及智能体: {list(set([r['agent_role'] for r in final_result.get('agent_results', {}).values()]))}")
    
    if final_result.get("final_report"):
        report = final_result["final_report"]
        print(f"报告质量: {report.get('overall_quality_score', 0):.2f}/1.0")
        print(f"完成时间: {report.get('completion_time')}")

多智能体协作系统示例

测试用例1：市场调研任务

[2026-03-25 21:42:49.085338] 接收新任务: task_20260325_214249
[2026-03-25 21:42:49.085716] 分解任务
[Manager 张经理] 分解任务: AI市场调研
[2026-03-25 21:42:49.086414] 分配子任务
[Researcher 李研究员] 收集信息: 信息收集
[Analyst 王分析师] 分析数据: 数据分析
[Writer 赵作家] 撰写报告: 报告撰写
[2026-03-25 21:42:49.087128] 执行任务
[Researcher 李研究员] 收集信息: 信息收集
[Analyst 王分析师] 分析数据: 数据分析
[Writer 赵作家] 撰写报告: 报告撰写
[2026-03-25 21:42:49.087847] 评审结果
[Reviewer 钱评审] 评审结果: 评审任务结果: 信息收集
[Reviewer 钱评审] 评审结果: 评审任务结果: 数据分析
[Reviewer 钱评审] 评审结果: 评审任务结果: 报告撰写
任务ID: task_20260325_214249
任务状态: completed
完成子任务数: 3
消息数量: 11
冲突数量: 2
最终报告质量评分: 0.76
报告摘要: 任务 task_20260325_214249 已完成，整体质量评分: 0.76/1.0

测试用例2：产品分析任务

[2026-03-25 21:42:49.089065] 接收新任务: task_20260325_214249
[2026-03-25 21:42:49.089311] 分解任务
[Manager 张经理] 分解任务: 竞品分析
[2026-03-25 21:42:49.089732] 分配子任务
[Analyst 王分析师] 分析数据: 数据分析
[2026-03-25 21:42:49.090293] 执行任务
[Analyst 王分析师] 分析数据: 数据分析
[2026-03-25 21:42:49.090645] 评审结果
[Reviewer 钱评审] 评审结果: 评审任务结果: 数据分析
任务ID: task_20260325_214249
任务状态: completed
完成子任务数: 1

消息日志（最近5条）:
[2026-03-25 21:42:49.089115] system: 任务 task_20260325_214249 已接收
[2026-03-25 21:42:49.089397] task_assignment: 任务已分解为 1 个子任务
[2026-03-25 21:42:49.090085] task_assignment: 任务 subtask_1 已分配给 analyst
[2026-03-25 21:42:49.090388] progress_update: 任务 subtask_1 已完成 (analyst)
[2026-03-25 21:42:49.090858] result_submission: 任务完成，最终报告已生成，质量评分: 0.74

======================================================================
多智能体系统案例总结:

1. 角色化智能体设计
2. 任务分解和分配机制
3. 智能体间通信和协作
4. 冲突检测和解决
5. 质量评审和报告生成
   ======================================================================

演示系统监控和分析:

执行任务 1: 研究区块链技术发展趋势…
[2026-03-25 21:42:49.093878] 接收新任务: task_20260325_214249
[2026-03-25 21:42:49.094200] 分解任务
[Manager 张经理] 分解任务: 测试任务1
[2026-03-25 21:42:49.094601] 分配子任务
[Researcher 李研究员] 收集信息: 信息收集
[2026-03-25 21:42:49.094907] 执行任务
[Researcher 李研究员] 收集信息: 信息收集
[2026-03-25 21:42:49.095324] 评审结果
[Reviewer 钱评审] 评审结果: 评审任务结果: 信息收集

执行任务 2: 分析新能源汽车市场…
[2026-03-25 21:42:49.096023] 接收新任务: task_20260325_214249
[2026-03-25 21:42:49.096265] 分解任务
[Manager 张经理] 分解任务: 测试任务2
[2026-03-25 21:42:49.096615] 分配子任务
[Analyst 王分析师] 分析数据: 数据分析
[2026-03-25 21:42:49.096927] 执行任务
[Analyst 王分析师] 分析数据: 数据分析
[2026-03-25 21:42:49.097314] 评审结果
[Reviewer 钱评审] 评审结果: 评审任务结果: 数据分析

执行任务 3: 评估远程办公工具效果…
[2026-03-25 21:42:49.098054] 接收新任务: task_20260325_214249
[2026-03-25 21:42:49.098281] 分解任务
[Manager 张经理] 分解任务: 测试任务3
[2026-03-25 21:42:49.098618] 分配子任务
[Analyst 王分析师] 分析数据: 数据分析
[2026-03-25 21:42:49.098858] 执行任务
[Analyst 王分析师] 分析数据: 数据分析
[2026-03-25 21:42:49.099202] 评审结果
[Reviewer 钱评审] 评审结果: 评审任务结果: 数据分析

执行任务 4: 调研健康科技应用场景…
[2026-03-25 21:42:49.099939] 接收新任务: task_20260325_214249
[2026-03-25 21:42:49.100154] 分解任务
[Manager 张经理] 分解任务: 测试任务4
[2026-03-25 21:42:49.100672] 分配子任务
[Researcher 李研究员] 收集信息: 信息收集
[2026-03-25 21:42:49.101163] 执行任务
[Researcher 李研究员] 收集信息: 信息收集
[2026-03-25 21:42:49.101645] 评审结果
[Reviewer 钱评审] 评审结果: 评审任务结果: 信息收集

============================================================
多智能体系统性能报告

总任务数: 4
成功任务数: 4
失败任务数: 0
成功率: 100.0%

平均子任务数: 1.0
平均冲突数: 0.5
平均质量评分: 0.77/1.0

智能体工作量分布:
researcher: 2 个任务
analyst: 2 个任务

演示智能体协作场景:
模拟一个需要多个智能体协作的复杂任务…
[2026-03-25 21:42:49.102930] 接收新任务: task_20260325_214249
[2026-03-25 21:42:49.103176] 分解任务
[Manager 张经理] 分解任务: 数字化转型评估
[2026-03-25 21:42:49.103529] 分配子任务
[Researcher 李研究员] 收集信息: 信息收集
[Analyst 王分析师] 分析数据: 数据分析
[Writer 赵作家] 撰写报告: 报告撰写
[2026-03-25 21:42:49.103906] 执行任务
[Researcher 李研究员] 收集信息: 信息收集
[Analyst 王分析师] 分析数据: 数据分析
[Writer 赵作家] 撰写报告: 报告撰写
[2026-03-25 21:42:49.104376] 评审结果
[Reviewer 钱评审] 评审结果: 评审任务结果: 信息收集
[Reviewer 钱评审] 评审结果: 评审任务结果: 数据分析
[Reviewer 钱评审] 评审结果: 评审任务结果: 报告撰写

复杂任务执行结果:
任务状态: completed
涉及智能体: [‘analyst’, ‘researcher’, ‘writer’]
报告质量: 0.76/1.0
完成时间: 2026-03-25 21:42:49.104552