diff --git a/ai_knowledge_base_v2/apps/admin-web/src/App.vue b/ai_knowledge_base_v2/apps/admin-web/src/App.vue
index e39e234..64b132f 100644
--- a/ai_knowledge_base_v2/apps/admin-web/src/App.vue
+++ b/ai_knowledge_base_v2/apps/admin-web/src/App.vue
@@ -310,6 +310,7 @@ function modelSceneLabel(scene: string) {
   return ({
     background_report: "周期报告",
     background_summary: "摘要沉淀",
+    fixed_info: "固定信息问答",
     knowledge_grounded: "知识问答",
     general_chat: "通用对话",
   } as Record<string, string>)[scene] || scene || "未分类";
@@ -1459,11 +1460,14 @@ function formatRecordDateTime(value: string, boundary: "start" | "end") {
               <el-input v-model="modelForm.usageScenarios" placeholder="例如：正式对话、摘要、固定信息、报告批处理" />
             </el-form-item>
             <el-form-item label="可用能力">
-              <div class="entitlement-capabilities">
-                <el-checkbox v-model="modelForm.allowFixedInfo" :true-value="1" :false-value="0">固定信息</el-checkbox>
-                <el-checkbox v-model="modelForm.allowDeepChat" :true-value="1" :false-value="0">深度对话</el-checkbox>
-                <el-checkbox v-model="modelForm.allowSummary" :true-value="1" :false-value="0">摘要沉淀</el-checkbox>
-                <el-checkbox v-model="modelForm.allowReport" :true-value="1" :false-value="0">周期报告</el-checkbox>
+              <div>
+                <div class="entitlement-capabilities">
+                  <el-checkbox v-model="modelForm.allowFixedInfo" :true-value="1" :false-value="0">固定信息</el-checkbox>
+                  <el-checkbox v-model="modelForm.allowDeepChat" :true-value="1" :false-value="0">深度对话</el-checkbox>
+                  <el-checkbox v-model="modelForm.allowSummary" :true-value="1" :false-value="0">摘要沉淀</el-checkbox>
+                  <el-checkbox v-model="modelForm.allowReport" :true-value="1" :false-value="0">周期报告</el-checkbox>
+                </div>
+                <small class="model-routing-help">默认主模型勾选某项能力时会优先承担该场景；如需专用模型接管，请取消主模型的对应能力，并为专用模型开启该能力。</small>
               </div>
             </el-form-item>
             <el-form-item label="备注">
diff --git a/ai_knowledge_base_v2/apps/admin-web/src/components/AgentManagementView.vue b/ai_knowledge_base_v2/apps/admin-web/src/components/AgentManagementView.vue
index 3c61b50..7498bb2 100644
--- a/ai_knowledge_base_v2/apps/admin-web/src/components/AgentManagementView.vue
+++ b/ai_knowledge_base_v2/apps/admin-web/src/components/AgentManagementView.vue
@@ -33,6 +33,7 @@ const historyDetailOpen = ref(false);
 const selectedHistory = ref<PromptDetail | null>(null);
 const historyDetailLoading = ref(false);
 const agentDebugTrace = ref<Record<string, any>[]>([]);
+const lastDebugRoute = ref<{ modelName?: string; routeReason?: string } | null>(null);
 const agentPreviewMessages = ref<{
   role: "user" | "assistant" | "system";
   content: string;
@@ -104,7 +105,9 @@ async function load() {
     applyPrompt(promptResult);
     models.value = modelRows;
     knowledge.value = knowledgeRows;
-    agentForm.modelId = modelRows.find((item) => item.enabled === 1)?.id ?? modelRows[0]?.id;
+    agentForm.modelId = modelRows.find((item) => item.isDefault === 1)?.id
+      ?? modelRows.find((item) => item.enabled === 1)?.id
+      ?? modelRows[0]?.id;
     applyRuntimeConfig(formalConfig);
     applyDebugModelDefaults(agentForm.modelId);
     agentForm.knowledgeIds = knowledgeRows
@@ -313,6 +316,7 @@ async function debugAgent() {
   agentForm.question = "";
   agentDebugging.value = true;
   agentDebugTrace.value = [];
+  lastDebugRoute.value = null;
   agentDebugAbortController.value = new AbortController();
   scrollAgentPreview();
   try {
@@ -348,7 +352,8 @@ async function debugAgent() {
       },
       (result) => {
         agentDebugTrace.value = result.retrievalTrace || [];
-        ElMessage.success(result.message || "Agent 调试完成");
+        lastDebugRoute.value = { modelName: result.modelName, routeReason: result.routeReason };
+        ElMessage.success(`${result.message || "Agent 调试完成"} · ${result.modelName || selectedModelName.value}`);
       },
       agentDebugAbortController.value.signal,
     );
@@ -383,6 +388,7 @@ function scrollAgentPreview() {
 function clearAgentPreview() {
   agentPreviewMessages.value = [{ role: "assistant", content: "预览已清空，可以继续发起新的 Agent 调试。" }];
   agentDebugTrace.value = [];
+  lastDebugRoute.value = null;
 }
 
 function changeTypeLabel(value?: string) {
@@ -588,7 +594,7 @@ function errorMessage(error: unknown, fallback: string) {
     <aside class="agent-preview-panel">
       <div class="agent-preview-head">
         <div><h3>调试预览</h3><small>{{ selectedKnowledgeSummary }} · {{ selectedDebugUserLabel }}</small></div>
-        <div class="agent-preview-head-actions"><span>{{ selectedModelName }}</span><el-button link :disabled="agentDebugging" @click="clearAgentPreview">清空</el-button></div>
+        <div class="agent-preview-head-actions"><span :title="lastDebugRoute?.routeReason || ''">{{ lastDebugRoute?.modelName || selectedModelName }}</span><el-button link :disabled="agentDebugging" @click="clearAgentPreview">清空</el-button></div>
       </div>
       <div ref="agentPreviewChat" class="agent-preview-chat">
         <article v-for="(message, index) in agentPreviewMessages" :key="`${message.role}-${index}`" class="agent-preview-message-row" :class="message.role">
diff --git a/ai_knowledge_base_v2/apps/admin-web/src/config/systemSettings.ts b/ai_knowledge_base_v2/apps/admin-web/src/config/systemSettings.ts
index 45d22e3..f66b6fc 100644
--- a/ai_knowledge_base_v2/apps/admin-web/src/config/systemSettings.ts
+++ b/ai_knowledge_base_v2/apps/admin-web/src/config/systemSettings.ts
@@ -159,6 +159,13 @@ export const systemSettingSections: SystemSettingSection[] = [
         defaultValue: true,
         description: "后台始终记录引用；此项控制用户端是否展示。",
       },
+      {
+        key: "fixed_info_model_routing_enabled",
+        label: "固定信息模型分流",
+        type: "switch",
+        defaultValue: true,
+        description: "仅当本轮实际召回内容全部来自固定信息类知识库时使用对应低成本模型；关闭后统一使用默认主模型。",
+      },
       {
         key: "chat_max_active_requests",
         label: "问答最大并发数",
diff --git a/ai_knowledge_base_v2/apps/admin-web/src/styles.css b/ai_knowledge_base_v2/apps/admin-web/src/styles.css
index fbc5744..4692aa3 100644
--- a/ai_knowledge_base_v2/apps/admin-web/src/styles.css
+++ b/ai_knowledge_base_v2/apps/admin-web/src/styles.css
@@ -534,6 +534,7 @@ textarea {
 .storage-grid strong { margin-top: 8px; font-size: 20px; }
 .model-capability-tags { display: flex; flex-wrap: wrap; gap: 6px; }
 .model-default-tag { margin-left: 6px; }
+.model-routing-help { display: block; margin-top: 7px; color: #71827d; line-height: 1.55; }
 .cleanup-controls { display: flex; align-items: center; gap: 8px; white-space: nowrap; }
 .retention-panel { display: grid; grid-template-columns: minmax(260px, 1fr) 180px auto auto; align-items: center; gap: 12px; padding: 14px 16px; margin-bottom: 14px; border: 1px solid #dfe8e5; border-radius: 8px; background: #fff; }
 .retention-panel div span { display: block; margin-top: 4px; color: #667a73; font-size: 12px; }
diff --git a/ai_knowledge_base_v2/apps/admin-web/src/types/api.ts b/ai_knowledge_base_v2/apps/admin-web/src/types/api.ts
index 92b48fb..b610960 100644
--- a/ai_knowledge_base_v2/apps/admin-web/src/types/api.ts
+++ b/ai_knowledge_base_v2/apps/admin-web/src/types/api.ts
@@ -243,6 +243,8 @@ export interface AgentDebugResult {
   knowledgeIds?: string;
   retrievalTrace?: Record<string, unknown>[];
   retrievalLogId?: number;
+  routeReason?: string;
+  questionType?: string;
 }
 
 export interface AgentDebugStreamComplete {
@@ -252,6 +254,8 @@ export interface AgentDebugStreamComplete {
   knowledgeIds?: string;
   retrievalTrace?: Record<string, unknown>[];
   retrievalLogId?: number;
+  routeReason?: string;
+  questionType?: string;
 }
 
 export interface KnowledgeVersion {
diff --git a/ai_knowledge_base_v2/apps/backend/app/services/agent_debug_service.py b/ai_knowledge_base_v2/apps/backend/app/services/agent_debug_service.py
index 818cba5..c73cd04 100644
--- a/ai_knowledge_base_v2/apps/backend/app/services/agent_debug_service.py
+++ b/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()
diff --git a/ai_knowledge_base_v2/apps/backend/app/services/chat_service.py b/ai_knowledge_base_v2/apps/backend/app/services/chat_service.py
index b244eae..ac0a8ae 100644
--- a/ai_knowledge_base_v2/apps/backend/app/services/chat_service.py
+++ b/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
diff --git a/ai_knowledge_base_v2/apps/backend/app/services/chat_stream_service.py b/ai_knowledge_base_v2/apps/backend/app/services/chat_stream_service.py
index 48cb167..e2a912f 100644
--- a/ai_knowledge_base_v2/apps/backend/app/services/chat_stream_service.py
+++ b/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)
diff --git a/ai_knowledge_base_v2/apps/backend/app/services/model_routing_service.py b/ai_knowledge_base_v2/apps/backend/app/services/model_routing_service.py
index b0d5e9e..7703443 100644
--- a/ai_knowledge_base_v2/apps/backend/app/services/model_routing_service.py
+++ b/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", "启用"}
diff --git a/ai_knowledge_base_v2/apps/backend/app/services/model_service.py b/ai_knowledge_base_v2/apps/backend/app/services/model_service.py
index f30dc2c..b6c9fa5 100644
--- a/ai_knowledge_base_v2/apps/backend/app/services/model_service.py
+++ b/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():
diff --git a/ai_knowledge_base_v2/apps/backend/app/services/model_stream_service.py b/ai_knowledge_base_v2/apps/backend/app/services/model_stream_service.py
index f9f9e3e..30707b7 100644
--- a/ai_knowledge_base_v2/apps/backend/app/services/model_stream_service.py
+++ b/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]:
diff --git a/ai_knowledge_base_v2/apps/backend/tests/test_agent_runtime_config.py b/ai_knowledge_base_v2/apps/backend/tests/test_agent_runtime_config.py
index 0070fd6..ca2f032 100644
--- a/ai_knowledge_base_v2/apps/backend/tests/test_agent_runtime_config.py
+++ b/ai_knowledge_base_v2/apps/backend/tests/test_agent_runtime_config.py
@@ -21,7 +21,7 @@ from app.services.model_stream_service import (
     _stream_configured_model_async,
 )
 from app.services.model_service import ModelClientService, _copy_model_with_overrides, _max_output_tokens
-from app.services.rag_service import PromptService, RagResult
+from app.services.rag_service import PromptService, RagResult, RetrievedChunk
 
 
 def _database() -> Session:
@@ -148,10 +148,90 @@ def test_disabled_stream_returns_one_complete_chunk():
 
     with patch("app.services.model_stream_service._call_configured_model", return_value=answer):
         chunks = asyncio.run(collect())
-
     assert chunks == [answer]
 
 
+def test_agent_preview_uses_same_fixed_info_route_when_default_model_is_selected(monkeypatch):
+    async def fixed_chunks():
+        yield "本周三晚八点上课。"
+
+    async def build_result(_db, _payload):
+        return RagResult(
+            question="本周什么时候上课？",
+            knowledge_scopes=[],
+            chunks=[
+                RetrievedChunk(
+                    knowledge_id=1,
+                    knowledge_name="当前安排",
+                    title="本周安排",
+                    content="本周三晚八点上课。",
+                    knowledge_type="fixed",
+                )
+            ],
+            prompt="本周什么时候上课？",
+            tool_trace=[],
+        )
+
+    captured = {}
+
+    def debug_stream(model, _rag_result, _overrides, **kwargs):
+        captured["model"] = model.model_name
+        captured["fallback"] = kwargs["fallback_model"].model_name
+        captured["routeReason"] = kwargs["route_reason"]
+        return AsyncStreamingModelResponse(
+            model_id=model.id,
+            model_name=model.model_name,
+            input_token=10,
+            chunks=fixed_chunks(),
+            route_reason=kwargs["route_reason"],
+            question_type=kwargs["question_type"],
+        )
+
+    with _database() as db:
+        admin = _admin()
+        main = _model()
+        main.is_default = 1
+        main.allow_fixed_info = 0
+        fixed = ModelConfig(
+            id=2,
+            provider="fixed",
+            display_name="固定信息模型",
+            api_type="openai_compatible",
+            model_name="fixed-model",
+            base_url="https://example.com/v1",
+            api_url="",
+            api_key="encrypted",
+            auth_type="bearer",
+            max_token=8192,
+            stream_enabled=1,
+            timeout_second=30,
+            enabled=1,
+            is_default=0,
+            allow_fixed_info=1,
+        )
+        db.add_all([admin, main, fixed])
+        db.commit()
+        monkeypatch.setattr(AgentDebugService, "build_result", build_result)
+        monkeypatch.setattr(ModelStreamService, "debug_stream_async", debug_stream)
+        payload = AgentDebugRequest(
+            promptContent="你是测试助手",
+            modelId=main.id,
+            question="本周什么时候上课？",
+        )
+
+        async def collect_events():
+            return [item async for item in AgentDebugService.stream(payload, db, admin)]
+
+        events = asyncio.run(collect_events())
+
+        complete = next(item for item in events if item["type"] == "complete")
+        assert captured["model"] == "fixed-model"
+        assert captured["fallback"] == "production-model"
+        assert "仅召回固定信息类" in captured["routeReason"]
+        assert complete["modelName"] == "fixed-model"
+        assert complete["questionType"] == "fixed_info"
+        assert any(item.get("tool") == "model_route" for item in complete["retrievalTrace"])
+
 def test_debug_stream_setting_overrides_model_without_changing_it():
     model = _model()
     debug_model = _copy_model_with_overrides(model, {"stream_enabled": 0})
diff --git a/ai_knowledge_base_v2/apps/backend/tests/test_model_routing.py b/ai_knowledge_base_v2/apps/backend/tests/test_model_routing.py
index c4018fe..2d4b3b7 100644
--- a/ai_knowledge_base_v2/apps/backend/tests/test_model_routing.py
+++ b/ai_knowledge_base_v2/apps/backend/tests/test_model_routing.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 from decimal import Decimal
 
 import pytest
@@ -16,6 +17,9 @@ from app.models.logs import AiRequestLog
 from app.schemas.admin import DefaultModelRequest, EnableModelRequest
 from app.services.model_routing_service import ModelRoutingService
 from app.services.model_service import ModelClientService
+from app.services.model_stream_service import ModelStreamService
+from app.services.external_errors import ExternalServiceError
+from app.services.rag_service import RagResult, RetrievedChunk
 from app.services.tracked_generation_service import TrackedGenerationService
 
 
@@ -36,6 +40,7 @@ def _model(
     is_default: int,
     allow_report: int,
     allow_summary: int,
+    allow_fixed_info: int = 1,
 ) -> ModelConfig:
     return ModelConfig(
         id=model_id,
@@ -48,7 +53,7 @@ def _model(
         is_default=is_default,
         allow_report=allow_report,
         allow_summary=allow_summary,
-        allow_fixed_info=1,
+        allow_fixed_info=allow_fixed_info,
         allow_deep_chat=1,
         input_price_per_1k=Decimal("0.002"),
         output_price_per_1k=Decimal("0.006"),
@@ -86,6 +91,127 @@ def test_background_scenario_falls_back_to_default_model():
         assert "回退默认主模型" in route.reason
 
 
+def test_fixed_info_chat_routes_only_when_all_recalled_knowledge_is_fixed():
+    with _db() as db:
+        main = _model(
+            1,
+            "main-model",
+            is_default=1,
+            allow_report=1,
+            allow_summary=1,
+            allow_fixed_info=0,
+        )
+        fixed = _model(
+            2,
+            "fixed-model",
+            is_default=0,
+            allow_report=0,
+            allow_summary=0,
+            allow_fixed_info=1,
+        )
+        db.add_all([main, fixed])
+        db.commit()
+
+        fixed_route = ModelRoutingService.resolve_chat(db, ["fixed", "fixed"])
+        mixed_route = ModelRoutingService.resolve_chat(db, ["fixed", "course"])
+        no_hit_route = ModelRoutingService.resolve_chat(db, [])
+
+        assert fixed_route.model is fixed
+        assert fixed_route.question_type == "fixed_info"
+        assert mixed_route.model is main
+        assert mixed_route.question_type == "knowledge_grounded"
+        assert "混合类型" in mixed_route.reason
+        assert no_hit_route.model is main
+        assert no_hit_route.question_type == "general_chat"
+
+        db.add(SystemConfig(config_key="fixed_info_model_routing_enabled", config_value="false"))
+        db.commit()
+        disabled_route = ModelRoutingService.resolve_chat(db, ["fixed"])
+        assert disabled_route.model is main
+        assert disabled_route.question_type == "fixed_info"
+        assert "分流开关已关闭" in disabled_route.reason
+
+
+def test_fixed_info_non_stream_call_falls_back_to_main_model_on_provider_failure(monkeypatch):
+    with _db() as db:
+        main = _model(1, "main-model", is_default=1, allow_report=1, allow_summary=1, allow_fixed_info=0)
+        fixed = _model(2, "fixed-model", is_default=0, allow_report=0, allow_summary=0, allow_fixed_info=1)
+        db.add_all([main, fixed, SystemConfig(config_key="mock_model_enabled", config_value="false")])
+        db.commit()
+        calls: list[str] = []
+
+        def call(model, _rag_result, *, allow_no_hit=False):
+            calls.append(model.model_name)
+            if model.id == fixed.id:
+                raise ExternalServiceError("固定信息模型暂时不可用", provider="model")
+            return "主模型回退回答"
+
+        monkeypatch.setattr("app.services.model_service._call_configured_model", call)
+        completion = ModelClientService.complete(db, _rag_result("fixed"))
+
+        assert calls == ["fixed-model", "main-model"]
+        assert completion.model_id == main.id
+        assert completion.answer == "主模型回退回答"
+        assert completion.question_type == "fixed_info"
+        assert "运行时回退默认主模型" in (completion.route_reason or "")
+
+
+def test_fixed_info_async_stream_falls_back_before_first_output(monkeypatch):
+    async def collect(response):
+        return [chunk async for chunk in response.chunks]
+
+    with _db() as db:
+        main = _model(1, "main-model", is_default=1, allow_report=1, allow_summary=1, allow_fixed_info=0)
+        fixed = _model(2, "fixed-model", is_default=0, allow_report=0, allow_summary=0, allow_fixed_info=1)
+        db.add_all([main, fixed, SystemConfig(config_key="mock_model_enabled", config_value="false")])
+        db.commit()
+
+        async def stream(model, _rag_result):
+            if model.id == fixed.id:
+                raise ExternalServiceError("固定信息模型暂时不可用", provider="model")
+            yield "主模型流式回退回答"
+
+        monkeypatch.setattr("app.services.model_stream_service._stream_configured_model_async", stream)
+        response = ModelStreamService.stream_async(db, _rag_result("fixed"))
+        chunks = asyncio.run(collect(response))
+
+        assert chunks == ["主模型流式回退回答"]
+        assert response.model_id == main.id
+        assert response.question_type == "fixed_info"
+        assert "运行时回退默认主模型" in (response.route_reason or "")
+
+
+def test_fixed_info_stream_does_not_restart_after_partial_output(monkeypatch):
+    with _db() as db:
+        main = _model(1, "main-model", is_default=1, allow_report=1, allow_summary=1, allow_fixed_info=0)
+        fixed = _model(2, "fixed-model", is_default=0, allow_report=0, allow_summary=0, allow_fixed_info=1)
+        db.add_all([main, fixed, SystemConfig(config_key="mock_model_enabled", config_value="false")])
+        db.commit()
+        calls: list[str] = []
+
+        async def stream(model, _rag_result):
+            calls.append(model.model_name)
+            if model.id == fixed.id:
+                yield "已经输出的部分"
+                raise ExternalServiceError("输出中断", provider="model")
+            yield "不应重新生成"
+
+        async def collect(response):
+            chunks = []
+            with pytest.raises(ExternalServiceError):
+                async for chunk in response.chunks:
+                    chunks.append(chunk)
+            return chunks
+
+        monkeypatch.setattr("app.services.model_stream_service._stream_configured_model_async", stream)
+        response = ModelStreamService.stream_async(db, _rag_result("fixed"))
+        chunks = asyncio.run(collect(response))
+
+        assert chunks == ["已经输出的部分"]
+        assert calls == ["fixed-model"]
+        assert response.model_id == fixed.id
+
+
 def test_tracked_background_generation_records_actual_route_tokens_and_cost():
     with _db() as db:
         db.add(SystemConfig(config_key="mock_model_enabled", config_value="true"))
@@ -152,3 +278,20 @@ def test_model_pool_keeps_one_default_and_rejects_disabling_last_default():
                 current_admin=admin,
             )
         assert exc_info.value.status_code == 400
+
+
+def _rag_result(knowledge_type: str) -> RagResult:
+    return RagResult(
+        question="本周上课时间是什么？",
+        knowledge_scopes=[],
+        chunks=[
+            RetrievedChunk(
+                knowledge_id=1,
+                knowledge_name="当前安排",
+                title="本周安排",
+                content="本周三晚八点上课。",
+                knowledge_type=knowledge_type,
+            )
+        ],
+        prompt="本周上课时间是什么？",
+    )
diff --git a/ai_knowledge_base_v2/deployment/DEPLOYMENT_GUIDE.md b/ai_knowledge_base_v2/deployment/DEPLOYMENT_GUIDE.md
index adfd100..7e9e910 100644
--- a/ai_knowledge_base_v2/deployment/DEPLOYMENT_GUIDE.md
+++ b/ai_knowledge_base_v2/deployment/DEPLOYMENT_GUIDE.md
@@ -172,6 +172,8 @@ PERIODIC_REPORT_MAX_ATTEMPTS=3
 
 周期报告会选择“周期报告”能力已开启的可用模型，主题摘要和成长档案会选择“摘要沉淀”能力已开启的可用模型。若找不到匹配模型，会自动回退默认主模型，不会因为分流配置缺失直接中断任务。
 
+固定信息问答采用保守分流：只有本轮最终召回并采用的知识全部属于固定信息类时，才选择“固定信息”能力模型；混合召回、未命中和其他知识问答仍使用默认主模型。场景模型在尚未输出任何内容前调用失败，会自动重试默认主模型；已经输出部分内容后不会重新生成，避免重复内容。
+
 部署迁移后，旧版本原来启用的模型会自动成为默认主模型。新增其他模型时建议按以下顺序操作：
 
 1. 保存模型并执行“测试”；
@@ -180,6 +182,8 @@ PERIODIC_REPORT_MAX_ATTEMPTS=3
 4. 如需让低成本模型承担报告，应取消默认主模型的“周期报告”能力，避免默认模型优先命中；
 5. 在数据看板“模型使用与成本”中核对实际模型和成本。
 
+如线上发现固定信息模型质量或稳定性异常，可在“系统配置 / AI 问答”关闭“固定信息模型分流”，下一次提问立即恢复为默认主模型，无需重新部署。后台 Agent 预览选择默认主模型时会复用正式分流规则；显式选择非默认模型时视为人工调试覆盖，不执行自动分流。
+
 停用或删除唯一默认主模型会被后端拒绝，必须先启用并设置替代主模型。该限制用于避免生产聊天突然变成无模型可用。
 
 ## 回滚原则
diff --git a/ai_knowledge_base_v2/docs/qianwen_product_todo.md b/ai_knowledge_base_v2/docs/qianwen_product_todo.md
index 7d0fca9..74289d9 100644
--- a/ai_knowledge_base_v2/docs/qianwen_product_todo.md
+++ b/ai_knowledge_base_v2/docs/qianwen_product_todo.md
@@ -651,6 +651,7 @@ AI 日志增加：
 
 - 2026-07-31：一期已新增模型输入/输出千 Token 单价、币种、适用场景和可用能力字段；AI 请求日志记录模型 ID、估算成本、币种、问题类型、知识命中和路由原因；数据看板展示筛选范围内估算成本。暂未自动切换模型，避免影响正式回答稳定性，后续再基于这些字段做模型分流。
 - 2026-07-31：二期先完成低风险后台任务分流。模型管理支持“多个可用模型 + 一个默认主模型”；周期报告和主题摘要按能力标签选模型，无匹配时回退默认主模型；正式聊天、追问改写和检索重排仍固定使用默认主模型。后台生成会记录实际模型、分流原因、Token、耗时和估算成本，数据看板增加按调用场景、模型和币种拆分的成本明细。固定信息和正式聊天的自动分流暂不启用，待后台任务运行稳定并核对成本后再做。
+- 2026-07-31：固定信息问答已接入保守自动分流。只有本轮最终采用的召回内容全部来自固定信息类知识库时，才选择“固定信息”能力模型；未命中、非固定信息召回、固定信息与其他类型混合召回均继续使用默认主模型。场景模型在输出任何内容前失败时自动回退主模型，已开始输出后不重放，防止用户看到重复回答。用户端 AI 日志和后台 Agent 预览均展示实际模型、问题类型和路由原因；系统设置可通过 `fixed_info_model_routing_enabled` 即时关闭该分流。
 
 ---