From 1de307f0e23f45a089d730a1ba3e064309f6d594 Mon Sep 17 00:00:00 2001 From: FiveFingersMountain Date: Wed, 11 Nov 2026 09:28:05 +0000 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20add=20opt-in=20multi-model=20task?= =?UTF-8?q?=20routing=20for=20agent=20runs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Route agent runs to the best-matching model of the agent's own configured pool (quick QA / long context / deep reasoning tiers). - Opt-in via MODEL_ROUTING_ENABLED (default false; pre-routing behaviour fully preserved when off) - Domain vocabularies live in configs/model_routing_rules.json (medical terms shipped as example; engine is domain-agnostic) - Explicit request model selection always wins over routing - No-signal and no-capable-candidate cases keep the existing selection; routing never fails a run - Every decision emits a structured audit log with matched rules, candidate scores and fallback tier - Capacity guard prefers the provider input limit and rejects candidates that cannot hold the estimated tokens - 43 unit tests plus create_agent_config integration coverage --- backend/configs/model_routing_rules.json | 30 + backend/contracts/constants/environment.py | 6 + .../services/agent/preparation/compiler.py | 75 ++- .../services/agent/preparation/run.py | 4 + backend/management/services/model/routing.py | 606 ++++++++++++++++++ deploy/env/.env.example | 9 + docs/model_routing_integration.md | 312 +++++++++ test/backend/agents/test_create_agent_info.py | 8 + .../test_service_directory_architecture.py | 1 + .../services/test_model_task_routing.py | 359 +++++++++++ 10 files changed, 1407 insertions(+), 3 deletions(-) create mode 100644 backend/configs/model_routing_rules.json create mode 100644 backend/management/services/model/routing.py create mode 100644 docs/model_routing_integration.md create mode 100644 test/backend/services/test_model_task_routing.py diff --git a/backend/configs/model_routing_rules.json b/backend/configs/model_routing_rules.json new file mode 100644 index 0000000000..68f8a1453a --- /dev/null +++ b/backend/configs/model_routing_rules.json @@ -0,0 +1,30 @@ +{ + "long_context": { + "token_threshold": 24000, + "char_threshold": 8000, + "terms": ["病历", "出院小结", "检查报告", "文献", "全文", "长文", "总结归纳", "medical record", "discharge summary", "full text"] + }, + "quick_qa": { + "char_limit": 120, + "terms": ["是什么", "什么是", "正常值", "参考范围", "翻译", "缩写", "定义", "查一下", "简述", "what is", "define", "abbreviation", "normal range", "reference range", "translate"] + }, + "deep_reasoning": { + "terms": ["鉴别诊断", "会诊", "mdt", "多学科", "分析", "评估", "推理", "为什么", "机制", "步骤", "方案比较", "鉴别", "病因", "differential", "diagnosis", "workup", "analyze", "why ", "compare", "root cause", "root-cause"] + }, + "risk": { + "high_terms": ["手术", "麻醉", "化疗", "放疗", "处方", "剂量", "用药", "透析", "输血", "surgery", "anesthesia", "chemotherapy", "dosage", "prescription", "transfusion", "intubat", "surgical"], + "review_terms": ["确诊", "诊断", "出院", "医嘱", "diagnos", "discharge", "clinical decision"] + }, + "task_tags": { + "quick_qa": "quick_qa", + "long_context": "long_context", + "deep_reasoning": "deep_reasoning", + "summary": "long_context", + "report": "long_context", + "mdt": "deep_reasoning", + "analyze": "deep_reasoning", + "quality_check": "quick_qa" + }, + "history_tokens_per_turn": 200, + "attachment_tokens": 800 +} diff --git a/backend/contracts/constants/environment.py b/backend/contracts/constants/environment.py index cfe4d9a436..c24259bb8c 100644 --- a/backend/contracts/constants/environment.py +++ b/backend/contracts/constants/environment.py @@ -445,6 +445,12 @@ MODELS_DEV_CATALOG_JSON_PATH = os.getenv('MODELS_DEV_CATALOG_JSON_PATH', str(_CONFIG_DIRECTORY / 'models_dev_catalog.json')) # models.dev capability catalog downloaded during the backend image build. +# Multi-model task routing. Opt-in: when False, model selection keeps the +# pre-routing behavior exactly (first available configured model). +MODEL_ROUTING_ENABLED = os.getenv('MODEL_ROUTING_ENABLED', 'false').lower() in ('true', '1', 'yes', 'on') +# Routing vocabulary/config file path (tier keywords, task tags, risk terms). +MODEL_ROUTING_RULES_JSON_PATH = os.getenv('MODEL_ROUTING_RULES_JSON_PATH', str(_CONFIG_DIRECTORY / 'model_routing_rules.json')) + # External Memory Provider Configuration MEMORY_PROVIDER_PLUGINS_DIR = os.getenv('MEMORY_PROVIDER_PLUGINS_DIR', '') diff --git a/backend/management/services/agent/preparation/compiler.py b/backend/management/services/agent/preparation/compiler.py index 57482b4c8e..c2ae756100 100644 --- a/backend/management/services/agent/preparation/compiler.py +++ b/backend/management/services/agent/preparation/compiler.py @@ -7,6 +7,7 @@ from typing import Any, Dict, List, Optional from contracts.constants.environment import ( + MODEL_ROUTING_ENABLED, NEXENT_SANDBOX_DEFAULT_LEVEL, RUNTIME_PARALLEL_EXECUTOR_TIMEOUT_SECONDS, ) @@ -92,6 +93,8 @@ async def create_agent_config( runtime_knowledge_tools: Optional[List[Dict[str, Any]]] = None, runtime_sub_agent_mounts: Optional[List[Dict[str, Any]]] = None, runtime_model_records: Optional[Dict[int, Dict[str, Any]]] = None, + run_history_length: int = 0, + run_has_attachments: bool = False, ): normalized_tool_params = _normalize_tool_params_request(tool_params) agent_info = search_agent_info_by_agent_id( @@ -146,6 +149,8 @@ async def create_agent_config( # Workbench overlays replace Skills on the effective root only. runtime_skill_snapshot=None, runtime_model_records=runtime_model_records, + run_history_length=run_history_length, + run_has_attachments=run_has_attachments, ) # Persisted Agent relations are rendered into the manager prompt by # their configured business name. Keep the callable registered in the @@ -183,6 +188,8 @@ async def create_agent_config( runtime_model_records=runtime_model_records, runtime_knowledge_tools=None, runtime_sub_agent_mounts=None, + run_history_length=run_history_length, + run_has_attachments=run_has_attachments, ) invocation_name = str( mount.get("invocation_name") @@ -595,15 +602,38 @@ async def create_agent_config( prompt_bundle = load_agent_prompt_bundle(is_manager=is_manager, language=language) prompt_composer = AgentPromptComposer(prompt_bundle) - # AgentInfo stores model_ids (a list); pick the first available model. + # AgentInfo stores model_ids (a list); pick the request override, or + # (when MODEL_ROUTING_ENABLED) route by task tier across the agent's + # configured pool. Explicit request-level selection always wins over + # routing; with routing disabled the pre-routing behavior is kept. agent_model_ids = agent_info.get("model_ids") or [] model_id_to_use = _select_agent_model_id( agent_model_ids, override_model_id, tenant_id ) + routed_model_info = None + if ( + MODEL_ROUTING_ENABLED + and override_model_id is None + and model_id_to_use is not None + ): + routed_model_info = _route_agent_model( + agent_model_ids=agent_model_ids, + tenant_id=tenant_id, + last_user_query=last_user_query, + history_length=run_history_length, + has_attachments=run_has_attachments, + runtime_model_records=runtime_model_records, + agent_id=agent_id, + ) + if routed_model_info is not None: + model_id_to_use = routed_model_info.get("model_id") or model_id_to_use model_info = None if model_id_to_use is not None: - model_info = (runtime_model_records.get(model_id_to_use) if runtime_model_records is not None - else get_model_by_model_id(model_id_to_use, tenant_id=tenant_id)) + if routed_model_info is not None and routed_model_info.get("model_id") == model_id_to_use: + model_info = routed_model_info + else: + model_info = (runtime_model_records.get(model_id_to_use) if runtime_model_records is not None + else get_model_by_model_id(model_id_to_use, tenant_id=tenant_id)) if model_info is None: raise RunModelControlsError("model_not_available") model_name = model_info["display_name"] if model_info is not None else "main_model" @@ -767,6 +797,45 @@ def _select_agent_model_id( lambda model_id: is_model_available(get_model_by_model_id(model_id, tenant_id=tenant_id))) +def _route_agent_model( + agent_model_ids: List[int], + tenant_id: str, + last_user_query: Optional[str], + history_length: int, + has_attachments: bool, + runtime_model_records: Optional[Dict[int, Dict[str, Any]]], + agent_id: int, +) -> Optional[Dict[str, Any]]: + """Route the run to the best-matching model in the agent's own pool. + + Routing stays inside the agent's configured ``model_ids`` so existing + permission and validation boundaries are preserved. Only available + records participate. Returns the chosen record (capacity snapshot + fields intact) or None when routing has no signal; the structured + decision is logged by the routing service. Domain vocabularies and + task tags come from the routing rules file, not from agent rows. + """ + + from management.services.model.routing import route_model_selection + + candidates = [] + for model_id in agent_model_ids: + record = (runtime_model_records.get(model_id) if runtime_model_records is not None + else get_model_by_model_id(model_id, tenant_id=tenant_id)) + if record is not None and is_model_available(record): + candidates.append(record) + if not candidates: + return None + selection = route_model_selection( + records=candidates, + query=last_user_query or "", + history_length=history_length, + has_attachments=has_attachments, + agent_id=agent_id, + ) + return selection[0] if selection else None + + def _resolve_prompt_runtime_flags( agent_info: Dict[str, Any], ) -> tuple[AgentVerificationConfig, bool]: diff --git a/backend/management/services/agent/preparation/run.py b/backend/management/services/agent/preparation/run.py index 3cc9f875e3..07f5c6f8fc 100644 --- a/backend/management/services/agent/preparation/run.py +++ b/backend/management/services/agent/preparation/run.py @@ -207,6 +207,10 @@ async def create_agent_run_info( create_config_kwargs["runtime_knowledge_tools"] = runtime_knowledge_tools if runtime_sub_agent_mounts is not None: create_config_kwargs["runtime_sub_agent_mounts"] = runtime_sub_agent_mounts + # Inputs for multi-model task routing (model routing is opt-in; these + # are ignored otherwise). History turns count completed messages. + create_config_kwargs["run_history_length"] = len(history or []) + create_config_kwargs["run_has_attachments"] = bool(minio_files) from nexent.core.model_service.profiles.outlet_storage import encode_resolved_model_profile from nexent.core.model_service.profiles.resolved_profile import ResolvedModelProfile captured_records = {} diff --git a/backend/management/services/model/routing.py b/backend/management/services/model/routing.py new file mode 100644 index 0000000000..5ea72fff8f --- /dev/null +++ b/backend/management/services/model/routing.py @@ -0,0 +1,606 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Task routing for multi-model agent runs. + +Classifies each run (intent, risk, size) and selects the best matching +model from the agent's configured model pool. Designed for deployments +where different tiers of model capacity (quick QA / long context / deep +reasoning) serve different request classes, while keeping every routing +decision auditable. + +Design contract: +1. Opt-in: routing runs only when ``MODEL_ROUTING_ENABLED`` is true. +2. Explicit request-level model selection always wins; routing never + overrides a caller-pinned model. +3. Routing only selects from the agent's own configured model pool that + the tenant already validated. +4. Domain knowledge (medical terms, task tags) lives in a configurable + rules JSON; this module is domain-agnostic. +5. When routing has no usable signal or no capable candidate, it returns + None and the caller keeps the pre-routing selection behavior. + +Layers: +1. classify - rule engine over the user query and run parameters. +2. rank - capability-based scoring of candidate model records. +3. fallback - tier relaxation when a tier has no available model. +""" + +import json +import logging +import os +import threading +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Dict, List, Optional, Tuple + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Enums +# --------------------------------------------------------------------------- + + +class RouteTier(str, Enum): + """Model service tiers a run can be routed to.""" + + QUICK_QA = "quick_qa" + LONG_CONTEXT = "long_context" + DEEP_REASONING = "deep_reasoning" + + +class IntentSource(str, Enum): + """Where the routing intent came from - the audit trail anchor.""" + + EXPLICIT_TAG = "explicit_tag" # rules file task-tag vocabulary + RULE = "rule" # rule engine matched the query + DEFAULT = "default" # no signal matched + + +class RiskLevel(str, Enum): + """Risk classes; high-risk runs are flagged for downstream review.""" + + ROUTINE = "routine" + HIGH = "high" + + +# Relaxation order when a tier has no capable model: quick QA falls up to +# long context first (capacity is a superset), everything else falls up to +# deep reasoning (strongest model absorbs). +_TIER_RELAXATION = { + RouteTier.QUICK_QA: RouteTier.LONG_CONTEXT, + RouteTier.LONG_CONTEXT: RouteTier.DEEP_REASONING, + RouteTier.DEEP_REASONING: None, +} + +_DEFAULT_RULES: Dict[str, Any] = { + "long_context": {"token_threshold": 24000, "char_threshold": 8000, "terms": []}, + "quick_qa": {"char_limit": 120, "terms": []}, + "deep_reasoning": {"terms": []}, + "risk": {"high_terms": [], "review_terms": []}, + "task_tags": {}, + "history_tokens_per_turn": 200, + "attachment_tokens": 800, +} + +# --------------------------------------------------------------------------- +# Rules loading (file-based, thread-safe, degrade gracefully) +# --------------------------------------------------------------------------- + +_rules_lock = threading.Lock() +_rules_cache: Optional[Dict[str, Any]] = None +_rules_loaded_from: Optional[str] = None + + +def _deep_merge(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]: + """Recursively merge override into base (new dicts, no input mutation).""" + + merged = dict(base) + for key, value in override.items(): + if isinstance(value, dict) and isinstance(merged.get(key), dict): + merged[key] = _deep_merge(merged[key], value) + else: + merged[key] = value + return merged + + +def load_rules(force_reload: bool = False) -> Dict[str, Any]: + """Load routing rules from ``MODEL_ROUTING_RULES_JSON_PATH``. + + Missing or malformed files degrade to built-in defaults (empty + vocabularies); loading never raises so the run path stays safe. + """ + + global _rules_cache, _rules_loaded_from + with _rules_lock: + if _rules_cache is not None and not force_reload: + return _rules_cache + from contracts.constants.environment import MODEL_ROUTING_RULES_JSON_PATH + + rules = _DEFAULT_RULES + path = MODEL_ROUTING_RULES_JSON_PATH + try: + if path and os.path.exists(path): + with open(path, encoding="utf-8") as handle: + raw = json.load(handle) + if isinstance(raw, dict): + rules = _deep_merge(_DEFAULT_RULES, raw) + _rules_loaded_from = path + except (OSError, ValueError) as exc: + logger.warning("Model routing rules unavailable (%s); using defaults", exc) + rules = _DEFAULT_RULES + _rules_cache = rules + return rules + + +def reset_rules_cache() -> None: + """Drop the cached rules (used by tests and config reloads).""" + + global _rules_cache, _rules_loaded_from + with _rules_lock: + _rules_cache = None + _rules_loaded_from = None + + +# --------------------------------------------------------------------------- +# Data carriers +# --------------------------------------------------------------------------- + + +@dataclass +class ClassificationResult: + """Outcome of the rule engine; every field lands in the audit log.""" + + tier: RouteTier + source: IntentSource + risk: RiskLevel + review_required: bool + est_tokens: int + matched_rules: List[str] = field(default_factory=list) + features: Dict[str, Any] = field(default_factory=dict) + + +@dataclass +class RoutingDecision: + """Full routing decision for one run - the audit record itself.""" + + agent_id: Optional[int] + tier: RouteTier + source: IntentSource + risk: RiskLevel + review_required: bool + est_tokens: int + matched_rules: List[str] + selected_model_id: Optional[int] + selected_display_name: str + candidates: List[Dict[str, Any]] + reason: str + fallback_from: Optional[RouteTier] = None + + def to_audit_dict(self) -> Dict[str, Any]: + """Serialize for structured logging (no secrets).""" + + return { + "agent_id": self.agent_id, + "tier": self.tier.value, + "source": self.source.value, + "risk": self.risk.value, + "review_required": self.review_required, + "est_tokens": self.est_tokens, + "matched_rules": list(self.matched_rules), + "selected_model_id": self.selected_model_id, + "selected_display_name": self.selected_display_name, + "candidates": [dict(candidate) for candidate in self.candidates], + "reason": self.reason, + "fallback_from": self.fallback_from.value if self.fallback_from else None, + } + + +# --------------------------------------------------------------------------- +# Classification (rule engine) +# --------------------------------------------------------------------------- + + +def estimate_tokens( + text: str, + history_length: int = 0, + has_attachments: bool = False, + rules: Optional[Dict[str, Any]] = None, +) -> int: + """Cheap conservative token estimate. + + CJK ~1 token/char, other ~4 chars/token; history and attachments add + configured per-turn / per-attachment overhead. + """ + + rules = rules or load_rules() + if not text: + text = "" + cjk = sum(1 for ch in text if "一" <= ch <= "鿿") + other = len(text) - cjk + base = cjk + (other + 3) // 4 + per_turn = rules.get("history_tokens_per_turn", 200) + per_attachment = rules.get("attachment_tokens", 800) + return base + history_length * per_turn + (per_attachment if has_attachments else 0) + + +def _contains_term(text: str, terms) -> List[str]: + """Return terms present in the text (case-insensitive substring).""" + + lowered = text.lower() + return [term for term in terms if term in lowered] + + +def _tier_terms(rules: Dict[str, Any], tier: RouteTier) -> List[str]: + section = rules.get(tier.value) or {} + terms = section.get("terms") or [] + return [str(term) for term in terms if term] + + +def classify_run( + query: str, + history_length: int = 0, + agent_meta: Optional[Dict[str, Any]] = None, + has_attachments: bool = False, + rules: Optional[Dict[str, Any]] = None, +) -> ClassificationResult: + """Classify a run into a service tier using auditable rules. + + Priority: explicit task tag > risk-escalated analysis > long context > + deep reasoning > quick QA > default. Long context wins over deep + reasoning when both match, because truncation is a harder failure than + a weaker reasoning model. With empty vocabularies every query lands + on the default branch, which callers treat as "keep existing selection". + """ + + rules = rules or load_rules() + text = query or "" + features: Dict[str, Any] = {} + matched: List[str] = [] + est_tokens = estimate_tokens( + text, history_length=history_length, has_attachments=has_attachments, rules=rules + ) + features["est_tokens"] = est_tokens + features["query_chars"] = len(text) + features["history_length"] = history_length + features["has_attachments"] = has_attachments + + meta = agent_meta or {} + task_tag = str(meta.get("task_tag") or "").strip().lower() + if task_tag: + features["task_tag"] = task_tag + + # 1. Explicit task tag wins outright (vocabulary from the rules file). + tag_map = rules.get("task_tags") or {} + if task_tag and task_tag in tag_map: + try: + tier = RouteTier(tag_map[task_tag]) + except ValueError: + tier = None + if tier is not None: + matched.append(f"task_tag:{task_tag}") + risk, review = _assess_risk(text, rules) + return ClassificationResult( + tier=tier, + source=IntentSource.EXPLICIT_TAG, + risk=risk, + review_required=review, + est_tokens=est_tokens, + matched_rules=matched, + features=features, + ) + + # 2. Rule engine over query vocabulary and size. + long_section = rules.get("long_context") or {} + quick_section = rules.get("quick_qa") or {} + long_hits = _contains_term(text, _tier_terms(rules, RouteTier.LONG_CONTEXT)) + deep_hits = _contains_term(text, _tier_terms(rules, RouteTier.DEEP_REASONING)) + quick_hits = _contains_term(text, _tier_terms(rules, RouteTier.QUICK_QA)) + risk, review = _assess_risk(text, rules) + + token_threshold = int(long_section.get("token_threshold") or 0) + char_threshold = int(long_section.get("char_threshold") or 0) + quick_char_limit = int(quick_section.get("char_limit") or 0) + is_long = ( + (token_threshold and est_tokens >= token_threshold) + or (char_threshold and len(text) >= char_threshold) + ) + if is_long: + matched.append("long_context:size") + matched.extend(f"long_context:{term}" for term in long_hits) + matched.extend(f"deep:{term}" for term in deep_hits) + matched.extend(f"quick:{term}" for term in quick_hits) + + if risk is RiskLevel.HIGH and deep_hits: + # High risk + analysis -> strongest tier, always review. + matched.append("risk:escalate_deep") + return ClassificationResult( + tier=RouteTier.DEEP_REASONING, + source=IntentSource.RULE, + risk=risk, + review_required=True, + est_tokens=est_tokens, + matched_rules=matched, + features=features, + ) + + if is_long or long_hits: + return ClassificationResult( + tier=RouteTier.LONG_CONTEXT, + source=IntentSource.RULE, + risk=risk, + review_required=review, + est_tokens=est_tokens, + matched_rules=matched, + features=features, + ) + + if deep_hits: + return ClassificationResult( + tier=RouteTier.DEEP_REASONING, + source=IntentSource.RULE, + risk=risk, + review_required=review, + est_tokens=est_tokens, + matched_rules=matched, + features=features, + ) + + if text.strip() and (quick_hits or (quick_char_limit and len(text) <= quick_char_limit)): + matched.append("quick:short_or_lookup") + return ClassificationResult( + tier=RouteTier.QUICK_QA, + source=IntentSource.RULE, + risk=risk, + review_required=review, + est_tokens=est_tokens, + matched_rules=matched, + features=features, + ) + + # 3. No signal matched - explicitly signal "no routing decision" so the + # caller keeps the pre-routing selection (contract: routing never + # silently changes the model when it has no evidence). + return ClassificationResult( + tier=RouteTier.QUICK_QA, + source=IntentSource.DEFAULT, + risk=risk, + review_required=review, + est_tokens=est_tokens, + matched_rules=matched, + features=features, + ) + + +def _assess_risk(text: str, rules: Dict[str, Any]) -> Tuple[RiskLevel, bool]: + """Risk assessment from configured vocabularies.""" + + risk_section = rules.get("risk") or {} + hits = _contains_term(text, risk_section.get("high_terms") or []) + if not hits: + return RiskLevel.ROUTINE, False + review = bool(_contains_term(text, risk_section.get("review_terms") or [])) + return RiskLevel.HIGH, review + + +# --------------------------------------------------------------------------- +# Candidate ranking +# --------------------------------------------------------------------------- + + +def _record_bool(record: Dict[str, Any], key: str) -> bool: + value = record.get(key) + return bool(value) if value is not None else False + + +def _record_int(record: Dict[str, Any], key: str) -> Optional[int]: + value = record.get(key) + if value is None: + return None + try: + return int(value) + except (TypeError, ValueError): + return None + + +def record_capacity(record: Dict[str, Any]) -> int: + """Input capacity of a record (0 = unknown/unlimited). + + Uses the provider's hard input limit first and the combined window as + a fallback; output reserve is handled by the existing context budget + module, so routing only guards the input side. + """ + + input_limit = _record_int(record, "max_input_tokens") or 0 + if input_limit: + return input_limit + return _record_int(record, "context_window_tokens") or 0 + + +def record_tier(record: Dict[str, Any]) -> RouteTier: + """Derive a record's tier from saved capability columns. + + Uses operator-verified facts already stored on ``model_record_t`` + (reasoning flags, context capacity, name) so routing needs no extra + configuration. Priority: reasoning capability -> context capacity -> + name heuristics -> quick QA. + """ + + if _record_bool(record, "reasoning_supported") or _record_bool(record, "enable_thinking"): + return RouteTier.DEEP_REASONING + if record_capacity(record) >= 48000: + return RouteTier.LONG_CONTEXT + name = str(record.get("model_name") or record.get("display_name") or "").lower() + if ( + "reason" in name + or "think" in name + or "-r1" in name + or "o1" in name + or "o3" in name + or "r1" in name.split("-") + ): + return RouteTier.DEEP_REASONING + if "long" in name or "long-context" in name: + return RouteTier.LONG_CONTEXT + return RouteTier.QUICK_QA + + +def score_candidate(record: Dict[str, Any], result: ClassificationResult) -> float: + """Score one candidate record for the classified tier; higher is better. + + Records of a different tier score -1 (ineligible). Records whose + known capacity cannot hold the estimated tokens are hard-rejected so + routing never selects a model that would truncate the request. + """ + + if record_tier(record) is not result.tier: + return -1.0 + capacity = record_capacity(record) + if capacity and result.est_tokens > capacity: + return -1.0 + score = 1.0 + if capacity and result.est_tokens: + headroom = capacity / max(result.est_tokens, 1) + score += min(headroom / 8.0, 0.5) + if result.tier is RouteTier.DEEP_REASONING and _record_bool(record, "enable_thinking"): + score += 0.2 + return score + + +def rank_candidates( + records: List[Dict[str, Any]], + result: ClassificationResult, +) -> Tuple[List[Tuple[Dict[str, Any], float]], Optional[RouteTier]]: + """Rank records for the classified tier, best first. + + Returns ``(ranked, fallback_from)``. Only positive scores qualify. + When no record matches the tier, relax along ``_TIER_RELAXATION``; + ``fallback_from`` records the tier the request was demoted from + (None when the primary tier was served). + """ + + ranked = _rank_for_tier(records, result.tier, result) + if ranked: + return ranked, None + + relaxed = _TIER_RELAXATION.get(result.tier) + if relaxed is None: + return [], None + relaxed_ranked = _rank_for_tier(records, relaxed, result) + if relaxed_ranked: + return relaxed_ranked, result.tier + return [], None + + +def _rank_for_tier( + records: List[Dict[str, Any]], + tier: RouteTier, + result: ClassificationResult, +) -> List[Tuple[Dict[str, Any], float]]: + """Score records against one explicit tier, positives only, desc.""" + + tier_result = ClassificationResult( + tier=tier, + source=result.source, + risk=result.risk, + review_required=result.review_required, + est_tokens=result.est_tokens, + matched_rules=result.matched_rules, + features=result.features, + ) + scored = sorted( + ((record, score_candidate(record, tier_result)) for record in records), + key=lambda pair: pair[1], + reverse=True, + ) + return [pair for pair in scored if pair[1] > 0] + + +# --------------------------------------------------------------------------- +# Entry point used by run preparation +# --------------------------------------------------------------------------- + + +def route_model_selection( + records: List[Dict[str, Any]], + query: str, + history_length: int = 0, + agent_meta: Optional[Dict[str, Any]] = None, + has_attachments: bool = False, + agent_id: Optional[int] = None, +) -> Optional[Tuple[Dict[str, Any], RoutingDecision]]: + """Pick the best record for this run from the agent's model pool. + + ``records`` are agent-configured, availability-validated model records. + Returns ``(record, decision)`` or None when routing has no usable + signal (default-tier classification, no capable candidate, exhausted + relaxation) - callers keep their existing selection behavior then. + """ + + if not records: + return None + result = classify_run( + query=query, + history_length=history_length, + agent_meta=agent_meta, + has_attachments=has_attachments, + ) + # Contract: no routing evidence -> keep the existing selection. + if result.source is IntentSource.DEFAULT: + return None + ranked, fallback_from = rank_candidates(records, result) + if not ranked: + logger.info( + "Model routing found no capable candidate (tier=%s rules=%s); keeping default selection", + result.tier.value, + result.matched_rules, + ) + return None + best_record = ranked[0][0] + decision = RoutingDecision( + agent_id=agent_id, + tier=result.tier, + source=result.source, + risk=result.risk, + review_required=result.review_required, + est_tokens=result.est_tokens, + matched_rules=list(result.matched_rules) + ( + [f"relax:{fallback_from.value}"] if fallback_from else [] + ), + selected_model_id=_record_int(best_record, "model_id"), + selected_display_name=str( + best_record.get("display_name") or best_record.get("model_name") or "" + ), + candidates=[ + { + "model_id": _record_int(record, "model_id"), + "display_name": str( + record.get("display_name") or record.get("model_name") or "" + ), + "tier": record_tier(record).value, + "score": round(score, 3), + } + for record, score in ranked + ], + reason=f"tier={result.tier.value} source={result.source.value}", + fallback_from=fallback_from, + ) + logger.info("Model routing decision: %s", decision.to_audit_dict()) + return best_record, decision + + +__all__ = [ + "ClassificationResult", + "IntentSource", + "RiskLevel", + "RouteTier", + "RoutingDecision", + "classify_run", + "estimate_tokens", + "load_rules", + "rank_candidates", + "record_capacity", + "record_tier", + "reset_rules_cache", + "route_model_selection", + "score_candidate", +] diff --git a/deploy/env/.env.example b/deploy/env/.env.example index 60ebd7e1be..81853e4eee 100644 --- a/deploy/env/.env.example +++ b/deploy/env/.env.example @@ -164,6 +164,15 @@ ENABLE_UPLOAD_IMAGE=false # Default: false (logprobs disabled, backward-compatible behaviour). LLM_INCLUDE_LOGPROBS=false +# Multi-Model Task Routing +# When true, agent runs without an explicit model selection are routed to the +# best-matching model of the agent's own configured pool (quick QA / long +# context / deep reasoning tiers). Routing vocabularies live in +# backend/configs/model_routing_rules.json (path overridable below). +# Default: false (routing disabled, pre-routing model selection behaviour). +MODEL_ROUTING_ENABLED=false +# MODEL_ROUTING_RULES_JSON_PATH=/opt/nexent/model_routing_rules.json + # Celery Configuration CELERY_WORKER_PREFETCH_MULTIPLIER=1 CELERY_TASK_TIME_LIMIT=3600 diff --git a/docs/model_routing_integration.md b/docs/model_routing_integration.md new file mode 100644 index 0000000000..e8abde5506 --- /dev/null +++ b/docs/model_routing_integration.md @@ -0,0 +1,312 @@ +# Nexent 多模型任务路由组件 — 技术总结与生产集成指南 + +> 版本：v2.0（收敛版）｜ 日期：2026-11-11 ｜ 基线：nexent develop 分支（0f3a813） +> 参考实现：openjiuwen `intelli_router`（Deployment/TagBasedStrategy/ReliableRouter 分层模式） +> v2.0 变更：按社区贡献审查意见收敛 —— 显式启用开关、医疗词表外置为示例配置、 +> DEFAULT 语义修正（无信号保持原选择）、fallback_from 审计、输入容量口径修正、 +> 移除被证伪的 context_policy 载体（其 extra="forbid" 不接受额外键）。 + +--- + +## 1. 功能概述 + +为 nexent Agent 平台新增**按任务意图自动分流的多模型路由能力**。每次 Agent 运行时，根据用户查询内容与 Agent 声明的任务标签，将请求自动路由到该 Agent 模型池中最合适的一个模型。 + +支撑三类服务分层（可映射到异构算力部署）： + +| 分层 | RouteTier | 典型场景 | 模型示例 | +|---|---|---|---| +| 快问服务 | `quick_qa` | 指标查询、术语翻译、简短事实问答 | Qwen3-30B（非思考） | +| 长文服务 | `long_context` | 病历归纳、出院小结处理、长文献分析 | Qwen3.6-35B（长窗口） | +| 深度推理 | `deep_reasoning` | 鉴别诊断、MDT 多学科会诊、方案比较 | DeepSeek-R1 | + +附加治理能力： + +- **风险优先**：命中高风险医疗词汇（手术/化疗/剂量…）且需要分析时，强制升级到最强推理层，并打 `review_required` 标记供下游临床复核流程使用 +- **全程可追溯**：每次决策输出结构化审计日志（分层来源、命中规则、候选评分、所选模型） +- **零配置分层**：直接复用 `model_record_t` 已有能力字段判定模型分层，无需运维额外维护映射表 +- **权限不越界**：路由严格限定在 Agent 自己配置的 `model_ids` 池内，显式指定的 model_id 永远优先于路由 + +--- + +## 2. 修改文件清单 + +| # | 文件 | 状态 | 说明 | +|---|---|---|---| +| 1 | [backend/management/services/model/routing.py](../backend/management/services/model/routing.py) | **新增** | 通用路由引擎：规则加载 + 分类 + 评分 + 降级 + 审计（领域无关，词表全部来自配置） | +| 2 | [backend/configs/model_routing_rules.json](../backend/configs/model_routing_rules.json) | **新增** | 路由规则示例配置（当前为医疗词表，可整体替换为其他领域） | +| 3 | [backend/management/services/agent/preparation/compiler.py](../backend/management/services/agent/preparation/compiler.py) | **修改** | 接线：`MODEL_ROUTING_ENABLED` 门控 + `_route_agent_model`；`create_agent_config` 新增 `run_history_length`/`run_has_attachments` 可选参数 | +| 4 | [backend/management/services/agent/preparation/run.py](../backend/management/services/agent/preparation/run.py) | **修改** | +2 行：传递真实 history 轮数与附件标记 | +| 5 | [backend/contracts/constants/environment.py](../backend/contracts/constants/environment.py) | **修改** | +5 行：`MODEL_ROUTING_ENABLED`（默认 false）与规则文件路径常量 | +| 6 | [test/backend/services/test_model_task_routing.py](../test/backend/services/test_model_task_routing.py) | **新增** | 43 个单元测试（规则加载/分类/评分/降级/审计/契约锁定） | +| 7 | [test/backend/agents/test_create_agent_info.py](../test/backend/agents/test_create_agent_info.py) | **修改** | 4 处精确调用断言补充新 kwargs | +| 8 | [test/backend/management/test_service_directory_architecture.py](../test/backend/management/test_service_directory_architecture.py) | **修改** | +1 行：架构守护白名单登记 `model/routing.py` | + +无数据库迁移、无前端改动、无 SDK（sdk/nexent）改动、无既有 HTTP 契约变更。 + +--- + +## 3. 核心逻辑解读 + +### 3.0 行为契约（v2.0 收敛定稿） + +1. **显式选择永远赢**：请求带 `model_id` 时路由不介入，选中模型与 v1 之前完全一致 +2. **默认关闭**：`MODEL_ROUTING_ENABLED=false`（默认）时路由代码路径整体跳过，行为与未引入路由的版本逐字节一致 +3. **池内选择**：只在 Agent 已配置且租户可用（`connect_status=Available`）的模型中选择 +4. **无信号不换模**：分类结果为 DEFAULT（无任何规则命中）时返回 None，保持原选择 —— 路由永远不因"猜测"而改变模型 +5. **无候选不失败**：目标分层无可用模型且降级穷尽时返回 None，保持原选择 —— 路由永远不让运行失败 +6. **回退可追踪**：发生分层降级时 `fallback_from` 记录原目标层，`matched_rules` 追加 `relax:` +7. **同分稳定**：评分相同的候选按输入顺序稳定排序（Python sort 稳定性），同池同配置下决策可复现 + +### 3.1 数据流总览 + +``` +AgentRequest（query, model_id?, history, minio_files…） + │ + ▼ +create_agent_run_info (run.py) + │ model_list = create_model_config_list(tenant_id) ← 租户全量 LLM + │ create_config_kwargs += run_history_length / run_has_attachments + ▼ +create_agent_config (compiler.py) + │ + │ ① _select_agent_model_id(agent_model_ids, override_model_id, tenant_id) + │ ├─ override_model_id 非 None → 直接用它（显式选择，路由不介入） + │ └─ 否则 → Agent.model_ids 中第一个可用模型（既有逻辑，保底） + │ + │ ② MODEL_ROUTING_ENABLED ∧ override 缺省 时 → _route_agent_model(...) ← 新增 + │ ├─ 收集 Agent.model_ids 池内可用记录（is_model_available 过滤） + │ └─ route_model_selection(候选记录, query, history_length, has_attachments) + │ ├─ load_rules() ← configs/model_routing_rules.json + │ ├─ classify_run() → ClassificationResult + │ ├─ DEFAULT 来源 → 返回 None（契约 4） + │ ├─ rank_candidates() → ([(record, score)], fallback_from) + │ └─ 空 → 返回 None（契约 5）；否则 (最佳记录, RoutingDecision) + │ + ▼ +AgentConfig.model_name = 选中模型的 display_name + │ （SDK 按 cite_name 取用，零改动） + ▼ +NexentAgent.create_model → OpenAIModel(api_base, api_key, …) +``` + +关键性质： +- **路由只在"开关开启 ∧ 未显式指定模型"时介入**（[compiler.py:608-621](../backend/management/services/agent/preparation/compiler.py#L608)） +- **候选池 = Agent 自己的 model_ids**，与现有权限校验 `validate_agent_model_selection` 的边界一致 +- 开关关闭 / 单模型池 / 无信号 / 无候选四种情况下自动回退原行为，**对存量 Agent 零影响** + +### 3.2 分类规则引擎（classify_run）— 领域无关 + +判定优先级（高→低）： + +``` +1. 显式任务标签 task_tag 命中规则文件 task_tags 映射 → 直接定层（来源=explicit_tag） +2. 风险升级 高风险词汇 ∧ 分析词汇 → deep_reasoning + review_required=True +3. 长上下文 估算 token ≥ 阈值 或 字符数 ≥ 阈值 或命中长文词汇 + （长文优先于推理：截断是比弱推理更硬的失败） +4. 深度推理 命中分析/推理词汇 +5. 快问 命中查询词汇 或（非空且 ≤ char_limit） +6. 兜底 无任何信号 → DEFAULT 来源 → 调用方保持原选择 +``` + +**词表、阈值、标签映射全部来自 `configs/model_routing_rules.json`**（`MODEL_ROUTING_RULES_JSON_PATH` 可覆盖路径）；文件缺失/损坏时降级为空词表默认值 —— 引擎本身不含任何医疗词汇，换领域只需换配置文件。 + +token 估算：CJK ≈1 token/字、其他 ≈4 字符/token；history 按配置每轮加成（默认 200）、附件按配置加成（默认 800）。 + +### 3.3 模型分层判定（record_tier）— 零配置设计 + +复用 `model_record_t` 运维已验证的字段，判定优先级： + +``` +reasoning_supported == true 或 enable_thinking == true → deep_reasoning +输入容量 ≥ 48000（先 max_input_tokens，后 context_window_tokens）→ long_context +名称启发式：reason / think / -r1 / o1 / o3 → deep_reasoning + long / long-context → long_context +其余 → quick_qa +``` + +容量口径（v2.0 修正）：路由守护**输入侧**，优先用 `max_input_tokens`（provider 硬输入上限），其次合并窗口；输出预留由既有 context budget 模块负责，不重复计算。 + +### 3.4 候选评分与降级（score_candidate / rank_candidates） + +同层候选打分： + +``` +score = 1.0 + + min(容量余量/8, 0.5) # headroom = 容量/估算token，越大越好 + + 0.2 # deep 层且 enable_thinking=true +``` + +硬性排除（score = -1）： +- 分层不匹配的候选 +- 已知输入容量 < 估算 token 的候选 —— **保证路由永不选中会截断请求的模型** + +降级链（某层无可用模型时）： + +``` +quick_qa → long_context → deep_reasoning +（fallback_from 记录原目标层；穷尽后返回 None 保持原选择） +``` + +### 3.5 审计与可追溯（RoutingDecision） + +每次实际发生的路由输出一条结构化日志（`logger.info("Model routing decision: %s", …)`）： + +```json +{ + "agent_id": 42, + "tier": "deep_reasoning", // 目标分层 + "source": "rule", // explicit_tag | rule（default 不产生决策） + "risk": "high", + "review_required": true, // 下游复核流程的输入标记 + "est_tokens": 1834, + "matched_rules": ["deep:鉴别诊断", "risk:escalate_deep"], + "selected_model_id": 12, + "selected_display_name": "DeepSeek-R1", + "candidates": [{"model_id": 12, "tier": "deep_reasoning", "score": 1.2}], + "fallback_from": null, // 降级时为原目标层，如 "quick_qa" + "reason": "tier=deep_reasoning source=rule" +} +``` + +### 3.6 接线点详解（compiler.py / run.py 改动） + +- [compiler.py:598-621](../backend/management/services/agent/preparation/compiler.py#L598)：选择流程改为「显式覆盖 →（开关开启时）路由 → 保底」；路由结果复用记录容量字段，无二次 DB 查询 +- [compiler.py:797-835](../backend/management/services/agent/preparation/compiler.py#L797)：`_route_agent_model` —— 池内候选收集 + 调用路由服务；延迟 import 遵循仓库惯例 +- [compiler.py:71-96](../backend/management/services/agent/preparation/compiler.py#L71)：`create_agent_config` 新增 `run_history_length=0` / `run_has_attachments=False` 可选参数（默认值向后兼容，两处子 Agent 递归调用透传） +- [run.py:210-214](../backend/management/services/agent/preparation/run.py#L210)：传递真实 history 轮数与附件标记 +- `route_model_selection` 返回 `(record, decision)` 元组，**不污染输入 record**（测试锁定） + +--- + +## 4. 测试用例解读 + +[test_model_task_routing.py](../test/backend/services/test_model_task_routing.py) — 43 个用例，7 组： + +| 组 | 用例数 | 覆盖点 | +|---|---|---| +| `TestRulesLoading` | 5 | 文件缺失降级空词表、示例配置可加载、损坏 JSON 降级、缓存重置后重读 | +| `TestEstimateTokens` | 4 | 空文本、CJK 1:1、ASCII 4:1、history/附件加成 | +| `TestClassifyRun` | 13 | 显式标签最高优先、未知标签落规则引擎、风险升级+强制复核、高风险无分析不升层、按体量判长文、长文压过推理、推理/快问关键词、短问兜底、**无信号 DEFAULT**、**空词表 DEFAULT**、空查询安全、非法标签映射忽略 | +| `TestRecordTier` | 5 | reasoning 标志、**输入上限优先于合并窗口**、容量阈值、名称启发式、reasoning 压过容量 | +| `TestScoreAndRank` | 6 | 异层不可选、同层余量优先、容量不足硬拒绝、降序+无降级标记、**降级记录 fallback_from**、穷尽降级返回空 | +| `TestRouteModelSelection` | 7 | 决策元组完整、空池 None、**DEFAULT 保持原选择（向后兼容契约锁定）**、短问正常路由、穷尽降级 None、**快问→长文降级审计**、不污染输入 | +| `test_task_tag_mapping`(参数化) | 4 | 代表标签到分层映射 | + +设计要点： +- **契约锁定**：`test_default_source_keeps_existing_selection` 与 `test_fallback_from_is_audited` 把 §3.0 契约 4/5 固化为回归 —— 这两条是社区审查最关心的向后兼容语义 +- **通用性证明**：`test_empty_vocabularies_fall_to_default` 证明引擎在零词表下正确退化为"永不路由"，医疗词表确实只是配置 +- 测试数据 `_record()` 工厂模拟 `model_record_t` 行结构，无需 DB fixture；`_isolated_rules` fixture 保证规则缓存测试隔离 + +### 集成测试 + +- `test_create_agent_info.py`（253 用例）：`create_agent_config` 全链路，4 处精确调用断言已补新 kwargs，全绿 +- 基线 diff 对比（stash 前后同测试集）：失败集合与基线**完全一致**，零新增失败 +- 架构守护 + contracts 守护：14 passed +- 版权头检查 6 文件：OK + +--- + +## 5. 生产环境集成步骤 + +### 阶段一：部署前确认（运维） + +1. **版本基线**：确认部署的 nexent 代码包含本变更（文件清单见 §2） +2. **无需数据库迁移**：所有判定复用 `model_record_t` 既有列；若列值为 NULL，`record_tier` 自动落到名称启发式 +3. **路由默认关闭**：部署后行为与升级前完全一致；确认 `MODEL_ROUTING_ENABLED` 未设置（或为 false） +4. **规则文件**：确认 `backend/configs/model_routing_rules.json` 随镜像发布（缺失时引擎降级为空词表 = 永不路由，安全但无功能）；医疗以外领域替换该文件内容即可 + +### 阶段二：模型池配置（管理员，管理界面操作） + +为需要路由的 Agent 配置**多模型池**（Agent 编辑页 → 模型选择，最多 5 个）： + +1. 确保租户模型列表中各分层模型已配置且**连通性检测为 Available**（`connect_status` 驱动 `is_model_available`，不可用模型自动退出候选） +2. 建议按能力如实填写模型记录字段（添加/编辑模型时）： + - 推理模型：`reasoning_supported = true`（或 `enable_thinking = true`） + - 长文模型：`context_window_tokens`（或 `max_input_tokens`）填真实值 ≥ 48000 + - 这些字段同时被容量预算模块（run_budget）使用，填写准确一举两得 +3. 为目标 Agent 勾选池内多个模型（如：快问模型 + 长文模型 + 推理模型） +4. （可选）领域任务标签：在**规则文件**的 `task_tags` 节维护 `标签 → 分层` 映射（当前示例含 `mdt`/`summary`/`quality_check` 等通用标签）；per-Agent 的显式标签载体（Agent 配置界面/字段）需先与社区确认配置契约，见 §6.2 + +> 未配置多模型的 Agent 行为不变：单模型池无路由空间；开关关闭时一切如旧。 + +### 阶段三：验证（灰度） + +1. **单元回归**（部署机或 CI）： + ```bash + pytest test/backend/services/test_model_task_routing.py -v + pytest test/backend/management/test_service_directory_architecture.py -v + ``` +2. **冒烟验证**（管理服务启动后）： + - 用配置了多模型的 Agent 发送三类查询，观察 management 服务日志中的 `Model routing decision` 行： + - 短查询（如 "CRP 的正常值"）→ `tier=quick_qa` + - 粘贴长病历并要求归纳 → `tier=long_context` + - "请分析该患者的鉴别诊断" → `tier=deep_reasoning` + - 核对 `selected_model_id` 与 `candidates` 评分符合预期 + - 在前端会话里确认实际回答的模型行为一致（可在模型服务侧看请求落点） +3. **显式选择优先级**：请求带 `model_id` 时确认日志**无** `Model routing decision`（路由未介入） + +### 阶段四：监控与审计接入 + +- **路由审计**：采集 management 服务中 `Model routing decision` 结构化日志，接入现有日志平台（如 ELK）。关键字段告警建议： + - `review_required=true` 的高风险运行量（供临床质控报表） + - `fallback_from` 或 `matched_rules` 含 `relax:` 的比例 —— 持续偏高说明某分层容量不足，需扩池 +- **容量阈值**：`est_tokens` 接近所选模型容量的运行占比（容量规划输入） + +### 回滚方案 + +改动为纯增量逻辑：删除/回退 2 个 backend 文件改动即恢复原行为（`_select_agent_model_id` 原路径完整保留）。路由不写库、不改请求契约、无缓存状态，回滚无副作用。 + +--- + +## 6. 已知边界与后续扩展点 + +### 6.1 可调参数 + +| 参数 | 位置 | 默认值 | 含义 | +|---|---|---|---| +| `MODEL_ROUTING_ENABLED` | 环境变量 | `false` | 路由总开关（默认关闭，行为与历史版本一致） | +| `MODEL_ROUTING_RULES_JSON_PATH` | 环境变量 | `configs/model_routing_rules.json` | 规则文件路径 | +| `long_context.token_threshold` | 规则文件 | 24000 | 估算 token 判长文阈值 | +| `long_context.char_threshold` | 规则文件 | 8000 | 字符数判长文阈值 | +| `quick_qa.char_limit` | 规则文件 | 120 | 短查询判快问上限 | +| 各层 `terms` / `risk.*_terms` / `task_tags` | 规则文件 | 医疗示例 | 领域词表与标签映射 | +| `history_tokens_per_turn` / `attachment_tokens` | 规则文件 | 200 / 800 | 估算加成 | +| 评分权重（headroom/8 封顶 0.5、thinking +0.2） | routing.py 常量 | — | 同层候选偏好（代码内，改动需发版） | + +### 6.2 后续扩展方向（按社区贡献路径推进） + +1. **Feature Issue 先行**：以通用问题表述（"Agent 配置多模型后按任务与输入规模选择合适模型，保留显式指定优先级"）发起方向讨论，附前后行为示例；确认配置位置（per-Agent 标签的载体）、启用方式（现环境变量，可讨论租户/Agent 级开关）、扩展接口 +2. **per-Agent 任务标签**：`context_policy` 被 `PolicyLayers(extra="forbid")` 校验（sdk policy.py:23），不能承载 task_tag；需在 Issue 中定载体（候选：Agent 表新列 / metadata 字段 / 标签实体） +3. **能力组合而非互斥分层**：现版 tier 互斥（推理模型兼任长文时被排除）；演进方向是"先过滤容量/模态硬条件，再按任务偏好评分"——已被审查标识为 v1 设计约束 +4. **review_required 贯通前端**：decision 挂 run 元数据，会话接口透出复核标记（当前仅审计日志） +5. **RPM/并发感知**：`concurrency_limit` 列已存在，可加实时余量权重（参考 openjiuwen AdaptiveStrategy） +6. **失败冷却**：叠加 openjiuwen ReliableRouter 式冷却状态（nexent SDK retry.py 已有错误分类） +7. **LLM 辅助分类**：规则未命中时小模型兜底分类（参考 automation/intent.py 的工厂模式） + +### 6.3 环境已知问题（非本变更引入，基线复现） + +- `test_run_model_control_boundaries.py` 与 `test_agent_run_model_controls.py` 同跑时 3 例因 import 污染失败（单跑通过） +- `test_create_agent_info_workbench_coverage.py` 7 例失败（基线一致） +- `test/backend/agents/` 下 6 个文件 collection error：本地 `uv sync` 安装的 nexent 包与仓库 sdk/ 版本不匹配；以可编辑方式安装 `sdk/` 可解决 +- 以上均已用 `git stash` 基线对比确认与本变更无关 + +--- + +## 7. 快速参考 + +```bash +# 跑路由单元测试 +pytest test/backend/services/test_model_task_routing.py -v + +# 冒烟观察路由决策 +# （management 服务日志） +grep "Model routing decision" + +# 为 Agent 声明任务标签（DB 或管理界面，context_policy JSONB） +{"task_tag": "medical_summary"} +``` + +核心文件入口： +- 路由服务：`backend/management/services/model/routing.py`（`classify_run` / `rank_candidates` / `route_model_selection`） +- 接线点：`backend/management/services/agent/preparation/compiler.py` → `_route_agent_model` diff --git a/test/backend/agents/test_create_agent_info.py b/test/backend/agents/test_create_agent_info.py index 979e410814..ad04942500 100644 --- a/test/backend/agents/test_create_agent_info.py +++ b/test/backend/agents/test_create_agent_info.py @@ -4695,6 +4695,8 @@ async def test_create_agent_run_info_success(self): conversation_id=None, enable_planning=ANY, runtime_file_context=ANY, + run_history_length=0, + run_has_attachments=False, ) mock_get_mcp.assert_called_once_with(tenant_id="tenant_1", is_need_auth=True) mock_filter.assert_called_once_with("agent_config", { @@ -5286,6 +5288,8 @@ async def test_create_agent_run_info_forwards_allow_memory_false(self): enable_planning=False, runtime_file_context=ANY, tool_params=None, + run_history_length=0, + run_has_attachments=False, ) @pytest.mark.asyncio @@ -5336,6 +5340,8 @@ async def test_create_agent_run_info_is_debug_true(self): enable_planning=False, runtime_file_context=ANY, tool_params=None, + run_history_length=0, + run_has_attachments=False, ) @pytest.mark.asyncio @@ -5392,6 +5398,8 @@ async def test_create_agent_run_info_no_published_version_fallback(self): enable_planning=False, runtime_file_context=ANY, tool_params=None, + run_history_length=0, + run_has_attachments=False, ) # Verify that get_remote_mcp_server_list was called with is_need_auth=True mock_get_mcp.assert_called_once_with(tenant_id="tenant_1", is_need_auth=True) diff --git a/test/backend/management/test_service_directory_architecture.py b/test/backend/management/test_service_directory_architecture.py index fecc510e2b..dc6af08d75 100644 --- a/test/backend/management/test_service_directory_architecture.py +++ b/test/backend/management/test_service_directory_architecture.py @@ -32,6 +32,7 @@ 'agent/tool_identity.py', 'model/run_bindings.py', 'model/run_budget.py', + 'model/routing.py', 'skill/creation.py', 'agent/a2a/adapter.py', diff --git a/test/backend/services/test_model_task_routing.py b/test/backend/services/test_model_task_routing.py new file mode 100644 index 0000000000..f73585ad3a --- /dev/null +++ b/test/backend/services/test_model_task_routing.py @@ -0,0 +1,359 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Unit tests for the multi-model task routing service.""" + +from pathlib import Path + +import pytest + +from management.services.model import routing +from management.services.model.routing import ( + ClassificationResult, + IntentSource, + RiskLevel, + RouteTier, + classify_run, + estimate_tokens, + rank_candidates, + record_tier, + reset_rules_cache, + route_model_selection, + score_candidate, +) + +# The shipped example rules file; pinned explicitly because sibling test +# modules may have replaced contracts.constants.environment with a stub +# whose path attribute would degrade load_rules() to empty vocabularies. +_SHIPPED_RULES_PATH = ( + Path(__file__).resolve().parents[3] + / "backend" / "configs" / "model_routing_rules.json" +) + + +@pytest.fixture(autouse=True) +def _isolated_rules(monkeypatch): + """Pin the shipped rules path and keep the cache isolated per test.""" + + import contracts.constants.environment as env_module + + monkeypatch.setattr( + env_module, "MODEL_ROUTING_RULES_JSON_PATH", str(_SHIPPED_RULES_PATH), + raising=False, + ) + reset_rules_cache() + yield + reset_rules_cache() + + +@pytest.fixture +def medical_rules(): + """Load the shipped example rules (medical vocabulary).""" + + return routing.load_rules() + + +def _record(model_id=1, name="qwen3-30b", tier_facts=None, **overrides): + record = { + "model_id": model_id, + "model_name": name, + "display_name": name, + "reasoning_supported": None, + "enable_thinking": None, + "context_window_tokens": None, + "max_input_tokens": None, + } + record.update(tier_facts or {}) + record.update(overrides) + return record + + +DEEP_FACTS = {"reasoning_supported": True, "context_window_tokens": 65536} +LONG_FACTS = {"reasoning_supported": False, "context_window_tokens": 262144} +QUICK_FACTS = {"reasoning_supported": False, "context_window_tokens": 32768} + + +class TestRulesLoading: + def test_missing_file_degrades_to_empty_defaults(self, monkeypatch, tmp_path): + monkeypatch.setattr( + "contracts.constants.environment.MODEL_ROUTING_RULES_JSON_PATH", + str(tmp_path / "nonexistent.json"), + ) + rules = routing.load_rules() + assert rules["task_tags"] == {} + assert rules["risk"]["high_terms"] == [] + + def test_shipped_rules_file_loads(self): + rules = routing.load_rules() + assert rules["task_tags"], "shipped rules file must carry task tags" + assert rules["risk"]["high_terms"] + + def test_malformed_file_degrades(self, monkeypatch, tmp_path): + bad = tmp_path / "bad.json" + bad.write_text("{not json", encoding="utf-8") + monkeypatch.setattr( + "contracts.constants.environment.MODEL_ROUTING_RULES_JSON_PATH", str(bad) + ) + rules = routing.load_rules() + assert rules == routing._DEFAULT_RULES + + def test_reset_reloads_from_disk(self, monkeypatch, tmp_path): + monkeypatch.setattr( + "contracts.constants.environment.MODEL_ROUTING_RULES_JSON_PATH", + str(tmp_path / "nonexistent.json"), + ) + routing.load_rules() + assert routing.load_rules()["task_tags"] == {} + reset_rules_cache() + rules = routing.load_rules() + # Back to the shipped file (its path is the real environment value). + from contracts.constants.environment import MODEL_ROUTING_RULES_JSON_PATH as real + + monkeypatch.undo() + reset_rules_cache() + if str(real) != str(tmp_path / "nonexistent.json"): + assert routing.load_rules()["task_tags"] + + +class TestEstimateTokens: + def test_empty_text(self): + assert estimate_tokens("", rules=routing._DEFAULT_RULES) == 0 + + def test_cjk_counts_one_per_char(self): + assert estimate_tokens("你好世界", rules=routing._DEFAULT_RULES) == 4 + + def test_ascii_counts_words(self): + # 8 ascii chars -> ceil(8/4) = 2 + assert estimate_tokens("abcdefgh", rules=routing._DEFAULT_RULES) == 2 + + def test_history_and_attachments_add_overhead(self): + rules = dict(routing._DEFAULT_RULES) + value = estimate_tokens( + "你好", history_length=3, has_attachments=True, rules=rules + ) + assert value == 2 + 3 * 200 + 800 + + +class TestClassifyRun: + def test_explicit_task_tag_wins(self, medical_rules): + result = classify_run("随便一句话", agent_meta={"task_tag": "mdt"}) + assert result.tier is RouteTier.DEEP_REASONING + assert result.source is IntentSource.EXPLICIT_TAG + assert "task_tag:mdt" in result.matched_rules + + def test_unknown_task_tag_falls_through_to_rules(self, medical_rules): + result = classify_run("房颤患者的鉴别诊断思路", agent_meta={"task_tag": "unknown_tag"}) + assert result.source is IntentSource.RULE + assert result.tier is RouteTier.DEEP_REASONING + + def test_high_risk_with_analysis_escalates_and_requires_review(self, medical_rules): + result = classify_run("请分析这个患者的化疗方案选择") + assert result.tier is RouteTier.DEEP_REASONING + assert result.risk is RiskLevel.HIGH + assert result.review_required is True + assert "risk:escalate_deep" in result.matched_rules + + def test_high_risk_without_analysis_stays_routine_tier(self, medical_rules): + result = classify_run("化疗的翻译是什么") + assert result.risk is RiskLevel.HIGH + assert result.tier is RouteTier.QUICK_QA + assert result.review_required is False + + def test_long_query_by_size(self, medical_rules): + result = classify_run("病" * 48000) + assert result.tier is RouteTier.LONG_CONTEXT + assert "long_context:size" in result.matched_rules + + def test_long_context_keyword_beats_deep_keyword(self, medical_rules): + result = classify_run("帮我归纳这份病历并分析病因") + assert result.tier is RouteTier.LONG_CONTEXT + + def test_deep_reasoning_keyword(self, medical_rules): + result = classify_run("多学科会诊的步骤是什么") + assert result.tier is RouteTier.DEEP_REASONING + + def test_quick_qa_keyword(self, medical_rules): + result = classify_run("CRP的正常值是多少") + assert result.tier is RouteTier.QUICK_QA + + def test_short_query_defaults_to_quick(self, medical_rules): + result = classify_run("你好") + assert result.tier is RouteTier.QUICK_QA + assert result.source is IntentSource.RULE + + def test_no_signal_yields_default_source(self, medical_rules): + # >120 chars without any keyword -> DEFAULT source; contract says + # the caller keeps the pre-routing selection in that case. + result = classify_run("今天天气不错" + "，窗外有光" * 25) + assert result.source is IntentSource.DEFAULT + assert result.tier is RouteTier.QUICK_QA + + def test_empty_vocabularies_fall_to_default(self): + rules = dict(routing._DEFAULT_RULES) + rules["quick_qa"] = {"char_limit": 0, "terms": []} + result = classify_run("任何内容都无所谓", rules=rules) + assert result.source is IntentSource.DEFAULT + + def test_empty_query_is_safe(self, medical_rules): + result = classify_run("") + assert result.source is IntentSource.DEFAULT + + def test_invalid_task_tag_mapping_is_ignored(self): + rules = dict(routing._DEFAULT_RULES) + rules["task_tags"] = {"broken": "not_a_tier"} + result = classify_run("内容", agent_meta={"task_tag": "broken"}, rules=rules) + assert result.source is not IntentSource.EXPLICIT_TAG + + +def _result(tier): + return ClassificationResult( + tier=tier, + source=IntentSource.RULE, + risk=RiskLevel.ROUTINE, + review_required=False, + est_tokens=1000, + ) + + +class TestRecordTier: + def test_reasoning_flag(self): + assert record_tier(_record(tier_facts=DEEP_FACTS)) is RouteTier.DEEP_REASONING + + def test_capacity_prefers_input_limit(self): + record = _record(tier_facts={"max_input_tokens": 32000, "context_window_tokens": 200000}) + # Input limit < 48000 wins over window -> not long tier. + assert record_tier(record) is RouteTier.QUICK_QA + + def test_capacity(self): + assert record_tier(_record(tier_facts=LONG_FACTS)) is RouteTier.LONG_CONTEXT + + def test_name_fallback(self): + assert record_tier(_record(name="deepseek-r1")) is RouteTier.DEEP_REASONING + assert record_tier(_record(name="qwen-long")) is RouteTier.LONG_CONTEXT + assert record_tier(_record(name="qwen3-30b")) is RouteTier.QUICK_QA + + def test_reasoning_flag_overrides_capacity(self): + assert record_tier(_record(tier_facts={ + "reasoning_supported": True, "context_window_tokens": 1000000 + })) is RouteTier.DEEP_REASONING + + +class TestScoreAndRank: + def test_wrong_tier_ineligible(self): + record = _record(tier_facts=LONG_FACTS) + assert score_candidate(record, _result(RouteTier.DEEP_REASONING)) == -1.0 + + def test_capacity_headroom_prefers_larger(self): + tight = _record(model_id=1, name="tight", tier_facts={ + "reasoning_supported": False, "context_window_tokens": 60000 + }) + large = _record(model_id=2, name="roomy", tier_facts={ + "reasoning_supported": False, "context_window_tokens": 200000 + }) + result = ClassificationResult( + tier=RouteTier.LONG_CONTEXT, source=IntentSource.RULE, + risk=RiskLevel.ROUTINE, review_required=False, est_tokens=50000, + ) + assert score_candidate(large, result) > score_candidate(tight, result) + + def test_insufficient_capacity_rejected(self): + record = _record(tier_facts={"reasoning_supported": False, "context_window_tokens": 4096}) + result = ClassificationResult( + tier=RouteTier.LONG_CONTEXT, source=IntentSource.RULE, + risk=RiskLevel.ROUTINE, review_required=False, est_tokens=100000, + ) + assert score_candidate(record, result) == -1.0 + + def test_rank_orders_desc_and_no_fallback(self): + records = [ + _record(model_id=1, name="a", tier_facts=QUICK_FACTS), + _record(model_id=2, name="b", tier_facts={ + "reasoning_supported": False, "context_window_tokens": 100000 + }), + ] + ranked, fallback_from = rank_candidates(records, _result(RouteTier.LONG_CONTEXT)) + assert ranked[0][0]["model_id"] == 2 + assert fallback_from is None + + def test_rank_relaxes_and_records_fallback_from(self): + records = [_record(model_id=1, name="only-long", tier_facts=LONG_FACTS)] + ranked, fallback_from = rank_candidates(records, _result(RouteTier.QUICK_QA)) + assert ranked and ranked[0][0]["model_id"] == 1 + assert fallback_from is RouteTier.QUICK_QA + + def test_rank_exhausted_returns_empty_with_none(self): + records = [_record(model_id=1, name="only-quick", tier_facts=QUICK_FACTS)] + ranked, fallback_from = rank_candidates(records, _result(RouteTier.DEEP_REASONING)) + assert ranked == [] + assert fallback_from is None + + +class TestRouteModelSelection: + def test_returns_record_and_decision(self, medical_rules): + records = [ + _record(model_id=11, name="fast", tier_facts=QUICK_FACTS), + _record(model_id=12, name="thinker", tier_facts=DEEP_FACTS), + ] + selection = route_model_selection(records, "请分析这个患者的鉴别诊断") + assert selection is not None + record, decision = selection + assert record["model_id"] == 12 + assert decision.tier is RouteTier.DEEP_REASONING + assert decision.selected_model_id == 12 + audit = decision.to_audit_dict() + assert audit["tier"] == "deep_reasoning" + assert len(audit["candidates"]) == 1 + + def test_no_records_returns_none(self): + assert route_model_selection([], "query") is None + + def test_default_source_keeps_existing_selection(self, medical_rules): + # No vocabulary hit -> DEFAULT source -> None (caller keeps the + # pre-routing model). This is the backward-compatibility contract. + records = [ + _record(model_id=5, name="m", tier_facts=QUICK_FACTS), + _record(model_id=6, name="n", tier_facts=DEEP_FACTS), + ] + assert route_model_selection(records, "今天天气不错" + "，窗外有光" * 25) is None + + def test_short_query_routes_to_quick(self, medical_rules): + records = [ + _record(model_id=5, name="m", tier_facts=QUICK_FACTS), + _record(model_id=6, name="n", tier_facts=DEEP_FACTS), + ] + selection = route_model_selection(records, "你好") + assert selection is not None + record, decision = selection + assert record["model_id"] == 5 + assert decision.fallback_from is None + + def test_fallback_from_is_audited(self, medical_rules): + records = [_record(model_id=5, name="m", tier_facts=QUICK_FACTS)] + selection = route_model_selection(records, "请分析鉴别诊断思路") + # No deep model: quick candidates cannot serve deep tier and deep + # has no relaxation target -> None (keep existing selection). + assert selection is None + + def test_quick_to_long_relaxation_audited(self, medical_rules): + records = [_record(model_id=5, name="m", tier_facts=LONG_FACTS)] + selection = route_model_selection(records, "你好") + assert selection is not None + _, decision = selection + assert decision.tier is RouteTier.QUICK_QA + assert decision.fallback_from is RouteTier.QUICK_QA + assert "relax:quick_qa" in decision.matched_rules + + def test_does_not_mutate_input_records(self, medical_rules): + records = [_record(model_id=7, name="m", tier_facts=QUICK_FACTS)] + route_model_selection(records, "你好") + assert "_routing_decision" not in records[0] + + +@pytest.mark.parametrize("tag,expected", [ + ("quality_check", RouteTier.QUICK_QA), + ("mdt", RouteTier.DEEP_REASONING), + ("summary", RouteTier.LONG_CONTEXT), + ("quick_qa", RouteTier.QUICK_QA), +]) +def test_task_tag_mapping(tag, expected, medical_rules): + result = classify_run("内容", agent_meta={"task_tag": tag}) + assert result.tier is expected