Managed Agent 平台与 AI 团队协作：从单兵作战到智能体军团的范式转变

"""
对比 Multi-Agent 和 Managed Agent 的实现差异
演示两种范式在任务处理上的本质区别
"""

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


# ===== Multi-Agent 范式：手动编排 =====

class SimpleMultiAgentSystem:
    """传统 Multi-Agent 系统：手动编排执行顺序"""

    def __init__(self):
        self.agents = {}

    def register_agent(self, name: str, capability: str):
        self.agents[name] = {
            "name": name,
            "capability": capability,
            "completed_tasks": 0,
            "skills": []
        }

    def execute_pipeline(self, task: str, agent_order: List[str]) -> Dict:
        """
        手动指定执行顺序的管道
        缺点：需要开发者预先知道哪个 Agent 做什么
        """
        result = {"task": task, "steps": []}
        current_input = task

        for agent_name in agent_order:
            if agent_name not in self.agents:
                raise ValueError(f"Agent {agent_name} not registered")

            agent = self.agents[agent_name]
            # 模拟 Agent 处理
            step_result = f"[{agent_name}({agent['capability']})] 处理: {current_input}"
            result["steps"].append(step_result)
            agent["completed_tasks"] += 1
            current_input = step_result

        return result


# ===== Managed Agent 范式：自动管理 =====

