feat: expose knowledge type priority traces

ai_knowledge_base_v2/apps/backend/app/api/admin_agent_records.py

@@ -14,6 +14,7 @@ from app.models.admin import Admin
 from app.models.knowledge import (
     HumanAttentionHistory,
     HumanAttentionRecord,
+    Knowledge,
     KnowledgeRetrievalCandidate,
     KnowledgeRetrievalLog,
 )
@@ -113,11 +114,19 @@ def retrieval_detail(
         .where(KnowledgeRetrievalCandidate.retrieval_log_id == log.id)
         .order_by(KnowledgeRetrievalCandidate.selected.desc(), KnowledgeRetrievalCandidate.id)
     ).all()
+    knowledge_ids = {item.knowledge_id for item in candidates}
+    knowledge_map = {
+        item.id: item
+        for item in db.scalars(select(Knowledge).where(Knowledge.id.in_(knowledge_ids))).all()
+    } if knowledge_ids else {}
     data = _retrieval_dict(log, detail=True)
     data["candidates"] = [
         {
             "id": item.id,
             "knowledgeId": item.knowledge_id,
+            "knowledgeName": knowledge_map[item.knowledge_id].name if item.knowledge_id in knowledge_map else "",
+            "knowledgeType": knowledge_map[item.knowledge_id].knowledge_type if item.knowledge_id in knowledge_map else "general",
+            "knowledgeTypeLabel": _knowledge_type_label(knowledge_map[item.knowledge_id].knowledge_type if item.knowledge_id in knowledge_map else "general"),
             "versionId": item.version_id,
             "chunkId": item.chunk_id,
             "sectionId": item.section_id,
@@ -131,6 +140,15 @@ def retrieval_detail(
     return api_success(data)
 
 
+def _knowledge_type_label(value: str) -> str:
+    return {
+        "fixed": "固定信息类·最高优先级",
+        "course": "课程知识库",
+        "qa": "答疑知识库",
+        "general": "通用知识库",
+    }.get(value, value or "通用知识库")
+
+
 @router.get("/attention/list")
 def attention_list(
     priority: str = Query(default=""),

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

@@ -154,7 +154,7 @@ class KnowledgeAgentService:
                     candidate_limit=candidate_limit,
                     prefer_fixed_information=cls._is_fixed_information_query(retrieval_question),
                 )
-                trace.append(cls._trace("search_knowledge", len(trace) + 1, {"queryTerms": terms, "knowledgeIds": selected_ids, "candidateLimit": candidate_limit}, {"candidateCount": len(candidates), "candidates": [cls._candidate_trace(x) for x in candidates]}, started))
+                trace.append(cls._trace("search_knowledge", len(trace) + 1, {"queryTerms": terms, "knowledgeIds": selected_ids, "candidateLimit": candidate_limit, "preferFixedInformation": cls._is_fixed_information_query(retrieval_question)}, {"candidateCount": len(candidates), "candidates": [cls._candidate_trace(x) for x in candidates]}, started))
                 await cls._rerank(db, retrieval_question, candidates, trace, started)
                 selected = cls._select_sections(
                     candidates,
@@ -391,7 +391,7 @@ class KnowledgeAgentService:
 
     @staticmethod
     def _candidate_trace(item: Candidate) -> dict:
-        return {"knowledgeId": item.knowledge.id, "knowledgeName": item.knowledge.name, "versionId": item.version.id, "chunkId": item.chunk.id, "sectionId": item.section.id, "title": item.chunk.title, "lexicalScore": item.lexical_score, "rerankScore": item.rerank_score, "selected": item.selected, "discardReason": item.discard_reason}
+        return {"knowledgeId": item.knowledge.id, "knowledgeName": item.knowledge.name, "knowledgeType": item.knowledge.knowledge_type, "knowledgeTypePriority": _knowledge_type_priority_label(item.knowledge.knowledge_type), "versionId": item.version.id, "chunkId": item.chunk.id, "sectionId": item.section.id, "title": item.chunk.title, "lexicalScore": item.lexical_score, "rerankScore": item.rerank_score, "selected": item.selected, "discardReason": item.discard_reason}
 
     @classmethod
     def search_knowledge(
@@ -732,6 +732,15 @@ def _json_trace_value(value):
     return str(value)
 
 
+def _knowledge_type_priority_label(value: str) -> str:
+    return {
+        "fixed": "最高优先级：固定信息类冲突时优先",
+        "course": "课程内容优先于答疑和通用知识",
+        "qa": "过往答疑参考，不覆盖课程原文",
+        "general": "背景补充，不覆盖业务知识",
+    }.get(value, "背景补充")
+
+
 def _heading_level(content: str) -> int | None:
     match = re.match(r"^(#{1,6})\s+", content.lstrip())
     return len(match.group(1)) if match else None