489 lines
13 KiB
Markdown
Executable File
489 lines
13 KiB
Markdown
Executable File
# 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. 前端显示 "执行成功" |