AI Agent实现语义化检索的架构设计与工程实践

1次阅读

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

传统的关键词检索技术如 TF-IDF 和 BM25 在处理复杂语义时存在明显短板。举个例子，在医疗领域的 NER（命名实体识别）任务中，”Apple” 可能是水果也可能是科技公司，传统方法无法区分这种语义歧义。

短文本问题 ：查询 ” 头疼怎么办 ” 和 ” 偏头痛治疗方案 ” 在关键词匹配中可能完全无交集
长尾效应 ：冷门专业术语（如 ”CRISPR-Cas9″）容易被常规分词器错误处理
多语言场景 ：同一概念的不同语言表达（如 ” 猫 ” 和 ”cat”）无法建立关联

BERT：适合短文本，但计算开销大，需处理 512 tokens 长度限制
Sentence-Transformer：专为句子嵌入优化，all-MiniLM-L6-v2 模型在速度和效果间取得平衡
Cohere：商业 API 适合无 GPU 环境，支持多语言但存在延迟问题

Faiss：Facebook 开源的 CPU/GPU 加速库，适合千万级向量
HNSW：基于图的算法，查询速度极快但内存占用高
PGvector：PostgreSQL 扩展，适合已有 PG 生态的场景

from sentence_transformers import SentenceTransformer
import torch

# 启用 GPU 加速
device = 'cuda' if torch.cuda.is_available() else 'cpu'
model = SentenceTransformer('all-MiniLM-L6-v2', device=device)

# 批量处理提升吞吐量
texts = ["量子计算原理", "区块链技术应用"]
embeddings = model.encode(texts, batch_size=128, show_progress_bar=True)

from rank_bm25 import BM25Okapi
import numpy as np

class HybridRetriever:
    def __init__(self, bm25_weight=0.3):
        self.bm25_weight = bm25_weight

    def query(self, text: str):
        # 语义检索分数
        semantic_score = faiss_index.search(embedding, k=10)
        # 关键词检索分数
        bm25_score = bm25.get_scores(tokenize(text))
        # 加权融合
        combined = semantic_score*(1-self.bm25_weight) + bm25_score*self.bm25_weight
        return np.argsort(combined)[::-1][:10]

import faiss

class DynamicIndex:
    def __init__(self, dim=384):
        self.index = faiss.IndexFlatIP(dim)

    def add_items(self, embeddings):
        if not self.index.is_trained:
            self.index.train(embeddings)
        self.index.add(embeddings)

    def remove_items(self, ids):
        # Faiss 原生不支持删除操作，需要重建索引
        new_index = faiss.IndexFlatIP(self.index.d)
        remaining = [i for i in range(self.index.ntotal) if i not in ids]
        new_index.add(self.index.reconstruct_batch(remaining))
        self.index = new_index