✨ feat: opt-in multi-model task routing for agent runs

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
+}

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', '')
 

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]:

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 = {}