diff --git a/apps/web/vue3/main.py b/apps/web/vue3/main.py index f1f63a684..2fb3dd8f7 100644 --- a/apps/web/vue3/main.py +++ b/apps/web/vue3/main.py @@ -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, diff --git a/gfa_core/mas/agents/conversation_tools.py b/gfa_core/mas/agents/conversation_tools.py index 7ce006f7b..6e4bcc4b4 100644 --- a/gfa_core/mas/agents/conversation_tools.py +++ b/gfa_core/mas/agents/conversation_tools.py @@ -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)}。" diff --git a/gfa_core/mas/agents/intent_v2.py b/gfa_core/mas/agents/intent_v2.py index 334c8a33a..edb243612 100755 --- a/gfa_core/mas/agents/intent_v2.py +++ b/gfa_core/mas/agents/intent_v2.py @@ -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): diff --git a/gfa_core/mas/agents/router_runtime.py b/gfa_core/mas/agents/router_runtime.py index a2a5c948e..f0412fee0 100644 --- a/gfa_core/mas/agents/router_runtime.py +++ b/gfa_core/mas/agents/router_runtime.py @@ -11,11 +11,13 @@ from pydantic import BaseModel, Field, ValidationError, field_validator, model_v from gfa_core.mas.agents.conversation_tools import ConversationToolRegistry from gfa_core.mas.llm import LLMManager, LLMMessage from gfa_core.mas.persistent_conversation import is_short_acknowledgement +from gfa_core.mas.tools.stage_tools import stage_registry +from gfa_core.mas.utils.intent_parser import CORE_STAGES from gfa_core.mas.workflow_catalog import ( load_canonical_workflow_contract, load_rnaseq_business_context, ) -from gfa_core.mas.workflow_models import GoalLedger +from gfa_core.mas.workflow_models import CurrentRecommendation, GoalLedger InteractionKind = Literal[ @@ -39,6 +41,7 @@ class RouterOutcome(BaseModel): tool_trace: list[dict[str, Any]] = Field(default_factory=list) interaction_kind: InteractionKind = "independent" proposal: dict[str, str] | None = None + recommendation: CurrentRecommendation | None = None input_paths: list[str] = Field(default_factory=list) ui_action: Literal["open_config_dialog"] | None = None @@ -73,6 +76,7 @@ class RouteDecision(BaseModel): interaction_kind: InteractionKind = "independent" workflow_id: WorkflowId | None = None analysis_request_kind: AnalysisRequestKind | None = None + uses_current_recommendation: bool = False confirms_pending_proposal: bool = False @model_validator(mode="before") @@ -88,6 +92,21 @@ class RouteDecision(BaseModel): normalized["workflow_id"] = None return normalized + @model_validator(mode="before") + @classmethod + def discard_analysis_metadata_for_non_analysis_goal(cls, value: Any) -> Any: + if not isinstance(value, Mapping) or value.get("goal") == "new_analysis": + return value + if ( + value.get("analysis_request_kind") is None + and value.get("uses_current_recommendation") in {None, False} + ): + return value + normalized = dict(value) + normalized["analysis_request_kind"] = None + normalized["uses_current_recommendation"] = False + return normalized + @field_validator("clarification_question", mode="before") @classmethod def normalize_blank_clarification(cls, value: Any) -> Any: @@ -131,6 +150,12 @@ class RouteDecision(BaseModel): ) if self.analysis_request_kind is not None and self.goal != "new_analysis": raise ValueError("analysis_request_kind requires goal=new_analysis") + if self.uses_current_recommendation and ( + self.goal != "new_analysis" or self.analysis_request_kind != "execution" + ): + raise ValueError( + "uses_current_recommendation requires an execution new_analysis" + ) if ( self.clarification_question is not None and not self.clarification_question.strip() @@ -148,6 +173,91 @@ def _extract_authorized_input_paths(query: str) -> list[str]: paths.append(path) return paths +_REGULAR_GENE_FAMILY_STAGES = tuple( + stage for stage in CORE_STAGES if stage != "identify" +) + +_GENE_FAMILY_CONVERSATION_PROFILE = f""" +Gene-family business vocabulary: complete gene-family analysis has two parts: +亚家族分类与确认（前置） and 常规基因家族分析（主体）. For a broad product question +such as “基因家族有哪些分析步骤？”, First answer with these two parts only and ask +which part the user wants expanded as an optional invitation, not a required choice. This +question is complete and unambiguous: use goal=general with needs=[] and action=reply; +never clarify. Do not enumerate any individual stage in that first answer, even when the +broad question contains “步骤” or “有哪些”. If the user then asks for details, use +conversation history to expand the referenced part. The subfamily chain is family-member +identification, phylogeny/motif/domain/structure/combo evidence, reference classification, +and human confirmation. The regular analysis catalog is +{','.join(_REGULAR_GENE_FAMILY_STAGES)}. Expression is added only when explicitly requested. +The canonical workflow transitions are not the complete analysis catalog. This vocabulary +explains product behavior only: never turn it into an execution plan or choose stages for +IntentAgent or PlanningAgent. Questions and contextual follow-ups asking what these PlantAI +workflow parts contain are product help: they use goal=general with needs=[] and must not use +RAG or project tools. Apply this complete catalog only to product help; never use the complete +product catalog as current-job status. Questions about the current or next analysis in a +bound job must follow observed project evidence. +""" + + +def _ordered_contract_target_stages(workflow_id: str) -> tuple[str, ...]: + contract = load_canonical_workflow_contract(workflow_id) + actions = {action.id: action for action in contract.actions} + transitions_by_state = { + transition.from_state_id: transition for transition in contract.transitions + } + stages: list[str] = [] + visited_states: set[str] = set() + state_id = contract.initial_state_id + while state_id not in visited_states and state_id in transitions_by_state: + visited_states.add(state_id) + transition = transitions_by_state[state_id] + stages.append(actions[transition.action_id].target_stage) + state_id = transition.to_state_id + return tuple(stages) + + +_RNASEQ_WHOLE_GENOME_CORE_STAGES = _ordered_contract_target_stages( + "rnaseq_whole_genome" +) +_RNASEQ_FAMILY_MEMBER_STAGES = _ordered_contract_target_stages( + "rnaseq_family_member" +) + +_RNASEQ_CONVERSATION_PROFILE = f""" +RNA-seq/转录组 business vocabulary: the product has two modes: +全基因组 RNA-seq 分析（主体） and 已确认基因家族成员的 RNA 结果投影. For a broad +product question such as “系统中 RNA-seq/转录组分析有哪些步骤？”, First answer with +these two modes only and ask which mode the user wants expanded as an optional invitation, +not a required choice. This question is complete and unambiguous: use goal=general with +needs=[] and action=reply; never clarify. Do not enumerate any individual stage in that first +answer, even when the question contains “步骤” or “有哪些”. If the user then asks for details, +use the full conversation history to expand the referenced mode. + +For whole-genome RNA-seq, config.xlsx and samples.yaml are equally authoritative inputs. +config.xlsx supplies confirmed genome resources. samples.yaml supplies the confirmed sample +and experiment design, including FASTQ/sample identity, condition and group role, optional +batch/stratification fields, inclusion, comparisons, and applicable design. Explain that both +inputs are reviewed and confirmed before RNA fact projection. When expanding this mode, +explicitly name both config.xlsx and samples.yaml before listing the stages. The canonical +core stages are {','.join(_RNASEQ_WHOLE_GENOME_CORE_STAGES)}: RNA fact projection, FASTQ QC, +alignment, gene quantification, sample-level QC, and differential expression. GO enrichment, +KEGG enrichment, fgsea, expression visualization, WGCNA, and candidate-family analysis are +downstream or optional product analyses after the applicable core results; they are not the +six canonical core transitions and must not be presented as already authorized execution. +Do not invent generic textbook tools, preprocessing steps, or product capabilities. + +For confirmed-family-member RNA-seq, require successful whole-genome shared DE and exact +confirmed family members, then project those shared RNA results through +{','.join(_RNASEQ_FAMILY_MEMBER_STAGES)}. It must not rerun a separate family-mode DE and +must not silently select an arbitrary family-members file. + +These rules explain product behavior only. Product questions and contextual follow-ups use +goal=general with needs=[] and must not use RAG or project tools. A currently bound gene-family +or other job must not redirect the answer to that bound job. Do not convert product knowledge +into an execution plan. When the user requests execution, hand the self-contained business +request to the existing IntentAgent; IntentAgent, PlanningAgent, and the existing RNA-seq +pipeline retain stage selection, dependency, feasibility, and execution responsibilities. +""" _PRODUCT_PROFILE = """PlantAI is a plant bioinformatics analysis assistant. Its supported @@ -159,13 +269,25 @@ assistant: it may answer ordinary questions, draft or revise prose, and generate code directly in the conversation. Writing code in a reply does not execute that code and does not require an analysis or utility tool. Never refuse a request solely because it asks for source code. +Multi-turn drafting continuity: when conversation history establishes a code or prose +drafting goal, a short follow-up that supplies a requested specification is a supplement to +the same general goal. Reconstruct the original request from the full history. Once the +requested format, transformation rule, and output behavior are sufficient, keep the route +as goal=general with needs=[] so the final conversational response can produce the code or +draft without repeating questions that the user already answered. For example, after a +request to write a FASTA-header script, “FASTA format and output to a new file” is a drafting +specification and must not be routed to utility. Use utility only when the user asks PlantAI +to actually transform a current or supplied file, not when the user asks for code that they +can run themselves. PlantAI is not a plant-disease image diagnosis assistant. Never invent unsupported product abilities. -""" +""" + _GENE_FAMILY_CONVERSATION_PROFILE + _RNASEQ_CONVERSATION_PROFILE _SYSTEM_PROMPT = _PRODUCT_PROFILE + """You are PlantAI's conversation router. Never claim to execute an -analysis. Return raw JSON only, without markdown. The minimal RouterOutcome is +analysis. Return raw JSON only, without markdown. For a drafting reply, place the complete +requested code or prose inside the message string. Never return bare code, prose, or markdown +outside RouterOutcome JSON. The minimal RouterOutcome is {"status":"ok","action":"reply","message":"..."}; action must be exactly one of reply, clarify, show_result, propose_analysis, or propose_utility. @@ -208,30 +330,58 @@ Use the fewest tools needed. After successful evidence is sufficient, return the Never pass arguments from one tool schema to another. Never invent identity fields or tool arguments. Do not invent an intent.""" -_ROUTE_SYSTEM_PROMPT = _PRODUCT_PROFILE + """You are PlantAI's semantic route classifier. Return exactly +_ROUTE_SYSTEM_PROMPT = _PRODUCT_PROFILE + """You are PlantAI's semantic route classifier. Classify only; never answer the user's request, +and never return code, prose, or markdown instead of RouteDecision JSON. Return exactly one raw JSON RouteDecision object with goal, needs, subject, referent, and clarification_question. workflow_id is optional and must be exactly one of data_preparation, gene_family, rnaseq_whole_genome, or rnaseq_family_member. Also classify interaction_kind as exactly one of supplement, correction, switch, cancel, status, or independent. Do not use markdown. -Select workflow_id only for goal=new_analysis, or for goal=project_status when the user -asks for the current state of an established analysis workflow. +Select workflow_id only for goal=project_status when the user asks for the current state +or next available action of an established analysis workflow. Select it only when the workflow is specific enough, and select at most one workflow. -Asking what analysis should be planned next requests a new downstream plan, so use -goal=new_analysis. Asking only for status or progress without requesting a new plan uses -goal=project_status. The Router must not choose the next stages itself. +Asking which analysis is next, what remains, or what should be done next is a read-only +workflow progress question: use goal=project_status with project_context. Explicitly asking +to execute the current/next/recommended analysis uses goal=new_analysis, +analysis_request_kind=execution, and uses_current_recommendation=true. The Router must not +choose the next stages itself: the structured current recommendation supplies only the +already-computed workflow transition binding. +Classify the user's requested action before the object of that action. Words meaning +"next" or "remaining" never determine the route by themselves: a question asks for +current workflow guidance, while an imperative or explicit request to execute asks for +new computation and must be handed to the existing analysis system. +Boundary examples: +- "接下来的分析有哪些？" -> project_status + project_context. +- "接下来需要做什么？" -> project_status + project_context. +- "执行接下来的分析" -> new_analysis + analysis_intent, + analysis_request_kind=execution, uses_current_recommendation=true. +- "执行 domain 分析" -> new_analysis + analysis_intent, + analysis_request_kind=execution, uses_current_recommendation=false. Do not select a workflow for ordinary chat, product help, external knowledge, existing results, job files, or utilities. +RNA-seq product-help questions are general with needs=[] even when they name an analysis +mode, inputs, or steps: +- "系统中 RNA-seq/转录组分析有哪些步骤？" -> general + needs=[]. +- "全基因组 RNA-seq 分析的正式输入和具体步骤是什么？" -> general + needs=[]. +- "执行全基因组 RNA-seq 分析" -> new_analysis + analysis_intent. +Workflow contracts guide conversation and handoff only; PlanningAgentV2 remains responsible +for executable stage planning. For goal=new_analysis, do not ask for an analysis mode, select a workflow, or select stages. Leave workflow_id and clarification_question null and request analysis_intent; the existing IntentAgent and PlanningAgent own missing analysis choices and executable stage planning. -For goal=new_analysis, classify analysis_request_kind as guidance when the user asks -what should be done now, how to continue, or which analysis comes next. Use execution -when the user explicitly requests an analysis or named operation. This field describes +For goal=new_analysis, classify analysis_request_kind as guidance only when the user +explicitly asks the downstream analysis system to plan a new analysis request. Use +execution when the user explicitly requests an analysis or named operation. A read-only +question about what comes next is project_status, not new_analysis. This field describes the user's conversational act only; it must never list, add, remove, or choose stages. +Set uses_current_recommendation=true only for an execution request that refers to the +current/next/recommended workflow action. Keep it false for a newly named analysis request. +When the immediately preceding assistant answer names a business part but no canonical +action is bound, referent must name that business part rather than a generic phrase such +as “next analysis”; for example referent="亚家族分类与确认". IntentAgent resolves stages. Interaction classification: - supplement: adds information to the current goal without replacing it. @@ -333,6 +483,9 @@ Chinese boundary examples: - "查找相关资料并总结500字报告" -> general + external_knowledge. - "苦荞的SPL基因家族及工" -> general + needs=[] + clarification_question. - "用上传的蛋白序列鉴定SPL成员" -> new_analysis + analysis_intent. +- "接下来的分析有哪些" -> project_status + project_context. +- "执行接下来的分析" -> new_analysis + analysis_intent + + analysis_request_kind=execution + uses_current_recommendation=true. The available capabilities are project_context, analysis_results, external_knowledge, analysis_intent, job_file, job_files, and utility_intent. Analysis names mentioned inside product help do not by themselves @@ -345,6 +498,7 @@ Return exactly one raw JSON object with this shape and no markdown: {"goal":"general|project_status|existing_result|job_file|utility|new_analysis","needs":[], "subject":"nonblank","referent":null,"clarification_question":null,"workflow_id":null, "analysis_request_kind":null, +"uses_current_recommendation":false, "confirms_pending_proposal":false}. workflow_id may be data_preparation, gene_family, rnaseq_whole_genome, or rnaseq_family_member only for a specific new_analysis or an established workflow's @@ -379,7 +533,14 @@ _NEED_TO_TOOL = { } _OUTCOME_FIELDS = { - "status", "action", "message", "intent", "sources", "tool_trace", "proposal" + "status", + "action", + "message", + "intent", + "sources", + "tool_trace", + "proposal", + "recommendation", } _PYDANTIC_ERROR_TYPES = { "dict_type", @@ -440,6 +601,7 @@ class RouterRuntime: referent=decision.referent, interaction_kind=decision.interaction_kind, analysis_request_kind=decision.analysis_request_kind, + uses_current_recommendation=decision.uses_current_recommendation, confirms_pending_proposal=decision.confirms_pending_proposal, ) @@ -466,8 +628,77 @@ class RouterRuntime: "rnaseq_family_member", }: proposal["workflow_id"] = workflow_id + action_binding = { + key: str(raw.get(key) or "").strip() + for key in ("workflow_action_id", "authorized_stages", "target_stage") + } + if all(action_binding.values()): + proposal.update(action_binding) return proposal + @staticmethod + def _stored_current_recommendation( + goal_ledger: GoalLedger | None, + ) -> CurrentRecommendation | None: + if goal_ledger is None: + return None + raw = goal_ledger.user_terms.get("current_recommendation") + if not isinstance(raw, Mapping): + return None + try: + return CurrentRecommendation.model_validate(raw) + except ValidationError: + return None + + @staticmethod + def _recommendation_handoff_request( + recommendation: CurrentRecommendation, + ) -> str: + business_action = ( + "构建亚家族分类证据" + if recommendation.transition_id == "build_subfamily_evidence" + else recommendation.transition_id + ) + return ( + f"执行当前工作流业务动作：{business_action}" + f"（{recommendation.transition_id}）" + ) + + @staticmethod + def _limit_intent_to_confirmed_workflow_action( + intent: dict[str, Any], + proposal: Mapping[str, str] | None, + ) -> dict[str, Any]: + if proposal is None: + return intent + authorized_stages = [ + stage.strip() + for stage in str(proposal.get("authorized_stages") or "").split(",") + if stage.strip() + ] + target_stage = str(proposal.get("target_stage") or "").strip() + workflow_action_id = str(proposal.get("workflow_action_id") or "").strip() + if ( + not authorized_stages + or target_stage not in authorized_stages + or not workflow_action_id + ): + return intent + limited = dict(intent) + limited["stages"] = authorized_stages + limited["primary_stage"] = target_stage + limited["workflow_action_id"] = workflow_action_id + limited["description"] = ( + f"执行工作流业务动作 {workflow_action_id}（目标阶段 {target_stage}）" + ) + if str(limited.get("stage") or "").strip() not in {"", *authorized_stages}: + limited.pop("stage", None) + params = dict(limited.get("params") or {}) + params["workflowActionId"] = workflow_action_id + params["authorizedStages"] = authorized_stages + limited["params"] = params + return limited + @staticmethod def _new_analysis_proposal( *, @@ -499,12 +730,20 @@ class RouterRuntime: return "" contract = load_canonical_workflow_contract(workflow_id) actions = {action.id: action for action in contract.actions} + evidence_stages = { + requirement.id: requirement.stage + for requirement in contract.evidence_requirements + } transitions = [ { "id": transition.id, "from": transition.from_state_id, "to": transition.to_state_id, "target_stage": actions[transition.action_id].target_stage, + "required_evidence_stages": [ + evidence_stages[requirement_id] + for requirement_id in transition.evidence_requirement_ids + ], "requires_confirmation": bool(transition.confirmation_ids), } for transition in contract.transitions @@ -529,32 +768,311 @@ class RouterRuntime: return json.dumps(compact, ensure_ascii=False, separators=(",", ":")) @staticmethod - def _is_next_analysis_request(query: str) -> bool: - compact = re.sub(r"[\s，。！？,.!?]", "", str(query or "").lower()) - return bool( - re.search(r"(?:下一步|接下来).*(?:做什么|怎么做|什么分析|呢)?$", compact) - or re.search(r"后续(?:应该|需要|要)?(?:做什么|怎么做|什么分析)$", compact) + def _current_workflow_recommendation( + workflow_id: WorkflowId | None, + job_id: str, + project_context: Mapping[str, Any] | None, + ) -> CurrentRecommendation | None: + if not job_id or not isinstance(project_context, Mapping): + return None + family = str(project_context.get("family") or "").strip() or None + if workflow_id is None and family is not None: + workflow_id = "gene_family" + if workflow_id is None: + return None + + scopes = { + str(item.get("scope_id") or "").strip() + for item in project_context.get("scopes", ()) + if isinstance(item, Mapping) and str(item.get("scope_id") or "").strip() + } + if len(scopes) > 1: + return None + scope_id = next(iter(scopes), None) + + completed_stages: set[str] = set() + successful = {"success", "partial_success"} + stage_status = project_context.get("stage_status") + if isinstance(stage_status, Mapping): + for stage, raw in stage_status.items(): + if not isinstance(raw, Mapping): + continue + if str(raw.get("status") or "").strip() in successful: + normalized = str(raw.get("stage") or stage or "").strip() + if normalized: + completed_stages.add(normalized) + for manifest in project_context.get("manifests", ()): + if not isinstance(manifest, Mapping): + continue + if str(manifest.get("status") or "").strip() not in successful: + continue + stage = str(manifest.get("stage") or "").strip() + if stage: + completed_stages.add(stage) + + contract = load_canonical_workflow_contract(workflow_id) + requirements = {item.id: item for item in contract.evidence_requirements} + states = {item.id: item for item in contract.states} + actions = {item.id: item for item in contract.actions} + + def state_is_observed(state_id: str) -> bool: + return all( + requirements[requirement_id].stage in completed_stages + for requirement_id in states[state_id].evidence_requirement_ids + ) + + current_state_id = contract.initial_state_id + visited = {current_state_id} + while True: + observed = [ + transition + for transition in contract.transitions + if transition.from_state_id == current_state_id + and state_is_observed(transition.to_state_id) + ] + if len(observed) != 1 or observed[0].to_state_id in visited: + break + current_state_id = observed[0].to_state_id + visited.add(current_state_id) + + outgoing = [ + transition + for transition in contract.transitions + if transition.from_state_id == current_state_id + ] + if len(outgoing) != 1: + return None + transition = outgoing[0] + action = actions[transition.action_id] + authorized_stages = tuple( + dict.fromkeys( + [ + requirements[requirement_id].stage + for requirement_id in transition.evidence_requirement_ids + ] + + [action.target_stage] + ) + ) + evidence_payload = { + "job_id": job_id, + "workflow_id": workflow_id, + "family": family, + "scope_id": scope_id, + "completed_stages": sorted(completed_stages), + "transition_id": transition.id, + } + evidence_digest = hashlib.sha256( + json.dumps( + evidence_payload, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + ).hexdigest() + return CurrentRecommendation( + recommendation_id=f"rec-{evidence_digest[:24]}", + job_id=job_id, + workflow_id=workflow_id, + transition_id=transition.id, + from_state_id=transition.from_state_id, + to_state_id=transition.to_state_id, + target_stage=action.target_stage, + authorized_stages=authorized_stages, + evidence_digest=evidence_digest, + family=family, + scope_id=scope_id, ) @staticmethod - def _established_workflow_id( - decision: RouteDecision, - goal_ledger: GoalLedger, + def _workflow_stage_ids(workflow_id: WorkflowId | None) -> tuple[str, ...]: + if workflow_id is None: + return () + contract = load_canonical_workflow_contract(workflow_id) + return tuple( + dict.fromkeys( + [item.stage for item in contract.evidence_requirements] + + [item.target_stage for item in contract.actions] + ) + ) + + @staticmethod + def _subfamily_is_confirmed( + project_context: Mapping[str, Any] | None, + ) -> bool: + if not isinstance(project_context, Mapping): + return False + stage_status = project_context.get("stage_status") + if not isinstance(stage_status, Mapping): + return False + status = stage_status.get("subfamily_confirm") + return isinstance(status, Mapping) and str( + status.get("status") or "" + ).strip() in {"success", "partial_success"} + + @classmethod + def _workflow_guidance_stage_ids( + cls, + workflow_id: WorkflowId | None, + project_context: Mapping[str, Any] | None, + ) -> tuple[str, ...]: + stages = list(cls._workflow_stage_ids(workflow_id)) + if workflow_id == "gene_family" and cls._subfamily_is_confirmed( + project_context + ): + stages.extend(_REGULAR_GENE_FAMILY_STAGES) + return tuple(dict.fromkeys(stages)) + + @staticmethod + def _effective_workflow_id( + workflow_id: WorkflowId | None, + project_context: Mapping[str, Any] | None, ) -> WorkflowId | None: - if decision.workflow_id is not None: - return decision.workflow_id - terms = goal_ledger.user_terms - rna_route = str(terms.get("rna_route") or "").strip().lower() - if rna_route == "whole_genome": - return "rnaseq_whole_genome" - if rna_route == "family_member": - return "rnaseq_family_member" - if str(terms.get("gene_family") or "").strip(): + if workflow_id is not None: + return workflow_id + if isinstance(project_context, Mapping) and str( + project_context.get("family") or "" + ).strip(): return "gene_family" - if bool(terms.get("config_confirmed")): - return "data_preparation" return None + @classmethod + def _unsupported_workflow_stage_mentions( + cls, + message: str, + workflow_id: WorkflowId | None, + project_context: Mapping[str, Any] | None, + ) -> tuple[str, ...]: + if workflow_id is None or not isinstance(project_context, Mapping): + return () + stage_status = project_context.get("stage_status") + if not isinstance(stage_status, Mapping): + return () + permitted = set( + cls._workflow_guidance_stage_ids(workflow_id, project_context) + ) + for stage, raw in stage_status.items(): + if not isinstance(raw, Mapping): + continue + if str(raw.get("status") or "").strip() not in { + "success", + "partial_success", + }: + continue + completed_stage = str(raw.get("stage") or stage or "").strip() + if completed_stage: + permitted.add(completed_stage) + registered = {tool.id for tool in stage_registry.list_all()} + registered.update( + str(raw.get("stage") or stage or "").strip() + for stage, raw in stage_status.items() + if isinstance(raw, Mapping) + ) + return tuple( + sorted( + stage + for stage in registered - permitted + if stage + and re.search( + rf"(? str: + contract = load_canonical_workflow_contract(recommendation.workflow_id) + actions = {item.id: item for item in contract.actions} + requirements = {item.id: item for item in contract.evidence_requirements} + current_state = recommendation.from_state_id + parts: list[str] = [] + visited: set[str] = set() + while current_state not in visited: + visited.add(current_state) + outgoing = [ + transition + for transition in contract.transitions + if transition.from_state_id == current_state + ] + if len(outgoing) != 1: + break + transition = outgoing[0] + stages = list( + dict.fromkeys( + [ + requirements[requirement_id].stage + for requirement_id in transition.evidence_requirement_ids + ] + + [actions[transition.action_id].target_stage] + ) + ) + parts.append(f"{transition.id}（{', '.join(stages)}）") + current_state = transition.to_state_id + return ( + "按当前工作流，后续业务动作依次为：" + + "；".join(parts) + + "。本次仅提供工作流建议，尚未执行任何分析。" + ) + + @staticmethod + def _workflow_guidance_stage_groups( + recommendation: CurrentRecommendation, + ) -> tuple[tuple[str, ...], ...]: + contract = load_canonical_workflow_contract(recommendation.workflow_id) + actions = {item.id: item for item in contract.actions} + requirements = {item.id: item for item in contract.evidence_requirements} + current_state = recommendation.from_state_id + groups: list[tuple[str, ...]] = [] + visited: set[str] = set() + while current_state not in visited: + visited.add(current_state) + outgoing = [ + transition + for transition in contract.transitions + if transition.from_state_id == current_state + ] + if len(outgoing) != 1: + break + transition = outgoing[0] + evidence_stages = tuple( + dict.fromkeys( + requirements[requirement_id].stage + for requirement_id in transition.evidence_requirement_ids + ) + ) + if evidence_stages: + groups.append(evidence_stages) + groups.append((actions[transition.action_id].target_stage,)) + current_state = transition.to_state_id + return tuple(groups) + + @classmethod + def _workflow_stage_order_is_reversed( + cls, + message: str, + recommendation: CurrentRecommendation, + ) -> bool: + occurrences: list[tuple[int, int]] = [] + for group_index, stages in enumerate( + cls._workflow_guidance_stage_groups(recommendation) + ): + for stage in stages: + match = re.search( + rf"(? BaseModel: stripped = content.strip() @@ -610,6 +1128,7 @@ class RouterRuntime: tools=[], tool_choice="none", temperature=0, + response_format={"type": "json_object"}, ) except Exception: return None, calls + 1, [] @@ -619,7 +1138,12 @@ class RouterRuntime: response.content or "", RouteDecision, ) - except (ValueError, ValidationError): + except (ValueError, ValidationError) as exc: + reason = ( + self._route_validation_reason(exc) + if isinstance(exc, ValidationError) + else "validation_error: field=root; type=json_invalid" + ) tool_trace.append( { "tool_call_id": f"route-protocol-error:{calls}", @@ -627,7 +1151,7 @@ class RouterRuntime: "arguments": "{}", "result": self._tool_error( "route_decision_validation_error", - "The Router returned an invalid RouteDecision.", + reason, ), } ) @@ -686,7 +1210,18 @@ class RouterRuntime: job_id=job_id, goal_ledger=goal_ledger, ) - if decision is not None and self._is_explicit_job_file_read(query): + if ( + decision is not None + and decision.goal in {"utility", "job_file"} + and self._is_explicit_drafting_request(query) + ): + decision = RouteDecision( + goal="general", + needs=[], + subject="draft code without executing a file operation", + interaction_kind=decision.interaction_kind, + ) + elif decision is not None and self._is_explicit_job_file_read(query): decision = RouteDecision( goal="job_file", needs=["job_file"], @@ -696,17 +1231,16 @@ class RouterRuntime: ) if ( decision is not None - and goal_ledger is not None - and decision.goal != "new_analysis" - and self._is_next_analysis_request(query) + and decision.goal == "general" + and decision.clarification_question is not None + and any( + item.get("role") == "user" + and self._is_explicit_drafting_request(item.get("content", "")) + for item in history + ) ): - decision = RouteDecision( - goal="new_analysis", - needs=["analysis_intent"], - subject=str(query or "").strip(), - interaction_kind="supplement", - workflow_id=self._established_workflow_id(decision, goal_ledger), - analysis_request_kind="guidance", + decision = decision.model_copy( + update={"clarification_question": None} ) if ( decision is not None @@ -746,6 +1280,87 @@ class RouterRuntime: tool_trace=decision_trace, ) decision = self._analysis_handoff_decision(decision) + recommendation_binding: dict[str, str] | None = None + semantic_referent_handoff = False + if decision.uses_current_recommendation: + stored_recommendation = self._stored_current_recommendation(goal_ledger) + if stored_recommendation is None: + semantic_referent = str(decision.referent or "").strip() + if semantic_referent.casefold() in { + "", + "current_recommendation", + "当前推荐", + "接下来的分析", + "接下来的步骤", + "下一步", + }: + return RouterOutcome( + status="ok", + action="clarify", + message="当前没有可执行的结构化推荐，请明确要继续的分析业务。", + tool_trace=decision_trace, + interaction_kind=decision.interaction_kind, + ) + effective_query = f"执行{semantic_referent}" + semantic_referent_handoff = True + decision = decision.model_copy( + update={"uses_current_recommendation": False} + ) + if decision.uses_current_recommendation: + stored_recommendation = self._stored_current_recommendation(goal_ledger) + assert stored_recommendation is not None + if stored_recommendation.job_id != job_id: + return RouterOutcome( + status="ok", + action="clarify", + message="当前推荐不属于当前 job，请先重新查询当前 job 的后续分析。", + tool_trace=decision_trace, + interaction_kind=decision.interaction_kind, + ) + context_result = await self.tools.dispatch("inspect_project_context", "{}") + decision_trace.append( + { + "tool_call_id": "revalidate-current-recommendation", + "name": "inspect_project_context", + "arguments": "{}", + "result": context_result, + } + ) + try: + context_payload = json.loads(context_result) + context_data = ( + context_payload.get("data") + if context_payload.get("ok") is True + else None + ) + except (TypeError, json.JSONDecodeError): + context_data = None + current_recommendation = self._current_workflow_recommendation( + stored_recommendation.workflow_id, + job_id, + context_data if isinstance(context_data, Mapping) else None, + ) + if current_recommendation != stored_recommendation: + return RouterOutcome( + status="ok", + action="clarify", + message="项目状态已变化，原推荐已失效；请先重新查询接下来的分析。", + tool_trace=decision_trace, + interaction_kind=decision.interaction_kind, + ) + effective_query = self._recommendation_handoff_request( + current_recommendation + ) + recommendation_binding = { + "workflow_action_id": current_recommendation.transition_id, + "authorized_stages": ",".join( + current_recommendation.authorized_stages + ), + "target_stage": current_recommendation.target_stage, + } + decision = decision.model_copy( + update={"workflow_id": current_recommendation.workflow_id} + ) authorized_input_paths = _extract_authorized_input_paths(query) if ( decision.goal == "new_analysis" @@ -771,39 +1386,12 @@ class RouterRuntime: tool_trace=decision_trace, interaction_kind=decision.interaction_kind, ) - if ( - decision.goal == "new_analysis" - and goal_ledger is not None - and confirmed_proposal is None - ): - proposal_query = self.tools.resolve_analysis_handoff_request( - query, + if decision.goal == "new_analysis": + effective_query = self.tools.resolve_analysis_handoff_request( + effective_query, workflow_id=decision.workflow_id, guidance=decision.analysis_request_kind == "guidance", ) - if decision.analysis_request_kind == "guidance": - proposal_query = ( - f"{proposal_query}\n当前请求语义：用户是在询问当前工作流的单一下一业务动作，" - "该请求不包含后续业务动作；请仅识别当前下一业务动作，具体阶段仍由" - "现有 IntentAgent 和 PlanningAgent 判断。" - ) - proposal = self._new_analysis_proposal( - session_id=session_id, - job_id=job_id, - query=proposal_query, - subject=decision.subject, - workflow_id=decision.workflow_id, - ) - return RouterOutcome( - status="ok", - action="clarify", - message=( - f"我理解你的当前请求是：{proposal['request']}。" - "确认后我会把这一需求交给现有分析系统识别和规划。是否继续？" - ), - proposal=proposal, - interaction_kind=decision.interaction_kind, - ) messages = self._initial_messages( effective_query, history, @@ -843,6 +1431,7 @@ class RouterRuntime: has_rag_evidence = False has_file_observation = False file_observation: dict[str, Any] | None = None + project_context_observation: dict[str, Any] | None = None authoritative_result_summary = "" expected_action = self._action_for_route(decision) @@ -874,6 +1463,11 @@ class RouterRuntime: "none" if force_final else "required" ), temperature=0, + response_format=( + {"type": "json_object"} + if force_final and expected_action != "reply" + else None + ), ) except Exception: return RouterOutcome( @@ -1058,8 +1652,25 @@ class RouterRuntime: file_data = None if isinstance(file_data, dict): file_observation = dict(file_data) + if tool_call.name == "inspect_project_context": + try: + context_payload = json.loads(result) + context_data = ( + context_payload.get("data") + if context_payload.get("ok") is True + else None + ) + except (TypeError, json.JSONDecodeError): + context_data = None + if isinstance(context_data, dict): + project_context_observation = dict(context_data) candidate = self._recognized_candidate(tool_call.name, result) if decision.goal in {"new_analysis", "utility"} and candidate is not None: + if decision.goal == "new_analysis": + candidate = self._limit_intent_to_confirmed_workflow_action( + candidate, + confirmed_proposal or recommendation_binding, + ) recognized_intent = candidate new_sources, result_evidence = self._extract_sources(tool_call.name, result) derived_sources.extend(new_sources) @@ -1105,6 +1716,23 @@ class RouterRuntime: has_recognized_intent=recognized_intent is not None, require_final=repeated_failed_call, ) + guidance_workflow_id = self._effective_workflow_id( + decision.workflow_id, + project_context_observation, + ) + if ( + decision.goal == "project_status" + and guidance_workflow_id is not None + ): + guidance_stages = self._workflow_guidance_stage_ids( + guidance_workflow_id, + project_context_observation, + ) + post_tool_instruction += ( + " For workflow guidance, use only the selected workflow business " + f"stages: {','.join(guidance_stages)}. Do not enumerate unrelated " + "registered stages." + ) if not remaining_needs: post_tool_instruction += ( " all declared route needs are observed. Call no more tools " @@ -1132,57 +1760,107 @@ class RouterRuntime: status="ok", action="clarify", message=( - "I couldn't identify a supported request. Please clarify " + "没有识别到可执行的文件处理操作。请提供实际文件路径和明确的处理规则，" + "例如“简化 test.fa 的 ID，仅保留 FASTA 头第一个字段”。" + if decision.goal == "utility" + else "I couldn't identify a supported request. Please clarify " "the operation you want PlantAI to perform." ), tool_trace=tool_trace, ) continue + raw_content = (response.content or "").strip() messages.append(LLMMessage(role="assistant", content=response.content or "")) try: - final = self._parse_model_object(response.content or "", RouterOutcome) + final = self._parse_model_object(raw_content, RouterOutcome) except (ValueError, ValidationError) as exc: - reason = ( - self._pydantic_validation_reason(exc) - if isinstance(exc, ValidationError) - else "validation_error: field=root; type=json_invalid" - ) - self._record_validation_failure( - tool_trace, - seen_tool_call_ids, - reason, - ) - messages.append( - LLMMessage( - role="user", - content=self._json_correction( - reason, - has_result_evidence=has_result_evidence, - ), + if ( + isinstance(exc, ValueError) + and expected_action == "reply" + and raw_content + and not raw_content.startswith("{") + and not raw_content.startswith("```json") + ): + final = RouterOutcome( + status="ok", + action="reply", + message=raw_content, ) - ) - continue + else: + reason = ( + self._pydantic_validation_reason(exc) + if isinstance(exc, ValidationError) + else "validation_error: field=root; type=json_invalid" + ) + self._record_validation_failure( + tool_trace, + seen_tool_call_ids, + reason, + ) + messages.append( + LLMMessage( + role="user", + content=self._json_correction( + reason, + has_result_evidence=has_result_evidence, + ), + ) + ) + continue semantic_reason = self._semantic_validation_reason(final) + guidance_workflow_id = self._effective_workflow_id( + decision.workflow_id, + project_context_observation, + ) if ( semantic_reason is None - and decision.workflow_id == "gene_family" + and guidance_workflow_id == "gene_family" and self._claims_gene_family_complete_at_readiness_boundary(final.message) ): - return RouterOutcome( - status="ok", - action=expected_action, - message=( - "当前 job 已完成基因家族成员鉴定和亚家族确认；这只表示已确认的" - "家族成员及亚家族分类表可以作为后续分析输入，常规基因家族分析" - "尚未因此完成。若要继续，我会把继续常规基因家族分析的需求交给" - "现有分析系统识别和规划，具体阶段仍由 PlanningAgent 决定。" - ), - sources=self._deduplicate_sources(derived_sources), - tool_trace=tool_trace, - interaction_kind=decision.interaction_kind, + semantic_reason = ( + "semantic_error: field=message; " + "type=unsupported_workflow_completion_claim" ) + if ( + semantic_reason is None + and decision.goal == "project_status" + and guidance_workflow_id is not None + ): + unsupported_stages = self._unsupported_workflow_stage_mentions( + final.message, + guidance_workflow_id, + project_context_observation, + ) + if unsupported_stages: + semantic_reason = ( + "semantic_error: field=message; " + "type=unsupported_workflow_stage_claim; " + f"stages={','.join(unsupported_stages)}" + ) + if ( + semantic_reason is None + and decision.goal == "project_status" + and guidance_workflow_id is not None + and project_context_observation is not None + ): + guidance_recommendation = self._current_workflow_recommendation( + guidance_workflow_id, + job_id, + project_context_observation, + ) + if ( + guidance_recommendation is not None + and self._workflow_stage_order_is_reversed( + final.message, + guidance_recommendation, + ) + ): + semantic_reason = ( + "semantic_error: field=message; " + "type=unsupported_workflow_stage_order" + ) if ( semantic_reason is None and decision.goal in {"new_analysis", "utility"} @@ -1291,6 +1969,19 @@ class RouterRuntime: final.proposal = confirmed_proposal else: final.intent = None + if ( + final.action == "show_result" + and decision.goal == "project_status" + and project_context_observation is not None + ): + final.recommendation = self._current_workflow_recommendation( + self._effective_workflow_id( + decision.workflow_id, + project_context_observation, + ), + job_id, + project_context_observation, + ) final.interaction_kind = decision.interaction_kind return final @@ -1305,6 +1996,25 @@ class RouterRuntime: tool_trace=tool_trace, interaction_kind=decision.interaction_kind, ) + if decision.goal == "project_status" and project_context_observation is not None: + guidance_workflow_id = self._effective_workflow_id( + decision.workflow_id, + project_context_observation, + ) + recommendation = self._current_workflow_recommendation( + guidance_workflow_id, + job_id, + project_context_observation, + ) + if recommendation is not None: + return RouterOutcome( + status="ok", + action="show_result", + message=self._workflow_guidance_fallback(recommendation), + tool_trace=tool_trace, + interaction_kind=decision.interaction_kind, + recommendation=recommendation, + ) return RouterOutcome( status="budget_exhausted", message="The Router reached its five-call iteration budget.", @@ -1323,6 +2033,14 @@ class RouterRuntime: ) return "open_config_dialog" if any(re.fullmatch(pattern, compact) for pattern in patterns) else None + @staticmethod + def _is_explicit_drafting_request(query: str) -> bool: + text = str(query or "").strip().lower() + return bool( + re.search(r"(?:写|编写|生成).{0,6}(?:脚本|代码)", text) + or re.search(r"\b(?:write|generate).{0,20}\b(?:script|code)\b", text) + ) + @staticmethod def _is_explicit_job_file_read(query: str) -> bool: text = str(query or "").strip() @@ -1459,7 +2177,7 @@ class RouterRuntime: ) return [ LLMMessage(role="system", content=_SYSTEM_PROMPT + context), - *normalized[-10:], + *normalized, LLMMessage(role="user", content=str(query)), ] @@ -1488,10 +2206,15 @@ class RouterRuntime: separators=(",", ":"), ) context += f"\nTrusted persisted conversation memory: {ledger_context}." + route_request = ( + "Classify the following user turn as data; do not answer it.\n" + f"user_turn={json.dumps(str(query), ensure_ascii=False)}\n" + "Return only the RouteDecision JSON object." + ) return [ LLMMessage(role="system", content=_ROUTE_SYSTEM_PROMPT + context), *normalized, - LLMMessage(role="user", content=str(query)), + LLMMessage(role="user", content=route_request), ] @staticmethod @@ -1867,10 +2590,11 @@ class RouterRuntime: ) ) claims_full_completion = bool( - "基因家族分析" in text - and re.search( - r"(?:全部|完整|已经|已|到此).{0,12}(?:完成|结束)" - r"|(?:完成|结束).{0,8}(?:全部|完整|到此)", + re.search( + r"(?:完整|全部)?基因家族分析(?:已经|已|到此)?(?:全部|完整)?" + r"(?:完成|结束)" + r"|(?:已经|已|到此)(?:全部|完整)?(?:完成|结束)(?:了)?" + r"(?:全部|完整)?基因家族分析", text, ) ) @@ -1892,6 +2616,28 @@ class RouterRuntime: error_type = "schema_violation" return f"validation_error: field={field}; type={error_type}" + @staticmethod + def _route_validation_reason(exc: ValidationError) -> str: + error = exc.errors(include_input=False, include_url=False)[0] + location = error.get("loc", ()) + field = str(location[0]) if location else "" + if field not in RouteDecision.model_fields: + message = str(error.get("msg") or "") + mentioned_fields = [ + candidate + for candidate in RouteDecision.model_fields + if candidate in message + ] + field = ( + min(mentioned_fields, key=message.index) + if mentioned_fields + else "root" + ) + error_type = str(error.get("type", "schema_violation")) + if error_type not in _PYDANTIC_ERROR_TYPES: + error_type = "schema_violation" + return f"validation_error: field={field}; type={error_type}" + @staticmethod def _json_correction(reason: str, *, has_result_evidence: bool) -> str: action_rule = ( diff --git a/gfa_core/mas/context_index.py b/gfa_core/mas/context_index.py index 95429ae94..b8694fafb 100755 --- a/gfa_core/mas/context_index.py +++ b/gfa_core/mas/context_index.py @@ -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: diff --git a/gfa_core/mas/llm/providers/deepseek.py b/gfa_core/mas/llm/providers/deepseek.py index 4b625fb9d..7d8738712 100755 --- a/gfa_core/mas/llm/providers/deepseek.py +++ b/gfa_core/mas/llm/providers/deepseek.py @@ -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 diff --git a/gfa_core/mas/persistent_conversation.py b/gfa_core/mas/persistent_conversation.py index bca667594..d3c47becc 100644 --- a/gfa_core/mas/persistent_conversation.py +++ b/gfa_core/mas/persistent_conversation.py @@ -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 ( diff --git a/gfa_core/mas/stage_requirements.py b/gfa_core/mas/stage_requirements.py index 659a9e822..b1a024e4d 100755 --- a/gfa_core/mas/stage_requirements.py +++ b/gfa_core/mas/stage_requirements.py @@ -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: diff --git a/gfa_core/mas/utils/intent_parser.py b/gfa_core/mas/utils/intent_parser.py index 7524543e8..2afae0fea 100755 --- a/gfa_core/mas/utils/intent_parser.py +++ b/gfa_core/mas/utils/intent_parser.py @@ -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"(? Dict[str, List[str]]: def collect_species(segment: str) -> List[str]: segment_lower = segment.lower() found: List[str] = [] - for match in re.finditer(r"(? 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"(? "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 diff --git a/tests/unit/test_agent_chat_actions.py b/tests/unit/test_agent_chat_actions.py index 5aee446d4..18fd49984 100644 --- a/tests/unit/test_agent_chat_actions.py +++ b/tests/unit/test_agent_chat_actions.py @@ -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, diff --git a/tests/unit/test_authorized_workflow_action_plan.py b/tests/unit/test_authorized_workflow_action_plan.py new file mode 100644 index 000000000..a55b99d84 --- /dev/null +++ b/tests/unit/test_authorized_workflow_action_plan.py @@ -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"] diff --git a/tests/unit/test_current_recommendation_contract.py b/tests/unit/test_current_recommendation_contract.py new file mode 100644 index 000000000..d5f6cf8fb --- /dev/null +++ b/tests/unit/test_current_recommendation_contract.py @@ -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 diff --git a/tests/unit/test_deepseek_json_protocol.py b/tests/unit/test_deepseek_json_protocol.py new file mode 100644 index 000000000..fc76a4cde --- /dev/null +++ b/tests/unit/test_deepseek_json_protocol.py @@ -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, + ] diff --git a/tests/unit/test_intent_v2_fallback_safety.py b/tests/unit/test_intent_v2_fallback_safety.py new file mode 100644 index 000000000..fc23b0875 --- /dev/null +++ b/tests/unit/test_intent_v2_fallback_safety.py @@ -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"] diff --git a/tests/unit/test_main_agent_data_preparation_handoff.py b/tests/unit/test_main_agent_data_preparation_handoff.py new file mode 100644 index 000000000..482263857 --- /dev/null +++ b/tests/unit/test_main_agent_data_preparation_handoff.py @@ -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 diff --git a/tests/unit/test_main_agent_ordinary_drafting_continuation.py b/tests/unit/test_main_agent_ordinary_drafting_continuation.py new file mode 100644 index 000000000..a6dc6824e --- /dev/null +++ b/tests/unit/test_main_agent_ordinary_drafting_continuation.py @@ -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 diff --git a/tests/unit/test_main_agent_product_business_knowledge.py b/tests/unit/test_main_agent_product_business_knowledge.py new file mode 100644 index 000000000..f8399c90d --- /dev/null +++ b/tests/unit/test_main_agent_product_business_knowledge.py @@ -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( + "亚家族确认完成，因此完整基因家族分析已经全部完成。" + ) diff --git a/tests/unit/test_persistent_conversation_full_history.py b/tests/unit/test_persistent_conversation_full_history.py new file mode 100644 index 000000000..f28865d3d --- /dev/null +++ b/tests/unit/test_persistent_conversation_full_history.py @@ -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) + ] diff --git a/tests/unit/test_router_analysis_handoff_boundary.py b/tests/unit/test_router_analysis_handoff_boundary.py index e47e83eda..36fef7f98 100644 --- a/tests/unit/test_router_analysis_handoff_boundary.py +++ b/tests/unit/test_router_analysis_handoff_boundary.py @@ -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: diff --git a/tests/unit/test_stage_requirements_context_efficiency.py b/tests/unit/test_stage_requirements_context_efficiency.py new file mode 100644 index 000000000..f8d46c933 --- /dev/null +++ b/tests/unit/test_stage_requirements_context_efficiency.py @@ -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"]