多Agent任务分解粒度：六引擎从粗到精实测

# TDAG核心循环伪代码
class TDAGPlanner:
    def decompose_and_generate(self, task):
        subtasks = self.llm_decompose(task)  # LLM驱动的分解
        agents = []
        for sub in subtasks:
            # 根据子任务描述动态生成Agent角色
            agent_spec = self.generate_agent_spec(sub)
            agent = self.create_agent(agent_spec)
            agents.append(agent)

        # 执行并监控，失败时动态调整
        while not all_complete(subtasks):
            result = self.execute_round(subtasks, agents)
            if result.has_failure:
                subtasks, agents = self.replan(result.failures)
        return aggregate(results)

# DynTaskMAS的DAG构建核心
class DynamicTaskGraph:
    def __init__(self):
        self.nodes = {}      # task_id -> TaskNode
        self.edges = {}      # dependency edges
        self.agent_pool = [] # available agents

    def add_task(self, task, parent_ids=None):
        node = TaskNode(task, status='pending')
        self.nodes[task.id] = node
        for pid in parent_ids:
            self.add_edge(pid, task.id)
        # 关键：根据当前负载决定是否进一步分解
        if self.estimate_complexity(node) > THRESHOLD:
            subtasks = self.split(node)
            return [self.add_task(st, [task.id]) for st in subtasks]

    def reschedule(self, failed_node):
        # 动态图重配置：失败节点的下游全部重新评估
        downstream = self.get_downstream(failed_node.id)
        for node in downstream:
            if node.status == 'pending':
                self.reassign_or_split(node)

# TALM树状分解与记忆检索
class TALMTree:
    def decompose(self, task, depth=0, max_depth=3):
        # 从长期记忆检索类似任务的分解方案
        historical = self.memory.search_similar(task.description)
        if historical and historical.success_rate > 0.85:
            return historical.decomposition  # 复用历史方案

        # 否则动态分解
        if depth >= max_depth:
            return [task]  # 叶子节点，不再分解

        children = self.llm_split(task)
        tree_nodes = []
        for child in children:
            subtree = self.decompose(child, depth+1, max_depth)
            tree_nodes.extend(subtree)

        # 记录分解方案用于未来复用
        self.memory.store(task, tree_nodes, pending=True)
        return tree_nodes

# MATDS约束感知分解
class MATDSDecomposer:
    def __init__(self, agent_registry):
        self.registry = agent_registry  # Agent能力画像

    def constrained_decompose(self, task):
        # 阶段1：获取可用Agent的能力约束
        constraints = self.registry.get_constraints()
        # 例：Agent_A: {tools: ['code', 'search'], max_tokens: 8k}
        #     Agent_B: {tools: ['browser'], max_tokens: 4k}

        # 阶段2：约束驱动的分解
        subtasks = []
        remaining = task
        for agent_id, cap in constraints.items():
            assignable = self.extract_assignable(remaining, cap)
            if assignable:
                subtasks.append(SubTask(assignable, agent_id))
                remaining = remaining - assignable

        # 阶段3：时间依赖排序
        return self.topological_sort(subtasks)

# Utility-Aware任务交换
class UtilityAgent:
    def evaluate_utility(self, subtask):
        """计算自身对某个子任务的效用值"""
        capability_match = self.compute_skill_match(subtask.required_skills)
        context_cost = len(subtask.context) / self.max_context
        historical_success = self.memory.get_success_rate(subtask.type)
        return (capability_match * 0.5 + 
                (1 - context_cost) * 0.3 + 
                historical_success * 0.2)

    def propose_exchange(self, my_task, other_agent, their_task):
        my_utility_theirs = self.evaluate_utility(their_task)
        their_utility_mine = other_agent.evaluate_utility(my_task)
        my_utility_mine = self.evaluate_utility(my_task)
        their_utility_theirs = other_agent.evaluate_utility(their_task)

        # 帕累托改进：交换后双方都不变差，至少一方变好
        return (my_utility_theirs >= my_utility_mine and 
                their_utility_mine >= their_utility_theirs)

graph TB
    subgraph "任务输入层"
        T[复杂任务输入]
        TC[任务复杂度评估器]
    end

    subgraph "分解决策层"
        TDAG_P[TDAG Planner]
        DTG[DynTaskMAS DAG Builder]
        DRAMA_M[DRAMA Modular Splitter]
        TALM_T[TALM Tree Decomposer]
        MATDS_C[MATDS Constraint Analyzer]
        UA_A[Utility-Aware Allocator]
    end

    subgraph "粒度控制器"
        GC[粒度评估器]
        DB[(粒度历史DB)]
        GC -->|complexity > threshold| GC
        GC -->|subtask/agent ratio check| DB
    end

    subgraph "Agent执行层"
        A1[Agent-1: 代码生成]
        A2[Agent-2: 文档分析]
        A3[Agent-3: 测试验证]
        A4[Agent-4: 搜索调研]
        A5[Agent-N: 动态生成]
    end

    subgraph "反馈与重分配层"
        HM[健康监控]
        RES[重分配引擎]
        MEM[(长期记忆)]
    end

    T --> TC
    TC -->|动态分解| TDAG_P
    TC -->|图构建| DTG
    TC -->|模块化| DRAMA_M
    TC -->|树状| TALM_T
    TC -->|约束感知| MATDS_C

    TDAG_P --> GC
    DTG --> GC
    DRAMA_M --> GC
    TALM_T --> GC
    MATDS_C --> GC
    UA_A --> GC

    GC --> A1 & A2 & A3 & A4 & A5

    A1 & A2 & A3 & A4 & A5 --> HM
    HM -->|失败/超时| RES
    RES --> GC
    HM --> MEM

    style GC fill:#ff6b6b,color:#fff
    style HM fill:#4ecdc4,color:#fff
    style RES fill:#45b7d1,color:#fff

graph LR
    A[输入任务] --> B{复杂度评估}
    B -->|低| C[不分解：单Agent执行]
    B -->|中| D[一级分解：2-5个子任务]
    B -->|高| E[二级分解：5-15个子任务]
    B -->|极高| F[三级分解：15-30个子任务]

    D --> G{子任务/Agent比}
    E --> G
    F --> G

    G -->|< 1.5x| H[合并：粒度过粗]
    G -->|1.5x - 3x| I[最优区间：执行]
    G -->|> 3x| J[聚合：粒度过细]

    H --> D
    J --> E

    style I fill:#51cf66,color:#000
    style H fill:#ffd43b,color:#000
    style J fill:#ffd43b,color:#000

graph TD
    START[开始选型] --> Q1{Agent数量?}
    Q1 -->|< 5个| Q2{任务是否重复性高?}
    Q1 -->|5-20个| Q3{Agent能力是否异构?}
    Q1 -->|> 20个| Q4{是否需要实时响应?}

    Q2 -->|是| TALM_R[选TALM: 记忆复用节省规划时间]
    Q2 -->|否| TDAG_R[选TDAG: 轻量级动态分解]

    Q3 -->|是| MATDS_R[选MATDS: 约束感知匹配]
    Q3 -->|否| DTG_R[选DynTaskMAS: DAG并行优化]

    Q4 -->|是| DRAMA_R[选DRAMA: 弹性容错编排]
    Q4 -->|否| UA_R[选Utility-Aware: 市场化最优分配]

    style TALM_R fill:#69db7c
    style TDAG_R fill:#69db7c
    style MATDS_R fill:#69db7c
    style DTG_R fill:#69db7c
    style DRAMA_R fill:#69db7c
    style UA_R fill:#69db7c