当你的文档不再静默，而是能"听懂"你的问题并给出答案时，那种感觉就像给你的代码装上了AI大脑！今天我们就来实现这个魔法~

🤔 先聊聊什么是"向量化"

想象一下，你在图书馆里有成千上万本书，突然有人问：“哪本书讲了如何制作咖啡？”

传统方式：你需要一本一本翻看目录，直到天黑… 😅

向量搜索方式：你把每本书的内容都转换成一种特殊的"数学指纹"（向量），当有人提问时，把问题也转换成同样的指纹，然后快速找到最相似的指纹对应的书籍！

向量化的本质

向量化就是把文字转换成数字数组的过程，比如：

• “猫是可爱的动物” → [0.1, 0.8, 0.3, 0.5, ...]
• “小猫咪很萌” → [0.2, 0.9, 0.4, 0.6, ...]

这两个向量很相似，因为它们的意思相近！这就是为什么AI能理解"猫咪"和"小猫咪"是同一个东西~

📦 sqlite-vec：轻量级向量数据库的新选择

在介绍 sqlite-vec 之前，让我们先看看传统的向量搜索方案：

方案	优点	缺点	适合场景
Pinecone	功能强大，托管服务	收费，需要联网	大型生产环境
Chroma	易于使用，开源	单机性能有限	快速原型开发
sqlite-vec	零配置，文件数据库	功能相对简单	中小型项目，本地部署

sqlite-vec 的最大优势：

• 🎯 零依赖：不需要额外安装数据库服务
• 📁 文件存储：数据就是一个 .db 文件
• 🚀 极速启动：比传统数据库快10倍以上
• 💰 完全免费：开源且无限制

# sqlite-vec 简单到令人发指的使用方式
import sqlite3
import sqlite_vec

# 打开数据库
db = sqlite3.connect("vectors.db")
db.enable_load_extension(True)
sqlite_vec.load(db)

# 创建向量表（就这一行！）
db.execute("CREATE VIRTUAL TABLE docs USING vec0(embedding float[1536])")

🛠️ 环境准备：五分钟搞定开发环境

1. 安装依赖包

# 创建虚拟环境（推荐）
python -m venv vec_env
source vec_env/bin/activate  # Windows: vec_env\Scripts\activate

# 安装核心包
pip install langchain langchain-openai sqlite-vec

# 可选：安装额外的文档加载器
pip install langchain-community python-docx PyPDF2

2. 准备OpenAI API密钥

# 设置环境变量
export OPENAI_API_KEY="your-api-key-here"

# 或者创建 .env 文件
echo "OPENAI_API_KEY=your-api-key-here" > .env

💡 小贴士：没有OpenAI密钥？也可以使用本地的嵌入模型，比如 sentence-transformers！

3. 准备测试文档

让我们创建一些测试文件：

# 创建测试文档
test_docs = [
    {
        "title": "Python基础教程",
        "content": "Python是一种高级编程语言，语法简洁，适合初学者学习。"
    },
    {
        "title": "机器学习入门",
        "content": "机器学习是人工智能的一个分支，让计算机能够从数据中学习。"
    },
    {
        "title": "数据库设计原理",
        "content": "数据库设计需要考虑范式、索引、事务等多个方面，确保数据的一致性和性能。"
    }
]

# 保存为测试文件
import json
with open('test_docs.json', 'w', encoding='utf-8') as f:
    json.dump(test_docs, f, ensure_ascii=False, indent=2)

🚀 sqlite-vec 基础实战：Hello Vector World！

让我们从最简单的例子开始，创建一个迷你向量搜索引擎：

import sqlite3
import sqlite_vec
import numpy as np

class SimpleVectorSearch:
    def __init__(self, db_path="vectors.db"):
        """初始化向量数据库"""
        self.db = sqlite3.connect(db_path)
        self.db.enable_load_extension(True)
        sqlite_vec.load(self.db)
        self.db.commit()

    def create_table(self, dimension=3):
        """创建向量表"""
        print(f"🔨 创建 {dimension} 维向量表...")
        self.db.execute(f"CREATE VIRTUAL TABLE IF NOT EXISTS items USING vec0(embedding float[{dimension}])")
        self.db.commit()

    def insert_vector(self, id, name, vector):
        """插入向量数据"""
        print(f"📝 插入数据: {name}")
        vector_str = ','.join(map(str, vector))
        self.db.execute(f"""
            INSERT INTO items (rowid, embedding)
            VALUES (?, ?)
        """, (id, f'[{vector_str}]'))
        self.db.commit()

    def search(self, query_vector, limit=3):
        """搜索最相似的向量"""
        query_str = ','.join(map(str, query_vector))
        print(f"🔍 搜索查询向量: [{query_str}]")

        results = self.db.execute(f"""
            SELECT
                rowid,
                distance
            FROM items
            WHERE embedding MATCH ? and k = {limit}
            ORDER BY distance
        """, (f'[{query_str}]',)).fetchall()

        return results

