feat: make chat model routing configurable

ai_knowledge_base_v2/apps/backend/tests/test_agent_runtime_config.py

@@ -232,6 +232,92 @@ def test_agent_preview_uses_same_fixed_info_route_when_default_model_is_selected
         assert complete["questionType"] == "fixed_info"
         assert any(item.get("tool") == "model_route" for item in complete["retrievalTrace"])
 
+
+def test_agent_preview_uses_conservative_simple_knowledge_route(monkeypatch):
+    async def answer_chunks():
+        yield "原生课程包含静水流深等练习。"
+
+    async def build_result(_db, _payload):
+        return RagResult(
+            question="原生里的作业都有哪些",
+            knowledge_scopes=[],
+            chunks=[
+                RetrievedChunk(
+                    knowledge_id=1,
+                    knowledge_name="原生课程",
+                    title="课程作业",
+                    content="静水流深静心。",
+                    knowledge_type="course",
+                )
+            ],
+            prompt="原生里的作业都有哪些",
+            tool_trace=[],
+        )
+
+    captured = {}
+
+    def debug_stream(model, _rag_result, _overrides, **kwargs):
+        captured["model"] = model.model_name
+        captured["questionType"] = kwargs["question_type"]
+        return AsyncStreamingModelResponse(
+            model_id=model.id,
+            model_name=model.model_name,
+            input_token=10,
+            chunks=answer_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_simple_knowledge = 0
+        simple = ModelConfig(
+            id=2,
+            provider="simple",
+            display_name="简单知识模型",
+            api_type="openai_compatible",
+            model_name="simple-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_simple_knowledge=1,
+        )
+        db.add_all(
+            [
+                admin,
+                main,
+                simple,
+                SystemConfig(config_key="chat_model_routing_mode", config_value="conservative"),
+            ]
+        )
+        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": "simple-model", "questionType": "simple_knowledge"}
+        assert complete["modelName"] == "simple-model"
+        assert complete["questionType"] == "simple_knowledge"
+
+
 def test_debug_stream_setting_overrides_model_without_changing_it():
     model = _model()
     debug_model = _copy_model_with_overrides(model, {"stream_enabled": 0})

ai_knowledge_base_v2/apps/backend/tests/test_model_routing.py

@@ -41,6 +41,7 @@ def _model(
     allow_report: int,
     allow_summary: int,
     allow_fixed_info: int = 1,
+    allow_simple_knowledge: int = 1,
 ) -> ModelConfig:
     return ModelConfig(
         id=model_id,
@@ -54,6 +55,7 @@ def _model(
         allow_report=allow_report,
         allow_summary=allow_summary,
         allow_fixed_info=allow_fixed_info,
+        allow_simple_knowledge=allow_simple_knowledge,
         allow_deep_chat=1,
         input_price_per_1k=Decimal("0.002"),
         output_price_per_1k=Decimal("0.006"),
@@ -132,6 +134,171 @@ def test_fixed_info_chat_routes_only_when_all_recalled_knowledge_is_fixed():
         assert "分流开关已关闭" in disabled_route.reason
 
 
+def test_simple_knowledge_routing_defaults_to_fixed_only_for_safe_rollout():
+    with _db() as db:
+        main = _model(
+            1,
+            "main-model",
+            is_default=1,
+            allow_report=1,
+            allow_summary=1,
+            allow_simple_knowledge=0,
+        )
+        simple = _model(
+            2,
+            "simple-model",
+            is_default=0,
+            allow_report=0,
+            allow_summary=0,
+            allow_simple_knowledge=1,
+        )
+        db.add_all([main, simple])
+        db.commit()
+
+        route = ModelRoutingService.resolve_chat(db, ["course"], "原生里的作业都有哪些")
+
+        assert ModelRoutingService.chat_routing_mode(db) == "fixed_only"
+        assert route.model is main
+        assert route.question_type == "knowledge_grounded"
+
+
+def test_conservative_mode_routes_only_clear_simple_knowledge_questions():
+    with _db() as db:
+        main = _model(
+            1,
+            "main-model",
+            is_default=1,
+            allow_report=1,
+            allow_summary=1,
+            allow_fixed_info=0,
+            allow_simple_knowledge=0,
+        )
+        simple = _model(
+            2,
+            "simple-model",
+            is_default=0,
+            allow_report=0,
+            allow_summary=0,
+            allow_fixed_info=0,
+            allow_simple_knowledge=1,
+        )
+        db.add_all(
+            [
+                main,
+                simple,
+                SystemConfig(config_key="chat_model_routing_mode", config_value="conservative"),
+            ]
+        )
+        db.commit()
+
+        simple_route = ModelRoutingService.resolve_chat(db, ["course", "qa"], "原生里的作业都有哪些")
+        personal_route = ModelRoutingService.resolve_chat(
+            db,
+            ["course"],
+            "我做阴影人格练习时身体很难受怎么办",
+        )
+        mixed_route = ModelRoutingService.resolve_chat(db, ["fixed", "course"], "课程作业有哪些")
+
+        assert simple_route.model is simple
+        assert simple_route.question_type == "simple_knowledge"
+        assert "保守简单知识规则" in simple_route.reason
+        assert personal_route.model is main
+        assert personal_route.question_type == "knowledge_grounded"
+        assert mixed_route.model is main
+        assert mixed_route.question_type == "knowledge_grounded"
+
+
+def test_chat_routing_off_keeps_fixed_information_on_main_model():
+    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="chat_model_routing_mode", config_value="off"),
+            ]
+        )
+        db.commit()
+
+        route = ModelRoutingService.resolve_chat(db, ["fixed"], "本周上课时间是什么")
+
+        assert route.model is main
+        assert route.question_type == "fixed_info"
+        assert "分流开关已关闭" in route.reason
+
+
+def test_user_stream_uses_conservative_simple_knowledge_route(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_simple_knowledge=0,
+        )
+        simple = _model(
+            2,
+            "simple-model",
+            is_default=0,
+            allow_report=0,
+            allow_summary=0,
+            allow_simple_knowledge=1,
+        )
+        db.add_all(
+            [
+                main,
+                simple,
+                SystemConfig(config_key="chat_model_routing_mode", config_value="conservative"),
+                SystemConfig(config_key="mock_model_enabled", config_value="false"),
+            ]
+        )
+        db.commit()
+
+        async def stream(model, _rag_result):
+            yield f"由{model.model_name}回答"
+
+        rag_result = RagResult(
+            question="原生里的作业都有哪些",
+            knowledge_scopes=[],
+            chunks=[
+                RetrievedChunk(
+                    knowledge_id=1,
+                    knowledge_name="原生课程",
+                    title="课程作业",
+                    content="静水流深静心。",
+                    knowledge_type="course",
+                )
+            ],
+            prompt="原生里的作业都有哪些",
+        )
+        monkeypatch.setattr("app.services.model_stream_service._stream_configured_model_async", stream)
+        response = ModelStreamService.stream_async(db, rag_result)
+        chunks = asyncio.run(collect(response))
+
+        assert chunks == ["由simple-model回答"]
+        assert response.model_id == simple.id
+        assert response.question_type == "simple_knowledge"
+
+
 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)

