feat: route fixed information chats by model

ai_knowledge_base_v2/apps/backend/app/services/agent_debug_service.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import asyncio
 from collections.abc import AsyncIterator
+from dataclasses import replace
 from types import SimpleNamespace
 
 from sqlalchemy.orm import Session
@@ -15,6 +16,7 @@ from app.services.entitlement_service import EntitlementService, entitlement_dic
 from app.services.growth_profile_service import GrowthProfileService
 from app.services.knowledge_agent_service import KnowledgeAgentService
 from app.services.model_stream_service import ModelStreamService
+from app.services.model_routing_service import ModelRoutingService
 from app.services.reasoning_policy_service import ReasoningPolicyService
 from app.services.rag_service import RagResult
 from app.services.topic_session_service import TopicSessionService
@@ -117,8 +119,8 @@ class AgentDebugService:
         db: Session,
         current_admin: Admin,
     ) -> AsyncIterator[dict]:
-        model = db.get(ModelConfig, payload.modelId)
-        if model is None:
+        requested_model = db.get(ModelConfig, payload.modelId)
+        if requested_model is None:
             yield {"type": "error", "message": "模型不存在"}
             return
         try:
@@ -130,10 +132,46 @@ class AgentDebugService:
                 "reasoningVisible": reasoning_visible,
             }
             rag_result = await cls.build_result(db, payload)
+            default_model = ModelRoutingService.default_model(db)
+            chat_route = ModelRoutingService.resolve_chat(
+                db,
+                [chunk.knowledge_type for chunk in rag_result.chunks],
+            )
+            automatic_route = default_model is not None and requested_model.id == default_model.id
+            if automatic_route:
+                model = chat_route.model or requested_model
+                route_reason = chat_route.reason
+            else:
+                model = requested_model
+                route_reason = "后台调试：管理员手动指定非默认模型，不执行自动模型分流"
+            route_trace = {
+                "tool": "model_route",
+                "order": len(rag_result.tool_trace or []) + 1,
+                "request": {
+                    "requestedModelId": requested_model.id,
+                    "automaticRoute": automatic_route,
+                    "knowledgeTypes": sorted({chunk.knowledge_type for chunk in rag_result.chunks}),
+                },
+                "status": "success",
+                "durationMs": 0,
+                "response": {
+                    "modelId": model.id,
+                    "modelName": model.display_name or model.model_name,
+                    "questionType": chat_route.question_type,
+                    "routeReason": route_reason,
+                },
+            }
+            rag_result = replace(
+                rag_result,
+                tool_trace=[*(rag_result.tool_trace or []), route_trace],
+            )
             model_response = ModelStreamService.debug_stream_async(
                 model,
                 rag_result,
                 cls.overrides(payload),
+                fallback_model=default_model if automatic_route else None,
+                route_reason=route_reason,
+                question_type=chat_route.question_type,
             )
             async for segment in ReasoningPolicyService.iter_segments(model_response.chunks):
                 if segment.kind == "content":
@@ -145,7 +183,7 @@ class AgentDebugService:
                 admin_id=current_admin.id,
                 module="agent",
                 action="debug_stream",
-                target_id=model.id,
+                target_id=model_response.model_id,
             )
             db.commit()
             yield {
@@ -156,6 +194,8 @@ class AgentDebugService:
                 "knowledgeIds": rag_result.knowledge_ids,
                 "retrievalTrace": rag_result.tool_trace or [],
                 "retrievalLogId": rag_result.retrieval_log_id,
+                "routeReason": model_response.route_reason,
+                "questionType": model_response.question_type,
             }
         except asyncio.CancelledError:
             db.rollback()

ai_knowledge_base_v2/apps/backend/app/services/chat_service.py

@@ -16,6 +16,7 @@ from app.services.external_errors import ExternalServiceError
 from app.services.growth_profile_service import GrowthProfileService
 from app.services.chat_context_service import ChatContextService
 from app.services.model_service import ModelClientService
+from app.services.model_routing_service import ModelRoutingService
 from app.services.rag_service import RagService
 from app.services.topic_session_service import TopicSessionService
 