class TaskPriority(Enum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    CRITICAL = 4


@dataclass
class ManagedTask:
    task_id: str
    description: str
    required_skills: List[str]
    priority: TaskPriority
    status: str = "pending"
    assigned_to: Optional[str] = None
    result: Optional[str] = None


@dataclass
class ManagedAgent:
    name: str
    skills: List[str]
    current_load: float = 0.0  # 0-1, 当前负载
    max_load: float = 1.0
    completed_tasks: int = 0
    performance_score: float = 1.0  # 0-1, 绩效评分
    accumulated_knowledge: List[str] = field(default_factory=list)


class ManagedAgentPlatform:
    """Managed Agent 平台：自动任务分配 + 进度追踪 + 技能累积"""

    def __init__(self):
        self.agents: Dict[str, ManagedAgent] = {}
        self.task_queue: List[ManagedTask] = []
        self.skill_registry: Dict[str, List[str]] = {}  # 技能 → 掌握的 Agent 列表
        self.completed_tasks: List[ManagedTask] = []

    def register_agent(self, agent: ManagedAgent):
        self.agents[agent.name] = agent
        # 更新技能注册表
        for skill in agent.skills:
            if skill not in self.skill_registry:
                self.skill_registry[skill] = []
            self.skill_registry[skill].append(agent.name)

    def submit_task(self, task: ManagedTask):
        self.task_queue.append(task)
        self._dispatch()  # 自动调度

    def _find_best_agent(self, task: ManagedTask) -> Optional[str]:
        """智能匹配：找到最适合的 Agent"""
        candidates = []

        for name, agent in self.agents.items():
            # 检查技能匹配度
            skill_match = sum(
                1 for s in task.required_skills if s in agent.skills
            ) / len(task.required_skills) if task.required_skills else 0

            # 检查可用性
            availability = 1.0 - (agent.current_load / agent.max_load)

            # 综合评分：技能匹配(60%) + 可用性(20%) + 历史绩效(20%)
            score = skill_match * 0.6 + availability * 0.2 + agent.performance_score * 0.2

            if skill_match > 0:  # 至少需要一个匹配技能
                candidates.append((name, score))

        if not candidates:
            return None

        # 选择评分最高的 Agent
        candidates.sort(key=lambda x: x[1], reverse=True)
        return candidates[0][0]

    def _dispatch(self):
        """自动任务调度"""
        pending = [t for t in self.task_queue if t.status == "pending"]
        # 按优先级排序
        pending.sort(key=lambda t: t.priority.value, reverse=True)

        for task in pending:
            best_agent = self._find_best_agent(task)
            if best_agent:
                task.status = "running"
                task.assigned_to = best_agent
                self.agents[best_agent].current_load += 0.3
                self._execute_task(task)

    def _execute_task(self, task: ManagedTask):
        """执行任务并累积经验"""
        agent = self.agents[task.assigned_to]

        # 模拟执行
        result = f"Agent '{agent.name}' 完成任务 '{task.description}'"
        task.result = result
        task.status = "completed"

        # 更新 Agent 状态
        agent.completed_tasks += 1
        agent.current_load = max(0, agent.current_load - 0.3)
        agent.performance_score = min(1.0, agent.performance_score + 0.05)

        # 技能累积
        for skill in task.required_skills:
            if skill not in agent.accumulated_knowledge:
                agent.accumulated_knowledge.append(skill)

        self.completed_tasks.append(task)

    def get_team_report(self) -> Dict:
        """生成团队报告"""
        return {
            "total_agents": len(self.agents),
            "total_completed": len(self.completed_tasks),
            "agents": {
                name: {
                    "completed": a.completed_tasks,
                    "skills": a.skills + a.accumulated_knowledge,
                    "performance": round(a.performance_score, 2),
                    "knowledge_count": len(a.accumulated_knowledge)
                }
                for name, a in self.agents.items()
            }
        }


# ===== 使用示例 =====

def demo():
    # --- Multi-Agent 方式 ---
    print("=" * 50)
    print("Multi-Agent 方式（手动编排）")
    print("=" * 50)

    ma = SimpleMultiAgentSystem()
    ma.register_agent("coder", "编码")
    ma.register_agent("reviewer", "代码审查")
    ma.register_agent("tester", "测试")

    result = ma.execute_pipeline(
        "开发用户登录功能",
        ["coder", "reviewer", "tester"]  # 需要手动指定顺序
    )
    for step in result["steps"]:
        print(f"  {step}")

    # --- Managed Agent 方式 ---
    print("\n" + "=" * 50)
    print("Managed Agent 方式（自动管理）")
    print("=" * 50)

    platform = ManagedAgentPlatform()
    platform.register_agent(ManagedAgent(
        name="FrontendBot",
        skills=["React", "TypeScript", "UI设计"]
    ))
    platform.register_agent(ManagedAgent(ManagedAgent(
        name="BackendBot",
        skills=["Python", "API设计", "数据库"]
    )))
    platform.register_agent(ManagedAgent(
        name="QABot",
        skills=["测试", "性能分析", "安全审查"]
    ))

    # 提交任务，平台自动分配
    platform.submit_task(ManagedTask(
        task_id="T001",
        description="开发用户登录 API",
        required_skills=["Python", "API设计"],
        priority=TaskPriority.HIGH
    ))

    platform.submit_task(ManagedTask(
        task_id="T002",
        description="实现登录页面 UI",
        required_skills=["React", "TypeScript"],
        priority=TaskPriority.MEDIUM
    ))

    platform.submit_task(ManagedTask(
        task_id="T003",
        description="安全渗透测试",
        required_skills=["安全审查", "测试"],
        priority=TaskPriority.CRITICAL
    ))

    report = platform.get_team_report()
    print(f"团队报告：{report['total_agents']} 个 Agent，"
          f"完成 {report['total_completed']} 个任务")
    for name, info in report["agents"].items():
        print(f"  {name}: 完成 {info['completed']} 个任务，"
              f"绩效 {info['performance']}，"
              f"累积知识 {info['knowledge_count']} 项")


if __name__ == "__main__":
    demo()

"""
Managed Agent 平台核心实现
包含：任务分解、Agent 调度、执行监控、经验累积
"""

import asyncio
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Callable, Any
from enum import Enum
import json
import time


class TaskStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    BLOCKED = "blocked"


@dataclass
class SubTask:
    id: str
    description: str
    required_skills: List[str]
    depends_on: List[str] = field(default_factory=list)
    status: TaskStatus = TaskStatus.PENDING
    assigned_to: Optional[str] = None
    result: Optional[str] = None
    error: Optional[str] = None


@dataclass
class AgentProfile:
    name: str
    skills: List[str]
    current_tasks: List[str] = field(default_factory=list)
    max_concurrent: int = 2
    completed_count: int = 0
    failed_count: int = 0
    skill_improvements: Dict[str, float] = field(default_factory=dict)


class ManagedAgentPlatform:
    """Managed Agent 平台核心实现"""

    def __init__(self, planner_fn=None, executor_fn=None):
        self.agents: Dict[str, AgentProfile] = {}
        self.tasks: Dict[str, SubTask] = {}
        self.task_dependencies: Dict[str, List[str]] = {}
        self.knowledge_base: Dict[str, List[Dict]] = {}
        self.execution_log: List[Dict] = []

        # 可插拔的规划器和执行器
        self.planner_fn = planner_fn or self._default_planner
        self.executor_fn = executor_fn or self._default_executor

    def register_agent(self, profile: AgentProfile):
        """注册 Agent"""
        self.agents[profile.name] = profile

    async def submit_project(self, project_name: str, description: str) -> List[str]:
        """提交项目：自动分解 + 调度 + 执行"""
        # 1. 任务分解
        subtasks = await self.planner_fn(project_name, description)

        # 2. 注册任务
        task_ids = []
        for st in subtasks:
            self.tasks[st.id] = st
            task_ids.append(st.id)

        # 3. 构建依赖图
        self._build_dependency_graph()

        # 4. 执行循环
        await self._execution_loop()

        return task_ids

    async def _default_planner(
        self, project_name: str, description: str
    ) -> List[SubTask]:
        """
        默认规划器：基于规则的简单分解
        生产环境应替换为 LLM 驱动的智能规划器
        """
        # 简化示例：固定分解模式
        return [
            SubTask(
                id=f"{project_name}-design",
                description=f"设计 {description} 的架构",
                required_skills=["架构设计", "系统设计"]
            ),
            SubTask(
                id=f"{project_name}-impl",
                description=f"实现 {description}",
                required_skills=["编码", "开发"],
                depends_on=[f"{project_name}-design"]
            ),
            SubTask(
                id=f"{project_name}-test",
                description=f"测试 {description}",
                required_skills=["测试", "QA"],
                depends_on=[f"{project_name}-impl"]
            ),
        ]

    def _build_dependency_graph(self):
        """构建任务依赖图"""
        for task_id, task in self.tasks.items():
            self.task_dependencies[task_id] = task.depends_on

    def _get_ready_tasks(self) -> List[str]:
        """获取可以执行的任务（依赖已满足）"""
        ready = []
        for task_id, task in self.tasks.items():
            if task.status != TaskStatus.PENDING:
                continue
            deps = self.task_dependencies.get(task_id, [])
            if all(
                self.tasks[d].status == TaskStatus.COMPLETED
                for d in deps
            ):
                ready.append(task_id)
        return ready

    def _assign_task(self, task_id: str) -> Optional[str]:
        """为任务分配合适的 Agent"""
        task = self.tasks[task_id]
        candidates = []

        for name, agent in self.agents.items():
            # 检查技能
            skill_match = sum(
                1 for s in task.required_skills if s in agent.skills
            ) / max(len(task.required_skills), 1)

            # 检查负载
            if len(agent.current_tasks) >= agent.max_concurrent:
                continue

            # 计算匹配分数
            success_rate = agent.completed_count / max(
                agent.completed_count + agent.failed_count, 1
            )
            score = skill_match * 0.7 + success_rate * 0.3

            candidates.append((name, score))

        if not candidates:
            return None

        candidates.sort(key=lambda x: x[1], reverse=True)
        return candidates[0][0]

    async def _execution_loop(self):
        """主执行循环"""
        max_iterations = 100
        iteration = 0

        while iteration < max_iterations:
            iteration += 1

            # 获取可执行任务
            ready = self._get_ready_tasks()
            if not ready:
                # 检查是否全部完成
                all_done = all(
                    t.status in (TaskStatus.COMPLETED, TaskStatus.FAILED)
                    for t in self.tasks.values()
                )
                if all_done:
                    break
                await asyncio.sleep(0.1)
                continue

            # 分配并执行
            for task_id in ready:
                agent_name = self._assign_task(task_id)
                if agent_name:
                    await self._execute_with_agent(task_id, agent_name)

        # 执行完成后提取经验
        self._extract_experience()

    async def _execute_with_agent(self, task_id: str, agent_name: str):
        """使用指定 Agent 执行任务"""
        task = self.tasks[task_id]
        agent = self.agents[agent_name]

        task.status = TaskStatus.RUNNING
        task.assigned_to = agent_name
        agent.current_tasks.append(task_id)

        try:
            # 注入相关知识
            relevant_knowledge = self._get_relevant_knowledge(
                task.required_skills
            )

            # 执行
            result = await self.executor_fn(
                task, agent, relevant_knowledge
            )

            task.result = result
            task.status = TaskStatus.COMPLETED
            agent.completed_count += 1

            # 记录执行日志
            self.execution_log.append({
                "task": task_id,
                "agent": agent_name,
                "status": "completed",
                "timestamp": time.time()
            })

        except Exception as e:
            task.error = str(e)
            task.status = TaskStatus.FAILED
            agent.failed_count += 1

        finally:
            agent.current_tasks.remove(task_id)

    def _get_relevant_knowledge(self, skills: List[str]) -> List[Dict]:
        """获取相关知识"""
        knowledge = []
        for skill in skills:
            if skill in self.knowledge_base:
                knowledge.extend(self.knowledge_base[skill])
        return knowledge

    def _extract_experience(self):
        """从已完成任务中提取经验"""
        for task_id, task in self.tasks.items():
            if task.status != TaskStatus.COMPLETED or not task.result:
                continue

            # 提取技能相关经验
            for skill in task.required_skills:
                if skill not in self.knowledge_base:
                    self.knowledge_base[skill] = []

                self.knowledge_base[skill].append({
                    "task": task.description,
                    "agent": task.assigned_to,
                    "result_summary": task.result[:100],
                    "timestamp": time.time()
                })

    async def _default_executor(
        self, task: SubTask, agent: AgentProfile, knowledge: List[Dict]
    ) -> str:
        """默认执行器：模拟执行"""
        await asyncio.sleep(0.05)  # 模拟执行时间
        return f"完成: {task.description}"

    def get_status_report(self) -> Dict:
        """生成项目状态报告"""
        total = len(self.tasks)
        completed = sum(
            1 for t in self.tasks.values()
            if t.status == TaskStatus.COMPLETED
        )
        failed = sum(
            1 for t in self.tasks.values()
            if t.status == TaskStatus.FAILED
        )

        return {
            "progress": f"{completed}/{total}",
            "success_rate": f"{completed/max(completed+failed,1)*100:.0f}%",
            "agents": {
                name: {
                    "completed": a.completed_count,
                    "failed": a.failed_count,
                    "active_tasks": len(a.current_tasks)
                }
                for name, a in self.agents.items()
            },
            "knowledge_base_size": {
                k: len(v) for k, v in self.knowledge_base.items()
            }
        }


async def demo_platform():
    """演示 Managed Agent 平台"""
    platform = ManagedAgentPlatform()

    # 注册 Agent 团队
    platform.register_agent(AgentProfile(
        name="Architect",
        skills=["架构设计", "系统设计", "技术选型"],
        max_concurrent=1
    ))
    platform.register_agent(AgentProfile(
        name="Developer",
        skills=["编码", "开发", "代码审查"],
        max_concurrent=3
    ))
    platform.register_agent(AgentProfile(
        name="Tester",
        skills=["测试", "QA", "性能分析"],
        max_concurrent=2
    ))

    # 提交项目
    task_ids = await platform.submit_project(
        "blog-system",
        "带用户认证的博客系统"
    )

    # 查看报告
    report = platform.get_status_report()
    print(json.dumps(report, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    asyncio.run(demo_platform())