基于向量搜索与LLM构建个人AI记忆库与自动化工作流实践

在实际项目中，我们经常需要处理大量零散的技术笔记、代码片段、会议纪要和待办事项。这些信息分散在各个地方，查找困难，更难以形成体系化的知识。作为一名工程师，你是否想过构建一个属于自己的、能够理解你所有技术文档的“第二大脑”？这个大脑不仅能帮你快速检索，还能基于你的历史记录，自动化处理一些重复性工作，比如根据你的代码风格生成新的函数，或者将会议纪要自动整理成待办清单。

这正是“个人AI记忆库与自动化工作流”的核心价值。它不是一个现成的产品，而是一个需要你亲手搭建的工程系统。其核心思想是：将你所有的非结构化文本数据（笔记、代码、邮件等）进行向量化存储，构建一个语义搜索库；然后，利用大语言模型的代码生成、文本理解和推理能力，基于这个记忆库为你提供智能问答、内容生成和流程自动化服务。

本文将聚焦于使用 OpenAI 的 Codex 模型（或其后续演进模型，如 GPT-3.5/4 的代码补全能力）作为核心引擎，来实践这一构想。我们将从零开始，搭建一个本地的个人知识库后端，集成向量数据库，并通过 API 调用 Codex 来实现“记忆检索 -> 智能生成 -> 自动化执行”的闭环。整个过程涉及环境搭建、数据预处理、向量化、API 集成和简单的自动化脚本编写。学完后，你将掌握构建此类 AI 增强型个人工具的核心方法论，并能根据自身需求进行定制和扩展。

1. 理解核心概念：记忆库、向量搜索与 AI 工作流引擎

在动手之前，需要厘清几个关键概念，这决定了我们系统的架构设计。

1.1 什么是个人 AI 记忆库？

个人 AI 记忆库不是一个简单的文件备份或云盘。它的核心特征是语义化和可计算。

• 语义化：存储的不是原始文件的副本，而是其内容的数学表示（向量嵌入），使得系统能够理解“Java 线程池参数配置”和“如何优化多线程性能”之间的语义关联，而不仅仅是关键词匹配。
• 可计算：记忆库的内容可以作为上下文（Context）输入给大语言模型（LLM），让模型基于你的个人历史数据进行推理、总结和生成，产出高度个性化的结果。

一个典型的记忆库包含以下层级：

1. 原始数据层：你的 Markdown 笔记、PDF 文档、代码文件、网页剪藏等。
2. 处理与存储层：将原始数据分块、清洗，并转换为向量，存入向量数据库（如 Chroma, Pinecone, Weaviate）。
3. 检索层：根据用户查询，从向量数据库中找出最相关的文本片段。
4. 应用层：将检索到的片段作为上下文，连同用户指令一起发送给 LLM（如 Codex），得到最终答案或执行动作。

1.2 Codex 与 GPT 系列模型在工程中的角色

Codex 是 OpenAI 基于 GPT-3 微调的、专门用于理解和生成代码的模型。虽然 OpenAI 后续推出了功能更通用的 GPT-3.5/4 模型，并且 Codex 的 API 访问方式有所变化，但其核心能力——将自然语言指令转化为代码——仍然是构建自动化工作流的关键。

在当前的技术栈中，我们通常使用gpt-3.5-turbo-instruct或gpt-4的补全（Completion）API 来执行类似 Codex 的代码生成任务。在本文的语境下，当我们提及“使用 Codex”，实际指的是使用 OpenAI 提供的、具备强大代码生成能力的 Completion API。

它的角色是工作流引擎：

• 输入：一段自然语言指令 + （可选的）从记忆库检索到的相关上下文。
• 处理：理解指令，结合上下文进行推理。
• 输出：生成可执行的代码片段、结构化的数据（如 JSON）、或下一步的行动指令。

例如，你可以询问：“根据我上个月写的‘用户服务’代码风格，生成一个处理用户注册的 RESTful API 控制器。”系统会先从记忆库中找到你关于“用户服务”的代码片段和笔记，然后将这些上下文和你的指令一起发给 Codex，生成符合你习惯的 Spring Boot 控制器代码。

1.3 自动化工作流的设计模式

自动化工作流不是魔法，而是由清晰触发条件和执行动作组成的链条。常见模式有：

• 问答机器人：用户提问 -> 检索记忆 -> 生成回答。
• 代码生成器：描述功能 -> 检索类似代码模式 -> 生成新代码。
• 文档整理器：输入杂乱文本 -> 提取关键信息 -> 生成结构化摘要或待办项。
• 定时任务：定期扫描新数据（如 RSS、邮箱） -> 处理并存入记忆库 -> 发送摘要通知。