ai_knowledge_base_v2/apps/backend/tests/test_production_readiness.py

@@ -3,6 +3,8 @@ from __future__ import annotations
 from datetime import date
 import asyncio
 import json
+from pathlib import Path
+import re
 import time
 
 import pytest
@@ -41,6 +43,19 @@ def test_secret_is_encrypted_and_masked():
     assert SecretService.masked(encrypted) == MASKED_SECRET
 
 
+def test_migration_revision_ids_fit_default_alembic_version_column():
+    versions_dir = Path(__file__).parents[1] / "alembic" / "versions"
+    revision_pattern = re.compile(r'^revision\s*=\s*["\']([^"\']+)', re.MULTILINE)
+    revision_ids = []
+    for migration in versions_dir.glob("*.py"):
+        match = revision_pattern.search(migration.read_text(encoding="utf-8"))
+        if match:
+            revision_ids.append((migration.name, match.group(1)))
+
+    too_long = [(filename, revision) for filename, revision in revision_ids if len(revision) > 32]
+    assert not too_long, f"Alembic revision ID 超过默认 VARCHAR(32): {too_long}"
+
+
 def test_rate_limit_blocks_after_limit():
     SecurityStateService.enforce_limit("test:rate", limit=2, window_seconds=60, message="too many")
     SecurityStateService.enforce_limit("test:rate", limit=2, window_seconds=60, message="too many")