PlantAI-dev/docs/multi_agent_genome_prepare.md

# 多智能体基因组预处理架构设计

## 当前问题

1. **单线程处理**：当前 `genome_prepare` 是顺序处理多个物种
2. **无智能决策**：无法自动选择最佳处理策略
3. **错误恢复弱**：一个物种失败不影响其他，但没有智能重试机制
4. **资源利用低**：没有并行处理多个物种

## 多智能体架构设计

### 架构概览

```
┌─────────────────────────────────────────────────────────────┐
│                    Master Agent (协调者)                     │
│  - 任务分解与分配                                             │
│  - 结果汇总                                                   │
│  - 异常处理                                                   │
└──────────────────────┬──────────────────────────────────────┘
                       │
        ┌──────────────┼──────────────┐
        │              │              │
┌───────▼──────┐ ┌────▼─────┐ ┌──────▼───────┐
│ Species Agent│ │ Species  │ │ Species Agent│
│    (Ath)     │ │ Agent    │ │    (Os)      │
│              │ │  (Ft)    │ │              │
└───────┬──────┘ └────┬─────┘ └──────┬───────┘
        │             │              │
        └─────────────┼──────────────┘
                      │
        ┌─────────────┼─────────────┐
        │             │             │
┌───────▼─────┐ ┌────▼────┐ ┌──────▼──────┐
| File Agent  │ │ File    │ │ Quality     │
│  (格式识别)  │ │ Agent   │ │ Agent       │
│             │ │ (解压)  │ │ (质控检查)  │
└─────────────┘ └─────────┘ └─────────────┘
```

### Agent角色定义

#### 1. Master Agent (主控智能体)

**职责**：
- 接收用户输入目录
- 扫描并识别所有物种
- 创建 Species Agent 子任务
- 监控所有子任务状态
- 汇总结果并生成报告

**状态机**：
```
[Idle] -> [Scanning] -> [Dispatching] -> [Monitoring] -> [Aggregating] -> [Complete]
                              |              |                |
                              v              v                v
                         [Failed]      [Retrying]        [Partial]
```

#### 2. Species Agent (物种智能体)

**职责**：
- 处理单个物种的所有文件
- 自动检测数据库类型
- 选择最佳处理策略
- 执行预处理流程
- 自我验证结果

**工具**：
- `detect_db_type()`: 检测JGI/Ensembl/NCBI
- `decompress_files()`: 解压压缩文件
- `extract_gene_ids()`: 提取基因ID
- `filter_gff()`: 过滤GFF文件
- `validate_output()`: 验证输出完整性

#### 3. File Agent (文件智能体)

**职责**：
- 处理单个文件的特定操作
- 格式转换
- 压缩/解压
- 格式验证

#### 4. Quality Agent (质控智能体)

**职责**：
- 检查结果文件完整性
- 验证序列格式
- 检查基因ID一致性
- 生成质控报告

### 通信协议

```python
# 消息类型
class MessageType(Enum):
    TASK_ASSIGN = "task_assign"      # 分配任务
    TASK_COMPLETE = "task_complete"  # 任务完成
    TASK_FAILED = "task_failed"      # 任务失败
    STATUS_QUERY = "status_query"    # 状态查询
    RESOURCE_REQUEST = "resource_request"  # 资源请求

# 消息格式
{
    "msg_id": "uuid",
    "from_agent": "agent_name",
    "to_agent": "agent_name",
    "type": "message_type",
    "payload": {...},
    "timestamp": "iso8601"
}
```

### 工作流程

```python
# 伪代码
async def master_workflow(input_dir: Path):
    # 1. 扫描输入目录
    species_list = await scan_input_dir(input_dir)
    
    # 2. 创建Species Agent
    agents = []
    for species in species_list:
        agent = SpeciesAgent(species)
        agents.append(agent)
    
    # 3. 并行执行
    results = await asyncio.gather(
        *[agent.run() for agent in agents],
        return_exceptions=True
    )
    
    # 4. 结果汇总
    report = aggregate_results(results)
    
    return report
```

### 智能决策示例

