Files

293 lines
9.3 KiB
Markdown
Raw Permalink Normal View History

2026-08-03 18:42:47 +08:00
# 多智能体基因组预处理架构设计
## 当前问题
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. **详细日志和报告** (每个物种独立报告)
这些改进可以在不改变现有架构的情况下实现,为未来的完整多智能体架构奠定基础。