diff --git a/ai_knowledge_base_v2/apps/backend/app/services/knowledge_agent_service.py b/ai_knowledge_base_v2/apps/backend/app/services/knowledge_agent_service.py index 516ed66..9c061b2 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/knowledge_agent_service.py +++ b/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: diff --git a/ai_knowledge_base_v2/apps/backend/app/services/rag_service.py b/ai_knowledge_base_v2/apps/backend/app/services/rag_service.py index 3e12d68..6c28ac2 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/rag_service.py +++ b/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: - if not any(chunk.knowledge_type == "fixed" for chunk in chunks): + def _knowledge_type_rules(chunks: list[RetrievedChunk]) -> str: + types = {chunk.knowledge_type for chunk in chunks} + if not types: return "" - return ( - "\n[固定信息优先规则]\n" - "本轮召回了固定信息类知识库。上课时间、带练安排、回放和音频、课程作业、会议链接、服务权益、" - "课程助理联系方式及其他当前安排，必须以固定信息类知识库中的最新生效内容为准。" - "如与课程、答疑或通用知识库冲突，忽略其他库的冲突内容；多个固定信息来源冲突时，以生效版本时间较新的内容为准。" - ) + rules = [ + "\n[知识库类型使用规则]\n", + "课程知识库用于课程定义、方法、步骤和老师观点；答疑知识库用于常见问题的直接解答，但不能覆盖课程原文；" + "通用知识库仅作背景补充，不能覆盖课程知识库和答疑知识库。", + ] + if "fixed" in types: + rules.append( + "本轮召回了固定信息类知识库。上课时间、带练安排、回放和音频、课程作业、会议链接、服务权益、" + "课程助理联系方式及其他当前安排，必须以固定信息类知识库中的最新生效内容为准。" + "如与课程、答疑或通用知识库冲突，忽略其他库的冲突内容；多个固定信息来源冲突时，以生效版本时间较新的内容为准。" + ) + return "".join(rules) @staticmethod def render_messages(messages: list[dict[str, str]]) -> str: diff --git a/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_agent.py b/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_agent.py index 359e6f9..bdf53cb 100644 --- a/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_agent.py +++ b/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(