我们的项目将实现前两种模式，并为你扩展后两种提供基础框架。

2. 环境准备与项目初始化

我们将使用 Python 作为主要开发语言，因为它有丰富的 AI 和数据处理库。项目将分为几个核心模块：数据加载器、文本分割器、向量化模块、向量数据库、检索器和 OpenAI 客户端。

2.1 基础环境与依赖

首先，确保你的 Python 版本在 3.8 以上。建议使用虚拟环境隔离项目依赖。

# 创建项目目录并进入 mkdir personal_ai_memory && cd personal_ai_memory # 创建虚拟环境（以 venv 为例） python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate

接下来，创建requirements.txt文件，列出核心依赖。我们将使用openai官方库、langchain框架（简化流程）和chromadb（轻量级向量数据库）。

# requirements.txt openai>=1.0.0 langchain>=0.1.0 langchain-openai>=0.0.5 chromadb>=0.4.0 tiktoken>=0.5.0 # 用于文本分词和计数 pypdf>=3.0.0 # 用于处理PDF python-dotenv>=1.0.0 # 用于管理环境变量

使用 pip 安装：

pip install -r requirements.txt

注意：langchain是一个快速发展的框架，其 API 在版本间可能有较大变化。本文基于相对稳定的模式编写，如果遇到导入错误，请查阅其官方文档调整。

2.2 获取并配置 OpenAI API 密钥

你需要一个有效的 OpenAI API 密钥。访问 OpenAI 平台网站，注册并创建 API Key。

安全警告：API Key 是私密凭证，绝不能提交到代码仓库。

我们将使用.env文件来管理密钥。在项目根目录创建.env文件：

# .env OPENAI_API_KEY=sk-your-actual-api-key-here

然后在代码中使用python-dotenv加载：

# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") if not OPENAI_API_KEY: raise ValueError("请在 .env 文件中设置 OPENAI_API_KEY 环境变量")

2.3 项目结构设计

一个清晰的项目结构有助于维护。建议如下：

personal_ai_memory/ ├── .env # 环境变量（忽略提交） ├── requirements.txt # 项目依赖 ├── config.py # 配置文件 ├── main.py # 主程序入口 ├── core/ # 核心模块 │ ├── __init__.py │ ├── document_loader.py # 文档加载器 │ ├── text_splitter.py # 文本分割器 │ ├── vector_store.py # 向量数据库封装 │ └── retrieval_qa.py # 检索与问答链 ├── workflows/ # 自动化工作流示例 │ ├── __init__.py │ ├── code_generator.py # 代码生成工作流 │ └── note_summarizer.py # 笔记总结工作流 ├── data/ # 存放原始文档（如笔记、代码） │ └── your_notes.md └── storage/ # 向量数据库持久化目录（由 Chroma 自动创建）

3. 构建个人记忆库：从文档到向量

记忆库的构建是基础，也是最耗时的步骤。我们需要将各种格式的文档转换为统一的文本块，并生成向量嵌入。

3.1 实现通用文档加载器

我们首先实现一个能处理多种格式的文档加载器。这里使用langchain提供的工具。

# core/document_loader.py from langchain_community.document_loaders import TextLoader, PyPDFLoader from langchain.schema import Document from typing import List, Union import os class DocumentLoader: """通用文档加载器，支持 txt, md, pdf 等格式""" @staticmethod def load_file(file_path: str) -> List[Document]: """ 根据文件后缀选择加载器，返回 Document 列表。 Document 是 langchain 的标准数据结构，包含 page_content 和 metadata。 """ if not os.path.exists(file_path): raise FileNotFoundError(f"文件不存在: {file_path}") ext = os.path.splitext(file_path)[1].lower() if ext in ['.txt', '.md', '.rst']: loader = TextLoader(file_path, encoding='utf-8') elif ext == '.pdf': loader = PyPDFLoader(file_path) else: # 可以扩展支持更多格式，如 Docx, HTML raise ValueError(f"暂不支持的文件格式: {ext}") documents = loader.load() # 为每个文档添加来源元数据 for doc in documents: doc.metadata['source'] = file_path return documents @staticmethod def load_directory(directory_path: str, allowed_extensions: List[str] = None) -> List[Document]: """ 加载目录下所有指定格式的文件。 """ if allowed_extensions is None: allowed_extensions = ['.txt', '.md', '.pdf'] all_documents = [] for root, _, files in os.walk(directory_path): for file in files: file_path = os.path.join(root, file) ext = os.path.splitext(file_path)[1].lower() if ext in allowed_extensions: try: docs = DocumentLoader.load_file(file_path) all_documents.extend(docs) print(f"成功加载: {file_path}") except Exception as e: print(f"加载文件失败 {file_path}: {e}") return all_documents

