Chroma混合语义检索实现解析：从技术选型到生产环境部署

1次阅读

没有评论

共计 2314 个字符，预计需要花费 6 分钟才能阅读完成。

在信息检索领域，传统方法主要分为两类：关键词搜索和语义搜索。这两种方法各有优缺点，但单独使用时都存在明显的局限性。

关键词搜索 ：基于精确匹配，检索速度快，但对同义词、近义词处理能力弱，无法理解查询意图。例如搜索 ”AI” 时，会错过包含 ” 人工智能 ” 但未明确提及 ”AI” 的文档。
语义搜索 ：能够理解查询的语义含义，解决词汇不匹配问题，但计算成本高，对特定领域术语的精确匹配能力不足，且需要大量标注数据进行模型训练。

关键词搜索优势
实现简单，无需复杂模型
检索速度快，适合大规模数据
对精确术语匹配效果好
关键词搜索劣势
无法处理语义相似性
受限于查询表述方式
难以处理拼写错误或变体
语义搜索优势
理解查询意图
处理同义词和概念匹配
对自然语言查询友好
语义搜索劣势
计算资源消耗大
需要大量训练数据
对小众术语可能表现不佳

Chroma 采用分层架构实现混合语义检索，主要包含以下组件：

预处理层
文本规范化（大小写转换、标点处理）
关键词提取（TF-IDF、BM25）
语义嵌入生成（Sentence-BERT 等模型）
索引层
构建双重索引结构：
- 倒排索引（关键词）
- 向量索引（语义）
采用 HNSW 等近似最近邻算法优化向量搜索
融合层

设计混合评分函数：

def hybrid_score(keyword_score, semantic_score, alpha=0.5):
    return alpha * keyword_score + (1 - alpha) * semantic_score

动态调整权重参数 α（0- 1 之间）

import chromadb
from sentence_transformers import SentenceTransformer

# 初始化 Chroma 客户端和嵌入模型
client = chromadb.Client()
collection = client.create_collection("hybrid_search")
embedder = SentenceTransformer('all-MiniLM-L6-v2')

# 添加文档
# 实际应用中应从数据库或文件加载
documents = [
    "人工智能是模拟人类智能的计算机系统",
    "机器学习是 AI 的一个子领域",
    "深度学习使用神经网络进行特征学习"
]

# 生成双重表示
ids = [str(i) for i in range(len(documents))]
embeddings = embedder.encode(documents).tolist()

# 添加到集合
collection.add(
    documents=documents,
    embeddings=embeddings,
    ids=ids
)

# 混合查询函数
def hybrid_query(query: str, alpha=0.5, top_k=5):
    # 关键词搜索（BM25 评分）keyword_results = collection.query(query_texts=[query],
        n_results=top_k,
        include=["documents", "distances"]
    )

    # 语义搜索（余弦相似度）query_embedding = embedder.encode(query).tolist()
    semantic_results = collection.query(query_embeddings=[query_embedding],
        n_results=top_k,
        include=["documents", "distances"]
    )

    # 结果融合
    combined = []
    for i in range(top_k):
        doc_id = keyword_results['ids'][0][i]
        kw_score = 1 - keyword_results['distances'][0][i]  # 转换为相似度
        sem_score = 1 - semantic_results['distances'][0][i]
        combined_score = alpha * kw_score + (1 - alpha) * sem_score
        combined.append({
            'id': doc_id,
            'document': keyword_results['documents'][0][i],
            'score': combined_score
        })

    # 按综合评分排序
    return sorted(combined, key=lambda x: x['score'], reverse=True)