知识库检索增强生成系统

系统架构

代码结构

核心代码示例

1. 强化元数据处理

# 文档标准化元数据处理
# 统一元数据标准化处理
for doc in docs:
    if not hasattr(doc, 'metadata') or not doc.metadata:
        doc.metadata = {}
        
    # 确保关键元数据字段存在且格式一致
    doc.metadata['source'] = norm_file_path  # 统一使用正斜杠
    doc.metadata['filename'] = safe_filename
    doc.metadata['file_type'] = file_extension[1:]  # 移除前导点
    doc.metadata['doc_id'] = str(uuid.uuid4())  # 为每个文档分配唯一ID
    doc.metadata['timestamp'] = int(time.time())
    
    # 对不同类型的文档添加特定元数据
    if 'page' not in doc.metadata and file_extension == '.pdf':
        doc.metadata['page'] = 1  # 默认页码

2. 批处理文档添加

# 将文档添加到集合中 - 使用批处理避免超时
batch_size = 16  # 批处理大小限制
total_batches = (len(texts) + batch_size - 1) // batch_size

print(f"开始批量添加文档，共 {len(texts)} 个文本块，分 {total_batches} 批处理")

for i in range(total_batches):
    start_idx = i * batch_size
    end_idx = min(start_idx + batch_size, len(texts))
    
    batch_texts = texts[start_idx:end_idx]
    batch_metadatas = metadatas[start_idx:end_idx]
    batch_ids = ids[start_idx:end_idx]
    
    print(f"添加第 {i+1}/{total_batches} 批文档，包含 {len(batch_texts)} 个文本块")
    collection.add(
        documents=batch_texts,
        metadatas=batch_metadatas,
        ids=batch_ids
    )

3. 知识库健康检查

# 检查知识库健康状态
def check_knowledge_base_health():
    """检查知识库健康状态并修复可能的问题"""
    try:
        if not os.path.exists("./chroma_db"):
            return {"status": "error", "message": "知识库不存在"}

        client = chromadb.PersistentClient(path="./chroma_db")
        
        # 获取所有IDs
        all_ids = collection.get(include=["metadatas"])["ids"]
        if not all_ids:
            return {"status": "warning", "message": "知识库中没有文档ID", "count": 0}
            
        # 随机抽查5个文档或全部(取较小值)
        sample_size = min(5, len(all_ids))
        import random
        sample_ids = random.sample(all_ids, sample_size)
        
        # 检查这些文档是否有完整的元数据和内容
        samples = collection.get(ids=sample_ids, include=["metadatas", "documents"])
        
        # 计算元数据和内容的完整性
        metadata_integrity = sum(1 for meta in samples["metadatas"] if meta and "source" in meta) / sample_size
        content_integrity = sum(1 for doc in samples["documents"] if doc and len(doc) > 10) / sample_size
        
        integrity_score = (metadata_integrity + content_integrity) / 2
        
        return {
            "status": "healthy" if integrity_score > 0.8 else "warning",
            "message": f"知识库健康度: {integrity_score*100:.1f}%",
            "count": doc_count,
            "metadata_integrity": f"{metadata_integrity*100:.1f}%",
            "content_integrity": f"{content_integrity*100:.1f}%"
        }
    except Exception as e:
        return {"status": "error", "message": f"知识库健康检查失败: {str(e)}"}

知识库健康诊断

应用场景