3.2 实现文本分割器

大语言模型有上下文长度限制（如 4K, 8K, 16K tokens）。我们必须将长文档分割成较小的、有语义的块。简单的按字符分割会破坏语义，推荐使用递归字符分割或按标记分割。

# core/text_splitter.py from langchain.text_splitter import RecursiveCharacterTextSplitter, TokenTextSplitter from langchain.schema import Document from typing import List class TextSplitter: """文本分割器，确保分割后的块保留一定语义""" def __init__(self, chunk_size: int = 1000, chunk_overlap: int = 200, separator: str = "\n\n"): """ Args: chunk_size: 每个文本块的最大字符数。 chunk_overlap: 块之间的重叠字符数，避免在句子中间切断。 separator: 优先用于分割的字符序列。 """ # 使用递归字符分割器，它尝试按段落、句子、单词的优先级分割 self._splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=chunk_overlap, separators=[separator, "\n", "。", "！", "？", "；", "，", " ", ""] ) def split_documents(self, documents: List[Document]) -> List[Document]: """将 Document 列表分割成更小的块""" return self._splitter.split_documents(documents)

关键参数解释：

• chunk_size：太小会丢失上下文，太大会超出模型限制并增加检索噪声。对于代码，500-1000 字符较好；对于普通文本，800-1500 字符较合适。
• chunk_overlap：重叠部分保证了上下文连贯性，例如一个概念在块末尾被提及，在下一个块开头还能看到，有助于模型理解。

3.3 集成向量数据库（Chroma）

我们将使用 ChromaDB，一个轻量级、可嵌入、开源的向量数据库。它非常适合个人项目。

# core/vector_store.py import chromadb from chromadb.config import Settings from langchain.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain.schema import Document from typing import List, Optional import os class VectorStoreManager: """向量数据库管理类，负责存储和检索向量""" def __init__(self, persist_directory: str = "./storage/chroma_db", collection_name: str = "personal_memory"): self.persist_directory = persist_directory self.collection_name = collection_name # 使用 OpenAI 的 text-embedding-ada-002 模型生成向量 self.embedding_function = OpenAIEmbeddings( model="text-embedding-ada-002", openai_api_key=os.getenv("OPENAI_API_KEY") ) # 初始化 Chroma 客户端，设置持久化路径 self.vector_store = Chroma( collection_name=collection_name, embedding_function=self.embedding_function, persist_directory=persist_directory, client_settings=Settings( chroma_db_impl="duckdb+parquet", persist_directory=persist_directory ) ) def add_documents(self, documents: List[Document]): """将文档块添加到向量数据库""" # Chroma.from_documents 会创建集合并添加文档 # 注意：如果集合已存在，此操作会添加新文档，但可能产生重复。 # 对于初始化或全量重建，可以先删除旧集合。 self.vector_store = Chroma.from_documents( documents=documents, embedding=self.embedding_function, collection_name=self.collection_name, persist_directory=self.persist_directory ) print(f"已添加 {len(documents)} 个文档块到向量数据库。") def similarity_search(self, query: str, k: int = 4) -> List[Document]: """语义搜索，返回最相关的 k 个文档块""" return self.vector_store.similarity_search(query, k=k) def delete_collection(self): """删除整个集合（谨慎操作）""" self.vector_store.delete_collection() print(f"集合 '{self.collection_name}' 已删除。")

3.4 构建记忆库的完整流程

现在，我们将上述模块串联起来，编写一个构建记忆库的脚本。

# build_memory.py import sys sys.path.append('.') from core.document_loader import DocumentLoader from core.text_splitter import TextSplitter from core.vector_store import VectorStoreManager from config import OPENAI_API_KEY import os def build_memory_from_directory(data_dir: str = "./data"): """ 从指定目录加载文档，处理并存入向量数据库。 """ print("开始构建个人记忆库...") # 1. 加载文档 print("步骤1: 加载文档...") raw_documents = DocumentLoader.load_directory(data_dir) print(f"共加载 {len(raw_documents)} 个原始文档。") # 2. 分割文本 print("步骤2: 分割文本...") splitter = TextSplitter(chunk_size=800, chunk_overlap=150) all_splits = splitter.split_documents(raw_documents) print(f"文档被分割成 {len(all_splits)} 个文本块。") # 3. 生成向量并存储 print("步骤3: 生成向量并存储到数据库...") vs_manager = VectorStoreManager() # 为避免重复，先删除旧集合（生产环境应考虑增量更新） try: vs_manager.delete_collection() except: pass # 集合可能不存在 vs_manager.add_documents(all_splits) print("个人记忆库构建完成！") print(f"向量数据库持久化在: {vs_manager.persist_directory}") if __name__ == "__main__": # 假设你的文档都放在 ./data 目录下 build_memory_from_directory("./data")