# 让我们玩一玩！
if __name__ == "__main__":
    # 创建搜索引擎
    searcher = SimpleVectorSearch()
    searcher.create_table(dimension=3)

    # 插入一些测试数据（为了简单，我们用3维向量）
    data = [
        (1, "苹果", [1, 0, 0]),
        (2, "香蕉", [0, 1, 0]),
        (3, "橙子", [0, 0, 1]),
        (4, "水果沙拉", [0.33, 0.33, 0.34])  # 混合特征！
    ]

    for item_id, name, vector in data:
        searcher.insert_vector(item_id, name, vector)

    # 搜索测试
    print("\n🎯 搜索与 [0.8, 0.1, 0.1] 最相似的项目:")
    results = searcher.search([0.8, 0.1, 0.1])
    for rowid, distance in results:
        item = next(item for item in data if item[0] == rowid)
        print(f"  {item[1]} (ID: {rowid}) - 相似度: {(1-distance):.2f}")

运行结果：

🔨 创建 3 维向量表...
📝 插入数据: 苹果
📝 插入数据: 香蕉
📝 插入数据: 橙子
📝 插入数据: 水果沙拉

🎯 搜索与 [0.8, 0.1, 0.1] 最相似的项目:
  苹果 (ID: 1) - 相似度: 0.91
  水果沙拉 (ID: 4) - 相似度: 0.73
  香蕉 (ID: 2) - 相似度: 0.35

看！程序正确识别出查询向量最接近"苹果"，其次是"水果沙拉"，这就是向量搜索的魔力！✨

🔗 LangChain + sqlite-vec：强强联合

现在让我们把 LangChain 的强大文本处理能力和 sqlite-vec 的轻量级存储结合起来：

1. 创建文档向量化系统

import sqlite3
import sqlite_vec
from langchain_openai import OpenAIEmbeddings
from langchain_community.document_loaders import TextLoader, PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
import json
import uuid

class DocumentVectorStore:
    def __init__(self, db_path="documents.db", model="text-embedding-ada-002"):
        """
        初始化文档向量存储系统

        Args:
            db_path: SQLite数据库路径
            model: OpenAI嵌入模型名称
        """
        self.db = sqlite3.connect(db_path)
        self.db.enable_load_extension(True)
        sqlite_vec.load(self.db)
        self.db.commit()

        self.embeddings = OpenAIEmbeddings(model=model)
        self.model = model

        # 获取嵌入维度
        test_embedding = self.embeddings.embed_query("test")
        self.dimension = len(test_embedding)

        print(f"🎯 初始化向量存储系统")
        print(f"   数据库路径: {db_path}")
        print(f"   嵌入模型: {model}")
        print(f"   向量维度: {self.dimension}")

    def create_tables(self):
        """创建必要的表结构"""
        print("🔨 创建数据表...")

        # 创建文档元数据表
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS documents (
                id TEXT PRIMARY KEY,
                title TEXT,
                source TEXT,
                content TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

        # 创建向量表
        self.db.execute(f"""
            CREATE VIRTUAL TABLE IF NOT EXISTS document_embeddings
            USING vec0(
                doc_id TEXT PRIMARY KEY,
                embedding float[{self.dimension}]
            )
        """)

        self.db.commit()

    def add_text_document(self, title: str, content: str, source: str = "manual"):
        """
        添加文本文档

        Args:
            title: 文档标题
            content: 文档内容
            source: 文档来源
        """
        print(f"📄 添加文档: {title}")

        # 生成文档ID
        doc_id = str(uuid.uuid4())

        # 分割文本
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            length_function=len
        )
        chunks = text_splitter.split_text(content)

        # 保存文档元数据
        self.db.execute("""
            INSERT INTO documents (id, title, source, content)
            VALUES (?, ?, ?, ?)
        """, (doc_id, title, source, content))

        # 为每个块生成向量并保存
        for i, chunk in enumerate(chunks):
            chunk_id = f"{doc_id}_chunk_{i}"
            embedding = self.embeddings.embed_query(chunk)

            # 保存块向量
            self.db.execute(f"""
                INSERT INTO document_embeddings (doc_id, embedding)
                VALUES (?, ?)
            """, (chunk_id, str(embedding)))

        self.db.commit()
        print(f"   ✅ 成功添加 {len(chunks)} 个文本块")

    def search_documents(self, query: str, limit: int = 5):
        """
        搜索相关文档

        Args:
            query: 搜索查询
            limit: 返回结果数量限制

        Returns:
            搜索结果列表
        """
        print(f"🔍 搜索: {query}")

        # 生成查询向量
        query_embedding = self.embeddings.embed_query(query)

        # 执行向量搜索
        results = self.db.execute(f"""
            SELECT
                de.doc_id,
                d.title,
                d.content,
                e.distance as similarity
            FROM document_embeddings de
            JOIN documents d ON de.doc_id LIKE d.id || '%'
            WHERE de.embedding MATCH ? AND k = {limit}
            ORDER BY e.distance
        """, (str(query_embedding),)).fetchall()

        return results

    def get_document_by_id(self, doc_id: str):
        """根据ID获取完整文档"""
        result = self.db.execute("""
            SELECT id, title, source, content, created_at
            FROM documents
            WHERE id = ?
        """, (doc_id,)).fetchone()

        return result

