feat: apply knowledge type retrieval priority

ai_knowledge_base_v2/apps/backend/tests/test_knowledge_agent.py

@@ -5,6 +5,7 @@ import json
 from datetime import datetime
 from types import SimpleNamespace
 
+import pytest
 from sqlalchemy import create_engine, select
 from sqlalchemy.orm import Session
 from sqlalchemy.pool import StaticPool
@@ -24,13 +25,34 @@ from app.services.knowledge_agent_service import Candidate, KnowledgeAgentServic
 from app.services.knowledge_pipeline_service import parse_sections
 
 
+@pytest.fixture(autouse=True)
+def _disable_shared_redis_cache(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Keep unit-test catalogs isolated from the running development database cache."""
+    monkeypatch.setattr(
+        "app.services.knowledge_catalog_cache_service.get_sync_redis_client",
+        lambda: None,
+    )
+
+
 def _database() -> Session:
     engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
     Base.metadata.create_all(engine)
     return Session(engine)
 
 
-def _add_published_knowledge(db: Session, *, knowledge_id: int, name: str, open_status: int = 1) -> Knowledge:
+def _add_published_knowledge(
+    db: Session,
+    *,
+    knowledge_id: int,
+    name: str,
+    open_status: int = 1,
+    knowledge_type: str = "course",
+    purpose: str = "解答家长与孩子沟通、学习动力相关课程问题",
+    applicable_questions: str = "孩子学习动力、亲子沟通",
+    core_topics: str = "家长情绪、倾听孩子、学习动力",
+    section_title: str = "家长沟通的第一步",
+    section_content: str = "家长和学习动力不足的孩子沟通时，第一步是先稳定自己的焦虑，再倾听孩子遇到的具体困难。",
+) -> Knowledge:
     knowledge = Knowledge(
         id=knowledge_id,
         name=name,
@@ -39,7 +61,7 @@ def _add_published_knowledge(db: Session, *, knowledge_id: int, name: str, open_
         status=open_status,
         source_status="normal",
         manifest_confirmed=1,
-        knowledge_type="course",
+        knowledge_type=knowledge_type,
         review_mode="manual",
     )
     db.add(knowledge)
@@ -68,10 +90,10 @@ def _add_published_knowledge(db: Session, *, knowledge_id: int, name: str, open_
         KnowledgeManifest(
             knowledge_id=knowledge.id,
             version_id=version.id,
-            purpose="解答家长与孩子沟通、学习动力相关课程问题",
-            applicable_questions="孩子学习动力、亲子沟通",
+            purpose=purpose,
+            applicable_questions=applicable_questions,
             inapplicable_questions="天气和交通",
-            core_topics="家长情绪、倾听孩子、学习动力",
+            core_topics=core_topics,
             boundaries="不输出完整课程资料",
             content_hash=f"manifest-{knowledge_id}",
             confirmed=1,
@@ -81,8 +103,8 @@ def _add_published_knowledge(db: Session, *, knowledge_id: int, name: str, open_
         knowledge_id=knowledge.id,
         version_id=version.id,
         section_key="S0001",
-        title="家长沟通的第一步",
-        content="家长和学习动力不足的孩子沟通时，第一步是先稳定自己的焦虑，再倾听孩子遇到的具体困难。",
+        title=section_title,
+        content=section_content,
         source_start=0,
         source_end=44,
         sort_order=1,
@@ -97,7 +119,7 @@ def _add_published_knowledge(db: Session, *, knowledge_id: int, name: str, open_
             section_id=section.id,
             title=section.title,
             content=section.content,
-            normalized_text="家长 沟通 学习 动力 孩子 焦虑 倾听",
+            normalized_text=section_content,
             keywords='["家长","沟通","学习动力"]',
             synonyms='["父母","交流"]',
             source_start=0,
@@ -135,6 +157,48 @@ def test_course_question_without_catalog_does_not_fall_back_to_general_knowledge
         assert result.chunks == []
 
 
+def test_fixed_information_question_without_catalog_does_not_fall_back_to_general_knowledge():
+    with _database() as db:
+        result = asyncio.run(KnowledgeAgentService.build_result(db, question="这周的会议链接在哪里？"))
+        assert result.allow_general_knowledge is False
+        assert result.chunks == []
+
+
+def test_fixed_information_is_retrieved_first_and_overrides_other_types():
+    with _database() as db:
+        _add_published_knowledge(
+            db,
+            knowledge_id=1,
+            name="课程历史安排",
+            section_title="练习一：课程会议安排",
+            section_content="课程作业沿用旧安排，会议链接为旧链接，周二晚上上课。",
+            purpose="课程上课时间和会议安排",
+            applicable_questions="上课时间、会议链接",
+            core_topics="课程安排",
+        )
+        _add_published_knowledge(
+            db,
+            knowledge_id=2,
+            name="当前固定信息",
+            knowledge_type="fixed",
+            section_title="本周上课安排",
+            section_content="本周上课时间为周三晚上，会议链接为最新链接。",
+            purpose="保存当前上课时间、会议链接和服务安排",
+            applicable_questions="上课时间、会议链接、带练安排",
+            core_topics="当前安排",
+        )
+
+        result = asyncio.run(KnowledgeAgentService.build_result(db, question="课程作业和会议链接都有什么？"))
+
+        assert result.allow_general_knowledge is False
+        assert result.chunks[0].knowledge_type == "fixed"
+        assert result.chunks[0].knowledge_name == "当前固定信息"
+        assert "固定信息类·最高优先级" in result.prompt
+        assert "如与课程、答疑或通用知识库冲突，忽略其他库的冲突内容" in result.prompt
+        decision = next(item for item in result.tool_trace if item["tool"] == "agent_decision")
+        assert decision["selectedKnowledgeIds"][0] == 2
+
+
 def test_course_question_searches_chunk_and_reads_parent_section():
     with _database() as db:
         _add_published_knowledge(db, knowledge_id=1, name="亲子课程")