@@ -142,19 +143,33 @@ class ChatService:
             completion = ModelClientService.complete(db, rag_result)
         except ExternalServiceError as exc:
             cost_ms = int((perf_counter() - started_at) * 1000)
+            failed_route = ModelRoutingService.resolve_chat(
+                db,
+                [chunk.knowledge_type for chunk in rag_result.chunks] if rag_result is not None else [],
+            )
             AiRequestLogService.write_failed(
                 db,
                 session_id=session.id,
                 message_id=user_message.id,
                 user_id=user.id,
-                model_name=None,
+                model_name=getattr(
+                    exc,
+                    "model_name",
+                    failed_route.model.model_name if failed_route.model is not None else None,
+                ),
                 prompt=rag_result.prompt if rag_result is not None else normalized_question,
                 knowledge_ids=rag_result.knowledge_ids if rag_result is not None else None,
                 retrieve_count=len(rag_result.chunks) if rag_result is not None else 0,
                 cost_ms=cost_ms,
                 error_message=str(exc),
                 retrieved_chunks=rag_result.chunks if rag_result is not None else None,
-                model_id=None,
+                model_id=getattr(
+                    exc,
+                    "model_id",
+                    failed_route.model.id if failed_route.model is not None else None,
+                ),
+                route_reason=getattr(exc, "route_reason", failed_route.reason),
+                question_type=getattr(exc, "question_type", failed_route.question_type),
             )
             db.commit()
             raise HTTPException(
@@ -206,6 +221,8 @@ class ChatService:
             cost_ms=cost_ms,
             retrieved_chunks=rag_result.chunks,
             model_id=completion.model_id,
+            route_reason=completion.route_reason,
+            question_type=completion.question_type,
         )
         db.commit()
         return completion.answer

ai_knowledge_base_v2/apps/backend/app/services/chat_stream_service.py

@@ -21,6 +21,7 @@ from app.services.external_errors import ExternalServiceError
 from app.services.growth_profile_service import GrowthProfileService
 from app.services.human_attention_service import HumanAttentionService
 from app.services.model_stream_service import ModelStreamService
+from app.services.model_routing_service import ModelRoutingService
 from app.services.rag_async_service import AsyncRagService
 from app.services.rag_service import RagService
 from app.services.topic_session_service import TopicSessionService
@@ -105,19 +106,22 @@ class ChatStreamService:
             raise
         except ExternalServiceError as exc:
             cost_ms = int((perf_counter() - started_at) * 1000)
+            failed_route = _model_log_context(db, rag_result, model_response)
             AiRequestLogService.write_failed(
                 db,
                 session_id=session.id,
                 message_id=user_message.id,
                 user_id=user.id,
-                model_name=model_response.model_name if model_response is not None else None,
+                model_name=failed_route[3],
                 prompt=rag_result.prompt if rag_result is not None else normalized_question,
                 knowledge_ids=rag_result.knowledge_ids if rag_result is not None else None,
                 retrieve_count=len(rag_result.chunks) if rag_result is not None else 0,
                 cost_ms=cost_ms,
                 error_message=str(exc),
                 retrieved_chunks=rag_result.chunks if rag_result is not None else None,
-                model_id=model_response.model_id if model_response is not None else None,
+                model_id=failed_route[2],
+                route_reason=failed_route[0],
+                question_type=failed_route[1],
             )
             _mark_retrieval_failed(db, rag_result, str(exc), cost_ms)
             db.commit()
@@ -126,19 +130,22 @@ class ChatStreamService:
         answer = "".join(answer_parts)
         if not answer.strip():
             cost_ms = int((perf_counter() - started_at) * 1000)
