PlantAI-dev/IDENTIFY_AGENT_FLOW.md

# identify 阶段多智能体数据流转详解

## 概述

本文档详细分析 identify 阶段从用户输入到最终输出的 10 个 Agent 数据流转，包括每个 Agent 的输入、输出、数据类型，以及错误处理机制。

---

## 1. 用户输入 -> Router Agent

### 输入数据
```json
{
  "query": "鉴定Ft.SPL基因家族",
  "user_input": {
    "prefix": "Ft.SPL",
    "stage": "identify",
    "species_name": "Fagopyrum tataricum"
  },
  "job_id": "job_xxx"
}
```

### Router Agent 处理
1. 调用 `RouterAgent.classify()` 判断模式: `analysis`
2. 如果是单 stage 模式（`user_input.stage` 存在），直接设置 intent:
   - `stages = ["identify"]`
   - `prefix = "Ft.SPL"`
3. 否则调用 `IntentAgentV2` 解析意图

### 输出数据（设置到 state）
```python
state["mode"] = "analysis"
state["intent_result"] = {
    "mode": "analysis",
    "stages": ["identify"],
    "prefix": "Ft.SPL",
    "single_stage_mode": True,
    "job_id": "job_xxx",
    "workspace": "/path/to/workspace/job_xxx"
}
state["next_node"] = "learning_recommend"
```

---

## 2. Router -> Learning Recommend Agent

### 输入数据（从 state 获取）
```python
intent = state.get("intent_result", {})
target_stages = intent.get("stages", ["identify"])  # ["identify"]
```

### Learning Recommend Agent 处理
- 查询历史执行记录，获取参数推荐
- 如果有高置信度历史记录，返回推荐参数
- 否则返回默认参数

### 输出数据（设置到 state）
```python
state["learning_recommendations"] = {
    "confidence": "high",  # 或 "medium" / "low"
    "params": {
        "evalue": "1e-5",
        "coverage": 0.5
    },
    "message": "基于历史执行推荐参数"
}
```

---

## 3. Learning Recommend -> Planning Agent

### 输入数据（从 state 获取）
```python
state["intent_result"]  # 包含 stages, prefix 等
state["learning_recommendations"]  # 参数推荐
state["completed_stages"]  # 已完成的阶段（从 manifest 查询）
```

### Planning Agent 处理
1. 重新查询 `completed_stages`（从文件系统 manifest）
2. 根据 intent 和 learning_recommendations 制定执行计划
3. 设置 stages_to_run = ["identify"]（排除已完成的）

### 输出数据（设置到 state）
```python
state["plan_result"] = {
    "stages": [
        {
            "id": "identify",
            "name": "基因鉴定",
            "params": {
                "evalue": "1e-5",
                "coverage": 0.5
            }
        }
    ],
    "prefix": "Ft.SPL",
    "workspace": "/path/to/workspace/job_xxx",
    "job_id": "job_xxx"
}
```

---

## 4. Planning -> PreFlight Agent

### 输入数据（从 state 获取）
```python
plan = state.get("plan_result", {})
job_id = state.get("job_id", "")
prefix = intent.get("prefix", "Ft.SPL")
```

### PreFlight Agent 处理
1. 检查输入文件是否存在：
   - `input/{prefix}.pep.fa`（参考序列）
   - `input/{species_prefix}.pep.fa`（物种序列）
2. 检查契约文件是否存在
3. 检查环境配置

### 输出数据（设置到 state）
```python
state["preflight_result"] = {
    "valid": True,  # 或 False
    "issues": [],   # 如果 valid=False，列出问题
    "checked_files": [
        "input/Ft.SPL.pep.fa: EXISTS",
        "input/Ft.pep.fa: EXISTS"
    ]
}
```

### 错误处理
- 如果 `preflight_result.valid = False`:
  - 记录 `state["errors"]`
  - Planning Agent 可能重新规划（目前未实现）

---

## 5. PreFlight -> Execution Agent 【关键】

### 输入数据（从 state 获取）
```python
plan = state.get("plan_result", {})
# plan 包含:
# - stages: [{"id": "identify", "params": {...}}]
# - prefix: "Ft.SPL"
# - workspace: "/path/to/workspace/job_xxx"
# - job_id: "job_xxx"
```

### Execution Agent 处理
1. 解析 plan，获取 `stages_to_run = ["identify"]`
2. 调用 `stage_registry.execute("identify", job_id, prefix, params)`
3. 执行 `python apps/cli/run_identify.py --prefix Ft.SPL --workspace ...`
4. 等待执行完成，读取 manifest

