fix: restore main-agent routing and workflow handoff

This commit is contained in:
wtx
2026-08-09 13:50:35 +08:00
parent 774490b47f
commit 0f81b5140c
21 changed files with 2687 additions and 208 deletions
+90 -14
View File
@@ -45,6 +45,7 @@ PROJECT_ROOT = Path(__file__).parent.parent.parent.parent
DEFAULT_RUNTIME_ENV = os.environ.get("BBAV5_DEFAULT_RUNTIME_ENV", "test4py311")
sys.path.insert(0, str(PROJECT_ROOT))
from gfa_core.project_paths import get_conda_env_bin, get_conda_env_rscript, get_configured_path, get_wgs_bcftools, get_wgs_java, get_wgs_samtools
from gfa_core.mas import confirmation_snapshot as _confirmation_snapshot
ASAW_BIN = get_conda_env_bin("ASAW")
ASAW_RSCRIPT = get_conda_env_rscript("ASAW")
@@ -703,9 +704,10 @@ def _confirmed_config_species_scope(job_id: str) -> Dict[str, list[str]]:
if not job_id:
return {}
try:
from gfa_core.mas.confirmation_snapshot import build_confirmation_snapshot
confirmation = build_confirmation_snapshot(PROJECT_ROOT, job_id)
confirmation = _confirmation_snapshot.build_confirmation_snapshot(
PROJECT_ROOT,
job_id,
)
except (OSError, ValueError):
return {}
if not confirmation.get("config_confirmed"):
@@ -878,7 +880,8 @@ def _species_role_conflict_message(conflicts: list[Dict[str, str]]) -> str:
return (
"检测到本轮请求与项目物种角色不一致:"
+ detail_text
+ "。系统已阻止执行,请按 config.xlsx 中已确认的角色重新描述分析范围。"
+ "。系统已阻止执行,请按 config.xlsx 中已确认的角色重新描述分析范围;"
"如需修改项目物种角色,请发送“打开 config.xlsx”。"
)
@@ -1088,6 +1091,8 @@ def _hydrate_goal_terms_from_current_job(
return hydrated
config_scope = _confirmed_config_species_scope(job_id)
if not config_scope and (INPUT_ROOT / job_id / "config.xlsx").is_file():
config_scope = _configured_species_scope(job_id)
for config_key, term_key in (
("targetSpecies", "target_species"),
("referenceSpecies", "reference_species"),
@@ -4983,12 +4988,6 @@ orchestrator = None
langgraph_orchestrator = None
if MAS_AVAILABLE:
try:
# 清除Python模块缓存,确保配置重新加载
import sys
modules_to_remove = [k for k in sys.modules.keys() if 'gfa_core.mas' in k]
for m in modules_to_remove:
del sys.modules[m]
# 加载MAS配置
from gfa_core.mas.config import config as mas_config
llm_config = mas_config.load_llm_config()
@@ -17634,6 +17633,11 @@ async def _run_agent_chat(
if isinstance(outcome.get("proposal"), dict)
else None
),
recommendation=(
outcome.get("recommendation")
if isinstance(outcome.get("recommendation"), dict)
else None
),
)
return {
@@ -17655,6 +17659,11 @@ async def _run_agent_chat(
"model_used": request.model,
"session_id": session_id,
"input_paths": list(outcome.get("input_paths") or []),
"recommendation": (
outcome.get("recommendation")
if isinstance(outcome.get("recommendation"), dict)
else None
),
"handoff_request": (
str((outcome.get("proposal") or {}).get("request") or "").strip()
if isinstance(outcome.get("proposal"), dict)
@@ -18496,6 +18505,41 @@ class AgentPlanRequest(BaseModel):
intent: dict = {} # 前端可传入已有的intent,避免重复调用IntentAgentV2
def _authorized_workflow_action_stages(intent: Dict[str, Any]) -> List[str]:
candidate = intent or {}
params = candidate.get("params") or {}
action_id = str(
candidate.get("workflow_action_id") or params.get("workflowActionId") or ""
).strip()
authorized = list(dict.fromkeys(_param_list(params.get("authorizedStages"))))
declared = set(_param_list(candidate.get("stages")))
if not action_id or not authorized or not set(authorized).issubset(declared):
return []
try:
from gfa_core.mas.workflow_catalog import load_canonical_workflow_contract
contract = load_canonical_workflow_contract("gene_family")
transition = next(item for item in contract.transitions if item.id == action_id)
actions = {item.id: item for item in contract.actions}
evidence_stages = {
item.id: item.stage for item in contract.evidence_requirements
}
contract_stages = [
evidence_stages[requirement_id]
for requirement_id in transition.evidence_requirement_ids
]
contract_stages.append(actions[transition.action_id].target_stage)
except (ImportError, StopIteration, KeyError):
return []
if set(authorized) != set(contract_stages):
return []
return [
stage
for stage in SUBFAMILY_PREDICTION_PLAN_STAGES
if stage in set(contract_stages)
]
SUBFAMILY_PREDICTION_PLAN_STAGES = [
"identify",
"phylogeny",
@@ -20578,6 +20622,7 @@ async def agent_plan(request: AgentPlanRequest):
request.intent = dict(request.intent or {})
request.intent["params"] = dict(request.intent.get("params") or {})
_apply_reconciled_species_to_intent(request.intent, plan_role_resolution)
authorized_workflow_stages = _authorized_workflow_action_stages(request.intent)
query_lower_for_context = (request.message or "").lower()
result_report_query = (
"结果报告" in query_lower_for_context
@@ -20647,7 +20692,7 @@ async def agent_plan(request: AgentPlanRequest):
and not _looks_like_gene_family_workflow_query(query_lower_for_context)
and not _requires_confirmed_subfamily_query(query_lower_for_context)
)
if subfamily_prediction_query_for_context:
if subfamily_prediction_query_for_context or authorized_workflow_stages:
requested_prefix_for_subfamily = request.prefix or ""
prefix_species, prefix_family = _prefix_species_family(requested_prefix_for_subfamily)
family_for_subfamily = _resolve_authoritative_gene_family(
@@ -20710,7 +20755,27 @@ async def agent_plan(request: AgentPlanRequest):
}
if family_for_subfamily and targets_for_subfamily:
prefix_for_subfamily = f"{targets_for_subfamily[0]}.{family_for_subfamily}" if len(targets_for_subfamily) == 1 else f"Target.{family_for_subfamily}"
deterministic_stages = ["identify", "phylogeny", "motif", "domain", "structure", "combo", "reference_classification"]
deterministic_stages = authorized_workflow_stages or [
"identify",
"phylogeny",
"motif",
"domain",
"structure",
"combo",
"reference_classification",
]
authorized_action_id = str(
(request.intent or {}).get("workflow_action_id")
or ((request.intent or {}).get("params") or {}).get(
"workflowActionId"
)
or ""
).strip()
plan_description = (
f"执行已确认的工作流业务动作 {authorized_action_id}"
if authorized_workflow_stages
else "执行 target/comparative + reference 亚家族分类"
)
classification_species_for_subfamily = list(targets_for_subfamily)
classification_species_for_subfamily, role_targets_for_subfamily, role_comps_for_subfamily = _split_subfamily_selected_species_by_config_roles(
request.job_id,
@@ -20732,6 +20797,12 @@ async def agent_plan(request: AgentPlanRequest):
"confirmPrefixBySpecies": confirm_prefix_by_species,
"subfamilyState": subfamily_state_for_plan,
}
if authorized_workflow_stages:
request_action_params = (request.intent or {}).get("params") or {}
plan_params["workflowActionId"] = authorized_action_id
plan_params["authorizedStages"] = _param_list(
request_action_params.get("authorizedStages")
)
if identify_selection_for_subfamily:
plan_params["identifySpeciesSelection"] = _identify_selection_with_default(
identify_selection_for_subfamily,
@@ -20801,7 +20872,7 @@ async def agent_plan(request: AgentPlanRequest):
"comparative_species": role_comps_for_subfamily,
"scopeMode": plan_params["scopeMode"],
"prefix": prefix_for_subfamily,
"description": "执行 target/comparative + reference 亚家族分类",
"description": plan_description,
"estimated_time": f"{estimated_minutes}分钟",
"estimated_total_time": f"{estimated_minutes}分钟",
"execution_strategy": "parallel",
@@ -20823,10 +20894,15 @@ async def agent_plan(request: AgentPlanRequest):
"comparative_species": role_comps_for_subfamily,
"scopeMode": plan_params["scopeMode"],
"stages": deterministic_stages,
"description": "执行目标物种+参考物种亚家族分类",
"description": plan_description,
"confidence": 0.98,
"prefix": prefix_for_subfamily,
"params": plan_params,
**(
{"workflow_action_id": authorized_action_id}
if authorized_action_id
else {}
),
},
"completed_stages": completed_stages,
"session_id": session_id,
+3 -38
View File
@@ -23,10 +23,7 @@ from gfa_core.mas.mutation_surface import ToolSurfaceEntry
from gfa_core.mas.stage_capabilities import get_stage_capability
from gfa_core.mas.tools.stage_status import SUCCESS_STATUSES
from gfa_core.mas.tools.stage_tools import stage_registry
from gfa_core.mas.workflow_catalog import (
load_canonical_workflow_contract,
load_rnaseq_business_context,
)
from gfa_core.mas.workflow_catalog import load_rnaseq_business_context
from gfa_core.mas.workflow_models import GoalLedger
@@ -1610,16 +1607,10 @@ def build_plantai_conversation_tools(
) -> str:
"""Adapt MainAgent memory to the original IntentAgent text contract."""
text = str(query or "").strip()
if (
text.startswith("执行数据准备(genome_prepare)")
and "当前业务动作已由数据准备契约唯一确定" in text
):
return text
if goal_ledger is None:
return text
terms = goal_ledger.user_terms
context: list[str] = []
canonical_request = ""
family = str(terms.get("gene_family") or "").strip()
if family:
context.append(f"基因家族 {family}")
@@ -1671,7 +1662,6 @@ def build_plantai_conversation_tools(
or "config.xlsx 已确认" in goal_ledger.approved_decisions
or formal_config_exists
)
genome_prepare_observed = False
subfamily_confirm_observed = False
if job_id:
final_dir = project_root / "workspace" / job_id / "final"
@@ -1682,8 +1672,6 @@ def build_plantai_conversation_tools(
continue
stage = str(manifest.get("stage") or "").strip()
status = manifest.get("status")
if stage == "genome_prepare" and status in {"success", "partial_success"}:
genome_prepare_observed = True
manifest_prefix = str(manifest.get("prefix") or "").strip()
family_matches = not family or bool(
re.search(
@@ -1698,36 +1686,13 @@ def build_plantai_conversation_tools(
and family_matches
):
subfamily_confirm_observed = True
if config_confirmed and not genome_prepare_observed:
context.append(
"正式 config.xlsx 已确认;这只表示配置已核对,不等于数据准备已经完成,"
"当前 job 尚无数据准备完成证据;具体分析意图和阶段仍由现有分析系统判断"
)
if guidance and workflow_id in {
"data_preparation",
"gene_family",
"rnaseq_whole_genome",
"rnaseq_family_member",
}:
contract = load_canonical_workflow_contract("data_preparation")
transition = next(
item
for item in contract.transitions
if item.from_state_id == "config_confirmed"
)
actions = {item.id: item for item in contract.actions}
target_stage = actions[transition.action_id].target_stage
canonical_request = (
f"执行数据准备({target_stage})\n"
"当前业务动作已由数据准备契约唯一确定"
)
if config_confirmed:
context.append("正式 config.xlsx 已确认")
if family and subfamily_confirm_observed:
context.append(
"当前 job 的 subfamily_confirm 已成功完成,家族成员和亚家族分类表已确认,"
"这是当前项目状态事实;具体分析意图和阶段仍由现有分析系统判断"
)
if canonical_request:
return canonical_request
if not context:
return text
return f"{text}\n已确认上下文:{';'.join(context)}。"
+110 -10
View File
@@ -4,7 +4,7 @@ IntentAgent V2 - LLM驱动的智能意图识别
"""
import json
import re
from typing import Dict, List, Optional, Any
from typing import Any, Dict, List, Mapping, Optional
from .base import AgentMessage, AgentState, BaseAgent, QueryMode
from gfa_core.mas.llm import LLMManager
from gfa_core.mas.stage_catalog import (
@@ -152,6 +152,82 @@ class IntentAgentV2(BaseAgent):
def __init__(self, llm_manager: LLMManager):
super().__init__(llm_manager)
@staticmethod
def _apply_structured_goal_context(
intent: Dict[str, Any],
goal_context: Mapping[str, Any] | None,
) -> Dict[str, Any]:
if not isinstance(goal_context, Mapping):
return intent
terms = goal_context.get("user_terms")
if not isinstance(terms, Mapping):
return intent
def species_list(value: Any) -> List[str]:
if isinstance(value, str):
values = value.split(",")
elif isinstance(value, (list, tuple, set)):
values = value
else:
values = ()
return list(
dict.fromkeys(str(item).strip() for item in values if str(item).strip())
)
normalized = dict(intent)
params = dict(normalized.get("params") or {})
targets = species_list(
params.get("targetSpecies") or normalized.get("target_species")
) or species_list(terms.get("target_species"))
references = species_list(
params.get("referenceSpecies") or normalized.get("reference_species")
) or species_list(terms.get("reference_species"))
comparatives = species_list(
params.get("comparativeSpecies") or normalized.get("comparative_species")
) or species_list(terms.get("comparative_species"))
references = [species for species in references if species not in targets]
comparatives = [
species
for species in comparatives
if species not in targets and species not in references
]
family = str(
normalized.get("gene_family") or terms.get("gene_family") or ""
).strip()
scope_mode = (
"target_plus_reference_plus_comparative"
if references and comparatives
else "target_plus_reference"
if references
else "target_plus_comparative"
if comparatives
else "target_only"
)
params.update(
{
"targetSpecies": targets,
"referenceSpecies": references,
"comparativeSpecies": comparatives,
"target_species": targets,
"scopeMode": scope_mode,
}
)
normalized.update(
{
"params": params,
"gene_family": family,
"target_species": targets,
"reference_species": references,
"comparative_species": comparatives,
"scopeMode": scope_mode,
}
)
if targets:
normalized["species"] = targets[0] if len(targets) == 1 else "Multi"
if normalized.get("species") and family:
normalized["prefix"] = f"{normalized['species']}.{family}"
return normalized
async def execute(self, state: AgentState, message: AgentMessage) -> AgentState:
"""
@@ -237,6 +313,11 @@ class IntentAgentV2(BaseAgent):
print(f"[IntentAgentV2] Protected species from parser: {base_intent['species']}")
intent_result.setdefault("task_kind", "pipeline")
intent_result.setdefault("taskType", intent_result.get("task_kind", "pipeline"))
intent_result = self._apply_structured_goal_context(
intent_result,
message.payload.get("goal_context"),
)
# 保留原始 stages,不过滤(留给 PlanningAgentV2 处理)
original_stages = intent_result.get("stages", [])
@@ -693,8 +774,17 @@ class IntentAgentV2(BaseAgent):
intent_data["stages"] = (
["rnaseq_de", "rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_fgsea", "rnaseq_wgcna"]
if has_rnaseq_keyword
else ["identify"]
else []
)
if not intent_data["stages"]:
intent_data.update(
{
"description": "未能可靠识别要执行的分析阶段",
"needs_clarification": True,
"clarification_type": "analysis_stage",
"clarification_question": "请说明要执行的具体分析阶段",
}
)
# 根据依赖关系补充缺失的阶段(关键修复:确保依赖展开)
original_stages = intent_data["stages"]
@@ -975,11 +1065,8 @@ class IntentAgentV2(BaseAgent):
has_rnaseq_keyword = any(
kw in query_lower for kw in ["rnaseq", "rna-seq", "rna seq", "转录组", "fastq"]
)
stages = (
["rnaseq_de", "rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_fgsea", "rnaseq_wgcna"]
if has_rnaseq_keyword
else ["identify"]
)
if has_rnaseq_keyword:
stages = ["rnaseq_de", "rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_fgsea", "rnaseq_wgcna"]
# 补充依赖(使用 stage_catalog 的主规划依赖链)
original_stages = stages.copy()
@@ -989,18 +1076,31 @@ class IntentAgentV2(BaseAgent):
gene_family = self._extract_gene_family_fallback(query)
species = "Multi" if any(kw in query.lower() for kw in ["基因家族成员鉴定", "基因家族鉴定", "identify"]) else ""
return {
result = {
"gene_family": gene_family,
"species": species,
"prefix": f"{species}.{gene_family}",
"stages": stages,
"description": f"执行{gene_family}基因家族的{', '.join(stages)}分析",
"description": (
f"执行{gene_family}基因家族的{', '.join(stages)}分析"
if stages
else "未能可靠识别要执行的分析阶段"
),
"requirements": [],
"confidence": 0.7,
"confidence": 0.7 if stages else 0.0,
"task_kind": "pipeline",
"taskType": "pipeline",
"mode": "fallback"
}
if not stages:
result.update(
{
"needs_clarification": True,
"clarification_type": "analysis_stage",
"clarification_question": "请说明要执行的具体分析阶段",
}
)
return result
def create_intent_agent_v2(llm_manager):
File diff suppressed because it is too large Load Diff
+15 -3
View File
@@ -251,7 +251,13 @@ class ProjectContextIndexer:
self.workspace_root = self.project_root / "workspace"
self.allow_legacy_manifest_migration = allow_legacy_manifest_migration
def build(self, job_id: str, prefix: str = "") -> ProjectContextIndex:
def build(
self,
job_id: str,
prefix: str = "",
*,
include_stage_status: bool = True,
) -> ProjectContextIndex:
workspace = self.workspace_root / job_id
canonical_input_dir = self.project_root / "input" / job_id
input_dir = canonical_input_dir if canonical_input_dir.exists() else workspace / "input"
@@ -275,7 +281,8 @@ class ProjectContextIndexer:
self._scan_files(index, final_dir, "final")
self._scan_stage_dirs(index, workspace)
self._load_manifests(index, final_dir)
self._load_stage_status(index, job_id, prefix or self._best_prefix(index))
if include_stage_status:
self._load_stage_status(index, job_id, prefix or self._best_prefix(index))
self._scan_logs(index, logs_dir)
self._derive_warnings(index)
return index
@@ -696,11 +703,16 @@ def build_project_context(
project_root: Path | str = PROJECT_ROOT,
*,
allow_legacy_manifest_migration: bool = True,
include_stage_status: bool = True,
) -> ProjectContextIndex:
return ProjectContextIndexer(
project_root,
allow_legacy_manifest_migration=allow_legacy_manifest_migration,
).build(job_id=job_id, prefix=prefix)
).build(
job_id=job_id,
prefix=prefix,
include_stage_status=include_stage_status,
)
def main() -> int:
+3
View File
@@ -62,6 +62,9 @@ class DeepSeekProvider(BaseLLMProvider):
tool_choice = kwargs.get("tool_choice")
if tool_choice is not None:
request_kwargs["tool_choice"] = tool_choice
response_format = kwargs.get("response_format")
if response_format is not None:
request_kwargs["response_format"] = response_format
response = self.client.chat.completions.create(**request_kwargs)
message = response.choices[0].message
+29 -7
View File
@@ -9,7 +9,7 @@ from typing import Any
from uuid import NAMESPACE_URL, UUID, uuid4, uuid5
from gfa_core.mas.agent_state_repository import SQLiteAgentStateRepository
from gfa_core.mas.workflow_models import GoalLedger
from gfa_core.mas.workflow_models import CurrentRecommendation, GoalLedger
_INTERACTION_KINDS = {
@@ -68,7 +68,7 @@ def _history(items: Any) -> tuple[dict[str, str], ...]:
content = str(item.get("content") or "")
if content:
normalized.append({"role": item["role"], "content": content})
return tuple(normalized[-10:])
return tuple(normalized)
class PersistentConversationStore:
@@ -207,7 +207,7 @@ class PersistentConversationStore:
),
expected_revision=ledger.revision,
)
stored_messages = self.repository.list_recent_messages(conversation_id, limit=10)
stored_messages = self.repository.list_messages(conversation_id)
if not stored_messages:
for item in _history(client_history):
persisted = self.repository.append_message(
@@ -220,10 +220,7 @@ class PersistentConversationStore:
conversation = self.repository.get_conversation(conversation_id)
assert conversation is not None
assert conversation.revision == persisted.revision
stored_messages = self.repository.list_recent_messages(
conversation_id,
limit=10,
)
stored_messages = self.repository.list_messages(conversation_id)
current_user_already_saved = bool(
stored_messages
and stored_messages[-1].role == "user"
@@ -264,6 +261,7 @@ class PersistentConversationStore:
action: str | None,
intent: dict[str, Any] | None = None,
proposal: dict[str, str] | None = None,
recommendation: Mapping[str, Any] | None = None,
) -> GoalLedger:
if interaction_kind not in _INTERACTION_KINDS:
raise ValueError("interaction_kind is invalid")
@@ -362,6 +360,20 @@ class PersistentConversationStore:
user_terms = dict(serialized["user_terms"])
if turn.job_id:
user_terms["active_job_id"] = turn.job_id
if action == "show_result":
if recommendation is None:
user_terms.pop("current_recommendation", None)
else:
current_recommendation = CurrentRecommendation.model_validate(
recommendation
)
if current_recommendation.job_id != turn.job_id:
raise ValueError("recommendation job binding conflict")
user_terms["current_recommendation"] = (
current_recommendation.model_dump(mode="json", exclude_none=True)
)
if interaction_kind in {"cancel", "correction", "switch"}:
user_terms.pop("current_recommendation", None)
if action == "clarify" and isinstance(proposal, dict):
normalized_proposal = {
key: str(proposal.get(key) or "").strip()
@@ -376,6 +388,16 @@ class PersistentConversationStore:
"rnaseq_family_member",
}:
normalized_proposal["workflow_id"] = workflow_id
action_binding = {
key: str(proposal.get(key) or "").strip()
for key in (
"workflow_action_id",
"authorized_stages",
"target_stage",
)
}
if all(action_binding.values()):
normalized_proposal.update(action_binding)
normalized_proposal["job_id"] = turn.job_id
user_terms["pending_proposal"] = normalized_proposal
if (
+23 -2
View File
@@ -14,6 +14,7 @@ flag enables decision-making.
"""
from __future__ import annotations
from copy import copy
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any, Iterable
@@ -93,6 +94,7 @@ def resolve_stage_requirements(
params: dict[str, Any] | None = None,
project_root: Path | str = PROJECT_ROOT,
shadow_only: bool = True,
project_context: ProjectContextIndex | None = None,
) -> StageRequirementReport:
"""Resolve stage requirements without changing planning or files."""
params = dict(params or {})
@@ -110,7 +112,16 @@ def resolve_stage_requirements(
report.messages.append("Cannot resolve requirements without job_id.")
return report
context = build_project_context(job_id=job_id, prefix=identity.prefix, project_root=project_root)
context = project_context
if context is None:
context = build_project_context(
job_id=job_id,
prefix=identity.prefix,
project_root=project_root,
include_stage_status=False,
)
else:
context = copy(context)
_apply_identity_to_context(context, identity)
if identity.stage == "identify":
@@ -151,18 +162,28 @@ def resolve_many_stage_requirements(
reports: dict[str, dict[str, Any]] = {}
params_by_stage = params_by_stage or {}
shared_params = shared_params or {}
contexts_by_prefix: dict[str, ProjectContextIndex] = {}
for stage in stages:
stage_params = {**shared_params, **(params_by_stage.get(stage) or {})}
if stage not in CORE_REQUIREMENT_STAGES:
continue
try:
effective_prefix = stage_params.get("prefix") or prefix
if effective_prefix not in contexts_by_prefix:
contexts_by_prefix[effective_prefix] = build_project_context(
job_id=job_id,
prefix=effective_prefix,
project_root=project_root,
include_stage_status=False,
)
report = resolve_stage_requirements(
stage=stage,
prefix=stage_params.get("prefix") or prefix,
prefix=effective_prefix,
job_id=job_id,
params=stage_params,
project_root=project_root,
shadow_only=shadow_only,
project_context=contexts_by_prefix[effective_prefix],
)
reports[stage] = report.to_dict()
except Exception as exc:
+33 -3
View File
@@ -304,8 +304,21 @@ SPECIES_KEYWORDS = {
"Cp": ["cp", "papaya", "木瓜"],
"Vv": ["vv", "grape", "葡萄"],
"Lg": ["lg"],
"Gm": ["gm"],
"Br": ["br"],
"Pt": ["pt"],
"Sb": ["sb"],
}
_SPECIES_CODE_PATTERN = (
r"(?<![A-Za-z0-9])(?:"
+ "|".join(
sorted((re.escape(code) for code in SPECIES_KEYWORDS), key=len, reverse=True)
)
+ r")(?![A-Za-z0-9])"
)
_ROLE_BOUND_SPECIES_CODE = r"[A-Z][A-Za-z0-9_-]{1,15}"
SPECIES_TO_LATIN = {
"At": "Arabidopsis thaliana",
"Os": "Oryza sativa",
@@ -348,7 +361,11 @@ def _species_mentions_with_roles(query: str) -> Dict[str, List[str]]:
def collect_species(segment: str) -> List[str]:
segment_lower = segment.lower()
found: List[str] = []
for match in re.finditer(r"(?<![A-Za-z0-9])(?:Ft|At|Os|Ls|Zm|Sl|Lc|Dl|Ppy|Py|Cp|Vv|Lg)(?![A-Za-z0-9])", segment, flags=re.IGNORECASE):
for match in re.finditer(
_SPECIES_CODE_PATTERN,
segment,
flags=re.IGNORECASE,
):
code = canonical.get(match.group(0).lower())
if code:
found.append(code)
@@ -371,6 +388,11 @@ def _species_mentions_with_roles(query: str) -> Dict[str, List[str]]:
),
("comparativeSpecies", r"比较物种|比较|对比|comparative|comparison|outgroup|外群"),
]
role_bound_marker_patterns = {
"targetSpecies": r"目标物种|目标基因组|target\s+(?:species|genome)",
"referenceSpecies": r"参考物种|参考基因组|reference\s+(?:species|genome)",
"comparativeSpecies": r"比较物种|比较基因组|comparative\s+(?:species|genome)|outgroup",
}
roles = {"targetSpecies": [], "referenceSpecies": [], "comparativeSpecies": [], "unassigned": []}
# A role may follow the species, as in "Ft是目标物种". Resolve
# comma-delimited clauses first so a path fragment such as ref_data cannot
@@ -380,6 +402,15 @@ def _species_mentions_with_roles(query: str) -> Dict[str, List[str]]:
for role, pattern in marker_patterns:
if re.search(pattern, clause_lower):
roles[role].extend(collect_species(clause))
role_bound_marker = role_bound_marker_patterns[role]
for role_bound_pattern in (
rf"(?i:(?:{role_bound_marker})\s*(?:为|是|作为|is|[::=])?\s*)({_ROLE_BOUND_SPECIES_CODE})",
rf"({_ROLE_BOUND_SPECIES_CODE})(?i:\s*(?:是|为|作为|is)?\s*(?:{role_bound_marker}))",
):
roles[role].extend(
match.group(1)
for match in re.finditer(role_bound_pattern, clause)
)
if any(roles[role] for role in ("targetSpecies", "referenceSpecies", "comparativeSpecies")):
assigned = set(roles["targetSpecies"] + roles["referenceSpecies"] + roles["comparativeSpecies"])
roles["unassigned"] = [sp for sp in collect_species(text) if sp not in assigned]
@@ -399,8 +430,7 @@ def _species_mentions_with_roles(query: str) -> Dict[str, List[str]]:
return {key: _unique_keep_order(value) for key, value in roles.items()}
mentions: List[tuple[int, str]] = []
code_pattern = r"(?<![A-Za-z0-9])(?:Ft|At|Os|Ls|Zm|Sl|Lc|Dl|Ppy|Py|Cp|Vv|Lg)(?![A-Za-z0-9])"
for match in re.finditer(code_pattern, text, flags=re.IGNORECASE):
for match in re.finditer(_SPECIES_CODE_PATTERN, text, flags=re.IGNORECASE):
code = canonical.get(match.group(0).lower())
if code:
mentions.append((match.start(), code))
+44
View File
@@ -2611,6 +2611,50 @@ class WorkflowStateSnapshot(_WorkflowModel):
return self
class CurrentRecommendation(_WorkflowModel):
recommendation_id: str
job_id: str
workflow_id: Literal[
"data_preparation",
"gene_family",
"rnaseq_whole_genome",
"rnaseq_family_member",
]
transition_id: str
from_state_id: str
to_state_id: str
target_stage: str
authorized_stages: tuple[str, ...] = Field(min_length=1)
evidence_digest: str
family: str | None = None
scope_id: str | None = None
@model_validator(mode="after")
def validate_recommendation(self) -> "CurrentRecommendation":
for field_name in (
"recommendation_id",
"job_id",
"transition_id",
"from_state_id",
"to_state_id",
"target_stage",
):
_require_nonblank(getattr(self, field_name), field_name)
if len(self.authorized_stages) != len(set(self.authorized_stages)):
raise ValueError("authorized_stages must contain unique stages")
if any(not stage.strip() for stage in self.authorized_stages):
raise ValueError("authorized_stages cannot contain blank stages")
if self.target_stage not in self.authorized_stages:
raise ValueError("target_stage must be authorized")
if not _SHA256_RE.fullmatch(self.evidence_digest):
raise ValueError("evidence_digest must be a sha256 hex digest")
for field_name in ("family", "scope_id"):
value = getattr(self, field_name)
if value is not None:
_require_nonblank(value, field_name)
return self
class GoalLedger(_WorkflowModel):
goal_ledger_version: Literal["goal_ledger_v1"] = "goal_ledger_v1"
conversation_id: str
+62
View File
@@ -157,6 +157,68 @@ async def test_agent_chat_persists_router_proposal_without_analysis_intent(
}
@pytest.mark.asyncio
async def test_agent_chat_exposes_and_persists_current_recommendation(
monkeypatch,
web_main,
tmp_path,
):
repository = SQLiteAgentStateRepository(tmp_path / "agent-state.sqlite3")
repository.initialize()
store = PersistentConversationStore(repository)
monkeypatch.setattr(web_main, "agent_conversation_store", store)
recommendation = {
"recommendation_id": "rec-build-subfamily-evidence",
"job_id": "job-recommendation",
"workflow_id": "gene_family",
"transition_id": "build_subfamily_evidence",
"from_state_id": "family_identified",
"to_state_id": "subfamily_evidence_ready",
"target_stage": "combo",
"authorized_stages": [
"domain",
"motif",
"phylogeny",
"structure",
"combo",
],
"evidence_digest": "a" * 64,
"family": "SPL",
"scope_id": "SPL.target-Ft.target_only",
}
_install_router(
monkeypatch,
{
"status": "ok",
"action": "show_result",
"message": "下一步是构建亚家族证据。",
"intent": None,
"recommendation": recommendation,
"sources": [],
"tool_trace": [],
},
[],
)
result = await web_main.agent_chat(
web_main.AgentChatRequest(
message="接下来的分析有哪些?",
session_id="recommendation-session",
job_id="job-recommendation",
history=[],
)
)
assert result["action"] == "show_result"
assert result["recommendation"] == recommendation
ledger = repository.get_goal_ledger(result["session_id"])
assert ledger is not None
assert (
ledger.model_dump(mode="json")["user_terms"]["current_recommendation"]
== recommendation
)
@pytest.mark.asyncio
async def test_agent_chat_exposes_confirmed_request_for_original_planner(
monkeypatch,
@@ -0,0 +1,137 @@
import asyncio
from apps.web.vue3 import main as api
def test_build_subfamily_evidence_plan_keeps_confirmed_stage_boundary():
intent = {
"workflow_action_id": "build_subfamily_evidence",
"stages": ["domain", "motif", "phylogeny", "structure", "combo"],
"params": {
"workflowActionId": "build_subfamily_evidence",
"authorizedStages": [
"domain",
"motif",
"phylogeny",
"structure",
"combo",
],
},
}
stages = api._authorized_workflow_action_stages(intent)
assert stages == ["phylogeny", "motif", "domain", "structure", "combo"]
assert "identify" not in stages
assert "reference_classification" not in stages
def test_unbound_plan_does_not_invent_an_authorized_stage_boundary():
assert api._authorized_workflow_action_stages(
{"stages": ["identify", "phylogeny", "reference_classification"]}
) == []
def test_agent_plan_honors_authorized_workflow_action_without_keyword_reclassification(
monkeypatch,
):
authorized = ["domain", "motif", "phylogeny", "structure", "combo"]
intent = {
"gene_family": "SPL",
"target_species": ["Ft"],
"reference_species": ["At"],
"stages": authorized,
"primary_stage": "combo",
"workflow_action_id": "build_subfamily_evidence",
"params": {
"workflowActionId": "build_subfamily_evidence",
"authorizedStages": authorized,
"targetSpecies": ["Ft"],
"referenceSpecies": ["At"],
},
}
monkeypatch.setattr(
api,
"_reconcile_confirmed_gene_family_species",
lambda *args, **kwargs: {"applicable": False, "conflicts": []},
)
monkeypatch.setattr(
api, "_resolve_authoritative_gene_family", lambda *args, **kwargs: "SPL"
)
monkeypatch.setattr(
api,
"_resolve_subfamily_request_species_roles",
lambda *args, **kwargs: {
"classification_species": ["Ft"],
"reference_species": ["At"],
},
)
monkeypatch.setattr(
api,
"_latest_gene_family_scope_defaults",
lambda *args, **kwargs: {
"target_species": ["Ft"],
"reference_species": ["At"],
"comparative_species": [],
},
)
monkeypatch.setattr(
api,
"_load_identify_species_selection",
lambda *args, **kwargs: {
"groups": {
"target": [{"species": "Ft"}],
"reference": [{"species": "At"}],
"comparative": [],
}
},
)
monkeypatch.setattr(
api,
"_subfamily_state_for_targets",
lambda *args, **kwargs: {"overall": "not_classified", "by_species": {}},
)
monkeypatch.setattr(
api, "_identity_completed_stages_for_plan", lambda *args, **kwargs: []
)
monkeypatch.setattr(
api,
"_resolve_contextual_gene_family_request",
lambda *args, **kwargs: {
"family": "SPL",
"prefix": "Ft.SPL",
"stages": authorized,
"targetSpecies": ["Ft"],
"referenceSpecies": ["At"],
"comparativeSpecies": [],
"scopeMode": "target_plus_reference",
"contextGate": {"valid": False, "issues": []},
"context_route": "gene_family_default_scope",
"requiresConfirmedSubfamily": False,
"skipSubfamilyLabels": False,
"subfamilyState": {
"overall": "not_classified",
"by_species": {"Ft": {"status": "not_classified"}},
},
},
)
response = asyncio.run(
api.agent_plan(
api.AgentPlanRequest(
message="执行接下来的步骤",
session_id="authorized-action-test",
job_id="job-test",
prefix="Ft.SPL",
intent=intent,
)
)
)
expected = ["phylogeny", "motif", "domain", "structure", "combo"]
assert response["plan"]["all_stages"] == expected
assert response["plan"]["stages"] == expected
assert response["intent"]["workflow_action_id"] == "build_subfamily_evidence"
assert response["intent"]["params"]["authorizedStages"] == authorized
assert "reference_classification" not in response["plan"]["all_stages"]
assert "subfamily_confirm" not in response["plan"]["all_stages"]
@@ -0,0 +1,450 @@
import asyncio
import json
from pydantic import BaseModel
from gfa_core.mas.agent_state_repository import SQLiteAgentStateRepository
from gfa_core.mas.agents.conversation_tools import ConversationToolRegistry
from gfa_core.mas.agents.router_runtime import RouterRuntime
from gfa_core.mas.llm import LLMResponse, LLMToolCall
from gfa_core.mas.persistent_conversation import PersistentConversationStore
from gfa_core.mas.workflow_models import CurrentRecommendation, GoalLedger
class _NoArgs(BaseModel):
pass
class _IntentArgs(BaseModel):
query: str
class _SequenceLLM:
def __init__(self, responses):
self.responses = list(responses)
async def acall(self, *_args, **_kwargs):
return self.responses.pop(0)
def _response(content="", *, tool_calls=None):
return LLMResponse(
content=content,
model="test-model",
provider="test-provider",
usage={},
tool_calls=list(tool_calls or []),
)
def _recommendation(job_id="job-current-recommendation"):
return CurrentRecommendation(
recommendation_id="rec-build-subfamily-evidence",
job_id=job_id,
workflow_id="gene_family",
transition_id="build_subfamily_evidence",
from_state_id="family_identified",
to_state_id="subfamily_evidence_ready",
target_stage="combo",
authorized_stages=("domain", "motif", "phylogeny", "structure", "combo"),
evidence_digest="a" * 64,
family="SPL",
scope_id="SPL.target-Ft.target_only",
)
def test_identified_project_context_projects_one_typed_recommendation():
recommendation = RouterRuntime._current_workflow_recommendation(
"gene_family",
"job-project-recommendation",
{
"family": "SPL",
"scopes": [{"scope_id": "SPL.target-Ft.target_only"}],
"manifests": [{"stage": "identify", "status": "success"}],
"stage_status": {
"identify": {"stage": "identify", "status": "success"}
},
},
)
assert isinstance(recommendation, CurrentRecommendation)
assert recommendation.transition_id == "build_subfamily_evidence"
assert recommendation.target_stage == "combo"
assert recommendation.authorized_stages == (
"domain",
"motif",
"phylogeny",
"structure",
"combo",
)
def test_next_analysis_question_stays_read_only_and_returns_typed_recommendation():
job_id = "job-next-analysis-question"
context = {
"job_id": job_id,
"family": "SPL",
"scopes": [{"scope_id": "SPL.target-Ft.target_only"}],
"manifests": [{"stage": "identify", "status": "success"}],
"stage_status": {
"identify": {"stage": "identify", "status": "success"}
},
}
registry = ConversationToolRegistry()
registry.register(
name="inspect_project_context",
description="Inspect current project context.",
args_model=_NoArgs,
handler=lambda _args: context,
permission="read_only",
)
llm = _SequenceLLM(
[
_response(
'{"goal":"project_status","needs":["project_context"],'
'"subject":"当前工作流的下一业务动作",'
'"referent":null,"clarification_question":null,'
'"interaction_kind":"status","workflow_id":"gene_family",'
'"analysis_request_kind":null,'
'"uses_current_recommendation":false,'
'"confirms_pending_proposal":false}'
),
_response(
tool_calls=[
LLMToolCall(
id="inspect-next-analysis-1",
name="inspect_project_context",
arguments="{}",
)
]
),
_response(
'{"status":"ok","action":"show_result",'
'"message":"下一步是构建亚家族证据。"}'
),
]
)
outcome = asyncio.run(
RouterRuntime(llm, registry, max_iterations=5).run(
query="接下来的分析有哪些?",
history=[],
session_id="session-next-analysis-question",
job_id=job_id,
goal_ledger=GoalLedger(
conversation_id="conversation-next-analysis-question",
revision=0,
user_terms={"active_job_id": job_id, "gene_family": "SPL"},
),
)
)
assert outcome.action == "show_result"
assert outcome.proposal is None
assert outcome.recommendation is not None
assert outcome.recommendation.transition_id == "build_subfamily_evidence"
def test_show_result_persists_and_cancel_clears_current_recommendation(tmp_path):
repository = SQLiteAgentStateRepository(tmp_path / "agent-state.sqlite3")
repository.initialize()
store = PersistentConversationStore(repository)
turn = store.begin_turn(
session_id="session-current-recommendation",
job_id="job-current-recommendation",
query="接下来的分析有哪些?",
client_history=[],
)
recommendation = _recommendation().model_dump(mode="json")
ledger = store.finish_turn(
turn,
assistant_message="下一步是构建亚家族证据。",
interaction_kind="status",
action="show_result",
recommendation=recommendation,
)
assert (
ledger.model_dump(mode="json")["user_terms"]["current_recommendation"]
== recommendation
)
cancel_turn = store.begin_turn(
session_id="session-current-recommendation",
job_id="job-current-recommendation",
query="取消",
client_history=[],
)
ledger = store.finish_turn(
cancel_turn,
assistant_message="已取消。",
interaction_kind="cancel",
action="reply",
)
assert "current_recommendation" not in ledger.user_terms
def test_pending_proposal_preserves_complete_workflow_action_binding(tmp_path):
repository = SQLiteAgentStateRepository(tmp_path / "agent-state.sqlite3")
repository.initialize()
store = PersistentConversationStore(repository)
turn = store.begin_turn(
session_id="session-action-binding",
job_id="job-action-binding",
query="执行当前推荐",
client_history=[],
)
proposal = {
"proposal_id": "proposal-action-binding",
"request": "执行 build_subfamily_evidence",
"subject": "执行当前推荐",
"workflow_id": "gene_family",
"workflow_action_id": "build_subfamily_evidence",
"authorized_stages": "domain,motif,phylogeny,structure,combo",
"target_stage": "combo",
}
ledger = store.finish_turn(
turn,
assistant_message="请确认。",
interaction_kind="supplement",
action="clarify",
proposal=proposal,
)
assert ledger.user_terms["pending_proposal"] == {
**proposal,
"job_id": "job-action-binding",
}
def test_execute_current_recommendation_revalidates_and_limits_intent():
job_id = "job-execute-current-recommendation"
context = {
"job_id": job_id,
"family": "SPL",
"scopes": [{"scope_id": "SPL.target-Ft.target_only"}],
"manifests": [{"stage": "identify", "status": "success"}],
"stage_status": {
"identify": {"stage": "identify", "status": "success"}
},
}
recommendation = RouterRuntime._current_workflow_recommendation(
"gene_family", job_id, context
)
registry = ConversationToolRegistry()
registry.register(
name="inspect_project_context",
description="Inspect current project context.",
args_model=_NoArgs,
handler=lambda _args: context,
permission="read_only",
)
registry.register(
name="recognize_analysis_intent",
description="Recognize an analysis intent.",
args_model=_IntentArgs,
handler=lambda _args: {
"gene_family": "SPL",
"stages": ["identify"],
"primary_stage": "identify",
"params": {"targetSpecies": ["Ft"], "referenceSpecies": ["At"]},
},
permission="pure",
)
ledger = GoalLedger(
conversation_id="conversation-execute-current-recommendation",
revision=0,
user_terms={
"active_job_id": job_id,
"current_recommendation": recommendation.model_dump(mode="json"),
},
)
llm = _SequenceLLM(
[
_response(
'{"goal":"new_analysis","needs":["analysis_intent"],'
'"subject":"执行当前推荐动作","referent":"current_recommendation",'
'"clarification_question":null,"interaction_kind":"supplement",'
'"analysis_request_kind":"execution",'
'"uses_current_recommendation":true,'
'"confirms_pending_proposal":false}'
),
_response(
tool_calls=[
LLMToolCall(
id="recognize-current-recommendation-1",
name="recognize_analysis_intent",
arguments=json.dumps(
{"query": "执行接下来的分析"}, ensure_ascii=False
),
)
]
),
_response(
'{"status":"ok","action":"propose_analysis",'
'"message":"已交给现有分析系统。"}'
),
]
)
outcome = asyncio.run(
RouterRuntime(llm, registry, max_iterations=5).run(
query="执行接下来的分析",
history=[],
session_id="session-execute-current-recommendation",
job_id=job_id,
goal_ledger=ledger,
)
)
assert outcome.action == "propose_analysis"
assert outcome.intent["stages"] == [
"domain",
"motif",
"phylogeny",
"structure",
"combo",
]
assert outcome.intent["primary_stage"] == "combo"
assert outcome.intent["workflow_action_id"] == "build_subfamily_evidence"
def test_semantic_business_referent_hands_off_when_workflow_is_terminal():
captured_queries = []
registry = ConversationToolRegistry()
registry.register(
name="recognize_analysis_intent",
description="Recognize an analysis intent.",
args_model=_IntentArgs,
handler=lambda args: captured_queries.append(args.query)
or {
"gene_family": "SPL",
"stages": ["promoter", "synteny", "ppi"],
"primary_stage": "promoter",
},
permission="pure",
)
llm = _SequenceLLM(
[
_response(
'{"goal":"new_analysis","needs":["analysis_intent"],'
'"subject":"执行上一轮说明的后续业务",'
'"referent":"常规基因家族分析",'
'"clarification_question":null,"interaction_kind":"supplement",'
'"workflow_id":"gene_family","analysis_request_kind":"execution",'
'"uses_current_recommendation":true,'
'"confirms_pending_proposal":false}'
),
_response(
tool_calls=[
LLMToolCall(
id="recognize-semantic-business-referent-1",
name="recognize_analysis_intent",
arguments=json.dumps(
{"query": "执行常规基因家族分析"}, ensure_ascii=False
),
)
]
),
_response(
'{"status":"ok","action":"propose_analysis",'
'"message":"分析请求已交给现有意图系统。"}'
),
]
)
outcome = asyncio.run(
RouterRuntime(llm, registry, max_iterations=5).run(
query="执行接下来的分析",
history=[
{
"role": "assistant",
"content": (
"亚家族分类与确认已经完成,"
"接下来进入常规基因家族分析。"
),
}
],
session_id="session-semantic-business-referent",
job_id="job-confirmed-family",
goal_ledger=GoalLedger(
conversation_id="conversation-semantic-business-referent",
revision=0,
user_terms={"active_job_id": "job-confirmed-family"},
),
)
)
assert outcome.action == "propose_analysis"
assert captured_queries == ["执行常规基因家族分析"]
assert outcome.intent["stages"] == ["promoter", "synteny", "ppi"]
def test_unresolved_current_recommendation_without_binding_still_clarifies():
llm = _SequenceLLM(
[
_response(
'{"goal":"new_analysis","needs":["analysis_intent"],'
'"subject":"执行当前推荐动作",'
'"referent":"current_recommendation",'
'"clarification_question":null,"interaction_kind":"supplement",'
'"workflow_id":"gene_family","analysis_request_kind":"execution",'
'"uses_current_recommendation":true,'
'"confirms_pending_proposal":false}'
)
]
)
outcome = asyncio.run(
RouterRuntime(llm, ConversationToolRegistry(), max_iterations=5).run(
query="执行接下来的分析",
history=[],
session_id="session-unresolved-recommendation",
job_id="job-unresolved-recommendation",
goal_ledger=GoalLedger(
conversation_id="conversation-unresolved-recommendation",
revision=0,
user_terms={"active_job_id": "job-unresolved-recommendation"},
),
)
)
assert outcome.action == "clarify"
assert "明确" in outcome.message
def test_execute_current_recommendation_rejects_cross_job_binding():
recommendation = _recommendation(job_id="job-original")
ledger = GoalLedger(
conversation_id="conversation-cross-job-recommendation",
revision=0,
user_terms={
"current_recommendation": recommendation.model_dump(mode="json")
},
)
llm = _SequenceLLM(
[
_response(
'{"goal":"new_analysis","needs":["analysis_intent"],'
'"subject":"执行当前推荐动作","referent":"current_recommendation",'
'"clarification_question":null,"interaction_kind":"supplement",'
'"analysis_request_kind":"execution",'
'"uses_current_recommendation":true,'
'"confirms_pending_proposal":false}'
)
]
)
outcome = asyncio.run(
RouterRuntime(llm, ConversationToolRegistry(), max_iterations=5).run(
query="执行接下来的分析",
history=[],
session_id="session-cross-job-recommendation",
job_id="job-other",
goal_ledger=ledger,
)
)
assert outcome.action == "clarify"
assert "当前 job" in outcome.message
+88
View File
@@ -0,0 +1,88 @@
import asyncio
from types import SimpleNamespace
from gfa_core.mas.agents.conversation_tools import ConversationToolRegistry
from gfa_core.mas.agents.router_runtime import RouterRuntime
from gfa_core.mas.llm import LLMMessage, LLMResponse
from gfa_core.mas.llm.providers.deepseek import DeepSeekProvider
def test_deepseek_provider_forwards_requested_json_mode():
captured = {}
class _Completions:
def create(self, **kwargs):
captured.update(kwargs)
return SimpleNamespace(
choices=[
SimpleNamespace(
message=SimpleNamespace(content='{"ok":true}', tool_calls=[]),
finish_reason="stop",
)
],
usage=SimpleNamespace(
prompt_tokens=1,
completion_tokens=1,
total_tokens=2,
),
)
provider = DeepSeekProvider(api_key="test-key")
provider.client = SimpleNamespace(
chat=SimpleNamespace(completions=_Completions())
)
provider.call(
[LLMMessage(role="user", content="Return JSON")],
response_format={"type": "json_object"},
)
assert captured["response_format"] == {"type": "json_object"}
def test_router_route_but_not_freeform_reply_requests_native_json_mode():
class _RecordingLLM:
def __init__(self):
self.calls = []
self.responses = [
LLMResponse(
content=(
'{"goal":"general","needs":[],"subject":"编写脚本",'
'"referent":null,"clarification_question":null,'
'"interaction_kind":"supplement"}'
),
model="test-model",
provider="test-provider",
usage={},
),
LLMResponse(
content=(
'{"status":"ok","action":"reply",'
'"message":"已生成脚本。"}'
),
model="test-model",
provider="test-provider",
usage={},
),
]
async def acall(self, _messages, **kwargs):
self.calls.append(kwargs)
return self.responses.pop(0)
llm = _RecordingLLM()
outcome = asyncio.run(
RouterRuntime(llm, ConversationToolRegistry(), max_iterations=5).run(
query="帮我写一个 FASTA 脚本",
history=[],
session_id="deepseek-json-protocol",
job_id="",
goal_ledger=None,
)
)
assert outcome.action == "reply"
assert [call.get("response_format") for call in llm.calls] == [
{"type": "json_object"},
None,
]
@@ -0,0 +1,70 @@
import asyncio
from types import SimpleNamespace
from gfa_core.mas.agents.base import AgentMessage, AgentState
from gfa_core.mas.agents.intent_v2 import IntentAgentV2
def test_fallback_does_not_turn_unrecognized_continuation_into_identify():
intent = IntentAgentV2(None)._fallback_recognition("执行接下来的分析")
assert intent["stages"] == []
assert intent["needs_clarification"] is True
assert intent["clarification_type"] == "analysis_stage"
def test_validation_does_not_turn_empty_model_stages_into_identify():
intent = IntentAgentV2(None)._validate_intent(
{
"gene_family": "SPL",
"species": "Ft",
"stages": [],
"params": {},
},
"执行接下来的步骤",
)
assert intent["stages"] == []
assert intent["needs_clarification"] is True
assert intent["clarification_type"] == "analysis_stage"
def test_fallback_still_recognizes_explicit_identify():
intent = IntentAgentV2(None)._fallback_recognition("执行 identify 分析")
assert intent["stages"] == ["identify"]
assert intent.get("needs_clarification") is not True
def test_empty_model_response_cannot_authorize_identify_and_keeps_goal_context():
class _EmptyLLM:
async def acall(self, *_args, **_kwargs):
return SimpleNamespace(content="")
state = AgentState(session_id="empty-intent", job_id="job-empty-intent")
message = AgentMessage(
sender="router",
receiver="intent_v2",
message_type="recognize_intent",
payload={
"query": "执行接下来的分析",
"job_id": "job-empty-intent",
"workspace": "/tmp/job-empty-intent",
"goal_context": {
"user_terms": {
"gene_family": "SPL",
"target_species": ["Ft"],
"reference_species": ["At"],
"comparative_species": [],
}
},
},
)
returned = asyncio.run(IntentAgentV2(_EmptyLLM()).execute(state, message))
intent = returned.results["results"]["intent"]
assert intent["stages"] == []
assert intent["needs_clarification"] is True
assert intent["target_species"] == ["Ft"]
assert intent["reference_species"] == ["At"]
@@ -0,0 +1,85 @@
import json
from gfa_core.mas.agents.conversation_tools import build_plantai_conversation_tools
from gfa_core.mas.llm import LLMManager
from gfa_core.mas.workflow_models import GoalLedger
def _registry(tmp_path, *, job_id="job-data-preparation-handoff"):
ledger = GoalLedger(
conversation_id="conversation-data-preparation-handoff",
revision=0,
user_terms={
"active_job_id": job_id,
"gene_family": "SPL",
"target_species": ["Ft"],
"reference_species": ["At"],
"config_confirmed": True,
},
)
return build_plantai_conversation_tools(
llm_manager=LLMManager({}),
project_root=tmp_path,
bound_job_id=job_id,
goal_ledger=ledger,
)
def test_analysis_handoff_is_not_rewritten_to_genome_prepare(tmp_path):
resolved = _registry(tmp_path).resolve_analysis_handoff_request(
"执行亚家族分类",
workflow_id="gene_family",
guidance=True,
)
assert resolved.startswith("执行亚家族分类\n")
assert "执行数据准备(genome_prepare)" not in resolved
assert "目标物种 Ft" in resolved
assert "参考物种 At" in resolved
assert "正式 config.xlsx 已确认" in resolved
def test_genome_prepare_manifest_does_not_change_analysis_handoff(tmp_path):
job_id = "job-manifest-independent-handoff"
registry = _registry(tmp_path, job_id=job_id)
before = registry.resolve_analysis_handoff_request(
"执行亚家族分类",
workflow_id="gene_family",
guidance=True,
)
final_dir = tmp_path / "workspace" / job_id / "final"
final_dir.mkdir(parents=True)
(final_dir / "genome_prepare.genome_prepare.manifest.json").write_text(
json.dumps({"stage": "genome_prepare", "status": "success"}),
encoding="utf-8",
)
after = registry.resolve_analysis_handoff_request(
"执行亚家族分类",
workflow_id="gene_family",
guidance=True,
)
assert after == before
def test_explicit_data_preparation_request_is_preserved(tmp_path):
resolved = _registry(tmp_path).resolve_analysis_handoff_request(
"执行数据准备",
workflow_id="data_preparation",
guidance=False,
)
assert resolved.startswith("执行数据准备\n")
assert "当前业务动作已由数据准备契约唯一确定" not in resolved
def test_next_step_question_is_not_rewritten_to_genome_prepare(tmp_path):
resolved = _registry(tmp_path).resolve_analysis_handoff_request(
"接下来需要做什么?",
workflow_id="gene_family",
guidance=True,
)
assert resolved.startswith("接下来需要做什么?\n")
assert "执行数据准备(genome_prepare)" not in resolved
@@ -0,0 +1,204 @@
import asyncio
from pydantic import BaseModel
from gfa_core.mas.agents.conversation_tools import ConversationToolRegistry
from gfa_core.mas.agents.router_runtime import RouterRuntime
from gfa_core.mas.llm import LLMResponse, LLMToolCall
_HISTORY = [
{"role": "user", "content": "如果我想写个脚本来修订我的蛋白id"},
{"role": "assistant", "content": "请提供输入格式、修订规则和输出要求。"},
{"role": "user", "content": ">id.02020202 | name:123 | xxx"},
{"role": "assistant", "content": "请确认修订规则。"},
{"role": "user", "content": "仅保留第一个字段(id.02020202)"},
{
"role": "assistant",
"content": "请确认输入文件格式,以及覆盖原文件还是输出到新文件?",
},
]
def _response(content: str) -> LLMResponse:
return LLMResponse(
content=content,
model="test-model",
provider="test-provider",
usage={},
)
class _SequenceLLM:
def __init__(self, responses):
self.responses = list(responses)
self.calls = 0
async def acall(self, _messages, **_kwargs):
self.calls += 1
return self.responses.pop(0)
class _QueryArgs(BaseModel):
query: str
def test_drafting_profile_preserves_short_specification_follow_ups():
messages = RouterRuntime._route_messages(
query="FASTA格式,输出到新文件",
history=_HISTORY,
session_id="drafting-continuation",
job_id="job-bound-to-gene-family",
goal_ledger=None,
)
prompt = " ".join(messages[0].content.split())
assert "Multi-turn drafting continuity" in prompt
assert "a supplement to the same general goal" in prompt
assert "must not be routed to utility" in prompt
assert "actually transform a current or supplied file" in prompt
def test_explicit_script_request_cannot_be_promoted_to_utility_by_router():
llm = _SequenceLLM(
[
_response(
'{"goal":"utility","needs":["utility_intent"],'
'"subject":"简化FASTA ID","referent":"test.fa",'
'"clarification_question":null,"interaction_kind":"independent"}'
),
_response(
'{"status":"ok","action":"reply",'
'"message":"下面给出一个只生成、不执行的 Python 脚本。"}'
),
]
)
outcome = asyncio.run(
RouterRuntime(llm, ConversationToolRegistry(), max_iterations=5).run(
query="帮我写脚本简化 test.fa 的id",
history=[],
session_id="explicit-script-request",
job_id="",
goal_ledger=None,
)
)
assert outcome.action == "reply"
assert outcome.tool_trace == []
assert llm.calls == 2
def test_drafting_follow_up_uses_responder_instead_of_router_clarification():
llm = _SequenceLLM(
[
_response(
'{"goal":"general","needs":[],"subject":"补充FASTA脚本规则",'
'"referent":"蛋白ID脚本",'
'"clarification_question":"请重新提供全部需求。",'
'"interaction_kind":"supplement"}'
),
_response(
'{"status":"ok","action":"reply",'
'"message":"已记录保留第一个字段;还需要确认输入格式和输出方式。"}'
),
]
)
outcome = asyncio.run(
RouterRuntime(llm, ConversationToolRegistry(), max_iterations=5).run(
query="仅保留第一个字段(id.02020202)",
history=_HISTORY[:4],
session_id="drafting-follow-up",
job_id="",
goal_ledger=None,
)
)
assert outcome.action == "reply"
assert outcome.message.startswith("已记录保留第一个字段")
assert llm.calls == 2
def test_locked_general_reply_accepts_bare_drafting_text_without_retry():
llm = _SequenceLLM(
[
_response(
'{"goal":"general","needs":[],"subject":"生成FASTA脚本",'
'"referent":null,"clarification_question":null,'
'"interaction_kind":"supplement"}'
),
_response("```python\nprint('FASTA')\n```"),
_response(
'{"status":"ok","action":"reply",'
'"message":"recovered after retry"}'
),
]
)
outcome = asyncio.run(
RouterRuntime(llm, ConversationToolRegistry(), max_iterations=5).run(
query="FASTA格式,输出到新文件",
history=_HISTORY,
session_id="bare-drafting-reply",
job_id="",
goal_ledger=None,
)
)
assert outcome.action == "reply"
assert outcome.message == "```python\nprint('FASTA')\n```"
assert outcome.tool_trace == []
assert llm.calls == 2
def test_unrecognized_file_utility_returns_actionable_chinese_clarification():
registry = ConversationToolRegistry()
registry.register(
name="recognize_utility_intent",
description="Recognize a supported utility request.",
args_model=_QueryArgs,
handler=lambda _args: {
"success": False,
"error_code": "utility_intent_not_found",
"message": "No supported utility request was recognized.",
},
permission="pure",
)
llm = _SequenceLLM(
[
_response(
'{"goal":"utility","needs":["utility_intent"],'
'"subject":"处理文件","referent":null,'
'"clarification_question":null,"interaction_kind":"independent"}'
),
LLMResponse(
content="",
model="test-model",
provider="test-provider",
usage={},
tool_calls=[
LLMToolCall(
id="utility-not-found-1",
name="recognize_utility_intent",
arguments='{"query":"帮我处理一下文件"}',
)
],
),
]
)
outcome = asyncio.run(
RouterRuntime(llm, registry, max_iterations=5).run(
query="帮我处理一下文件",
history=[],
session_id="unrecognized-utility",
job_id="job-utility",
goal_ledger=None,
)
)
assert outcome.action == "clarify"
assert "实际文件路径" in outcome.message
assert "处理规则" in outcome.message
assert "I couldn't identify" not in outcome.message
@@ -0,0 +1,113 @@
from gfa_core.mas.agents.router_runtime import RouteDecision, RouterRuntime
def _general_prompt(query: str, history=None) -> str:
return RouterRuntime._initial_messages(
query=query,
history=list(history or []),
session_id="product-business-session",
job_id="job-bound-to-other-workflow",
decision=RouteDecision(
goal="general",
needs=[],
subject="解释产品业务",
interaction_kind="independent",
),
goal_ledger=None,
)[0].content
def _route_prompt(query: str) -> str:
return RouterRuntime._route_messages(
query=query,
history=[],
session_id="product-business-route",
job_id="job-bound-to-other-workflow",
goal_ledger=None,
)[0].content
def test_gene_family_product_help_defines_two_level_business_view():
prompt = _general_prompt("基因家族有哪些分析步骤?")
compact = " ".join(prompt.split())
assert "亚家族分类与确认(前置)" in prompt
assert "常规基因家族分析(主体)" in prompt
assert "First answer with these two parts only" in prompt
assert "Do not enumerate any individual stage in that first answer" in compact
assert "must not use RAG or project tools" in compact
assert (
"phylogeny,motif,domain,structure,combo,promoter,tfbs,synteny,"
"interspecies_synteny,karyotype,pep,ortholog,kaks,ppi"
) in prompt
def test_rnaseq_product_help_defines_modes_and_authoritative_inputs():
prompt = _general_prompt("系统中 RNA-seq 分析有哪些步骤?")
compact = " ".join(prompt.split())
assert "全基因组 RNA-seq 分析(主体)" in prompt
assert "已确认基因家族成员的 RNA 结果投影" in prompt
assert "First answer with these two modes only" in compact
assert "config.xlsx and samples.yaml are equally authoritative inputs" in compact
assert (
"rna_fact_projection,rnaseq_qc,rnaseq_align,rnaseq_quant,"
"rnaseq_sample_qc,rnaseq_de"
) in prompt
assert "must not rerun a separate family-mode DE" in compact
def test_rnaseq_product_question_and_execution_have_distinct_route_examples():
prompt = _route_prompt("全基因组 RNA-seq 分析的正式输入和具体步骤是什么?")
assert (
'"全基因组 RNA-seq 分析的正式输入和具体步骤是什么?" -> '
"general + needs=[]."
) in prompt
assert (
'"执行全基因组 RNA-seq 分析" -> new_analysis + analysis_intent.'
) in prompt
def test_product_follow_up_keeps_all_prior_business_context():
history = [
{"role": "user", "content": "系统中转录组分析有哪些步骤?"},
{
"role": "assistant",
"content": "分为全基因组 RNA-seq 和已确认家族成员的 RNA 结果投影。",
},
{"role": "user", "content": "先说全基因组这一部分。"},
{
"role": "assistant",
"content": "需要先确认 config.xlsx 和 samples.yaml。",
},
]
messages = RouterRuntime._initial_messages(
query="那家族成员这一部分呢?",
history=history,
session_id="product-follow-up-session",
job_id="job-product-follow-up",
decision=RouteDecision(
goal="general",
needs=[],
subject="继续解释产品业务",
interaction_kind="supplement",
),
goal_ledger=None,
)
assert [message.content for message in messages[1:-1]] == [
item["content"] for item in history
]
def test_completed_subfamily_or_stage_list_is_not_full_analysis_completion():
assert not RouterRuntime._claims_gene_family_complete_at_readiness_boundary(
"亚家族分类与确认已经完成,接下来进入常规基因家族分析,包括 ppi。"
)
assert not RouterRuntime._claims_gene_family_complete_at_readiness_boundary(
"已完成的阶段包括 identify;后续还有常规基因家族分析。"
)
assert RouterRuntime._claims_gene_family_complete_at_readiness_boundary(
"亚家族确认完成,因此完整基因家族分析已经全部完成。"
)
@@ -0,0 +1,100 @@
from gfa_core.mas.agent_state_repository import SQLiteAgentStateRepository
from gfa_core.mas.agents.router_runtime import RouteDecision, RouterRuntime
from gfa_core.mas.persistent_conversation import PersistentConversationStore
def test_begin_turn_imports_and_returns_complete_valid_history(tmp_path):
repository = SQLiteAgentStateRepository(tmp_path / "agent-state.sqlite3")
repository.initialize()
store = PersistentConversationStore(repository)
client_history = [
{
"role": "user" if index % 2 == 0 else "assistant",
"content": f"历史消息-{index:02d}",
}
for index in range(14)
]
turn = store.begin_turn(
session_id="full-history-session",
job_id="job-full-history",
query="当前问题",
client_history=client_history,
)
assert list(turn.history) == client_history
stored = repository.list_messages(turn.conversation_id)
assert [item.content for item in stored] == [
*(item["content"] for item in client_history),
"当前问题",
]
def test_begin_turn_returns_all_persisted_messages_not_only_latest_ten(tmp_path):
repository = SQLiteAgentStateRepository(tmp_path / "agent-state.sqlite3")
repository.initialize()
store = PersistentConversationStore(repository)
first_turn = store.begin_turn(
session_id="persisted-full-history-session",
job_id="job-full-history",
query="第一个问题",
client_history=[],
)
store.finish_turn(
first_turn,
assistant_message="第一个回答",
interaction_kind="independent",
action="reply",
)
for index in range(1, 7):
turn = store.begin_turn(
session_id="persisted-full-history-session",
job_id="job-full-history",
query=f"问题-{index}",
client_history=[],
)
store.finish_turn(
turn,
assistant_message=f"回答-{index}",
interaction_kind="supplement",
action="reply",
)
final_turn = store.begin_turn(
session_id="persisted-full-history-session",
job_id="job-full-history",
query="最终问题",
client_history=[],
)
assert len(final_turn.history) == 14
assert final_turn.history[0]["content"] == "第一个问题"
assert final_turn.history[-1]["content"] == "回答-6"
def test_main_agent_prompt_receives_the_complete_history():
history = [
{
"role": "user" if index % 2 == 0 else "assistant",
"content": f"消息-{index}",
}
for index in range(12)
]
messages = RouterRuntime._initial_messages(
query="当前问题",
history=history,
session_id="full-history-session",
job_id="job-full-history",
decision=RouteDecision(
goal="general",
needs=[],
subject="继续普通问答",
interaction_kind="supplement",
),
goal_ledger=None,
)
assert [message.content for message in messages[1:-1]] == [
f"消息-{index}" for index in range(12)
]
@@ -1,26 +1,41 @@
from __future__ import annotations
import asyncio
import json
from typing import Any
import pytest
from pydantic import BaseModel
from gfa_core.mas.agents.conversation_tools import ConversationToolRegistry
from gfa_core.mas.agents.router_runtime import (
RouteDecision,
RouterRuntime,
)
from gfa_core.mas.llm import LLMResponse, LLMToolCall
from gfa_core.mas.workflow_models import GoalLedger
class _PassThroughTools:
@staticmethod
def resolve_analysis_handoff_request(
query: str,
*,
workflow_id: str | None = None,
guidance: bool = False,
) -> str:
return query
class _IntentArgs(BaseModel):
query: str
class _SequenceLLM:
def __init__(self, responses):
self.responses = list(responses)
async def acall(self, *_args, **_kwargs):
return self.responses.pop(0)
def _response(content="", *, tool_calls=None):
return LLMResponse(
content=content,
model="test-model",
provider="test-provider",
usage={},
tool_calls=list(tool_calls or []),
)
@pytest.mark.parametrize(
@@ -36,13 +51,43 @@ class _PassThroughTools:
("执行motif分析", "gene_family", None),
),
)
def test_new_analysis_proposal_preserves_original_request_without_main_agent_planning(
def test_new_analysis_hands_original_request_to_intent_without_main_agent_planning(
monkeypatch: pytest.MonkeyPatch,
query: str,
workflow_id: str,
clarification_question: str | None,
) -> None:
runtime = RouterRuntime(object(), _PassThroughTools(), max_iterations=5) # type: ignore[arg-type]
captured_queries = []
registry = ConversationToolRegistry()
registry.register(
name="recognize_analysis_intent",
description="Recognize an analysis intent.",
args_model=_IntentArgs,
handler=lambda args: captured_queries.append(args.query)
or {"stages": ["motif"], "primary_stage": "motif"},
permission="pure",
)
runtime = RouterRuntime(
_SequenceLLM(
[
_response(
tool_calls=[
LLMToolCall(
id="recognize-direct-analysis",
name="recognize_analysis_intent",
arguments=json.dumps({"query": query}, ensure_ascii=False),
)
]
),
_response(
'{"status":"ok","action":"propose_analysis",'
'"message":"分析请求已交给现有意图系统。"}'
),
]
),
registry,
max_iterations=5,
)
async def decide_route(**_: Any) -> tuple[RouteDecision, int, list[dict[str, Any]]]:
return (
@@ -74,11 +119,10 @@ def test_new_analysis_proposal_preserves_original_request_without_main_agent_pla
)
)
assert outcome.action == "clarify"
assert outcome.proposal is not None
assert outcome.proposal["request"] == query
assert "workflow_id" not in outcome.proposal
assert "全基因组还是" not in outcome.message
assert outcome.action == "propose_analysis"
assert outcome.proposal is None
assert captured_queries == [query]
assert outcome.intent["stages"] == ["motif"]
def test_new_analysis_decision_discards_main_agent_workflow_and_clarification() -> None:
@@ -0,0 +1,107 @@
from types import SimpleNamespace
from gfa_core.mas import context_index, stage_requirements
from gfa_core.mas.context_index import ProjectContextIndex
from gfa_core.mas.stage_requirements import (
resolve_many_stage_requirements,
resolve_stage_requirements,
)
def test_stage_requirements_does_not_load_unconsumed_global_stage_status(
tmp_path,
monkeypatch,
):
calls = []
def fake_get_stage_status(**kwargs):
calls.append(kwargs["stage"])
return SimpleNamespace(status="missing", manifest_path="", to_dict=lambda: {})
monkeypatch.setattr(context_index, "get_stage_status", fake_get_stage_status)
report = resolve_stage_requirements(
stage="identify",
prefix="Ft.SPL",
job_id="job-test",
project_root=tmp_path,
)
assert report.stage == "identify"
assert calls == []
def _empty_context(tmp_path, job_id):
workspace = tmp_path / "workspace" / job_id
return ProjectContextIndex(
job_id=job_id,
workspace=str(workspace),
input_dir=str(tmp_path / "input" / job_id),
final_dir=str(workspace / "final"),
logs_dir=str(workspace / "logs"),
)
def test_many_stage_requirements_reuses_context_without_mutating_shared_identity(
tmp_path,
monkeypatch,
):
calls = []
base_context = _empty_context(tmp_path, "job-test")
def fake_build_project_context(**kwargs):
calls.append(kwargs["prefix"])
return base_context
monkeypatch.setattr(
stage_requirements,
"build_project_context",
fake_build_project_context,
)
reports = resolve_many_stage_requirements(
stages=["identify", "phylogeny"],
prefix="Ft.SPL",
job_id="job-test",
params_by_stage={
"identify": {"targetSpecies": ["Ft"]},
"phylogeny": {"targetSpecies": ["Os"]},
},
project_root=tmp_path,
)
assert calls == ["Ft.SPL"]
assert base_context.target_species == []
assert reports["identify"]["identity"]["target_species"] == ["Ft"]
assert reports["phylogeny"]["identity"]["target_species"] == ["Os"]
def test_many_stage_requirements_caches_context_per_effective_prefix(
tmp_path,
monkeypatch,
):
calls = []
def fake_build_project_context(**kwargs):
calls.append(kwargs["prefix"])
return _empty_context(tmp_path, kwargs["job_id"])
monkeypatch.setattr(
stage_requirements,
"build_project_context",
fake_build_project_context,
)
resolve_many_stage_requirements(
stages=["identify", "phylogeny", "motif"],
prefix="Ft.SPL",
job_id="job-test",
params_by_stage={
"identify": {"prefix": "Ft.SPL"},
"phylogeny": {"prefix": "Ft.SPL"},
"motif": {"prefix": "At.SPL"},
},
project_root=tmp_path,
)
assert calls == ["Ft.SPL", "At.SPL"]