+            failed_route = _model_log_context(db, rag_result, model_response)
             AiRequestLogService.write_failed(
                 db,
                 session_id=session.id,
                 message_id=user_message.id,
                 user_id=user.id,
-                model_name=model_response.model_name if model_response is not None else None,
+                model_name=failed_route[3],
                 prompt=rag_result.prompt if rag_result is not None else normalized_question,
                 knowledge_ids=rag_result.knowledge_ids if rag_result is not None else None,
                 retrieve_count=len(rag_result.chunks) if rag_result is not None else 0,
                 cost_ms=cost_ms,
                 error_message="模型未返回有效内容",
                 retrieved_chunks=rag_result.chunks if rag_result is not None else None,
-                model_id=model_response.model_id if model_response is not None else None,
+                model_id=failed_route[2],
+                route_reason=failed_route[0],
+                question_type=failed_route[1],
             )
             _mark_retrieval_failed(db, rag_result, "模型未返回有效内容", cost_ms)
             db.commit()
@@ -188,6 +195,8 @@ class ChatStreamService:
             cost_ms=cost_ms,
             retrieved_chunks=rag_result.chunks if rag_result is not None else None,
             model_id=model_response.model_id if model_response is not None else None,
+            route_reason=model_response.route_reason if model_response is not None else None,
+            question_type=model_response.question_type if model_response is not None else None,
         )
         db.commit()
 
@@ -270,19 +279,22 @@ class ChatStreamService:
             raise
         except ExternalServiceError as exc:
             cost_ms = int((perf_counter() - started_at) * 1000)
+            failed_route = _model_log_context(db, rag_result, model_response)
             AiRequestLogService.write_failed(
                 db,
                 session_id=session.id,
                 message_id=user_message.id,
                 user_id=user.id,
-                model_name=model_response.model_name if model_response is not None else None,
+                model_name=failed_route[3],
                 prompt=rag_result.prompt if rag_result is not None else normalized_question,
                 knowledge_ids=rag_result.knowledge_ids if rag_result is not None else None,
                 retrieve_count=len(rag_result.chunks) if rag_result is not None else 0,
                 cost_ms=cost_ms,
                 error_message=str(exc),
                 retrieved_chunks=rag_result.chunks if rag_result is not None else None,
-                model_id=model_response.model_id if model_response is not None else None,
+                model_id=failed_route[2],
+                route_reason=failed_route[0],
+                question_type=failed_route[1],
             )
             db.commit()
             raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc
@@ -290,19 +302,22 @@ class ChatStreamService:
         answer = "".join(answer_parts)
         if not answer.strip():
             cost_ms = int((perf_counter() - started_at) * 1000)
+            failed_route = _model_log_context(db, rag_result, model_response)
             AiRequestLogService.write_failed(
                 db,
                 session_id=session.id,
                 message_id=user_message.id,
                 user_id=user.id,
-                model_name=model_response.model_name if model_response is not None else None,
+                model_name=failed_route[3],
                 prompt=rag_result.prompt if rag_result is not None else normalized_question,
                 knowledge_ids=rag_result.knowledge_ids if rag_result is not None else None,
                 retrieve_count=len(rag_result.chunks) if rag_result is not None else 0,
                 cost_ms=cost_ms,
                 error_message="模型未返回有效内容",
                 retrieved_chunks=rag_result.chunks if rag_result is not None else None,
-                model_id=model_response.model_id if model_response is not None else None,
+                model_id=failed_route[2],
+                route_reason=failed_route[0],
+                question_type=failed_route[1],
             )
             db.commit()
             raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="模型未返回有效内容")
@@ -387,6 +402,8 @@ def _write_success(
         cost_ms=cost_ms,
         retrieved_chunks=rag_result.chunks if rag_result is not None else None,
         model_id=model_response.model_id if model_response is not None else None,
+        route_reason=model_response.route_reason if model_response is not None else None,
+        question_type=model_response.question_type if model_response is not None else None,
     )
     if rag_result is not None and rag_result.retrieval_log_id:
         retrieval_log = db.get(KnowledgeRetrievalLog, rag_result.retrieval_log_id)
