admin/PlantAI-dev
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix: preserve species roles through confirmation

Browse Source
This commit is contained in:
potaser
2026-08-07 08:57:22 +08:00
parent 8c3d601072
commit 822baab367
4 changed files with 201 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/web/vue3/main.py
+19 -11
View File
@@ -777,18 +777,12 @@ def _reconcile_confirmed_gene_family_species(
job_id,
_param_list(routed_params.get("comparativeSpecies") or routed_intent.get("comparative_species")),
)
explicit_reference_set = set(explicit["referenceSpecies"])
explicit_assigned_set = set(
explicit["targetSpecies"]
+ explicit["referenceSpecies"]
+ explicit["comparativeSpecies"]
)
explicit_unassigned_set = set(explicit["unassigned"])
for species in structured_targets:
config_role = config_role_by_species.get(species)
if (
config_role in {None, "reference"}
and species not in explicit_reference_set
and species not in explicit_assigned_set
and species in explicit_unassigned_set
):
conflicts.append({
"code": "analysis_target_config_role_conflict",
@@ -865,10 +859,24 @@ def _species_role_conflict_message(conflicts: list[Dict[str, str]]) -> str:
config_role = role_labels.get(conflict.get("config_role") or "", conflict.get("config_role") or "未知")
requested_role = role_labels.get(conflict.get("requested_role") or "", conflict.get("requested_role") or "配置内部重复角色")
details.append(f"{species}：config.xlsx 为{config_role}，本轮为{requested_role}")
conflict_codes = {str(conflict.get("code") or "") for conflict in conflicts}
detail_text = "；".join(details)
if "config_species_role_conflict" in conflict_codes:
return (
"检测到 config.xlsx 内部物种角色冲突："
+ detail_text
+ "。请发送“打开 config.xlsx”，修正并重新确认物种角色后再继续。"
)
if "analysis_target_config_role_conflict" in conflict_codes:
return (
"检测到系统意图解析发生角色漂移："
+ detail_text
+ "。config.xlsx 未被修改，系统已阻止按错误范围执行，请重新提交当前分析请求。"
)
return (
"检测到物种角色不一致："
+ "；".join(details)
+ "。为避免按错误范围执行，请发送“打开 config.xlsx”，核对并重新确认物种角色后再继续。"
"检测到本轮请求与项目物种角色不一致："
+ detail_text
+ "。系统已阻止执行，请按 config.xlsx 中已确认的角色重新描述分析范围。"
)
gfa_core/mas/agents/intent_v2.py
+66 -10
View File
@@ -12,7 +12,13 @@ from gfa_core.mas.stage_catalog import (
STAGE_INFO as CATALOG_STAGE_INFO,
complete_dependencies as catalog_complete_dependencies,
)
from gfa_core.mas.utils.intent_parser import is_gwas_fmt_intent, parse_gene_family, resolve_gwas_intent, resolve_population_intent
from gfa_core.mas.utils.intent_parser import (
is_gwas_fmt_intent,
parse_gene_family,
parse_gene_family_scope_selection,
resolve_gwas_intent,
resolve_population_intent,
)
class IntentAgentV2(BaseAgent):
@@ -696,18 +702,68 @@ class IntentAgentV2(BaseAgent):
print(f"[IntentAgentV2] Stages expanded: {original_stages} -> {intent_data['stages']}")
query_lower = query.lower()
species_codes = re.findall(r"(?<![A-Za-z0-9])(?:Ft|At|Os|Sl|Zm|Gm|Br|Vv|Pt|Sb|Lc|Dl|Py|Cp|Lg)(?![A-Za-z0-9])", query, flags=re.IGNORECASE)
species_map = {code.lower(): code for code in ["Ft", "At", "Os", "Sl", "Zm", "Gm", "Br", "Vv", "Pt", "Sb", "Lc", "Dl", "Py", "Cp", "Lg"]}
target_species = []
for item in species_codes:
sp = species_map.get(item.lower())
if sp and sp not in target_species:
target_species.append(sp)
selection = parse_gene_family_scope_selection(query)
params = intent_data.setdefault("params", {})
def as_species_list(value: Any) -> List[str]:
if isinstance(value, str):
values = [item.strip() for item in value.split(",")]
elif isinstance(value, (list, tuple, set)):
values = [str(item).strip() for item in value]
else:
values = []
return list(dict.fromkeys(item for item in values if item))
target_species = as_species_list(selection.get("targetSpecies"))
reference_species = as_species_list(selection.get("referenceSpecies"))
comparative_species = as_species_list(selection.get("comparativeSpecies"))
has_query_scope = bool(target_species or reference_species or comparative_species)
prefix_species = provided_prefix.split(".", 1)[0].strip() if provided_prefix and "." in provided_prefix else ""
if not target_species and prefix_species:
target_species = [prefix_species]
if not target_species:
target_species = as_species_list(params.get("targetSpecies") or intent_data.get("target_species"))
if not reference_species:
reference_species = as_species_list(params.get("referenceSpecies") or intent_data.get("reference_species"))
if not comparative_species:
comparative_species = as_species_list(params.get("comparativeSpecies") or intent_data.get("comparative_species"))
if not target_species and not has_query_scope:
llm_species = str(intent_data.get("species") or "").strip()
if llm_species and llm_species != "Multi":
target_species = [llm_species]
reference_species = [sp for sp in reference_species if sp not in target_species]
comparative_species = [
sp for sp in comparative_species
if sp not in target_species and sp not in reference_species
]
scope_mode = (
"target_plus_reference_plus_comparative"
if reference_species and comparative_species
else "target_plus_reference"
if reference_species
else "target_plus_comparative"
if comparative_species
else "target_only"
)
intent_data["target_species"] = target_species
intent_data["reference_species"] = reference_species
intent_data["comparative_species"] = comparative_species
intent_data["scopeMode"] = scope_mode
params.update({
"targetSpecies": target_species,
"referenceSpecies": reference_species,
"comparativeSpecies": comparative_species,
"scopeMode": scope_mode,
"target_species": target_species,
})
explicit_species = bool(target_species)
if target_species:
intent_data["target_species"] = target_species
intent_data.setdefault("params", {})["target_species"] = target_species
intent_data["species"] = target_species[0] if len(target_species) == 1 else "Multi"
elif has_query_scope:
intent_data["species"] = ""
generic_identify = (
"identify" in intent_data.get("stages", [])
and not provided_prefix
tests/unit/test_intent_species_roles.py
+58
View File
@@ -1,3 +1,6 @@
import pytest
from gfa_core.mas.agents.intent_v2 import IntentAgentV2
from gfa_core.mas.utils.intent_parser import (
parse_explicit_species_role_mentions,
parse_gene_family_scope_selection,
@@ -30,3 +33,58 @@ def test_explicit_role_parser_keeps_unassigned_mentions_separate() -> None:
"comparativeSpecies": [],
"unassigned": ["Ppy"],
}
@pytest.mark.parametrize("species", ["Gm", "Br", "Pt", "Sb"])
def test_scope_parser_preserves_species_supported_by_legacy_intent(species: str) -> None:
scope = parse_gene_family_scope_selection(f"分析 {species} 的 SPL 基因家族")
assert scope["targetSpecies"] == [species]
def test_scope_parser_accepts_explicit_role_bound_custom_species_code() -> None:
scope = parse_gene_family_scope_selection("目标物种 Sv；参考物种 At")
assert scope["targetSpecies"] == ["Sv"]
assert scope["referenceSpecies"] == ["At"]
def test_scope_parser_does_not_treat_gene_family_as_species() -> None:
scope = parse_gene_family_scope_selection("目标物种 Ft；分析 SPL 基因家族")
assert scope["targetSpecies"] == ["Ft"]
def test_scope_parser_does_not_bind_custom_code_to_generic_goal_word() -> None:
scope = parse_gene_family_scope_selection("分析目标 SPL 基因家族")
assert scope["targetSpecies"] == []
def test_intent_validation_keeps_confirmed_target_and_reference_separate() -> None:
agent = object.__new__(IntentAgentV2)
intent = agent._validate_intent(
{"gene_family": "SPL", "stages": ["identify"], "params": {}},
(
"请按照你规划的步骤来帮我执行\n"
"已确认上下文：基因家族 SPL；目标物种 Ft；参考物种 At。"
),
)
assert intent["target_species"] == ["Ft"]
assert intent["reference_species"] == ["At"]
assert intent["species"] == "Ft"
def test_intent_validation_preserves_unlabelled_multi_target_request() -> None:
agent = object.__new__(IntentAgentV2)
intent = agent._validate_intent(
{"gene_family": "SPL", "stages": ["identify"], "params": {}},
"分析 Ft 和 Os 的 SPL 基因家族",
)
assert intent["target_species"] == ["Ft", "Os"]
assert intent["reference_species"] == []
assert intent["species"] == "Multi"
tests/unit/test_subfamily_prediction_plan.py
+58
View File
@@ -426,6 +426,64 @@ def test_confirmed_multi_species_roles_remain_separate(monkeypatch):
assert resolved["classification_species"] == ["Ft", "Ppy"]
def test_confirmation_ack_normalizes_structured_role_drift(monkeypatch):
main = _load_vue3_main()
monkeypatch.setattr(
main,
"_confirmed_config_species_scope",
lambda _job_id: {
"targetSpecies": ["Ft"],
"referenceSpecies": ["At"],
"comparativeSpecies": [],
},
)
resolved = main._reconcile_confirmed_gene_family_species(
"job-confirmed-ack",
"是",
{
"gene_family": "SPL",
"stages": ["identify"],
"target_species": ["Ft", "At"],
"params": {"targetSpecies": ["Ft", "At"]},
},
)
assert resolved["conflicts"] == []
assert resolved["target_species"] == ["Ft"]
assert resolved["reference_species"] == ["At"]
def test_unassigned_reference_species_analysis_request_remains_blocked(monkeypatch):
main = _load_vue3_main()
monkeypatch.setattr(
main,
"_confirmed_config_species_scope",
lambda _job_id: {
"targetSpecies": ["Ft"],
"referenceSpecies": ["At"],
"comparativeSpecies": [],
},
)
resolved = main._reconcile_confirmed_gene_family_species(
"job-unassigned-reference",
"分析 At 的 SPL 基因家族",
{
"gene_family": "SPL",
"stages": ["identify"],
"target_species": ["At"],
"params": {"targetSpecies": ["At"]},
},
)
assert any(
conflict["code"] == "analysis_target_config_role_conflict"
and conflict["species"] == "At"
for conflict in resolved["conflicts"]
)
def test_confirmed_config_duplicate_roles_are_blocked(monkeypatch):
main = _load_vue3_main()
monkeypatch.setattr(