从零构建ChatGPT知识库：新手必看的架构设计与实践指南

1次阅读

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

传统全文检索技术（如 Elasticsearch）依赖关键词匹配，面临两个核心问题：

语义鸿沟：无法理解 ” 汽车 ” 和 ” 机动车 ” 的等价性
语境缺失：搜索 ”Python” 时难以区分编程语言与蟒蛇

方案	适用场景	免费额度
Faiss	本地部署 / 千万级数据	完全开源
Pinecone	云服务 / 自动扩展	免费 5GB
Chroma	快速原型开发	完全开源

from openai import OpenAI
from typing import List

client = OpenAI()

def get_embedding(text: str, model="text-embedding-3-small") -> List[float]:
    """调用 OpenAI Embedding API 生成文本向量"""
    return client.embeddings.create(input=[text], 
        model=model
    ).data[0].embedding

from langchain.text_splitter import RecursiveCharacterTextSplitter

def chunk_document(text: str, chunk_size=500) -> List[str]:
    """
    文本分块处理
    :param text: 原始文本
    :param chunk_size: 单块 token 数
    :return: 文本块列表
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=50
    )
    return splitter.split_text(text)

import numpy as np
from typing import Tuple

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

    def search(self, query: str, top_k=5) -> List[Tuple[str, float]]:
        """
        混合检索实现
        :return: (文本, 综合得分) 元组列表
        """
        # 获取向量相似度得分
        vector_results = self.vector_db.similarity_search(query, k=top_k*2)

        # 获取 BM25 关键词得分
        bm25_results = bm25_search(query)

        # 融合得分计算
        combined = {}
        for doc, score in vector_results:
            combined[doc] = score * (1 - self.bm25_weight)

        for doc, score in bm25_results:
            combined[doc] = combined.get(doc, 0) + score * self.bm25_weight

        return sorted(combined.items(), key=lambda x: -x[1])[:top_k]

实现指数退避重试机制
使用异步请求批量处理
本地缓存已处理文本的 embedding

import backoff
import asyncio

@backoff.on_exception(backoff.expo, Exception, max_tries=3)
async def batch_embed(texts: List[str]) -> List[List[float]]:
    """批量处理文本向量化"""
    return await asyncio.gather(*[get_embedding(text) for text in texts
    ])

import faiss

def train_quantized_index(embeddings: np.ndarray) -> faiss.Index:
    """训练量化压缩索引"""
    d = embeddings.shape[1]  # 向量维度
    quantizer = faiss.IndexFlatL2(d)
    index = faiss.IndexIVFPQ(quantizer, d, 100, 8, 4)  # 100 个聚类中心, 8 位编码
    index.train(embeddings)
    index.add(embeddings)
    return index