@@ -409,6 +426,30 @@ def _write_success(
     db.commit()
 
 
+def _model_log_context(
+    db: Session,
+    rag_result,
+    model_response,
+) -> tuple[str | None, str | None, int | None, str | None]:
+    if model_response is not None:
+        return (
+            model_response.route_reason,
+            model_response.question_type,
+            model_response.model_id,
+            model_response.model_name,
+        )
+    route = ModelRoutingService.resolve_chat(
+        db,
+        [chunk.knowledge_type for chunk in rag_result.chunks] if rag_result is not None else [],
+    )
+    return (
+        route.reason,
+        route.question_type,
+        route.model.id if route.model is not None else None,
+        route.model.model_name if route.model is not None else None,
+    )
+
+
 def user_message_id_for_attention(db: Session, session_id: int, assistant_message_id: int) -> int:
     message_id = db.scalar(
         select(ChatMessage.id)

ai_knowledge_base_v2/apps/backend/app/services/model_routing_service.py

@@ -6,7 +6,7 @@ from typing import Literal
 from sqlalchemy import desc, select
 from sqlalchemy.orm import Session
 
-from app.models.ai_config import ModelConfig
+from app.models.ai_config import ModelConfig, SystemConfig
 
 ModelScenario = Literal["report", "summary", "fixed_info", "deep_chat"]
 
@@ -31,10 +31,11 @@ class ModelRoute:
     scenario: ModelScenario
     reason: str
     fallback_used: bool
+    question_type: str | None = None
 
 
 class ModelRoutingService:
-    """Centralizes deterministic model selection without changing live-chat routing."""
+    """Centralizes deterministic, auditable model selection and safe fallback."""
 
     @staticmethod
     def default_model(db: Session) -> ModelConfig | None:
@@ -87,3 +88,49 @@ class ModelRoutingService:
             reason=f"场景分流：{label}无可用模型",
             fallback_used=True,
         )
+
+    @classmethod
+    def resolve_chat(cls, db: Session, knowledge_types: list[str]) -> ModelRoute:
+        normalized_types = {item.strip().lower() for item in knowledge_types if item and item.strip()}
+        if normalized_types == {"fixed"}:
+            if not cls._config_bool(db, "fixed_info_model_routing_enabled", True):
+                return ModelRoute(
+                    model=cls.default_model(db),
+                    scenario="deep_chat",
+                    reason="正式聊天：仅召回固定信息类知识库，但固定信息模型分流开关已关闭；使用默认主模型",
+                    fallback_used=False,
+                    question_type="fixed_info",
+                )
+            route = cls.resolve(db, "fixed_info")
+            return ModelRoute(
+                model=route.model,
+                scenario=route.scenario,
+                reason=f"正式聊天：仅召回固定信息类知识库；{route.reason}",
+                fallback_used=route.fallback_used,
+                question_type="fixed_info",
+            )
+
+        default = cls.default_model(db)
+        if not normalized_types:
+            reason = "正式聊天：未召回知识库；使用默认主模型"
+            question_type = "general_chat"
+        elif "fixed" in normalized_types:
+            reason = "正式聊天：混合类型知识库召回；为保证综合判断使用默认主模型"
+            question_type = "knowledge_grounded"
+        else:
+            reason = "正式聊天：非固定信息类知识召回；使用默认主模型"
+            question_type = "knowledge_grounded"
+        return ModelRoute(
+            model=default,
+            scenario="deep_chat",
+            reason=reason if default is not None else f"{reason}，但当前无可用模型",
+            fallback_used=False,
+            question_type=question_type,
+        )
+
+    @staticmethod
+    def _config_bool(db: Session, key: str, default: bool) -> bool:
+        config = db.scalar(select(SystemConfig).where(SystemConfig.config_key == key))
+        if config is None or not config.config_value.strip():
+            return default
+        return config.config_value.strip().lower() in {"1", "true", "yes", "on", "启用"}

ai_knowledge_base_v2/apps/backend/app/services/model_service.py

@@ -13,7 +13,7 @@ from sqlalchemy.orm import Session
 from app.core.config import get_settings
 from app.models.ai_config import ModelConfig, SystemConfig
 from app.services.external_errors import ExternalServiceError
-from app.services.model_routing_service import ModelRoutingService, ModelScenario
+from app.services.model_routing_service import ModelRoute, ModelRoutingService, ModelScenario
 from app.services.rag_service import NO_HIT_ANSWER, RagResult
 from app.services.secret_service import SecretService
 
@@ -26,12 +26,17 @@ class ModelCompletion:
     input_token: int
     output_token: int
     route_reason: str | None = None
+    question_type: str | None = None
 
 
 class ModelClientService:
     @staticmethod
     def complete(db: Session, rag_result: RagResult) -> ModelCompletion:
-        model = ModelClientService._get_enabled_model(db)
+        route = ModelRoutingService.resolve_chat(
+            db,
+            [chunk.knowledge_type for chunk in rag_result.chunks],
+        )
+        model = route.model
         mock_model_enabled = _system_config_bool(db, "mock_model_enabled", get_settings().mock_model_enabled)
         if mock_model_enabled:
             model_name = model.model_name if model is not None else "mock-model"
@@ -40,13 +45,41 @@ class ModelClientService:
             if model is None:
                 raise ExternalServiceError("未启用可用模型，请先在模型管理中启用一个模型。", provider="model")
             model_name = model.model_name
-            answer = _call_configured_model(model, rag_result, allow_no_hit=True)
+            try:
+                answer = _call_configured_model(model, rag_result, allow_no_hit=True)
+            except ExternalServiceError as primary_error:
+                fallback = ModelRoutingService.default_model(db)
+                if fallback is None or fallback.id == model.id:
+                    _annotate_model_error(primary_error, model, route.reason, route.question_type)
+                    raise
+                fallback_reason = f"{route.reason}；场景模型调用失败，运行时回退默认主模型"
+                try:
+                    answer = _call_configured_model(fallback, rag_result, allow_no_hit=True)
+                except ExternalServiceError as fallback_error:
+                    _annotate_model_error(
+                        fallback_error,
+                        fallback,
+                        f"{fallback_reason}；默认主模型调用仍失败",
+                        route.question_type,
+                    )
+                    raise
+                model = fallback
+                model_name = fallback.model_name
+                route = ModelRoute(
+                    model=fallback,
+                    scenario=route.scenario,
+                    reason=fallback_reason,
+                    fallback_used=True,
+                    question_type=route.question_type,
+                )
         return ModelCompletion(
             answer=answer,
             model_id=model.id if model is not None else None,
             model_name=model_name,
             input_token=_rough_token_count(rag_result.prompt),
             output_token=_rough_token_count(answer),
+            route_reason=route.reason,
+            question_type=route.question_type,
         )
 
     @staticmethod
@@ -186,6 +219,18 @@ def _mock_answer(rag_result: RagResult) -> str:
     )
 
 
