From 42f55ac4d1e8d5e7569baf0f46d188699d6d5e74 Mon Sep 17 00:00:00 2001
From: MerCry <jiong1114>
Date: Tue, 10 Mar 2026 12:11:55 +0800
Subject: [PATCH] chore: add utility scripts and tool definitions for KB search
 and metadata testing [AC-UTILS]

---
 ai-service/check_kb_collections.py            |  47 ++++
 ai-service/check_qdrant.py                    |  81 +++++++
 ai-service/test_kb_metadata_search.py         | 166 ++++++++++++++
 ai-service/test_kb_search_with_metadata.py    | 217 ++++++++++++++++++
 ai-service/test_metadata_filter_only.py       | 172 ++++++++++++++
 ai-service/tools/high_risk_check.md           |  67 ++++++
 ai-service/tools/intent_hint.md               |  60 +++++
 ai-service/tools/kb_search_dynamic.md         |  58 +++++
 .../tools/list_document_metadata_fields.md    |  76 ++++++
 ai-service/tools/memory_recall.md             |  67 ++++++
 10 files changed, 1011 insertions(+)
 create mode 100644 ai-service/check_kb_collections.py
 create mode 100644 ai-service/check_qdrant.py
 create mode 100644 ai-service/test_kb_metadata_search.py
 create mode 100644 ai-service/test_kb_search_with_metadata.py
 create mode 100644 ai-service/test_metadata_filter_only.py
 create mode 100644 ai-service/tools/high_risk_check.md
 create mode 100644 ai-service/tools/intent_hint.md
 create mode 100644 ai-service/tools/kb_search_dynamic.md
 create mode 100644 ai-service/tools/list_document_metadata_fields.md
 create mode 100644 ai-service/tools/memory_recall.md

