PlantAI-dev/IDENTIFY_AGENT_FLOW.md

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

概述

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

---

1. 用户输入 -> Router Agent

输入数据

{
  "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）

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 获取）

intent = state.get("intent_result", {})
target_stages = intent.get("stages", ["identify"])  # ["identify"]

Learning Recommend Agent 处理

• 查询历史执行记录，获取参数推荐
• 如果有高置信度历史记录，返回推荐参数
• 否则返回默认参数

输出数据（设置到 state）

state["learning_recommendations"] = {
    "confidence": "high",  # 或 "medium" / "low"
    "params": {
        "evalue": "1e-5",
        "coverage": 0.5
    },
    "message": "基于历史执行推荐参数"
}

---

3. Learning Recommend -> Planning Agent

输入数据（从 state 获取）

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）

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 获取）

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）

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 获取）

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

执行结果结构（关键）

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）【问题点】

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 获取）

execution_results = state.get("execution_results", [])  # LIST

Fallback 机制（关键）

如果 execution_results 为空（LangGraph state 传递失败）：

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）

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 获取）

execution_results_list = state.get("execution_results", [])  # LIST
parse_result = state.get("parse_result", {})

Fallback 机制（再次确认）

如果 execution_results_list 为空：

# 再次从 manifest 加载（双重保险）

Validator Agent 处理

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

输出数据（设置到 state）

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

---

8. Validator -> Decision: _validate_decision

决策逻辑

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 获取）

execution_results = state.get("execution_results", [])
parse_result = state.get("parse_result", {})

Discovery Agent 处理

• 分析鉴定结果，生成科学发现报告
• 例如：发现新的 SPL 基因亚家族

输出数据（设置到 state）

state["discovery_result"] = {
    "findings": ["发现 15 个 SPL 基因，分为 3 个亚家族"],
    "should_continue": False,  # 是否继续迭代
    "updated_plan": None       # 如果继续迭代，新的计划
}

---

10. Discovery -> Decision: _discovery_decision

决策逻辑

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 获取）

execution_results = state.get("execution_results", [])
discovery_result = state.get("discovery_result", {})

Learning Agent 处理

• 记录本次执行的参数、结果
• 更新知识库
• 生成学习总结

输出数据（设置到 state）

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 处理

state["diagnosis_result"] = {
    "error_type": "missing_output_file",
    "suggested_fix": "检查输入文件路径",
    "retry_params": {...}
}
state["retry_count"] += 1

Diagnosis -> Decision: _diagnosis_decision

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

---

LangGraph State 传递问题详解

TypedDict 类型定义（state.py）

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. 前端显示 "执行成功"