+def _annotate_model_error(
+    error: ExternalServiceError,
+    model: ModelConfig,
+    route_reason: str,
+    question_type: str | None,
+) -> None:
+    error.model_id = model.id
+    error.model_name = model.model_name
+    error.route_reason = route_reason
+    error.question_type = question_type
+
+
 def _system_config_bool(db: Session, key: str, default: bool) -> bool:
     config = db.scalar(select(SystemConfig).where(SystemConfig.config_key == key))
     if config is None or not config.config_value.strip():

ai_knowledge_base_v2/apps/backend/app/services/model_stream_service.py

@@ -29,30 +29,35 @@ from app.services.model_service import (
     _system_and_turn_messages,
     _system_config_bool,
 )
-from app.services.model_routing_service import ModelRoutingService
+from app.services.model_routing_service import ModelRoute, ModelRoutingService
 from app.services.rag_service import RagResult
 
 
-@dataclass(frozen=True)
+@dataclass
 class StreamingModelResponse:
     model_id: int | None
     model_name: str
     input_token: int
     chunks: Iterator[str]
+    route_reason: str | None = None
+    question_type: str | None = None
 
 
-@dataclass(frozen=True)
+@dataclass
 class AsyncStreamingModelResponse:
     model_id: int | None
     model_name: str
     input_token: int
     chunks: AsyncIterator[str]