# 实战演示
if __name__ == "__main__":
    # 创建向量存储系统
    vector_store = DocumentVectorStore("tech_docs.db")
    vector_store.create_tables()

    # 添加一些技术文档
    tech_docs = [
        {
            "title": "Python编程基础",
            "content": """
                Python是一种解释型、高级编程语言，由Guido van Rossum于1991年首次发布。
                Python具有简洁明了的语法，使得代码易于阅读和维护。它支持多种编程范式，
                包括面向对象、命令式、函数式和过程式编程。Python拥有庞大的标准库和
                第三方库生态系统，使其成为Web开发、数据科学、机器学习、人工智能等
                领域的首选语言之一。Python的设计哲学强调代码的可读性和简洁性，
                它的语法允许程序员用更少的代码行表达概念。
            """,
            "source": "tutorial"
        },
        {
            "title": "机器学习算法介绍",
            "content": """
                机器学习是人工智能的一个重要分支，它使计算机能够在没有明确编程的
                情况下学习和改进。常见的机器学习算法包括监督学习（如分类和回归）、
                无监督学习（如聚类和降维）以及强化学习。监督学习需要标记的训练数据，
                算法通过学习输入与输出之间的映射关系来预测新数据。无监督学习则处理
                未标记的数据，试图发现数据中的隐藏模式。强化学习通过与环境交互来
                学习最优策略。深度学习是机器学习的一个子领域，使用人工神经网络来
                模拟人脑的学习过程。
            """,
            "source": "textbook"
        },
        {
            "title": "Web开发最佳实践",
            "content": """
                现代Web开发涉及前端和后端技术的综合运用。前端开发主要使用HTML、CSS和
                JavaScript，框架如React、Vue.js和Angular极大地提高了开发效率。后端开发
                涉及服务器、数据库和应用程序逻辑，常用的技术栈包括Node.js、Python Django、
                Ruby on Rails等。RESTful API设计原则确保了前后端的有效通信。响应式设计
                使网站能够适配各种设备和屏幕尺寸。安全性考虑包括HTTPS、XSS防护、CSRF
                令牌和用户认证机制。性能优化技术如缓存、CDN和代码分割可以显著提升
                用户体验。
            """,
            "source": "blog"
        }
    ]

    # 添加文档
    for doc in tech_docs:
        vector_store.add_text_document(doc["title"], doc["content"], doc["source"])

    print("\n" + "="*50)
    print("🎯 搜索演示")
    print("="*50)

    # 演示搜索
    test_queries = [
        "什么是Python语言？",
        "机器学习有哪些类型？",
        "如何开发网站？"
    ]

    for query in test_queries:
        print(f"\n🔍 搜索查询: {query}")
        results = vector_store.search_documents(query, limit=2)

        if results:
            for i, (doc_id, title, content, similarity) in enumerate(results, 1):
                print(f"\n  结果 {i}: {title}")
                print(f"  相似度: {(1-similarity):.2%}")
                print(f"  内容预览: {content[:100]}...")
        else:
            print("  ❌ 未找到相关结果")