### 执行结果结构（关键）
```python
result = {
    "status": "success",  # 或 "error"
    "manifest": {...},    # manifest.json 内容
    "manifest_path": "/path/to/manifest.json",
    "elapsed_sec": 45.2,
    "metrics": {
        "final_members": 15,
        "initial_candidates": 120
    }
}
```

### 输出数据（设置到 state）【问题点】
```python
execution_results = [
    {
        "stage": "identify",
        "result": {
            "status": "success",
            "manifest": {...},
            "manifest_path": "...",
            "message": "✅ Identify completed successfully"
        }
    }
]
state["execution_results"] = execution_results  # LIST 类型
```

### **LangGraph State 传递问题**
- `_execution_node` 设置 `state["execution_results"] = execution_results`（list）
- LangGraph 调用 `return state` 返回给下一个节点
- **TypedDict 类型定义必须匹配**: `execution_results: list[dict[str, Any]]`
- 如果类型不匹配，LangGraph 可能无法正确传递

---

## 6. Execution -> ResultParser Agent

### 输入数据（从 state 获取）
```python
execution_results = state.get("execution_results", [])  # LIST
```

### **Fallback 机制（关键）**
如果 `execution_results` 为空（LangGraph state 传递失败）：
```python
if not execution_results:
    # 从文件系统重新加载 manifest
    workspace = Path(f"/path/to/workspace/{job_id}/final")
    for manifest_file in workspace.glob(f"{prefix}.*.manifest.json"):
        manifest = json.load(manifest_file)
        execution_results.append({
            "stage": manifest.get("stage"),
            "result": {
                "status": manifest.get("status", "success"),
                "manifest": manifest
            }
        })
```

### ResultParser Agent 处理
- 解析 execution_results，生成结构化报告
- 提取关键指标：final_members, initial_candidates 等

### 输出数据（设置到 state）
```python
state["parse_result"] = {
    "stage": "identify",
    "status": "success",
    "summary": "鉴定到 15 个 Ft.SPL 基因家族成员",
    "metrics": {
        "final_members": 15,
        "initial_candidates": 120
    },
    "files": [
        "final/Ft.SPL.identify.pep.fa",
        "final/Ft.SPL.identify.manifest.json"
    ]
}
```

---

## 7. ResultParser -> Validator Agent 【关键决策点】

### 输入数据（从 state 获取）
```python
execution_results_list = state.get("execution_results", [])  # LIST
parse_result = state.get("parse_result", {})
```

### **Fallback 机制（再次确认）**
如果 `execution_results_list` 为空：
```python
# 再次从 manifest 加载（双重保险）
```

### Validator Agent 处理
1. 检查 `execution_result.status == "success"`
2. 检查输出文件是否存在
3. 检查数据合理性（final_members > 0）
4. 检查 manifest 完整性

### 输出数据（设置到 state）
```python
state["validation_result"] = {
    "overall_passed": True,  # 或 False
    "stage": "identify",
    "score": 95.0,
    "errors": [],     # 如果 overall_passed=False
    "warnings": [],   # 可能有警告但不影响通过
    "summary": "✅ 验证通过，所有检查项正常"
}
```

---

## 8. Validator -> Decision: _validate_decision

### 决策逻辑
```python
def _validate_decision(state: GraphState) -> str:
    validation = state.get("validation_result", {})
    
    if validation.get("overall_passed", False):
        return "success"  # -> Discovery Agent
    else:
        return "failure"  # -> Diagnosis Agent
```

### **关键问题根因**
如果 `validation_result` 为空 `{}`（LangGraph state 传递失败）：
- `validation.get("overall_passed", False)` 返回 `False`
- 决策走 `"failure"` -> Diagnosis Agent
- 触发错误诊断 -> retry

---

## 9. Validator(success) -> Discovery Agent

### 输入数据（从 state 获取）
```python
execution_results = state.get("execution_results", [])
parse_result = state.get("parse_result", {})
```

### Discovery Agent 处理
- 分析鉴定结果，生成科学发现报告
- 例如：发现新的 SPL 基因亚家族

### 输出数据（设置到 state）
```python
state["discovery_result"] = {
    "findings": ["发现 15 个 SPL 基因，分为 3 个亚家族"],
    "should_continue": False,  # 是否继续迭代
    "updated_plan": None       # 如果继续迭代，新的计划
}
```

---

## 10. Discovery -> Decision: _discovery_decision

### 决策逻辑
```python
def _discovery_decision(state: GraphState) -> str:
    discovery = state.get("discovery_result", {})
    iteration_count = state.get("iteration_count", 0)
    max_iterations = state.get("max_iterations", 3)
    
    if iteration_count >= max_iterations:
        return "learning"
    
    if discovery.get("should_continue", False):
        return "replan"  # -> Planning（循环迭代）
    else:
        return "learning"  # -> Learning Agent
```