+    route_reason: str | None = None
+    question_type: str | None = None
 
 
 class ModelStreamService:
     @staticmethod
     def stream(db: Session, rag_result: RagResult) -> StreamingModelResponse:
-        model = _get_enabled_model(db)
+        route = _chat_route(db, rag_result)
+        model = route.model
         mock_model_enabled = _system_config_bool(db, "mock_model_enabled", get_settings().mock_model_enabled)
 
         if mock_model_enabled:
@@ -62,23 +67,26 @@ class ModelStreamService:
                 model_name=model_name,
                 input_token=_rough_token_count(rag_result.prompt),
                 chunks=_display_chunks(model, _mock_answer(rag_result)),
+                route_reason=route.reason,
+                question_type=route.question_type,
             )
 
-        if model is None:
-            raise ExternalServiceError("未启用可用模型，请先在模型管理中启用一个模型。", provider="model")
-        if not (model.api_url or model.base_url) or not model.api_key:
-            raise ExternalServiceError("模型 Base URL/API URL 或 API Key 未配置", provider="model")
-
-        return StreamingModelResponse(
+        model, fallback, route_reason = _prepare_routed_model(db, route)
+        response = StreamingModelResponse(
             model_id=model.id,
             model_name=model.model_name,
             input_token=_rough_token_count(rag_result.prompt),
-            chunks=_stream_configured_model(model, rag_result),
+            chunks=iter(()),
+            route_reason=route_reason,
+            question_type=route.question_type,
         )
+        response.chunks = _stream_with_runtime_fallback(response, model, fallback, rag_result)
+        return response
 
     @staticmethod
     def stream_async(db: Session, rag_result: RagResult) -> AsyncStreamingModelResponse:
-        model = _get_enabled_model(db)
+        route = _chat_route(db, rag_result)
+        model = route.model
         mock_model_enabled = _system_config_bool(db, "mock_model_enabled", get_settings().mock_model_enabled)
 
         if mock_model_enabled:
@@ -88,39 +96,141 @@ class ModelStreamService:
                 model_name=model_name,
                 input_token=_rough_token_count(rag_result.prompt),
                 chunks=_async_display_chunks(model, _mock_answer(rag_result)),
+                route_reason=route.reason,
+                question_type=route.question_type,
             )
 