```python
class SpeciesAgent:
    async def decide_strategy(self) -> Strategy:
        """智能决策处理策略"""
        
        # 1. 检查文件完整性
        if not self.has_minimum_files():
            # 尝试从其他来源补充
            if await self.try_download_missing():
                return Strategy.DOWNLOAD_AND_PROCESS
            else:
                return Strategy.SKIP_WITH_WARNING
        
        # 2. 检测数据库类型
        db_type = self.detect_db_type()
        
        # 3. 根据类型选择策略
        if db_type == "jgi":
            if self.has_cds():
                return Strategy.JGI_FULL
            else:
                return Strategy.JGI_PROTEIN_ONLY
        elif db_type == "ensembl":
            return Strategy.ENSEMBL_LONGEST
        elif db_type == "ncbi":
            return Strategy.NCBI_REFSEQ
        else:
            # 未知类型，尝试通用策略
            return Strategy.GENERIC
    
    async def run(self):
        """执行处理"""
        strategy = await self.decide_strategy()
        
        try:
            result = await self.execute_strategy(strategy)
            
            # 自我验证
            if not self.validate_result(result):
                # 尝试修复
                result = await self.attempt_repair(result)
            
            return TaskResult(success=True, data=result)
            
        except Exception as e:
            # 尝试降级策略
            if strategy != Strategy.GENERIC:
                return await self.run_fallback()
            else:
                return TaskResult(success=False, error=str(e))
```

### 错误恢复机制

```python
class ErrorRecovery:
    """错误恢复策略"""
    
    @staticmethod
    async def handle_decompression_error(agent, file_path: Path):
        """处理解压错误"""
        # 尝试不同解压工具
        for tool in ['pigz', 'gzip', 'zcat']:
            try:
                result = await agent.decompress_with_tool(file_path, tool)
                if result.success:
                    return result
            except:
                continue
        
        # 所有工具都失败，标记为失败
        raise DecompressionError(f"Failed to decompress {file_path}")
    
    @staticmethod
    async def handle_missing_cds(agent):
        """处理缺少CDS文件"""
        # 策略1: 从基因组+GFF提取
        if agent.has_genome() and agent.has_gff():
            return await agent.extract_cds_from_genome()
        
        # 策略2: 使用protein作为占位
        logger.warning("Using protein as CDS placeholder")
        return await agent.use_protein_as_cds()
```

### 实现建议

#### 阶段1: 基础多智能体 (当前可实现)

1. **Species Agent 并行化**
   - 使用 `asyncio` 或 `multiprocessing` 并行处理多个物种
   - 每个 Species Agent 独立处理一个物种

2. **智能决策**
   - 根据文件自动选择处理策略
   - 错误时自动尝试修复

3. **Master Agent 协调**
   - 监控所有 Species Agent 状态
   - 汇总结果生成报告

#### 阶段2: 完整多智能体 (未来扩展)

1. **引入消息队列**
   - 使用 Redis/RabbitMQ 作为 Agent 间通信
   - 支持分布式部署

2. **LLM 增强决策**
   - 使用 LLM 分析错误原因
   - 智能推荐修复策略

3. **学习优化**
   - 记录每次处理的成功/失败
   - 优化策略选择算法

### 代码结构

```
bbav20311/
├── agents/
│   ├── __init__.py
│   ├── base_agent.py          # 基础Agent类
│   ├── master_agent.py        # 主控Agent
│   ├── species_agent.py       # 物种Agent
│   ├── file_agent.py          # 文件Agent
│   └── quality_agent.py       # 质控Agent
├── messaging/
│   ├── __init__.py
│   ├── message_bus.py         # 消息总线
│   └── protocol.py            # 通信协议
├── strategies/
│   ├── __init__.py
│   ├── base_strategy.py       # 基础策略
│   ├── jgi_strategy.py        # JGI策略
│   ├── ensembl_strategy.py    # Ensembl策略
│   └── ncbi_strategy.py       # NCBI策略
└── skills/
    └── genome_prepare_v2/     # 技能化封装
        ├── __init__.py
        ├── skill.yaml
        └── agent_workflow.py
```

### 立即可实施的改进

1. **并行处理多个物种** (使用 `asyncio`)
2. **智能策略选择** (根据文件自动选择)
3. **错误自动恢复** (尝试多种修复策略)
4. **详细日志和报告** (每个物种独立报告)

这些改进可以在不改变现有架构的情况下实现，为未来的完整多智能体架构奠定基础。