运行此脚本前，请在./data目录下放置一些你的 Markdown 笔记或 PDF 文件。然后执行：

python build_memory.py

如果一切顺利，你会看到控制台输出加载、分割和存储的进度，并在./storage/chroma_db目录下看到数据库文件。

4. 实现智能问答与代码生成工作流

记忆库建好后，我们就可以利用它来回答问题或生成代码了。核心是检索增强生成（Retrieval-Augmented Generation, RAG）模式。

4.1 实现检索增强生成（RAG）链

我们使用langchain的链（Chain）来简化 RAG 流程。

# core/retrieval_qa.py from langchain.chains import RetrievalQA from langchain_openai import OpenAI from core.vector_store import VectorStoreManager from config import OPENAI_API_KEY import os class MemoryQA: """基于记忆库的问答系统""" def __init__(self): # 初始化向量存储 self.vs_manager = VectorStoreManager() # 初始化 LLM，这里使用 gpt-3.5-turbo-instruct 模拟 Codex 的补全能力 # 对于纯代码生成，completion 模型更合适。 self.llm = OpenAI( model="gpt-3.5-turbo-instruct", # 或 "gpt-4" temperature=0.1, # 低 temperature 使输出更确定，适合代码 openai_api_key=OPENAI_API_KEY ) # 构建检索问答链 self.qa_chain = RetrievalQA.from_chain_type( llm=self.llm, chain_type="stuff", # 将检索到的所有文档“塞”进上下文 retriever=self.vs_manager.vector_store.as_retriever(search_kwargs={"k": 4}), return_source_documents=True # 返回来源文档，便于追溯 ) def ask(self, question: str) -> dict: """ 向记忆库提问。 返回一个字典，包含答案和来源文档。 """ result = self.qa_chain({"query": question}) return { "answer": result["result"], "source_documents": result["source_documents"] }

4.2 代码生成工作流示例

现在，我们创建一个专门用于代码生成的工作流。它会先检索记忆中相关的代码片段和规范，然后指导 LLM 生成新代码。

# workflows/code_generator.py from core.retrieval_qa import MemoryQA from langchain.prompts import PromptTemplate from langchain.chains import LLMChain from config import OPENAI_API_KEY import os class CodeGenerator: """代码生成工作流""" def __init__(self): self.qa_system = MemoryQA() # 定义一个更专注于代码生成的提示模板 self.code_prompt = PromptTemplate( input_variables=["context", "task"], template=""" 你是一个经验丰富的软件工程师，擅长根据已有的代码模式和规范编写新的代码。 以下是从我的个人知识库中检索到的相关代码片段和笔记： {context} 请根据以上上下文和我提供的任务描述，生成符合我个人编码风格的代码。 任务描述： {task} 要求： 1. 代码风格（如命名、注释、结构）应与上下文中的示例保持一致。 2. 只输出最终的代码块，不要额外解释。 3. 如果任务描述不清晰或上下文不足，请基于最佳实践生成，并添加 TODO 注释。 生成的代码： """ ) self.llm = self.qa_system.llm self.chain = LLMChain(llm=self.llm, prompt=self.code_prompt) def generate(self, task_description: str) -> str: """ 根据任务描述生成代码。 1. 先用任务描述本身检索相关记忆。 2. 将检索到的内容作为上下文，连同任务描述一起发送给 LLM。 """ # 步骤1：检索相关记忆 retrieval_result = self.qa_system.ask(task_description) context_text = "\n---\n".join([doc.page_content for doc in retrieval_result["source_documents"]]) # 步骤2：使用定制提示模板生成代码 generated_code = self.chain.run({ "context": context_text, "task": task_description }) return { "generated_code": generated_code.strip(), "sources": retrieval_result["source_documents"] }

4.3 运行与测试

创建一个主程序来测试我们的系统。

