feat: enforce knowledge type source hierarchy

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

@@ -45,6 +45,7 @@ BUSINESS_MARKERS = {
     "课堂", "练习", "静心", "觉察", "内在",
 }
 FIXED_INFORMATION_TYPE = "fixed"
+BUSINESS_TYPE_PRIORITY = {"course": 3, "qa": 2, "general": 1, "fixed": 0}
 FIXED_INFORMATION_MARKERS = {
     "上课时间", "上课安排", "开课时间", "带练", "回放", "音频", "课程作业", "作业",
     "会议", "会议链接", "直播链接", "链接", "服务权益", "权益", "课程助理", "助理",
@@ -603,15 +604,24 @@ class KnowledgeAgentService:
         if not catalog:
             return business or fixed_information, [], "当前没有可用正式知识库"
         terms = set(extract_terms(question))
-        ranked: list[tuple[bool, float, str, int]] = []
+        ranked: list[tuple[int, float, str, int]] = []
         for item in catalog:
             manifest_terms = set(extract_terms(" ".join(str(item.get(key, "")) for key in ("name", "purpose", "applicableQuestions", "coreTopics"))))
             overlap = len(terms & manifest_terms) / max(1, min(len(terms), 18))
             is_fixed = item.get("type") == FIXED_INFORMATION_TYPE
             if overlap > 0 or business or (fixed_information and is_fixed):
-                ranked.append((fixed_information and is_fixed, overlap, str(item.get("publishedAt") or ""), int(item["knowledgeId"])))
+                if fixed_information:
+                    type_priority = 4 if is_fixed else BUSINESS_TYPE_PRIORITY.get(str(item.get("type")), 0)
+                elif business:
+                    type_priority = BUSINESS_TYPE_PRIORITY.get(str(item.get("type")), 0)
+                else:
+                    type_priority = 0
+                ranked.append((type_priority, overlap, str(item.get("publishedAt") or ""), int(item["knowledgeId"])))
         ranked.sort(reverse=True)
-        selected = [item_id for preferred, score, _, item_id in ranked if preferred or score >= 0.04][:4]
+        selected = [
+            item_id for type_priority, score, _, item_id in ranked
+            if (fixed_information and type_priority == 4) or score >= 0.04
+        ][:4]
         if business and not selected:
             selected = [item_id for _, _, _, item_id in ranked[:3]]
         if fixed_information and selected:

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

@@ -134,7 +134,7 @@ class PromptService:
                 + "[不可关闭的最低安全规则 v1]\n"
                 + "现实危险、自伤伤人风险应优先建议立即寻求线下专业帮助；医疗、法律、财务问题不得给出替代专业意见的结论；"
                 + "不得伪造老师观点或课程内容；不得输出整篇课程文章、大段连续原文，也不得通过多轮拼接还原完整资料。"
-                + cls._fixed_information_rule(chunks),
+                + cls._knowledge_type_rules(chunks),
             }
         ]
 
@@ -161,11 +161,16 @@ class PromptService:
 
     @staticmethod
     def _format_chunk(index: int, chunk: RetrievedChunk) -> str:
+        type_labels = {
+            "fixed": "固定信息类·最高优先级",
+            "course": "课程知识库",
+            "qa": "答疑知识库",
+            "general": "通用知识库",
+        }
+        priority = type_labels.get(chunk.knowledge_type, "通用知识")
         if chunk.knowledge_type == "fixed":
-            priority = "固定信息类·最高优先级"
             published = f"\n生效版本时间：{chunk.published_at}" if chunk.published_at else ""
         else:
-            priority = "普通知识"
             published = ""
         return (
             f"[已回读完整章节 {index}｜{priority}] {chunk.title}\n"
@@ -173,15 +178,22 @@ class PromptService:
         )
 
     @staticmethod
-    def _fixed_information_rule(chunks: list[RetrievedChunk]) -> str:
+    def _knowledge_type_rules(chunks: list[RetrievedChunk]) -> str:
-        if not any(chunk.knowledge_type == "fixed" for chunk in chunks):
+        types = {chunk.knowledge_type for chunk in chunks}
+        if not types:
             return ""
-        return (
+        rules = [
-            "\n[固定信息优先规则]\n"
+            "\n[知识库类型使用规则]\n",
+            "课程知识库用于课程定义、方法、步骤和老师观点；答疑知识库用于常见问题的直接解答，但不能覆盖课程原文；"
+            "通用知识库仅作背景补充，不能覆盖课程知识库和答疑知识库。",
+        ]
+        if "fixed" in types:
+            rules.append(
                 "本轮召回了固定信息类知识库。上课时间、带练安排、回放和音频、课程作业、会议链接、服务权益、"
                 "课程助理联系方式及其他当前安排，必须以固定信息类知识库中的最新生效内容为准。"
                 "如与课程、答疑或通用知识库冲突，忽略其他库的冲突内容；多个固定信息来源冲突时，以生效版本时间较新的内容为准。"
             )
+        return "".join(rules)
 
     @staticmethod
     def render_messages(messages: list[dict[str, str]]) -> str:

ai_knowledge_base_v2/apps/backend/tests/test_knowledge_agent.py

@@ -164,6 +164,34 @@ def test_fixed_information_question_without_catalog_does_not_fall_back_to_genera
         assert result.chunks == []
 
 
+def test_course_type_is_preferred_over_qa_for_course_source_questions():
+    catalog = [
+        {
+            "knowledgeId": 2,
+            "name": "课程常见答疑",
+            "type": "qa",
+            "purpose": "回答课程练习问题",
+            "applicableQuestions": "课程练习",
+            "coreTopics": "静心练习",
+            "publishedAt": "2026-07-17T10:00:00",
+        },
+        {
+            "knowledgeId": 1,
+            "name": "课程原文",
+            "type": "course",
+            "purpose": "说明课程练习方法",
+            "applicableQuestions": "课程练习",
+            "coreTopics": "静心练习",
+            "publishedAt": "2026-07-16T10:00:00",
+        },
+    ]
+
+    need_knowledge, selected_ids, _ = KnowledgeAgentService._decide("课程里的静心练习怎么做？", catalog)
+
+    assert need_knowledge is True
+    assert selected_ids[:2] == [1, 2]
+
+
 def test_fixed_information_is_retrieved_first_and_overrides_other_types():
     with _database() as db:
         _add_published_knowledge(