---

## 11. Discovery -> Learning Agent

### 输入数据（从 state 获取）
```python
execution_results = state.get("execution_results", [])
discovery_result = state.get("discovery_result", {})
```

### Learning Agent 处理
- 记录本次执行的参数、结果
- 更新知识库
- 生成学习总结

### 输出数据（设置到 state）
```python
state["final_output"] = {
    "response": "鉴定完成，发现 15 个 SPL 基因",
    "artifacts": [
        "final/Ft.SPL.identify.pep.fa",
        "final/Ft.SPL.identify.manifest.json"
    ],
    "metrics": {...}
}
state["status"] = "completed"
```

---

## 错误处理路径

### Validator 验证失败 -> Diagnosis Agent

### Diagnosis Agent 处理
```python
state["diagnosis_result"] = {
    "error_type": "missing_output_file",
    "suggested_fix": "检查输入文件路径",
    "retry_params": {...}
}
state["retry_count"] += 1
```

### Diagnosis -> Decision: _diagnosis_decision
```python
if retry_count < max_retries:
    return "retry"  # -> Execution Agent（重新执行）
else:
    return "fail"   # -> END（失败结束）
```

---

## LangGraph State 传递问题详解

### TypedDict 类型定义（state.py）
```python
class GraphState(TypedDict):
    execution_results: list[dict[str, Any]]  # LIST 类型（已修复）
    validation_result: dict[str, Any]        # DICT 类型
    ...
```

### LangGraph 内部机制
LangGraph 使用 TypedDict 进行类型检查：
1. 节点函数返回 `state`（dict）
2. LangGraph 检查返回值是否符合 TypedDict 定义
3. 如果类型不匹配，可能导致：
   - 数据丢失（字段被忽略）
   - 类型转换错误
   - 默认值覆盖

### 已修复的问题
之前 `execution_results` 定义为 `dict[str, Any]`，但实际使用 `list[dict]`，导致：
- `_execution_node` 设置 list
- `_validator_node` 获取空 `{}` 或 `[]`
- fallback 机制从 manifest 加载（但可能加载失败）
- `_validate_decision` 判断 failure

---

## 数据流转图（简化版）

```
用户输入
    ↓
[Router] → state["intent_result"] = {stages, prefix, mode}
    ↓
[Learning Recommend] → state["learning_recommendations"] = {params}
    ↓
[Planning] → state["plan_result"] = {stages, prefix, workspace}
    ↓
[PreFlight] → state["preflight_result"] = {valid, issues}
    ↓
[Execution] → state["execution_results"] = [{stage, result}]  ← 关键
    ↓
[ResultParser] → state["parse_result"] = {summary, metrics}  ← 有 fallback
    ↓
[Validator] → state["validation_result"] = {overall_passed}   ← 有 fallback
    ↓
[_validate_decision] → "success" / "failure"
    ↓
[Discovery] (success) → state["discovery_result"]
    ↓
[Learning] → state["final_output"]
    ↓
END

OR (failure):
[Validator] → "failure"
    ↓
[Diagnosis] → state["diagnosis_result"] + retry_count++
    ↓
[_diagnosis_decision] → "retry" (if retry < max) / "fail" (if max reached)
    ↓
[Execution] (retry) → 重新执行
```

---

## 关键数据类型总结

| 字段名 | 类型 | 设置节点 | 使用节点 |
|--------|------|----------|----------|
| `intent_result` | `dict[str, Any]` | Router | Planning |
| `learning_recommendations` | `dict[str, Any]` | Learning Recommend | Planning |
| `plan_result` | `dict[str, Any]` | Planning | PreFlight, Execution |
| `preflight_result` | `dict[str, Any]` | PreFlight | (记录 errors) |
| `execution_results` | `list[dict[str, Any]]` | Execution | ResultParser, Validator, Diagnosis |
| `parse_result` | `dict[str, Any]` | ResultParser | Validator |
| `validation_result` | `dict[str, Any]` | Validator | _validate_decision |
| `diagnosis_result` | `dict[str, Any]` | Diagnosis | _diagnosis_decision |
| `discovery_result` | `dict[str, Any]` | Discovery | _discovery_decision |

---

## 测试验证步骤

1. 启动 Celery 服务
2. 前端触发 identify 任务
3. 观察 Celery 日志：
   - `[Execution] DEBUG: state['execution_results'] set: 1 items`
   - `[Validator] ✅ 使用execution_results进行验证`
   - `[Validator] overall_passed: True`
4. 检查 `workspace/job_xxx/logs/` 只有 1 个日志文件
5. 前端显示 "执行成功"