diff --git a/ai-service/check_kb_collections.py b/ai-service/check_kb_collections.py
new file mode 100644
index 0000000..72c97f2
--- /dev/null
+++ b/ai-service/check_kb_collections.py
@@ -0,0 +1,47 @@
+"""
+检查知识库特定集合中的元数据
+"""
+import asyncio
+from app.core.qdrant_client import get_qdrant_client
+
+
+async def check_kb_collections():
+    client = await get_qdrant_client()
+    qdrant = await client.get_client()
+    
+    print("=" * 60)
+    print("检查知识库集合中的元数据")
+    print("=" * 60)
+    
+    # 检查所有集合
+    collections = await qdrant.get_collections()
+    
+    for col in collections.collections:
+        print(f"\n{'='*60}")
+        print(f"集合: {col.name}")
+        print('='*60)
+        
+        info = await qdrant.get_collection(col.name)
+        print(f"向量数: {info.points_count}")
+        
+        if info.points_count > 0:
+            results = await qdrant.scroll(
+                collection_name=col.name,
+                limit=3,
+                with_payload=True,
+            )
+            
+            for i, point in enumerate(results[0], 1):
+                payload = point.payload or {}
+                metadata = payload.get("metadata", {})
+                text = payload.get("text", "")[:80]
+                kb_id = payload.get("kb_id", "N/A")
+                
+                print(f"\n  [{i}] ID: {str(point.id)[:8]}...")
+                print(f"      KB ID: {kb_id}")
+                print(f"      元数据: {metadata if metadata else '(空)'}")
+                print(f"      内容: {text}...")
+
+
+if __name__ == "__main__":
+    asyncio.run(check_kb_collections())
diff --git a/ai-service/check_qdrant.py b/ai-service/check_qdrant.py
new file mode 100644
index 0000000..249330b
--- /dev/null
+++ b/ai-service/check_qdrant.py
@@ -0,0 +1,81 @@
+"""
+检查 Qdrant 中的集合和数据
+"""
+import asyncio
+from app.core.qdrant_client import get_qdrant_client
+
+
+async def check_qdrant():
+    client = await get_qdrant_client()
+    qdrant = await client.get_client()
+    
+    print("=" * 60)
+    print("Qdrant 状态检查")
+    print("=" * 60)
+    
+    # 1. 列出所有集合
+    collections = await qdrant.get_collections()
+    print(f"\n现有集合 ({len(collections.collections)}):")
+    for col in collections.collections:
+        print(f"  - {col.name}")
+    
+    # 2. 检查特定集合
+    tenant_id = "szmp@ash@2026"
+    collection_name = client.get_collection_name(tenant_id)
+    print(f"\n目标集合: {collection_name}")
+    
+    exists = await qdrant.collection_exists(collection_name)
+    print(f"集合存在: {exists}")
+    
+    if exists:
+        # 获取集合信息
+        info = await qdrant.get_collection(collection_name)
+        print(f"\n集合信息:")
+        print(f"  向量数: {info.points_count}")
+        vectors_config = info.config.params.vectors
+        if isinstance(vectors_config, dict):
+            print(f"  向量配置: {list(vectors_config.keys())}")
+        else:
+            print(f"  向量大小: {vectors_config.size}")
+        
+        # 获取一些样本点
+        if info.points_count > 0:
+            print(f"\n样本数据 (前3条):")
+            from qdrant_client.models import ScrollRequest
+            results = await qdrant.scroll(
+                collection_name=collection_name,
+                limit=3,
+                with_payload=True,
+            )
+            for i, point in enumerate(results[0], 1):
+                print(f"\n  [{i}] ID: {point.id}")
+                payload = point.payload or {}
+                metadata = payload.get("metadata", {})
+                print(f"      元数据: {metadata}")
+                print(f"      内容: {payload.get('text', '')[:100]}...")
+    else:
+        print("\n集合不存在，可能原因:")
+        print("  1. 还没有上传过文档")
+        print("  2. 集合名称格式不匹配")
+        print("\n尝试查找其他相关集合...")
+        
+        for col in collections.collections:
+            if tenant_id.replace("@", "_") in col.name or "szmp" in col.name:
+                print(f"\n找到相关集合: {col.name}")
+                info = await qdrant.get_collection(col.name)
+                print(f"  向量数: {info.points_count}")
+                
+                if info.points_count > 0:
+                    results = await qdrant.scroll(
+                        collection_name=col.name,
+                        limit=2,
+                        with_payload=True,
+                    )
+                    for point in results[0]:
+                        payload = point.payload or {}
+                        metadata = payload.get("metadata", {})
+                        print(f"  样本元数据: {metadata}")
+
+
+if __name__ == "__main__":
+    asyncio.run(check_qdrant())
diff --git a/ai-service/test_kb_metadata_search.py b/ai-service/test_kb_metadata_search.py
new file mode 100644
index 0000000..8dfa223
--- /dev/null
+++ b/ai-service/test_kb_metadata_search.py
@@ -0,0 +1,166 @@
+"""
+测试KB元数据过滤查询
+"""
+import asyncio
+import json
+from app.core.database import async_session_maker
+from app.services.mid.metadata_filter_builder import MetadataFilterBuilder
+from app.services.mid.default_kb_tool_runner import DefaultKbToolRunner
+from app.core.qdrant_client import get_qdrant_client
+
+
+async def test_metadata_filter():
+    """测试元数据过滤器构建"""
+    tenant_id = "szmp@ash@2026"
+    
+    # 测试上下文 - 模拟用户查询"初二数学痛点"
+    test_context = {
+        "grade": "初二",
+        "subject": "通用",
+        "kb_scene": "痛点"
+    }
+    
+    print("=" * 60)
+    print("测试元数据过滤器构建")
+    print("=" * 60)
+    print(f"租户: {tenant_id}")
+    print(f"查询上下文: {json.dumps(test_context, ensure_ascii=False)}")
+    print()
+    
+    async with async_session_maker() as session:
+        # 1. 测试过滤器构建
+        filter_builder = MetadataFilterBuilder(session)
+        result = await filter_builder.build_filter(tenant_id, test_context)
+        
+        print("过滤器构建结果:")
+        print(f"  成功: {result.success}")
+        print(f"  应用的过滤器: {json.dumps(result.applied_filter, ensure_ascii=False, indent=2)}")
+        print(f"  缺失的必填字段: {result.missing_required_slots}")
+        print(f"  调试信息: {json.dumps(result.debug_info, ensure_ascii=False, indent=2)}")
+        print()
+        
+        # 2. 获取可过滤字段列表
+        filter_schema = await filter_builder.get_filter_schema(tenant_id)
+        print("可过滤字段配置:")
+        for field in filter_schema:
+            print(f"  - {field['field_key']}: {field['label']} (类型: {field['type']}, 必填: {field['required']})")
+            if field['options']:
+                print(f"    选项: {field['options']}")
+        print()
+
+
+async def test_kb_search():
+    """测试KB向量检索（带元数据过滤）"""
+    tenant_id = "szmp@ash@2026"
+    kb_id = "your_kb_id"  # 需要替换为实际的知识库ID
+    
+    # 测试查询
+    query = "初二学生数学学习有什么困难"
+    
+    # 测试上下文
+    context = {
+        "grade": "初二",
+        "subject": "数学",
+        "kb_scene": "痛点"
+    }
+    
+    print("=" * 60)
+    print("测试KB向量检索（带元数据过滤）")
+    print("=" * 60)
+    print(f"租户: {tenant_id}")
+    print(f"知识库: {kb_id}")
+    print(f"查询: {query}")
+    print(f"上下文: {json.dumps(context, ensure_ascii=False)}")
+    print()
+    
+    async with async_session_maker() as session:
+        # 1. 先构建过滤器
+        filter_builder = MetadataFilterBuilder(session)
+        filter_result = await filter_builder.build_filter(tenant_id, context)
+        
+        print(f"过滤器: {json.dumps(filter_result.applied_filter, ensure_ascii=False)}")
+        print()
+        
+        # 执行检索 - 使用更长的超时时间
+        from app.services.mid.timeout_governor import TimeoutGovernor
+        from app.services.mid.default_kb_tool_runner import KbToolConfig
+        
+        config = KbToolConfig(
+            enabled=True,
+            top_k=5,
+            timeout_ms=10000,  # 10秒超时
+            min_score_threshold=0.5,
+        )
+        kb_runner = DefaultKbToolRunner(
+            timeout_governor=TimeoutGovernor(),
+            config=config,
+        )
+        
+        # 获取可用的KB列表
+        from app.services.knowledge_base_service import KnowledgeBaseService
+        kb_service = KnowledgeBaseService(session)
+        kbs = await kb_service.list_knowledge_bases(tenant_id)
+        
+        if not kbs:
+            print("未找到知识库，请先创建知识库并上传文档")
+            return
+            
+        print(f"找到 {len(kbs)} 个知识库:")
+        for kb in kbs:
+            print(f"  - {kb.name} (ID: {kb.id})")
+        print()
+        
+        # 使用第一个知识库进行测试
+        test_kb_id = str(kbs[0].id)
+        print(f"使用知识库: {kbs[0].name} (ID: {test_kb_id})")
+        print()
+        
+        # 执行检索
+        result = await kb_runner.execute(
+            tenant_id=tenant_id,
+            query=query,
+            metadata_filter=filter_result.applied_filter
+        )
+        
+        print("检索结果:")
+        print(f"  成功: {result.success}")
+        print(f"  命中数: {len(result.hits)}")
+        print(f"  回退原因: {result.fallback_reason_code}")
+        print()
+        
+        if result.hits:
+            print("命中文档:")
+            for i, hit in enumerate(result.hits, 1):
+                print(f"\n  [{i}] 分数: {hit.score:.4f}")
+                print(f"      内容: {hit.text[:200]}...")
+                print(f"      元数据: {json.dumps(hit.metadata, ensure_ascii=False)}")
+        else:
+            print("未命中任何文档")
+            print("\n可能原因:")
+            print("  1. 知识库中没有匹配的文档")
+            print("  2. 元数据过滤器过于严格")
+            print("  3. 向量相似度阈值过高")
+
+
+async def main():
+    print("\n" + "=" * 60)
+    print("KB元数据过滤查询测试")
+    print("=" * 60 + "\n")
+    
+    try:
+        # 测试1: 过滤器构建
+        await test_metadata_filter()
+        
+        print("\n" + "=" * 60 + "\n")
+        
+        # 测试2: 向量检索
+        await test_kb_search()
+        
+    except Exception as e:
+        print(f"\n测试失败: {e}")
+        import traceback
+        print(traceback.format_exc())
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/ai-service/test_kb_search_with_metadata.py b/ai-service/test_kb_search_with_metadata.py
new file mode 100644
index 0000000..cd83700
--- /dev/null
+++ b/ai-service/test_kb_search_with_metadata.py
@@ -0,0 +1,217 @@
+"""
+测试KB元数据过滤查询（使用正确的知识库集合）
+"""
+import asyncio
+import json
+from app.core.database import async_session_maker
+from app.core.qdrant_client import get_qdrant_client
+from app.services.embedding import get_embedding_provider
+
+
+async def test_kb_search_with_filter():
+    """测试带元数据过滤的KB检索"""
+    tenant_id = "szmp@ash@2026"
+    kb_id = "30c19c84-8f69-4768-9d23-7f4a5bc3627a"  # 客服咨询知识库
+    
+    query = "初二学生学习困难"
+    
+    # 测试过滤器
+    metadata_filter = {
+        "grade": {"$eq": "初二"},
+        "kb_scene": {"$eq": "痛点"}
+    }
+    
+    print("=" * 60)
+    print("测试带元数据过滤的KB检索")
+    print("=" * 60)
+    print(f"租户: {tenant_id}")
+    print(f"知识库: {kb_id}")
+    print(f"查询: {query}")
+    print(f"过滤器: {json.dumps(metadata_filter, ensure_ascii=False)}")
+    print()
+    
+    # 1. 获取 embedding
+    print("[1] 生成查询向量...")
+    embedding_provider = await get_embedding_provider()
+    query_vector = await embedding_provider.embed(query)
+    print(f"    向量维度: {len(query_vector)}")
+    print()
+    
+    # 2. 搜索 Qdrant
+    print("[2] 搜索 Qdrant...")
+    client = await get_qdrant_client()
+    
+    # 使用知识库特定集合
+    collection_name = client.get_kb_collection_name(tenant_id, kb_id)
+    print(f"    集合: {collection_name}")
+    
+    # 先获取所有数据（不带过滤）
+    all_results = await client.search(
+        tenant_id=tenant_id,  # 这里会被转换为集合名
+        query_vector=query_vector,
+        limit=10,
+        score_threshold=0.01,
+    )
+    print(f"    原始命中: {len(all_results)} 条")
+    print()
+    
+    # 3. 应用元数据过滤
+    print("[3] 应用元数据过滤...")
+    filtered_results = []
+    for hit in all_results:
+        payload = hit.get("payload", {})
+        hit_metadata = payload.get("metadata", {})
+        
+        match = True
+        for field_key, condition in metadata_filter.items():
+            hit_value = hit_metadata.get(field_key)
+            
+            if isinstance(condition, dict):
+                if "$eq" in condition:
+                    if hit_value != condition["$eq"]:
+                        match = False
+                        break
+                elif "$in" in condition:
+                    if hit_value not in condition["$in"]:
+                        match = False
+                        break
+            else:
+                if hit_value != condition:
+                    match = False
+                    break
+        
+        if match:
+            filtered_results.append(hit)
+    
+    print(f"    过滤后命中: {len(filtered_results)} 条")
+    print()
+    
+    # 4. 显示结果
+    print("=" * 60)
+    print("检索结果")
+    print("=" * 60)
+    
+    if filtered_results:
+        for i, hit in enumerate(filtered_results, 1):
+            payload = hit.get("payload", {})
+            metadata = payload.get("metadata", {})
+            text = payload.get("text", "")
+            score = hit.get("score", 0)
+            
+            print(f"\n[{i}] 相似度: {score:.4f}")
+            print(f"    年级: {metadata.get('grade', 'N/A')}")
+            print(f"    学科: {metadata.get('subject', 'N/A')}")
+            print(f"    场景: {metadata.get('kb_scene', 'N/A')}")
+            print(f"    内容: {text}")
+    else:
+        print("\n未命中任何文档")
+        print("\n可能原因:")
+        print("  1. 向量相似度太低")
+        print("  2. 元数据不匹配")
+        print("  3. 数据不在主集合中")
+        
+        # 显示原始命中（用于调试）
+        print("\n原始命中（未过滤前）:")
+        for i, hit in enumerate(all_results[:3], 1):
+            payload = hit.get("payload", {})
+            metadata = payload.get("metadata", {})
+            text = payload.get("text", "")[:60]
+            score = hit.get("score", 0)
+            print(f"  [{i}] 分数:{score:.3f} 元数据:{metadata} {text}...")
+
+
+async def test_different_filters():
+    """测试不同过滤条件"""
+    tenant_id = "szmp@ash@2026"
+    kb_id = "30c19c84-8f69-4768-9d23-7f4a5bc3627a"
+    
+    # 从 Qdrant 直接获取所有数据
+    client = await get_qdrant_client()
+    qdrant = await client.get_client()
+    collection_name = client.get_kb_collection_name(tenant_id, kb_id)
+    
+    print("\n" + "=" * 60)
+    print("测试不同过滤条件")
+    print("=" * 60)
+    print(f"集合: {collection_name}")
+    print()
+    
+    # 获取所有数据
+    results = await qdrant.scroll(
+        collection_name=collection_name,
+        limit=100,
+        with_payload=True,
+    )
+    
+    all_hits = []
+    for point in results[0]:
+        all_hits.append({
+            "id": str(point.id),
+            "payload": point.payload or {}
+        })
+    
+    print(f"总数据量: {len(all_hits)} 条")
+    print()
+    
+    # 显示所有数据的元数据
+    print("所有数据的元数据:")
+    for hit in all_hits:
+        payload = hit["payload"]
+        metadata = payload.get("metadata", {})
+        text = payload.get("text", "")[:50]
+        print(f"  - {metadata}: {text}...")
+    
+    # 测试各种过滤条件
+    test_cases = [
+        {"name": "初二", "filter": {"grade": {"$eq": "初二"}}},
+        {"name": "痛点", "filter": {"kb_scene": {"$eq": "痛点"}}},
+        {"name": "初二+痛点", "filter": {"grade": {"$eq": "初二"}, "kb_scene": {"$eq": "痛点"}}},
+        {"name": "通用学科", "filter": {"subject": {"$eq": "通用"}}},
+    ]
+    
+    print("\n过滤测试:")
+    for case in test_cases:
+        filter_def = case["filter"]
+        
+        # 应用过滤
+        filtered = []
+        for hit in all_hits:
+            payload = hit["payload"]
+            metadata = payload.get("metadata", {})
+            
+            match = True
+            for field_key, condition in filter_def.items():
+                hit_value = metadata.get(field_key)
+                
+                if isinstance(condition, dict) and "$eq" in condition:
+                    if hit_value != condition["$eq"]:
+                        match = False
+                        break
+                else:
+                    if hit_value != condition:
+                        match = False
+                        break
+            
+            if match:
+                filtered.append(hit)
+        
+        print(f"  {case['name']}: {len(filtered)}/{len(all_hits)} 条命中")
+
+
+async def main():
+    print("\n" + "=" * 60)
+    print("KB元数据过滤查询测试")
+    print("=" * 60 + "\n")
+    
+    try:
+        await test_kb_search_with_filter()
+        await test_different_filters()
+        
+    except Exception as e:
+        print(f"\n测试失败: {e}")
+        import traceback
+        print(traceback.format_exc())
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/ai-service/test_metadata_filter_only.py b/ai-service/test_metadata_filter_only.py
new file mode 100644
index 0000000..0dc84ae
--- /dev/null
+++ b/ai-service/test_metadata_filter_only.py
@@ -0,0 +1,172 @@
+"""
+测试元数据过滤逻辑（不依赖向量检索）
+"""
+import json
+
+
+def apply_metadata_filter(hits, metadata_filter):
+    """应用元数据过滤条件"""
+    filtered = []
+    for hit in hits:
+        payload = hit.get("payload", {})
+        hit_metadata = payload.get("metadata", {})
+        
+        match = True
+        for field_key, condition in metadata_filter.items():
+            hit_value = hit_metadata.get(field_key)
+            
+            if isinstance(condition, dict):
+                if "$eq" in condition:
+                    if hit_value != condition["$eq"]:
+                        match = False
+                        break
+                elif "$in" in condition:
+                    if hit_value not in condition["$in"]:
+                        match = False
+                        break
+            else:
+                if hit_value != condition:
+                    match = False
+                    break
+        
+        if match:
+            filtered.append(hit)
+    
+    return filtered
+
+
+# 模拟 Qdrant 返回的 hits（带 metadata）
+test_hits = [
+    {
+        "id": "1",
+        "score": 0.85,
+        "payload": {
+            "text": "初二数学学习困难：几何证明题难以理解",
+            "metadata": {
+                "grade": "初二",
+                "subject": "数学",
+                "kb_scene": "痛点"
+            }
+        }
+    },
+    {
+        "id": "2",
+        "score": 0.82,
+        "payload": {
+            "text": "初二英语学习：词汇量不足导致阅读理解困难",
+            "metadata": {
+                "grade": "初二",
+                "subject": "英语",
+                "kb_scene": "痛点"
+            }
+        }
+    },
+    {
+        "id": "3",
+        "score": 0.78,
+        "payload": {
+            "text": "初一数学基础：有理数运算不熟练",
+            "metadata": {
+                "grade": "初一",
+                "subject": "数学",
+                "kb_scene": "痛点"
+            }
+        }
+    },
+    {
+        "id": "4",
+        "score": 0.75,
+        "payload": {
+            "text": "初二数学学习方案：建立错题本",
+            "metadata": {
+                "grade": "初二",
+                "subject": "数学",
+                "kb_scene": "学习方案"
+            }
+        }
+    },
+]
+
+
+print("=" * 60)
+print("元数据过滤逻辑测试")
+print("=" * 60)
+print(f"\n测试数据: {len(test_hits)} 条文档")
+for hit in test_hits:
+    meta = hit["payload"]["metadata"]
+    print(f"  [{hit['id']}] {meta['grade']} {meta['subject']} {meta['kb_scene']}: {hit['payload']['text'][:30]}...")
+
+
+# 测试1: 过滤初二数学痛点
+print("\n" + "=" * 60)
+print("测试1: 过滤 '初二数学痛点'")
+print("=" * 60)
+filter1 = {
+    "grade": {"$eq": "初二"},
+    "subject": {"$eq": "数学"},
+    "kb_scene": {"$eq": "痛点"}
+}
+print(f"过滤器: {json.dumps(filter1, ensure_ascii=False)}")
+result1 = apply_metadata_filter(test_hits, filter1)
+print(f"结果: {len(result1)} 条命中")
+for hit in result1:
+    print(f"  - {hit['payload']['text']}")
+
+
+# 测试2: 过滤所有初二文档
+print("\n" + "=" * 60)
+print("测试2: 过滤 '初二' (所有学科)")
+print("=" * 60)
+filter2 = {
+    "grade": {"$eq": "初二"}
+}
+print(f"过滤器: {json.dumps(filter2, ensure_ascii=False)}")
+result2 = apply_metadata_filter(test_hits, filter2)
+print(f"结果: {len(result2)} 条命中")
+for hit in result2:
+    meta = hit["payload"]["metadata"]
+    print(f"  - [{meta['subject']}] {hit['payload']['text'][:40]}...")
+
+
+# 测试3: 过滤所有痛点
+print("\n" + "=" * 60)
+print("测试3: 过滤 '痛点' (所有年级学科)")
+print("=" * 60)
+filter3 = {
+    "kb_scene": {"$eq": "痛点"}
+}
+print(f"过滤器: {json.dumps(filter3, ensure_ascii=False)}")
+result3 = apply_metadata_filter(test_hits, filter3)
+print(f"结果: {len(result3)} 条命中")
+for hit in result3:
+    meta = hit["payload"]["metadata"]
+    print(f"  - [{meta['grade']} {meta['subject']}] {hit['payload']['text'][:40]}...")
+
+
+# 测试4: 空过滤器（不过滤）
+print("\n" + "=" * 60)
+print("测试4: 空过滤器（返回所有）")
+print("=" * 60)
+filter4 = {}
+print(f"过滤器: {json.dumps(filter4, ensure_ascii=False)}")
+result4 = apply_metadata_filter(test_hits, filter4)
+print(f"结果: {len(result4)} 条命中")
+
+
+# 测试5: 严格过滤（无匹配）
+print("\n" + "=" * 60)
+print("测试5: 严格过滤 '初三物理痛点'（无匹配）")
+print("=" * 60)
+filter5 = {
+    "grade": {"$eq": "初三"},
+    "subject": {"$eq": "物理"},
+    "kb_scene": {"$eq": "痛点"}
+}
+print(f"过滤器: {json.dumps(filter5, ensure_ascii=False)}")
+result5 = apply_metadata_filter(test_hits, filter5)
+print(f"结果: {len(result5)} 条命中")
+
+
+print("\n" + "=" * 60)
+print("测试完成！元数据过滤逻辑工作正常。")
+print("=" * 60)
diff --git a/ai-service/tools/high_risk_check.md b/ai-service/tools/high_risk_check.md
new file mode 100644
index 0000000..7079dd5
--- /dev/null
+++ b/ai-service/tools/high_risk_check.md
@@ -0,0 +1,67 @@
+---
+name: high_risk_check
+description: 高风险场景检测工具，检测退款、投诉、隐私承诺、转人工等高风险场景
+triggers:
+  - 用户消息可能涉及退款请求
+  - 用户表达投诉或升级意愿
+  - 用户要求敏感隐私承诺
+  - 用户明确要求转人工
+  - 用户情绪激动或表达不满
+anti_triggers:
+  - 已完成高风险判定且结果未变化
+  - 当前仅需知识检索，无风险迹象
+  - 用户问题与风险场景无关
+  - 已进入人工服务流程
+tools:
+  - high_risk_check
+---
+
+## 使用指南
+
+### 何时使用
+当用户消息可能涉及退款、投诉升级、隐私承诺、转人工等高风险场景时使用。
+
+### 何时不使用
+当已完成高风险判定且结果未变化，或当前仅需知识检索时不要重复调用。
+
+### 支持的高风险场景
+
+| 场景 | 标识 | 说明 |
+|-----|------|------|
+| 退款 | refund | 用户要求退款 |
+| 投诉升级 | complaint_escalation | 用户威胁投诉或升级 |
+| 隐私敏感承诺 | privacy_sensitive_promise | 用户要求敏感信息承诺 |
+| 转人工 | transfer | 用户要求转人工服务 |
+
+### 参数说明
+
+| 参数 | 类型 | 必填 | 说明 |
+|-----|------|-----|------|
+| message | string | 是 | 用户消息原文 |
+| tenant_id | string | 是 | 租户 ID（系统自动注入） |
+| domain | string | 否 | 业务域（可选） |
+| scene | string | 否 | 场景标识（可选） |
+| context | object | 否 | 上下文（仅 routing_signal 字段会被消费） |
+
+### 示例调用
+
+```json
+{
+  "message": "我要投诉你们并且现在就给我退款，不然我去12315",
+  "scene": "open_consult"
+}
+```
+
+### 结果解释
+
+- `matched=true`: 检测到高风险场景，优先按 `recommended_mode` 执行
+- `risk_scenario`: 匹配的风险场景类型
+- `rule_id`: 触发的规则 ID
+- `fallback_reason_code`: 降级原因码
+
+### 注意事项
+
+1. tenant_id 由系统自动注入，模型不要填写
+2. 该工具只消费 routing_signal 角色的字段
+3. 检测到高风险后应优先处理，不要继续常规流程
+4. 结果应作为重要参考，影响后续路由决策
diff --git a/ai-service/tools/intent_hint.md b/ai-service/tools/intent_hint.md
new file mode 100644
index 0000000..5873008
--- /dev/null
+++ b/ai-service/tools/intent_hint.md
@@ -0,0 +1,60 @@
+---
+name: intent_hint
+description: 轻量意图识别与路由建议工具，为策略路由提供软信号
+triggers:
+  - 用户意图不明确或存在歧义
+  - 需要给 policy_router 提供软路由信号
+  - 用户表达复杂，可能涉及多个意图
+  - 需要判断应该进入哪种处理模式
+anti_triggers:
+  - 已经明确进入固定模式/流程模式
+  - 已有确定的意图结果
+  - 用户问题简单明确，无需意图分析
+  - 已完成意图识别且上下文未变化
+tools:
+  - intent_hint
+---
+
+## 使用指南
+
+### 何时使用
+当用户意图不明确、需要给 policy_router 提供软路由信号时使用。
+
+### 何时不使用
+当已经明确进入固定模式/流程模式，或已有确定意图结果时不重复调用。
+
+### 参数说明
+
+| 参数 | 类型 | 必填 | 说明 |
+|-----|------|-----|------|
+| message | string | 是 | 用户输入原文 |
+| tenant_id | string | 是 | 租户 ID（系统自动注入） |
+| history | array | 否 | 会话历史（可选） |
+| top_n | integer | 否 | 返回建议数量（可选） |
+| context | object | 否 | 上下文字段（仅 routing_signal 字段会被消费） |
+
+### 示例调用
+
+```json
+{
+  "message": "我想退款，但是也想先咨询下怎么处理",
+  "top_n": 3,
+  "context": {"order_status": "delivered", "channel": "web"}
+}
+```
+
+### 结果解释
+
+关注输出中的以下字段：
+- `intent`: 识别出的意图
+- `confidence`: 置信度
+- `suggested_mode`: 建议的处理模式
+
+**重要**: 该工具只提供建议，不做最终决策。
+
+### 注意事项
+
+1. tenant_id 由系统自动注入，模型不要填写
+2. context 中只有 routing_signal 角色的字段会被消费
+3. 该工具是轻量级的，不会执行复杂推理
+4. 结果应作为参考，最终决策由 policy_router 做出
diff --git a/ai-service/tools/kb_search_dynamic.md b/ai-service/tools/kb_search_dynamic.md
new file mode 100644
index 0000000..7976379
--- /dev/null
+++ b/ai-service/tools/kb_search_dynamic.md
@@ -0,0 +1,58 @@
+---
+name: kb_search_dynamic
+description: 知识库动态检索工具，支持元数据驱动过滤
+triggers:
+  - 用户问题需要知识库事实支撑回答
+  - 需要按租户元数据动态过滤知识条目
+  - 涉及产品知识、政策条款、FAQ查询
+  - 用户询问具体业务规则或流程
+anti_triggers:
+  - 纯闲聊或问候语
+  - 仅流程确认（如"好的"、"明白了"）
+  - 已有充分 KB 结果且无需补充
+  - 用户问题与知识库内容无关
+tools:
+  - kb_search_dynamic
+---
+
+## 使用指南
+
+### 何时使用
+当需要知识库事实支撑回答，且需按租户元数据动态过滤时使用。
+
+### 何时不使用
+当用户问题不依赖知识库（纯闲聊/仅流程确认）或已有充分 KB 结果时不重复调用。
+
+### 参数说明
+
+| 参数 | 类型 | 必填 | 说明 |
+|-----|------|-----|------|
+| query | string | 是 | 检索查询文本 |
+| tenant_id | string | 是 | 租户 ID（系统自动注入） |
+| scene | string | 否 | 场景标识，如 open_consult |
+| top_k | integer | 否 | 返回条数，默认5 |
+| context | object | 否 | 上下文，用于动态过滤字段 |
+
+### 示例调用
+
+```json
+{
+  "query": "退款到账一般要多久",
+  "scene": "open_consult",
+  "top_k": 5,
+  "context": {"product_line": "vip_course", "region": "beijing"}
+}
+```
+
+### 结果解释
+
+- `success=true` 且 `hits` 非空：命中知识，可直接使用
+- `missing_required_slots` 非空：应先向用户补采信息
+- `fallback_reason_code` 存在：需降级处理
+
+### 注意事项
+
+1. tenant_id 由系统自动注入，模型不要填写
+2. context 字段用于传递动态过滤条件，如产品线、地区等
+3. 避免重复调用：如果上一轮已有充分结果，不要再次调用
+4. 查询文本应保持用户原意，不要过度改写
diff --git a/ai-service/tools/list_document_metadata_fields.md b/ai-service/tools/list_document_metadata_fields.md
new file mode 100644
index 0000000..31e7bf5
--- /dev/null
+++ b/ai-service/tools/list_document_metadata_fields.md
@@ -0,0 +1,76 @@
+---
+name: list_document_metadata_fields
+description: 列出当前知识库文档中使用的元数据字段及其常见取值，用于后续的知识库搜索过滤
+triggers:
+  - 需要了解知识库中有哪些可用的元数据过滤字段
+  - 需要知道某个字段有哪些可选值
+  - 构造知识库搜索请求前需要确定过滤条件
+  - 用户询问可以按什么条件筛选知识库内容
+anti_triggers:
+  - 已知可用的过滤字段
+  - 不需要元数据过滤
+  - 用户问题与知识库无关
+tools:
+  - list_document_metadata_fields
+---
+
+## 使用指南
+
+### 何时使用
+当需要了解知识库中有哪些可用的元数据过滤字段时使用。这个工具会返回当前文档中实际使用的元数据字段及其常见取值。
+
+### 何时不使用
+当已知可用的过滤字段，或不需要元数据过滤时不需要调用。
+
+### 参数说明
+
+| 参数 | 类型 | 必填 | 说明 |
+|-----|------|-----|------|
+| tenant_id | string | 否 | 租户 ID（系统自动注入） |
+| kb_id | string | 否 | 知识库 ID，用于限定范围 |
+| include_values | boolean | 否 | 是否包含常见值列表，默认 true |
+| top_n | integer | 否 | 每个字段返回的常见值数量，默认 10 |
+
+### 示例调用
+
+```json
+{
+  "include_values": true,
+  "top_n": 5
+}
+```
+
+### 结果解释
+
+返回结果包含：
+- `fields`: 字段列表，每个字段包含：
+  - `field_key`: 字段标识
+  - `field_type`: 字段类型（string/number/boolean/enum/array_enum）
+  - `label`: 字段显示名称
+  - `description`: 用途说明
+  - `common_values`: 常见取值列表
+  - `value_count`: 该字段在多少文档中出现
+  - `is_filterable`: 是否可用于过滤
+  - `options`: 预定义的可选值（如果有）
+- `total_documents`: 文档总数
+
+### 使用场景
+
+1. **构造搜索过滤条件前**
+   - 先调用此工具了解可用字段
+   - 根据返回的字段信息构造过滤条件
+
+2. **用户询问筛选条件**
+   - 用户问"可以按什么条件筛选？"
+   - 调用此工具返回可用字段
+
+3. **确定字段取值范围**
+   - 需要知道某个字段有哪些可选值
+   - 调用此工具获取 common_values
+
+### 注意事项
+
+1. tenant_id 由系统自动注入，模型不要填写
+2. 返回的字段是从实际文档中聚合得到的，反映了真实使用情况
+3. common_values 是从文档中统计的常见值，不是预定义的可选值
+4. 如果字段有预定义的 options，则只能使用 options 中的值
diff --git a/ai-service/tools/memory_recall.md b/ai-service/tools/memory_recall.md
new file mode 100644
index 0000000..af5d4f9
--- /dev/null
+++ b/ai-service/tools/memory_recall.md
@@ -0,0 +1,67 @@
+---
+name: memory_recall
+description: 记忆召回工具，读取用户可用记忆包（profile/facts/preferences/summary/slots)
+triggers:
+  - 需要补全用户画像信息
+  - 需要获取用户历史事实
+  - 需要了解用户偏好设置
+  - 需要获取用户槽位信息
+  - 避免重复追问用户已知信息
+anti_triggers:
+  - 当前轮次已有完整上下文且无需个性化记忆支撑
+  - 用户首次对话，无历史记忆
+  - 当前任务不依赖用户历史信息
+tools:
+  - memory_recall
+---
+
+## 使用指南
+
+### 何时使用
+当需要补全用户画像、历史事实、偏好、槽位，避免重复追问时使用。
+
+### 何时不使用
+当当前轮次已经有完整上下文且无需个性化记忆支撑时可不调用。
+
+### 召回范围
+
+| 范围 | 说明 |
+|-----|------|
+| profile | 用户画像信息 |
+| facts | 用户历史事实 |
+| preferences | 用户偏好设置 |
+| summary | 会话摘要 |
+| slots | 用户槽位信息 |
+
+### 参数说明
+
+| 参数 | 类型 | 必填 | 说明 |
+|-----|------|-----|------|
+| tenant_id | string | 否 | 租户 ID（系统自动注入) |
+| user_id | string | 否 | 用户 ID |
+| session_id | string | 否 | 会话 ID |
+| recall_scope | array | 否 | 召回范围，如 ["profile", "facts"] |
+| max_recent_messages | integer | 否 | 历史回填窗口大小 |
+
+### 示例调用
+```json
+{
+  "recall_scope": ["profile", "facts", "preferences", "summary", "slots"],
+}
+```
+
+### 结果解释
+关注以下字段：
+- `profile`: 用户画像信息
+- `facts`: 用户历史事实
+- `preferences`: 用户偏好
+- `slots`: 用户槽位
+- `missing_slots`: 缺失的槽位
+
+若 `fallback_reason_code` 存在，需降级处理。
+
+### 注意事项
+1. tenant_id 由系统自动注入，模型不要填写
+2. recall_scope 可根据需要选择范围，不必全部召回
+3. 召回的记忆信息应用于个性化回复，避免重复追问
+4. 如果用户信息已完整，可以不调用此工具