知识库¶
知识库是 Unifiles 三层架构的第三层,负责将提取的内容组织、索引并提供语义搜索能力。这是构建 RAG(检索增强生成)应用的核心。
知识库概念¶
- 知识库:文档的逻辑集合,每个知识库可以有独立的分块策略和搜索配置
- 文档:已提取内容的引用,一个文件可以添加到多个知识库
- 分块:文档被切分成的语义单元,用于精确检索
- 嵌入:分块的向量表示,支持语义相似度搜索
SDK 使用¶
创建知识库¶
from unifiles import UnifilesClient
client = UnifilesClient(api_key="sk_...")
# 创建知识库
kb = client.knowledge_bases.create(
name="legal-docs",
description="法律文档知识库",
chunking_strategy={
"type": "semantic", # fixed | semantic | paragraph
"chunk_size": 512, # 目标分块大小(token数)
"overlap": 50 # 重叠token数
},
embedding_model="text-embedding-3-small" # 可选,默认使用系统配置
)
print(f"知识库已创建: {kb.id}")
print(f"名称: {kb.name}")
列出知识库¶
# 获取所有知识库
kbs = client.knowledge_bases.list(limit=50)
for kb in kbs.items:
print(f"{kb.id}: {kb.name}")
print(f" 文档数: {kb.document_count}")
print(f" 总分块: {kb.chunk_count}")
获取知识库详情¶
kb = client.knowledge_bases.get(kb_id)
print(f"名称: {kb.name}")
print(f"描述: {kb.description}")
print(f"分块策略: {kb.chunking_strategy}")
print(f"文档数量: {kb.document_count}")
print(f"分块数量: {kb.chunk_count}")
print(f"创建时间: {kb.created_at}")
更新知识库¶
删除知识库¶
文档管理¶
添加文档到知识库¶
# 添加已提取的文件到知识库
doc = client.knowledge_bases.documents.create(
kb_id=kb.id,
file_id=file.id, # 必须是已完成提取的文件
title="合同模板v2.0", # 可选,自定义标题
metadata={ # 可选,自定义元数据
"category": "contract",
"version": "2.0",
"department": "legal"
}
)
print(f"文档已添加: {doc.id}")
print(f"状态: {doc.status}") # pending -> indexing -> indexed
等待索引完成¶
# 等待文档索引完成
doc = doc.wait(timeout=300)
if doc.status == "indexed":
print(f"索引完成!")
print(f"分块数量: {doc.chunk_count}")
else:
print(f"索引失败: {doc.error}")
列出知识库中的文档¶
docs = client.knowledge_bases.documents.list(
kb_id=kb.id,
limit=50
)
for doc in docs.items:
print(f"{doc.id}: {doc.title}")
print(f" 状态: {doc.status}")
print(f" 分块数: {doc.chunk_count}")
获取文档详情¶
doc = client.knowledge_bases.documents.get(
kb_id=kb.id,
document_id=doc.id
)
print(f"标题: {doc.title}")
print(f"文件ID: {doc.file_id}")
print(f"分块数: {doc.chunk_count}")
print(f"元数据: {doc.metadata}")
删除文档¶
搜索¶
语义搜索¶
语义搜索使用向量相似度匹配,理解查询的语义而非仅仅关键词:
results = client.knowledge_bases.search(
kb_id=kb.id,
query="违约条款有哪些?",
top_k=5, # 返回最相关的5个结果
threshold=0.7 # 相似度阈值(0-1)
)
print(f"找到 {len(results.chunks)} 个相关片段")
for chunk in results.chunks:
print(f"\n相似度: {chunk.score:.2f}")
print(f"来源: {chunk.document_title}")
print(f"内容: {chunk.content[:200]}...")
混合搜索¶
结合语义搜索和关键词搜索,提高召回率:
results = client.knowledge_bases.hybrid_search(
kb_id=kb.id,
query="违约金计算方法",
vector_weight=0.7, # 语义搜索权重
keyword_weight=0.3, # 关键词搜索权重
top_k=10
)
for chunk in results.chunks:
print(f"综合得分: {chunk.score:.2f}")
print(f"语义得分: {chunk.vector_score:.2f}")
print(f"关键词得分: {chunk.keyword_score:.2f}")
print(f"内容: {chunk.content[:200]}...")
带过滤的搜索¶
使用元数据过滤搜索结果:
results = client.knowledge_bases.search(
kb_id=kb.id,
query="合同条款",
top_k=10,
filter={
"metadata.category": "contract",
"metadata.version": {"$gte": "2.0"}
}
)
支持的过滤操作符:
| 操作符 | 说明 | 示例 |
|---|---|---|
$eq |
等于(默认) | {"field": "value"} |
$ne |
不等于 | {"field": {"$ne": "value"}} |
$gt |
大于 | {"field": {"$gt": 10}} |
$gte |
大于等于 | {"field": {"$gte": 10}} |
$lt |
小于 | {"field": {"$lt": 10}} |
$lte |
小于等于 | {"field": {"$lte": 10}} |
$in |
在列表中 | {"field": {"$in": ["a", "b"]}} |
REST API¶
知识库操作¶
创建知识库:
POST /v1/knowledge-bases
Authorization: Bearer sk_...
Content-Type: application/json
{
"name": "legal-docs",
"description": "法律文档知识库",
"chunking_strategy": {
"type": "semantic",
"chunk_size": 512,
"overlap": 50
}
}
响应:
{
"id": "kb_abc123",
"name": "legal-docs",
"description": "法律文档知识库",
"chunking_strategy": {
"type": "semantic",
"chunk_size": 512,
"overlap": 50
},
"document_count": 0,
"chunk_count": 0,
"created_at": "2024-01-15T10:30:00Z"
}
列出知识库:
获取知识库:
更新知识库:
PATCH /v1/knowledge-bases/{kb_id}
Authorization: Bearer sk_...
Content-Type: application/json
{
"name": "legal-docs-v2",
"description": "更新后的描述"
}
删除知识库:
文档操作¶
添加文档:
POST /v1/knowledge-bases/{kb_id}/documents
Authorization: Bearer sk_...
Content-Type: application/json
{
"file_id": "file_xyz789",
"title": "合同模板v2.0",
"metadata": {
"category": "contract"
}
}
响应:
{
"id": "doc_def456",
"kb_id": "kb_abc123",
"file_id": "file_xyz789",
"title": "合同模板v2.0",
"status": "pending",
"metadata": {
"category": "contract"
},
"created_at": "2024-01-15T10:35:00Z"
}
列出文档:
获取文档:
删除文档:
搜索操作¶
语义搜索:
POST /v1/knowledge-bases/{kb_id}/search
Authorization: Bearer sk_...
Content-Type: application/json
{
"query": "违约条款有哪些?",
"top_k": 5,
"threshold": 0.7
}
响应:
{
"query": "违约条款有哪些?",
"chunks": [
{
"id": "chunk_001",
"document_id": "doc_def456",
"document_title": "合同模板v2.0",
"content": "第十条 违约责任\n\n1. 甲方违约...",
"score": 0.92,
"metadata": {
"page": 5,
"section": "第十条"
}
}
],
"total": 1
}
混合搜索:
POST /v1/knowledge-bases/{kb_id}/hybrid-search
Authorization: Bearer sk_...
Content-Type: application/json
{
"query": "违约金计算方法",
"vector_weight": 0.7,
"keyword_weight": 0.3,
"top_k": 10
}
分块策略¶
选择合适的分块策略对 RAG 效果至关重要。
fixed(固定大小分块)¶
按固定 token 数切分,简单高效:
kb = client.knowledge_bases.create(
name="my-kb",
chunking_strategy={
"type": "fixed",
"chunk_size": 512, # 每块512 tokens
"overlap": 50 # 重叠50 tokens
}
)
适用场景:
- 结构不规则的文档
- 需要快速索引
- 对精确边界要求不高
semantic(语义分块)¶
基于语义边界智能切分:
kb = client.knowledge_bases.create(
name="my-kb",
chunking_strategy={
"type": "semantic",
"chunk_size": 512,
"overlap": 50
}
)
适用场景:
- 结构化文档(合同、论文)
- 需要保持语义完整性
- 对搜索质量要求高
paragraph(段落分块)¶
按段落边界切分:
kb = client.knowledge_bases.create(
name="my-kb",
chunking_strategy={
"type": "paragraph",
"max_chunk_size": 1024, # 最大块大小
"min_chunk_size": 100 # 最小块大小
}
)
适用场景:
- 段落清晰的文档
- 需要保持段落完整
- 文档格式一致
分块策略对比¶
| 策略 | 速度 | 语义完整性 | 适用场景 |
|---|---|---|---|
| fixed | 最快 | 一般 | 大规模快速索引 |
| semantic | 中等 | 最好 | 高质量问答 |
| paragraph | 快 | 好 | 结构清晰的文档 |
搜索最佳实践¶
1. 选择合适的 top_k¶
# 简单问答:较少结果
results = client.knowledge_bases.search(
kb_id=kb.id,
query="公司注册资本是多少?",
top_k=3
)
# 复杂分析:更多结果
results = client.knowledge_bases.search(
kb_id=kb.id,
query="总结所有违约条款",
top_k=10
)
2. 使用阈值过滤低相关结果¶
# 高精度场景:较高阈值
results = client.knowledge_bases.search(
kb_id=kb.id,
query="具体条款",
threshold=0.8 # 只返回高相关结果
)
# 广泛探索:较低阈值
results = client.knowledge_bases.search(
kb_id=kb.id,
query="相关内容",
threshold=0.5 # 返回更多可能相关的结果
)
3. 混合搜索提高召回率¶
# 当语义搜索可能遗漏关键词时,使用混合搜索
results = client.knowledge_bases.hybrid_search(
kb_id=kb.id,
query="ISO9001认证", # 包含专有名词
vector_weight=0.6,
keyword_weight=0.4 # 提高关键词权重
)
4. 利用元数据过滤¶
# 按部门过滤
results = client.knowledge_bases.search(
kb_id=kb.id,
query="审批流程",
filter={"metadata.department": "finance"}
)
# 按时间范围过滤
results = client.knowledge_bases.search(
kb_id=kb.id,
query="政策变更",
filter={
"metadata.year": {"$gte": 2023}
}
)
完整工作流示例¶
from unifiles import UnifilesClient
client = UnifilesClient(api_key="sk_...")
# 1. 创建知识库
kb = client.knowledge_bases.create(
name="company-docs",
description="公司文档知识库",
chunking_strategy={
"type": "semantic",
"chunk_size": 512,
"overlap": 50
}
)
# 2. 上传并提取文件
files_to_process = ["policy.pdf", "handbook.docx", "faq.md"]
for path in files_to_process:
# 上传
file = client.files.upload(path)
# 提取
extraction = client.extractions.create(file_id=file.id)
extraction.wait()
# 添加到知识库
doc = client.knowledge_bases.documents.create(
kb_id=kb.id,
file_id=file.id
)
doc.wait()
print(f"已处理: {path}")
# 3. 搜索
results = client.knowledge_bases.search(
kb_id=kb.id,
query="年假政策是什么?",
top_k=5
)
# 4. 使用搜索结果(例如传给LLM)
context = "\n\n".join([
f"【{chunk.document_title}】\n{chunk.content}"
for chunk in results.chunks
])
print("检索到的上下文:")
print(context)
下一步¶
- Webhooks - 配置索引完成等事件的通知
- LangChain 集成 - 将知识库接入 LangChain
- 分块策略详解 - 深入了解分块优化