基于ChatGPT构建个人知识库：从数据收集到智能检索的全流程实战

1次阅读

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

当代开发者面临三大知识管理痛点：
1. 信息孤岛问题：技术文档、会议纪要、代码片段分散在 Notion、GitHub、本地文件等不同平台
2. 非结构化数据检索难：PDF 技术白皮书、网页博客等内容无法通过简单关键词匹配精准定位
3. 语义鸿沟：传统搜索依赖字面匹配，无法理解 ”Python 多线程任务调度 ” 与 ”asyncio 任务管理 ” 的语义关联

关系型数据库：
优势：ACID 事务支持，适合结构化数据
局限：无法有效存储高维向量，相似度计算效率低
向量数据库(Pinecone/ChromaDB)：
优势：原生支持余弦相似度等向量运算，毫秒级检索千万级数据
典型场景：pinecone==2.2.4 的 pod 型部署适合企业级应用

规则匹配：
实现：正则表达式或布尔查询(如 Elasticsearch)
缺陷：需要预定义规则，无法处理同义词和语义扩展
语义嵌入(OpenAI Embeddings)：
实现：text-embedding-3-small生成 1536 维向量
优势：捕获 ” 机器学习 ” 与 ”ML” 等语义关联，支持多语言

# 需安装 pyPDF2==3.0.1
from PyPDF2 import PdfReader

def extract_pdf_text(file_path):
    """
    处理含特殊字符的 PDF 文档
    :param file_path: PDF 文件路径
    :return: 经过 Unicode 标准化处理的文本
    """text =""
    try:
        reader = PdfReader(file_path)
        for page in reader.pages:
            # 处理 PDF 中的软连字符(U+00AD)
            raw_text = page.extract_text()
            normalized = raw_text.replace('\xad', '')
            text += normalized
    except Exception as e:
        print(f"PDF 解析失败: {str(e)}")
    return text

# 需安装 bs4==0.0.1 和 html2text==2020.1.16
from bs4 import BeautifulSoup
import html2text

def clean_html(html_content):
    """
    转换 HTML 为纯净 Markdown 格式
    :param html_content: 原始 HTML 字符串
    :return: 去噪后的 Markdown 文本
    """soup = BeautifulSoup(html_content,'html.parser')
    # 移除广告等无关元素
    for element in soup(['script', 'style', 'nav']):
        element.decompose()
    converter = html2text.HTML2Text()
    converter.ignore_links = False
    return converter.handle(str(soup))

from langchain_text_splitters import RecursiveCharacterTextSplitter  # 需 0.0.2 以上版本

def semantic_chunking(text, chunk_size=1000, overlap=200):
    """
    基于语义重叠的分块实现
    :param text: 输入文本
    :param chunk_size: 块大小(token 数)
    :param overlap: 重叠区 token 数
    :return: 文本块列表
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        length_function=len,
        is_separator_regex=False,
    )
    # 自动按段落 / 句子边界分割
    return splitter.split_text(text)

分块参数优化建议：
– 技术文档：chunk_size=1200, overlap=150
– 对话记录：chunk_size=800, overlap=100

模型	维度	价格(每千 token)	适用场景
text-embedding-3-small	1536	$0.00002	通用知识库
text-embedding-3-large	3072	$0.00013	高精度语义匹配
ada-002	1536	$0.00010	已弃用，不推荐新项目

import openai
from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3))
async def batch_embed(texts, model="text-embedding-3-small"):
    """
    带重试机制的异步批处理嵌入
    :param texts: 文本列表(单批不超过 2048 个 token)
    :param model: 选择的嵌入模型
    :return: 向量列表
    """
    try:
        response = await openai.Embedding.create(
            input=texts,
            model=model,
            request_timeout=30
        )
        return [item["embedding"] for item in response["data"]]
    except Exception as e:
        print(f"嵌入失败: {str(e)}")
        raise

import chromadb  # 需 0.4.15 以上版本
from chromadb.config import Settings

# 初始化时启用 HSW 索引加速
client = chromadb.Client(Settings(
    chroma_db_impl="duckdb+parquet",
    persist_directory="./vector_store"
))

collection = client.create_collection(
    name="tech_docs",
    metadata={"hnsw:space": "cosine"}  # 使用余弦相似度
)

# 批量插入文档
collection.add(documents=["文档 1 内容", "文档 2 内容"],
    embeddings=[[0.1,...,0.5], [...]],  # 实际嵌入向量
    ids=["doc1", "doc2"]
)

from langchain.chains import RetrievalQA
from langchain.llms import OpenAI  # 需 0.0.5 以上版本

# 加载本地向量库
retriever = collection.as_retriever(
    search_type="mmr",  # 最大边际相关算法
    search_kwargs={"k": 5}
)

qa_chain = RetrievalQA.from_chain_type(llm=OpenAI(temperature=0),
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True
)

# 执行语义搜索
result = qa_chain.run("如何在 Python 中实现异步日志记录?")
print(result["result"])
print("来源文档:", result["source_documents"])

传输层加密：使用 HTTPS+SSL 验证所有 API 调用
静态数据加密：
敏感文档预处理前使用 AWS KMS(boto3==1.28.58)加密
本地向量库启用 SQLite 透明加密
访问控制：
为 OpenAI API 设置组织级访问策略
向量数据库配置 RBAC

冷启动加速：

# 预生成嵌入缓存示例
import pickle
from pathlib import Path

CACHE_DIR = Path("./embed_cache")

def get_embedding_cached(text, model):
    cache_file = CACHE_DIR / f"{hash(text)}.pkl"
    if cache_file.exists():
        return pickle.load(cache_file.open("rb"))
    emb = openai.Embedding.create(input=text, model=model)
    pickle.dump(emb, cache_file.open("wb"))
    return emb

混合模型策略：
简单查询使用gpt-3.5-turbo($0.002/1k tokens)
复杂推理切换gpt-4($0.06/1k tokens)

症状：问答结果不完整或包含无关内容
诊断：
使用 tiktoken==0.5.1 计算实际 token 数
检查分块是否截断了完整句子

修复：

# 动态调整分块示例
def dynamic_chunk(text, max_tokens=1024):
    tokens = count_tokens(text)  # 实现 token 计数
    if tokens <= max_tokens:
        return [text]
    # 按句子分割后再合并
    sentences = split_by_sentences(text)
    chunks = []
    current_chunk = ""
    for sent in sentences:
        if count_tokens(current_chunk + sent) > max_tokens:
            chunks.append(current_chunk)
            current_chunk = sent
        else:
            current_chunk += " " + sent
    return chunks

MRR(平均倒数排名)：

def calculate_mrr(correct_docs, result_docs):
    """
    :param correct_docs: 正确答案 ID 列表
    :param result_docs: 实际返回 ID 列表
    :return: MRR 分数(0-1)
    """
    for rank, doc in enumerate(result_docs, 1):
        if doc.id in correct_docs:
            return 1.0 / rank
    return 0.0