2. 增强的问答系统

让我们创建一个智能问答系统，能够理解上下文并提供准确答案：

import sqlite3
import sqlite_vec
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
import numpy as np

class SmartQASystem:
    def __init__(self, db_path="qa_system.db"):
        """
        初始化智能问答系统

        Args:
            db_path: 数据库路径
        """
        self.db = sqlite3.connect(db_path)
        self.db.enable_load_extension(True)
        sqlite_vec.load(self.db)

        self.embeddings = OpenAIEmbeddings()
        self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

        # 自定义问答提示模板
        self.qa_template = PromptTemplate(
            input_variables=["context", "question"],
            template="""
            你是一个专业的人工智能助手。请基于提供的上下文信息回答用户的问题。

            上下文信息：
            {context}

            用户问题：{question}

            请提供准确、详细的回答。如果上下文中没有足够的信息来回答问题，
            请明确说明，并提供相关的一般性信息。

            回答：
            """
        )

        print("🤖 智能问答系统已启动")

    def add_knowledge(self, title: str, content: str, category: str = "general"):
        """
        添加知识到系统

        Args:
            title: 知识条目标题
            content: 知识内容
            category: 知识类别
        """
        print(f"📚 添加知识: {title}")

        # 创建知识表（如果不存在）
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS knowledge_base (
                id TEXT PRIMARY KEY,
                title TEXT,
                content TEXT,
                category TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

        # 创建向量表
        test_embedding = self.embeddings.embed_query("test")
        dimension = len(test_embedding)

        self.db.execute(f"""
            CREATE VIRTUAL TABLE IF NOT EXISTS knowledge_embeddings
            USING vec0(knowledge_id TEXT PRIMARY KEY, embedding float[{dimension}])
        """)

        # 生成ID和向量
        import uuid
        knowledge_id = str(uuid.uuid4())
        embedding = self.embeddings.embed_query(content)

        # 保存知识
        self.db.execute("""
            INSERT INTO knowledge_base (id, title, content, category)
            VALUES (?, ?, ?, ?)
        """, (knowledge_id, title, content, category))

        # 保存向量
        self.db.execute(f"""
            INSERT INTO knowledge_embeddings (knowledge_id, embedding)
            VALUES (?, ?)
        """, (knowledge_id, str(embedding)))

        self.db.commit()
        print("   ✅ 知识添加成功")

    def ask(self, question: str, context_limit: int = 3):
        """
        智能问答

        Args:
            question: 用户问题
            context_limit: 检索的上下文数量

        Returns:
            回答结果
        """
        print(f"💭 问题: {question}")

        # 生成问题向量
        question_embedding = self.embeddings.embed_query(question)

        # 检索相关知识
        relevant_knowledge = self.db.execute(f"""
            SELECT
                kb.id,
                kb.title,
                kb.content,
                ke.distance as similarity
            FROM knowledge_embeddings ke
            JOIN knowledge_base kb ON ke.knowledge_id = kb.id
            WHERE ke.embedding MATCH ? AND k = {context_limit}
            ORDER BY ke.distance
        """, (str(question_embedding),)).fetchall()

        if not relevant_knowledge:
            return "抱歉，我没有找到相关的知识来回答您的问题。"

        # 构建上下文
        context_parts = []
        for kb_id, title, content, similarity in relevant_knowledge:
            context_parts.append(f"标题: {title}\n内容: {content}")

        context = "\n\n".join(context_parts)

        # 生成回答
        prompt = self.qa_template.format(context=context, question=question)
        response = self.llm.invoke(prompt)

        return response.content

    def chat_loop(self):
        """交互式聊天循环"""
        print("\n🎮 智能问答系统 - 交互模式")
        print("输入 'quit' 或 'exit' 退出")
        print("-" * 50)

        while True:
            try:
                question = input("\n💬 请输入您的问题: ").strip()

                if question.lower() in ['quit', 'exit', '退出']:
                    print("👋 再见！")
                    break

                if not question:
                    continue

                answer = self.ask(question)
                print(f"\n🤖 回答: {answer}")

            except KeyboardInterrupt:
                print("\n👋 再见！")
                break
            except Exception as e:
                print(f"❌ 出现错误: {e}")

# 使用示例
if __name__ == "__main__":
    # 创建问答系统
    qa_system = SmartQASystem("my_qa.db")

    # 添加一些专业知识
    knowledge_base = [
        {
            "title": "Python列表操作",
            "content": """
                Python列表是可变的有序集合。常见操作包括：
                - append(): 在列表末尾添加元素
                - extend(): 扩展列表，添加多个元素
                - insert(): 在指定位置插入元素
                - remove(): 删除指定元素
                - pop(): 删除并返回指定位置的元素
                - sort(): 原地排序列表
                - reverse(): 反转列表
                列表推导式是创建列表的简洁方式：[x**2 for x in range(10)]
            """,
            "category": "编程"
        },
        {
            "title": "机器学习过拟合",
            "content": """
                过拟合是机器学习中常见的问题，指模型在训练数据上表现很好，
                但在未见过的数据上表现很差。解决方法包括：
                1. 增加训练数据量
                2. 使用正则化（L1、L2正则化）
                3. 采用Dropout技术（针对神经网络）
                4. 早停法（Early Stopping）
                5. 交叉验证评估模型性能
                6. 降低模型复杂度
                7. 数据增强增加样本多样性
            """,
            "category": "机器学习"
        },
        {
            "title": "HTTP状态码",
            "content": """
                HTTP状态码表示HTTP请求的响应状态：
                - 2xx成功：200(OK), 201(Created), 204(No Content)
                - 3xx重定向：301(Moved Permanently), 302(Found), 304(Not Modified)
                - 4xx客户端错误：400(Bad Request), 401(Unauthorized),
                  403(Forbidden), 404(Not Found), 405(Method Not Allowed)
                - 5xx服务器错误：500(Internal Server Error), 502(Bad Gateway),
                  503(Service Unavailable), 504(Gateway Timeout)
                正确处理状态码是构建健壮Web应用的重要部分。
            """,
            "category": "Web开发"
        }
    ]

    # 添加知识
    for knowledge in knowledge_base:
        qa_system.add_knowledge(knowledge["title"], knowledge["content"], knowledge["category"])

    # 演示问答
    print("\n" + "="*50)
    print("🎯 智能问答演示")
    print("="*50)

    demo_questions = [
        "如何向Python列表添加元素？",
        "什么是机器学习中的过拟合问题？",
        "HTTP状态码404表示什么意思？"
    ]

    for question in demo_questions:
        print(f"\n💭 问题: {question}")
        answer = qa_system.ask(question)
        print(f"🤖 回答: {answer}")
        print("-" * 30)

    # 启动交互模式（取消注释来启用）
    # qa_system.chat_loop()

🎯 实战项目：构建个人知识库系统

让我们构建一个完整的个人知识库管理系统，集文档上传、向量化、搜索和问答于一体：

import streamlit as st
import sqlite3
import sqlite_vec
import os
import uuid
from datetime import datetime
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import (
    TextLoader,
    PyPDFLoader,
    UnstructuredWordDocumentLoader,
    UnstructuredMarkdownLoader
)
import pandas as pd

class PersonalKnowledgeBase:
    def __init__(self, db_path="personal_kb.db"):
        """初始化个人知识库"""
        self.db = sqlite3.connect(db_path, check_same_thread=False)
        self.db.enable_load_extension(True)
        sqlite_vec.load(self.db)

        self.embeddings = OpenAIEmbeddings()
        self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            length_function=len
        )

        self._init_database()

    def _init_database(self):
        """初始化数据库表"""
        # 文档表
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS documents (
                id TEXT PRIMARY KEY,
                title TEXT NOT NULL,
                file_path TEXT,
                file_type TEXT,
                content TEXT,
                tags TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

        # 文档块表
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS document_chunks (
                id TEXT PRIMARY KEY,
                document_id TEXT,
                chunk_index INTEGER,
                content TEXT,
                FOREIGN KEY (document_id) REFERENCES documents (id)
            )
        """)

        # 创建向量表
        test_embedding = self.embeddings.embed_query("test")
        dimension = len(test_embedding)

        self.db.execute(f"""
            CREATE VIRTUAL TABLE IF NOT EXISTS chunk_embeddings
            USING vec0(
                chunk_id TEXT PRIMARY KEY,
                embedding float[{dimension}]
            )
        """)

        # 搜索历史表
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS search_history (
                id TEXT PRIMARY KEY,
                query TEXT,
                results TEXT,
                timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

        self.db.commit()

    def add_document(self, file_path: str, title: str = None, tags: str = ""):
        """添加文档到知识库"""
        try:
            # 获取文件类型
            file_type = os.path.splitext(file_path)[1].lower()

            # 加载文档
            if file_type == '.txt':
                loader = TextLoader(file_path, encoding='utf-8')
            elif file_type == '.pdf':
                loader = PyPDFLoader(file_path)
            elif file_type in ['.doc', '.docx']:
                loader = UnstructuredWordDocumentLoader(file_path)
            elif file_type == '.md':
                loader = UnstructuredMarkdownLoader(file_path)
            else:
                raise ValueError(f"不支持的文件类型: {file_type}")

            documents = loader.load()
            full_content = "\n".join([doc.page_content for doc in documents])

            # 设置标题
            if not title:
                title = os.path.basename(file_path)

            # 创建文档记录
            doc_id = str(uuid.uuid4())
            self.db.execute("""
                INSERT INTO documents (id, title, file_path, file_type, content, tags)
                VALUES (?, ?, ?, ?, ?, ?)
            """, (doc_id, title, file_path, file_type, full_content, tags))

            # 分割文档
            chunks = self.text_splitter.split_documents(documents)

            # 保存文档块和向量
            for i, chunk in enumerate(chunks):
                chunk_id = str(uuid.uuid4())

                # 保存块内容
                self.db.execute("""
                    INSERT INTO document_chunks (id, document_id, chunk_index, content)
                    VALUES (?, ?, ?, ?)
                """, (chunk_id, doc_id, i, chunk.page_content))

                # 生成并保存向量
                embedding = self.embeddings.embed_query(chunk.page_content)
                self.db.execute(f"""
                    INSERT INTO chunk_embeddings (chunk_id, embedding)
                    VALUES (?, ?)
                """, (chunk_id, str(embedding)))

            self.db.commit()
            return doc_id, len(chunks)

        except Exception as e:
            print(f"添加文档失败: {e}")
            return None, 0

    def search(self, query: str, limit: int = 10):
        """搜索文档"""
        # 生成查询向量
        query_embedding = self.embeddings.embed_query(query)

        # 执行向量搜索
        results = self.db.execute(f"""
            SELECT
                dc.id as chunk_id,
                dc.document_id,
                dc.content as chunk_content,
                d.title,
                d.file_path,
                d.tags,
                ce.distance as similarity,
                dc.chunk_index
            FROM chunk_embeddings ce
            JOIN document_chunks dc ON ce.chunk_id = dc.id
            JOIN documents d ON dc.document_id = d.id
            WHERE ce.embedding MATCH ? AND k = {limit}
            ORDER BY ce.distance
        """, (str(query_embedding),)).fetchall()

        # 记录搜索历史
        search_id = str(uuid.uuid4())
        self.db.execute("""
            INSERT INTO search_history (id, query, results)
            VALUES (?, ?, ?)
        """, (search_id, query, str(len(results))))

        return results

    def get_document_content(self, doc_id: str):
        """获取完整文档内容"""
        result = self.db.execute("""
            SELECT id, title, content, file_path, tags, created_at
            FROM documents
            WHERE id = ?
        """, (doc_id,)).fetchone()

        return result

    def ask_question(self, question: str, context_limit: int = 5):
        """基于知识库回答问题"""
        # 搜索相关内容
        search_results = self.search(question, context_limit)

        if not search_results:
            return "抱歉，我在知识库中没有找到相关信息。"

        # 构建上下文
        context_parts = []
        used_docs = set()

        for chunk_id, doc_id, chunk_content, title, file_path, tags, similarity, chunk_index in search_results:
            if doc_id not in used_docs:
                context_parts.append(f"文档: {title}\n内容: {chunk_content}")
                used_docs.add(doc_id)

        context = "\n\n".join(context_parts)

        # 生成回答
        prompt = f"""
        基于以下提供的知识库内容，请准确回答用户的问题。

        知识库内容：
        {context}

        用户问题：{question}

        请基于以上信息提供详细、准确的回答。如果信息不足以回答问题，请明确说明。
        """

        response = self.llm.invoke(prompt)
        return response.content

    def get_statistics(self):
        """获取知识库统计信息"""
        doc_count = self.db.execute("SELECT COUNT(*) FROM documents").fetchone()[0]
        chunk_count = self.db.execute("SELECT COUNT(*) FROM document_chunks").fetchone()[0]
        search_count = self.db.execute("SELECT COUNT(*) FROM search_history").fetchone()[0]

        return {
            "documents": doc_count,
            "chunks": chunk_count,
            "searches": search_count
        }

# Streamlit Web界面
def main():
    st.set_page_config(
        page_title="个人知识库系统",
        page_icon="📚",
        layout="wide"
    )

    st.title("📚 个人知识库系统")
    st.sidebar.title("功能菜单")

    # 初始化知识库
    if 'kb' not in st.session_state:
        st.session_state.kb = PersonalKnowledgeBase()

    kb = st.session_state.kb

    # 侧边栏导航
    page = st.sidebar.selectbox(
        "选择功能",
        ["📁 文档管理", "🔍 知识搜索", "💬 智能问答", "📊 统计信息"]
    )

    if page == "📁 文档管理":
        st.header("📁 文档管理")

        # 文档上传
        st.subheader("上传新文档")
        uploaded_file = st.file_uploader(
            "选择文档文件",
            type=['txt', 'pdf', 'doc', 'docx', 'md'],
            help="支持文本、PDF、Word和Markdown文件"
        )

        if uploaded_file:
            title = st.text_input("文档标题", value=uploaded_file.name)
            tags = st.text_input("标签 (用逗号分隔)", placeholder="例如: 机器学习,Python,教程")

            if st.button("添加到知识库"):
                with st.spinner("正在处理文档..."):
                    # 保存临时文件
                    temp_path = f"temp_{uploaded_file.name}"
                    with open(temp_path, "wb") as f:
                        f.write(uploaded_file.getbuffer())

                    # 添加到知识库
                    doc_id, chunk_count = kb.add_document(temp_path, title, tags)

                    # 清理临时文件
                    os.remove(temp_path)

                    if doc_id:
                        st.success(f"✅ 文档添加成功！共生成 {chunk_count} 个文本块。")
                    else:
                        st.error("❌ 文档添加失败，请检查文件格式。")

        # 显示已有文档
        st.subheader("已有文档")
        documents = kb.db.execute("""
            SELECT id, title, file_path, tags, created_at
            FROM documents
            ORDER BY created_at DESC
        """).fetchall()

        if documents:
            df = pd.DataFrame(documents, columns=["ID", "标题", "文件路径", "标签", "创建时间"])
            st.dataframe(df[["标题", "标签", "创建时间"]], use_container_width=True)
        else:
            st.info("暂无文档，请先上传文档。")

    elif page == "🔍 知识搜索":
        st.header("🔍 知识搜索")

        query = st.text_input("输入搜索关键词", placeholder="例如: 机器学习算法")
        limit = st.slider("搜索结果数量", min_value=1, max_value=20, value=10)

        if st.button("搜索") and query:
            with st.spinner("正在搜索..."):
                results = kb.search(query, limit)

                if results:
                    st.success(f"找到 {len(results)} 个相关结果：")

                    for i, (chunk_id, doc_id, chunk_content, title, file_path, tags, similarity, chunk_index) in enumerate(results, 1):
                        similarity_score = (1 - similarity) * 100
                        with st.expander(f"📄 {title} (相似度: {similarity_score:.1f}%)"):
                            st.write(chunk_content)
                            st.caption(f"来源: {title} | 标签: {tags}")
                else:
                    st.warning("未找到相关内容，请尝试其他关键词。")

    elif page == "💬 智能问答":
        st.header("💬 智能问答")

        st.info("💡 基于您的知识库内容回答问题，系统会自动查找相关信息并生成答案。")

        question = st.text_input("输入您的问题", placeholder="例如: Python中的列表推导式是什么？")
        context_limit = st.slider("参考文档数量", min_value=1, max_value=10, value=5)

        if st.button("提问") and question:
            with st.spinner("正在思考..."):
                answer = kb.ask_question(question, context_limit)
                st.success("🤖 回答：")
                st.write(answer)

    elif page == "📊 统计信息":
        st.header("📊 知识库统计")

        stats = kb.get_statistics()

        col1, col2, col3 = st.columns(3)

        with col1:
            st.metric("📄 文档总数", stats["documents"])

        with col2:
            st.metric("🧩 文本块总数", stats["chunks"])

        with col3:
            st.metric("🔍 搜索次数", stats["searches"])

        # 搜索历史
        st.subheader("最近搜索记录")
        search_history = kb.db.execute("""
            SELECT query, timestamp
            FROM search_history
            ORDER BY timestamp DESC
            LIMIT 10
        """).fetchall()

        if search_history:
            df_history = pd.DataFrame(search_history, columns=["搜索查询", "时间"])
            st.dataframe(df_history, use_container_width=True)
        else:
            st.info("暂无搜索记录。")

if __name__ == "__main__":
    main()

🎓 进阶技巧与最佳实践

1. 性能优化

class OptimizedVectorStore:
    def __init__(self, db_path):
        self.db = sqlite3.connect(db_path)
        self.db.enable_load_extension(True)
        sqlite_vec.load(self.db)

        # 性能优化设置
        self.db.execute("PRAGMA journal_mode = WAL")  # 提高并发性能
        self.db.execute("PRAGMA synchronous = NORMAL")  # 平衡性能和安全性
        self.db.execute("PRAGMA cache_size = 10000")  # 增大缓存

    def batch_insert_embeddings(self, embeddings_batch):
        """批量插入向量，提高性能"""
        self.db.execute("BEGIN TRANSACTION")
        try:
            for embedding_data in embeddings_batch:
                self.db.execute("""
                    INSERT INTO embeddings (id, embedding, metadata)
                    VALUES (?, ?, ?)
                """, embedding_data)
            self.db.commit()
        except Exception as e:
            self.db.rollback()
            raise e

2. 向量索引优化

def optimize_vector_index(self):
    """优化向量索引"""
    # 创建额外的索引来加速查询
    self.db.execute("""
        CREATE INDEX IF NOT EXISTS idx_documents_category
        ON documents(category)
    """)

    # 定期优化数据库
    self.db.execute("VACUUM")
    self.db.execute("ANALYZE")

3. 错误处理和重试机制

import time
from functools import wraps

def retry_on_failure(max_retries=3, delay=1):
    """装饰器：失败重试机制"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise e
                    print(f"第 {attempt + 1} 次尝试失败，{delay}秒后重试...")
                    time.sleep(delay)
            return None
        return wrapper
    return decorator

class RobustVectorStore:
    @retry_on_failure(max_retries=3, delay=2)
    def add_document(self, document):
        """添加文档，带重试机制"""
        # 实现代码...
        pass

4. 监控和日志

import logging

class MonitoredVectorStore:
    def __init__(self, db_path):
        self.setup_logging()
        # ... 其他初始化代码

    def setup_logging(self):
        """设置日志"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('vector_store.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)

    def search_documents(self, query):
        """带监控的搜索"""
        start_time = time.time()

        try:
            results = self._perform_search(query)
            duration = time.time() - start_time

            self.logger.info(f"搜索完成: 查询='{query}', 结果数={len(results)}, 耗时={duration:.2f}s")
            return results

        except Exception as e:
            self.logger.error(f"搜索失败: 查询='{query}', 错误={str(e)}")
            raise

🔮 未来展望和发展方向

1. 多模态向量搜索

• 图像向量化（CLIP模型）
• 音频向量化
• 跨模态搜索（文字搜图片，图片搜文字）

2. 分布式部署

• sqlite-vec集群化
• 向量数据分片策略
• 负载均衡

3. 实时更新

• 增量向量化
• 实时索引更新
• 版本控制

4. 隐私保护

• 本地向量化
• 端到端加密
• 差分隐私

📋 总结

恭喜你！现在你已经掌握了使用LangChain + sqlite-vec构建文档向量搜索系统的核心技能：

🎯 核心要点回顾

1. 向量化基础：理解了什么是向量搜索及其应用场景
2. sqlite-vec：掌握了这个轻量级向量数据库的使用方法
3. LangChain集成：学会了如何将强大的文本处理能力与向量存储结合
4. 实战应用：构建了完整的智能问答和知识库系统
5. 进阶技巧：了解了性能优化、错误处理等最佳实践

🚀 下一步建议

1. 扩展功能：尝试添加更多文档格式支持（如Excel、PPT）
2. 用户界面：使用Streamlit或FastAPI构建Web应用
3. 性能优化：实现缓存机制和批量处理
4. 生产部署：考虑Docker容器化和云部署

💡 最后的小贴士

• 向量搜索不是银弹，对某些结构化查询，传统数据库可能更合适
• 注意数据隐私，特别是使用云端嵌入服务时
• 定期备份你的向量数据库
• 保持对新技术的关注，向量搜索领域发展很快！

---

🎉 恭喜你完成了这个教程！ 现在你已经具备了构建智能文档搜索系统的基础能力。记住，最好的学习方式是实践 - 赶快动手试试吧！

相关资源：

• sqlite-vec官方文档
• LangChain文档
• OpenAI API文档

有问题？ 欢迎在评论区留言，或者查看常见问题解答！📚✨