# main.py from workflows.code_generator import CodeGenerator from core.retrieval_qa import MemoryQA import sys def main(): print("=== 个人 AI 记忆库与工作流系统 ===") mode = input("请选择模式: 1-智能问答, 2-代码生成 (输入 1 或 2): ").strip() if mode == "1": qa_system = MemoryQA() print("问答模式已启动。输入 'quit' 退出。") while True: question = input("\n你的问题: ").strip() if question.lower() == 'quit': break answer_data = qa_system.ask(question) print(f"\n答案: {answer_data['answer']}") print(f"\n来源文档:") for i, doc in enumerate(answer_data['source_documents'][:2]): # 显示前两个来源 print(f"[{i+1}] {doc.metadata.get('source', '未知')} (片段): {doc.page_content[:200]}...") elif mode == "2": code_gen = CodeGenerator() print("代码生成模式已启动。输入 'quit' 退出。") while True: task = input("\n描述你想要生成的代码功能: ").strip() if task.lower() == 'quit': break result = code_gen.generate(task) print(f"\n生成的代码:\n```\n{result['generated_code']}\n```") if result['sources']: print(f"\n参考了以下记忆:") for doc in result['sources'][:2]: print(f"- {doc.metadata.get('source', '未知')}") else: print("无效选择。") if __name__ == "__main__": main()

运行python main.py，选择模式进行测试。

测试示例 1（智能问答）：

• 假设你的记忆库中有关于“Python 装饰器”的笔记。
• 提问：“Python 装饰器有什么作用？请用我笔记里的例子解释。”
• 系统会检索相关笔记，并生成结合上下文的回答。

测试示例 2（代码生成）：

• 假设你的记忆库中有你写的“使用 FastAPI 创建用户登录接口”的代码。
• 输入任务：“参考我之前的 FastAPI 用户登录代码，写一个用户注销的接口。”
• 系统会检索到你的登录接口代码，并生成风格类似的注销接口代码。

5. 关键配置、参数详解与性能调优

构建和运行系统时，以下参数对效果和成本有直接影响。

5.1 文本分割参数调优表

5.2 OpenAI API 模型与参数选择

5.3 向量检索相关参数

6. 常见问题排查与解决方案

在实际搭建和运行过程中，你可能会遇到以下问题。

6.1 环境与依赖问题

6.2 数据处理与检索问题

6.3 API 使用与性能问题

7. 生产环境最佳实践与扩展方向

将个人玩具项目升级为稳定可用的生产工具，还需要考虑以下方面。

7.1 安全与成本控制

• API 密钥管理：绝对不要将.env文件提交到 Git。使用.gitignore忽略它。在生产服务器上，使用环境变量或专业的密钥管理服务。
• 用量监控与限流：OpenAI API 按 token 收费。实现简单的用量日志，监控每日消耗。对于个人使用，可以在代码中设置每月预算上限，超出后自动禁用。
• 输入输出审查：避免向模型发送敏感信息（密码、密钥、个人隐私）。对模型的输出，特别是代码，要进行安全审查后再执行。

7.2 系统健壮性提升

• 增量更新记忆库：每次重新构建整个库效率低下。实现增量更新逻辑，仅处理新增或修改的文件。可以为每个文档块计算哈希值，仅存储和索引哈希值变化的块。
• 错误处理与重试：网络请求、API 调用都可能失败。使用tenacity等库为关键操作（如openai.Completion.create）添加重试机制。
• 日志记录：记录关键操作（如文档加载、向量化、查询）和错误信息，便于排查问题。可以使用 Python 的logging模块。

7.3 扩展工作流类型

本文实现了问答和代码生成，你可以在此基础上扩展更多自动化场景：

1. 会议纪要处理器：连接日历 API，在会议结束后自动将录音转文字，总结要点并生成待办事项存入记忆库。
2. 学习助手：当你阅读技术文章时，浏览器插件自动抓取内容，处理后存入记忆库，并生成知识卡片提问你。
3. 自动化代码审查：与 Git 钩子结合，在提交代码时，自动检索项目规范和历史代码，对本次提交生成审查意见。
4. 日报/周报生成器：检索你过去一天/一周的笔记、代码提交记录、完成的任务，自动生成工作总结。

7.4 技术栈升级选型

随着需求复杂，可以考虑升级部分组件：

构建个人 AI 记忆库与自动化工作流是一个迭代过程。从最小可行产品（MVP）开始，先让一两个核心流程跑通，再逐步加入新的数据源、工作流和优化。这个系统的价值会随着你数据的积累和使用的深入而指数级增长，最终成为你不可或缺的“第二大脑”。