MerCry
/
ai-robot-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: add utility scripts and tool definitions for KB search and metadata testing [AC-UTILS]

This commit is contained in:
MerCry 2026-03-10 12:11:55 +08:00
parent 3b354ba041
commit 42f55ac4d1
10 changed files with 1011 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
ai-service/check_kb_collections.py Normal file
View File

@ -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())

81
ai-service/check_qdrant.py Normal file
View File

@ -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())

166
ai-service/test_kb_metadata_search.py Normal file
View File

@ -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())

217
ai-service/test_kb_search_with_metadata.py Normal file
View File

@ -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())

172
ai-service/test_metadata_filter_only.py Normal file
View File

@ -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)

67
ai-service/tools/high_risk_check.md Normal file
View File

@ -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. 结果应作为重要参考，影响后续路由决策

60
ai-service/tools/intent_hint.md Normal file
View File

@ -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 做出

58
ai-service/tools/kb_search_dynamic.md Normal file
View File

@ -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. 查询文本应保持用户原意，不要过度改写

76
ai-service/tools/list_document_metadata_fields.md Normal file
View File

@ -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 中的值

67
ai-service/tools/memory_recall.md Normal file
View File

@ -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. 如果用户信息已完整，可以不调用此工具