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一致性
• 生成质控报告

通信协议

# 消息类型
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"
}

工作流程

# 伪代码
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

智能决策示例

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))

错误恢复机制

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. 详细日志和报告 (每个物种独立报告)

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