ChatGPT作为个人知识库的实践指南：从数据整理到高效检索

1次阅读

没有评论

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

在个人开发者和小型团队的知识管理实践中，普遍存在以下典型问题：

信息碎片化 ：知识分散在聊天记录、邮件、本地文档等多个孤岛中
检索效率低 ：传统文件系统依赖精确匹配，无法理解查询意图
维护成本高 ：需要手动建立复杂的分类体系和标签规则

与传统知识库工具对比：

Notion/Confluence 等工具仍依赖人工整理，无法自动建立语义关联
本地搜索工具如 Everything 仅支持关键词匹配，缺乏语义理解能力
维基类系统需要严格的内容结构，学习曲线陡峭

数据采集层 ：支持 Markdown/PDF/ 网页等多种格式的原始数据输入
预处理管道 ：
文本清洗（去除特殊字符、标准化编码）
语言检测（过滤非目标语言内容）
分块处理（保持语义完整性的段落分割）
向量化引擎 ：
使用 OpenAI text-embedding-3-small 模型
每个 chunk 生成 1536 维向量
检索系统 ：
FAISS 索引构建（IVF256,PQ16 参数配置）
近似最近邻搜索（内积相似度计算）

与传统关键词检索对比：

维度	关键词检索	语义检索
查询理解	字面匹配	意图理解
召回范围	精确词频	概念关联
适配场景	结构化文档	非结构化内容

import re
from pathlib import Path

def clean_text(text: str) -> str:
    """标准化文本内容"""
    # 合并连续空白字符
    text = re.sub(r'\s+', ' ', text)
    # 移除特殊控制字符
    text = text.encode('ascii', errors='ignore').decode()
    return text.strip()

class DocumentChunker:
    def __init__(self, chunk_size=512):
        self.chunk_size = chunk_size

    def process(self, file_path: Path) -> list[str]:
        """时间复杂度 O(n)，空间复杂度 O(n)"""
        with open(file_path, 'r') as f:
            text = clean_text(f.read())

        chunks = []
        current_chunk = []
        current_len = 0

        for sent in text.split('.'):
            sent = sent.strip()
            if not sent:
                continue

            sent_len = len(sent.split())
            if current_len + sent_len > self.chunk_size:
                chunks.append(' '.join(current_chunk))
                current_chunk = []
                current_len = 0

            current_chunk.append(sent)
            current_len += sent_len

        if current_chunk:
            chunks.append(' '.join(current_chunk))

        return chunks

import faiss
import numpy as np
from openai import OpenAI

class VectorSearcher:
    def __init__(self, dim=1536):
        self.index = faiss.IndexFlatIP(dim)
        self.client = OpenAI()

    def add_documents(self, texts: list[str]):
        """构建索引，时间复杂度 O(n*k) k= 向量维度"""
        embeddings = self._get_embeddings(texts)
        self.index.add(np.array(embeddings))

    def search(self, query: str, top_k=3) -> list[tuple]:
        """检索耗时约 5 -50ms/query"""
        query_embed = self._get_embeddings([query])[0]
        D, I = self.index.search(np.array([query_embed]), top_k)
        return [(i, d) for i, d in zip(I[0], D[0])]

    def _get_embeddings(self, texts: list[str]) -> list[list[float]]:
        """调用 OpenAI API，速率限制需处理"""
        try:
            resp = self.client.embeddings.create(
                input=texts,
                model="text-embedding-3-small"
            )
            return [d.embedding for d in resp.data]
        except Exception as e:
            print(f"Embedding error: {str(e)}")
            return []