-        if model is None:
-            raise ExternalServiceError("未启用可用模型，请先在模型管理中启用一个模型。", provider="model")
-        if not (model.api_url or model.base_url) or not model.api_key:
-            raise ExternalServiceError("模型 Base URL/API URL 或 API Key 未配置", provider="model")
-
-        return AsyncStreamingModelResponse(
+        model, fallback, route_reason = _prepare_routed_model(db, route)
+        response = AsyncStreamingModelResponse(
             model_id=model.id,
             model_name=model.model_name,
             input_token=_rough_token_count(rag_result.prompt),
-            chunks=_stream_configured_model_async(model, rag_result),
+            chunks=_empty_async_chunks(),
+            route_reason=route_reason,
+            question_type=route.question_type,
         )
+        response.chunks = _stream_with_runtime_fallback_async(response, model, fallback, rag_result)
+        return response
 
     @staticmethod
     def debug_stream_async(
         model: ModelConfig,
         rag_result: RagResult,
         overrides: dict[str, Any],
+        *,
+        fallback_model: ModelConfig | None = None,
+        route_reason: str = "后台调试：管理员手动指定模型",
+        question_type: str | None = None,
     ) -> AsyncStreamingModelResponse:
         debug_model = _copy_model_with_overrides(model, overrides)
-        if not (debug_model.api_url or debug_model.base_url) or not debug_model.api_key:
-            raise ExternalServiceError("模型 Base URL/API URL 或 API Key 未配置", provider="model")
-        return AsyncStreamingModelResponse(
-            model_id=model.id,
-            model_name=model.model_name,
-            input_token=_rough_token_count(rag_result.prompt),
-            chunks=_stream_configured_model_async(debug_model, rag_result),
+        debug_fallback = (
+            _copy_model_with_overrides(fallback_model, overrides)
+            if fallback_model is not None and fallback_model.id != model.id
+            else None
         )
+        if not _is_configured(debug_model):
+            if debug_fallback is None or not _is_configured(debug_fallback):
+                raise ExternalServiceError("模型 Base URL/API URL 或 API Key 未配置", provider="model")
+            debug_model = debug_fallback
+            route_reason = f"{route_reason}；场景模型配置不可用，运行时回退默认主模型"
+        response = AsyncStreamingModelResponse(
+            model_id=debug_model.id,
+            model_name=debug_model.model_name,
+            input_token=_rough_token_count(rag_result.prompt),
+            chunks=_empty_async_chunks(),
+            route_reason=route_reason,
+            question_type=question_type,
+        )
+        response.chunks = _stream_with_runtime_fallback_async(
+            response,
+            debug_model,
+            debug_fallback,
+            rag_result,
+        )
+        return response
 
 
-def _get_enabled_model(db: Session) -> ModelConfig | None:
-    return ModelRoutingService.default_model(db)
+def _chat_route(db: Session, rag_result: RagResult) -> ModelRoute:
+    return ModelRoutingService.resolve_chat(
+        db,
+        [chunk.knowledge_type for chunk in rag_result.chunks],
+    )
+
+
+def _prepare_routed_model(
+    db: Session,
+    route: ModelRoute,
+) -> tuple[ModelConfig, ModelConfig | None, str]:
+    model = route.model
+    if model is None:
+        raise ExternalServiceError("未启用可用模型，请先在模型管理中启用一个模型。", provider="model")
+    fallback = _runtime_fallback_model(db, model)
+    if _is_configured(model):
+        return model, fallback, route.reason
+    if fallback is None or not _is_configured(fallback):
+        raise ExternalServiceError("模型 Base URL/API URL 或 API Key 未配置", provider="model")
+    return fallback, fallback, f"{route.reason}；场景模型配置不可用，运行时回退默认主模型"
+
+
+def _runtime_fallback_model(db: Session, selected: ModelConfig) -> ModelConfig | None:
+    default = ModelRoutingService.default_model(db)
+    if default is None or default.id == selected.id:
+        return None
+    return default
+
+
+def _is_configured(model: ModelConfig) -> bool:
+    return bool((model.api_url or model.base_url) and model.api_key)
+
+
+def _stream_with_runtime_fallback(
+    response: StreamingModelResponse,
+    model: ModelConfig,
+    fallback: ModelConfig | None,
+    rag_result: RagResult,
+) -> Iterator[str]:
+    emitted = False
+    try:
+        for chunk in _stream_configured_model(model, rag_result):
+            if chunk:
+                emitted = True
+            yield chunk
+        return
+    except ExternalServiceError:
+        if emitted or fallback is None or fallback.id == model.id or not _is_configured(fallback):
+            raise
+    response.model_id = fallback.id
+    response.model_name = fallback.model_name
+    response.route_reason = f"{response.route_reason}；场景模型调用失败，运行时回退默认主模型"
+    yield from _stream_configured_model(fallback, rag_result)
+
+
+async def _stream_with_runtime_fallback_async(
+    response: AsyncStreamingModelResponse,
+    model: ModelConfig,
+    fallback: ModelConfig | None,
+    rag_result: RagResult,
+) -> AsyncIterator[str]:
+    emitted = False
+    try:
+        async for chunk in _stream_configured_model_async(model, rag_result):
+            if chunk:
+                emitted = True
+            yield chunk
+        return
+    except ExternalServiceError:
+        if emitted or fallback is None or fallback.id == model.id or not _is_configured(fallback):
+            raise
+    response.model_id = fallback.id
+    response.model_name = fallback.model_name
+    response.route_reason = f"{response.route_reason}；场景模型调用失败，运行时回退默认主模型"
+    async for chunk in _stream_configured_model_async(fallback, rag_result):
+        yield chunk
+
+
+async def _empty_async_chunks() -> AsyncIterator[str]:
+    if False:
+        yield ""
 
 
 def _stream_configured_model(model: ModelConfig, rag_result: RagResult) -> Iterator[str]: