在项目迭代和日常开发中我们常常需要向AI助手如ChatGPT、Claude、DeepSeek等反复提供项目背景、代码片段、API文档等上下文信息。每次开启新对话都像“从头开始”手动复制粘贴既低效又容易遗漏关键细节。你是否也渴望一个能自动记录、整理并智能关联工作上下文的“AI记忆卡”本文将分享一套完整的解决方案通过构建一个轻量级的本地“AI记忆卡”系统自动收集你的工作进度、代码变更、会议纪要和灵感碎片并在与AI交互时自动附上相关上下文。这套方案基于开源工具链无需复杂部署适合开发者、产品经理和任何需要频繁使用AI辅助工作的知识工作者。学完后你将掌握从环境搭建、数据收集、向量化存储到智能检索集成的全流程实战能力。1. 背景与核心概念为什么需要“AI记忆卡”1.1 当前AI协作的痛点现代AI大模型在代码生成、问题排查、方案设计等方面表现出色但其对话本质上是“无状态”的。这意味着上下文丢失每次新对话AI都不知道你之前讨论过什么、项目进展到哪一步。信息重复输入你需要反复解释项目结构、技术栈、业务规则。知识碎片化有价值的工作记录、决策思路散落在各个聊天记录和本地文件中难以系统化利用。1.2 “AI记忆卡”是什么“AI记忆卡”是一个比喻它指的是一套本地化、自动化的工作上下文管理与注入系统。其核心功能是自动收集监控你的工作环境如代码仓库、笔记文档、通讯工具自动抓取结构化信息。智能存储将收集的信息转化为向量Embedding存入向量数据库实现语义化检索。按需注入在你与AI助手对话时系统自动根据你的问题从记忆库中检索最相关的上下文片段并拼接到提示词Prompt中实现“有记忆”的对话。1.3 核心价值与应用场景对开发者自动关联当前Git提交、代码片段、错误日志让AI精准理解bug上下文。对团队共享项目文档、API设计稿、会议纪要让AI基于最新团队共识提供建议。对个人链接你的知识库、学习笔记、待办清单让AI成为你的个性化第二大脑。 与手动整理相比这套系统实现了从“被动喂养”到“主动关联”的转变显著提升AI协作的深度和效率。2. 环境准备与版本说明我们将使用Python作为主要开发语言结合一系列轻量级开源库。请确保你的环境满足以下要求。2.1 基础环境操作系统Windows 10/11, macOS 10.15, 或主流Linux发行版如Ubuntu 20.04。Python版本 3.8 - 3.11。推荐使用3.9或3.10以获得最佳兼容性。包管理工具pip通常随Python安装。版本控制Git用于监控代码变更。可选IDEVS Code, PyCharm等。2.2 核心依赖库及版本我们将创建一个requirements.txt文件来管理依赖。以下是核心库及其作用# 核心框架与异步 fastapi0.104.1 uvicorn[standard]0.24.0 pydantic2.5.0 # 文件监控与系统交互 watchfiles0.20.0 python-dotenv1.0.0 # 文本处理与加载 langchain0.0.340 langchain-community0.0.10 # 包含多种文档加载器 unstructured0.12.0 # 用于解析多种文档格式 # 向量数据库与嵌入模型 chromadb0.4.18 # 轻量级向量数据库 sentence-transformers2.2.2 # 本地运行嵌入模型 # 或者使用OpenAI嵌入需API Key # openai1.3.0 # 前端界面可选 streamlit1.28.0版本说明以上版本在撰写时经过测试能保证基本功能运行。由于开源库迭代迅速实际使用时若遇到兼容性问题可适当调整次要版本。重点在于理解各库的职责和配置方式。2.3 项目结构预览在开始前我们先规划项目目录这有助于理解后续的代码组织。ai_memory_card/ ├── .env # 环境变量如API密钥 ├── requirements.txt # 项目依赖 ├── app.py # FastAPI主应用入口 ├── config.py # 配置文件 ├── memory_core/ # 核心模块 │ ├── __init__.py │ ├── collector.py # 数据收集器 │ ├── embedder.py # 嵌入模型管理 │ ├── vector_store.py # 向量数据库操作 │ └── retriever.py # 检索器 ├── sources/ # 监控的数据源目录示例 │ ├── code/ │ ├── notes/ │ └── logs/ ├── storage/ # 向量数据库持久化目录 │ └── chroma_db/ └── scripts/ # 工具脚本 └── init_memory.py # 初始化记忆库脚本接下来我们开始搭建系统核心。3. 核心模块拆解与原理“AI记忆卡”系统主要由四个核心模块组成收集器Collector、嵌入器Embedder、向量存储Vector Store和检索器Retriever。3.1 收集器Collector自动捕获工作上下文收集器的职责是从不同源头抓取数据。我们采用基于事件如文件变化和基于轮询如定时读取Git日志的混合策略。关键设计文件系统监控使用watchfiles库监听sources/目录下的文件变动增、删、改实时触发处理流程。Git钩子集成在项目的.git/hooks/post-commit中注入脚本在每次提交后自动捕获提交信息、差异代码作为记忆片段。文档加载器利用LangChain的UnstructuredFileLoader、TextLoader等支持解析.txt,.md,.py,.pdf等多种格式。元数据附加为每一段文本记忆片段附加来源、时间戳、类型代码/笔记/日志等元数据便于后续筛选。3.2 嵌入器Embedder与向量存储这是实现语义检索的核心。我们将文本转换为计算机能理解的数值向量。工作流程文本分块长文档需要被切割成大小适中的片段如500字符同时保持语义连贯。LangChain的RecursiveCharacterTextSplitter是常用工具。向量化使用嵌入模型将文本块转换为固定维度的向量如384维或768维。我们优先选用本地模型如all-MiniLM-L6-v2它平衡了速度与精度且无需网络调用和付费。向量存储将向量及其对应的文本块、元数据存入ChromaDB。ChromaDB 是一个轻量级、可持久化的向量数据库支持按集合Collection组织数据非常适合个人或小团队使用。3.3 检索器Retriever智能关联问题与记忆当用户提出问题时检索器负责找到最相关的记忆片段。检索过程将用户问题同样转换为向量。在向量数据库中进行相似性搜索通常使用余弦相似度。返回相似度最高的前k个文本片段如前3个。混合检索除了向量检索还可以结合元数据过滤例如只检索“代码”类型的记忆或最近一周的记忆使结果更精准。4. 完整实战构建你的第一张“AI记忆卡”现在我们将一步步实现上述系统。请跟随操作并注意代码中的注释。4.1 初始化项目与环境首先创建项目目录并安装依赖。# 1. 创建项目目录 mkdir ai_memory_card cd ai_memory_card # 2. 创建虚拟环境推荐 python -m venv venv # Windows 激活: venv\Scripts\activate # macOS/Linux 激活: source venv/bin/activate # 3. 创建 requirements.txt 并写入上一节的依赖内容 # 可以使用编辑器创建或使用 echo 命令注意版本号 # 4. 安装依赖 pip install -r requirements.txt # 5. 创建项目结构所需目录 mkdir -p memory_core sources/{code,notes,logs} storage/chroma_db scripts4.2 编写核心配置文件创建config.py集中管理配置参数。# config.py import os from pathlib import Path from pydantic_settings import BaseSettings class Settings(BaseSettings): # 项目路径 BASE_DIR: Path Path(__file__).parent SOURCES_DIR: Path BASE_DIR / sources STORAGE_DIR: Path BASE_DIR / storage VECTOR_DB_PATH: Path STORAGE_DIR / chroma_db # 向量数据库配置 COLLECTION_NAME: str work_memory EMBEDDING_MODEL: str all-MiniLM-L6-v2 # 本地句子嵌入模型 # 如果使用OpenAI请取消注释并设置API Key # OPENAI_API_KEY: str # EMBEDDING_MODEL: str text-embedding-ada-002 # 文本处理配置 CHUNK_SIZE: int 500 # 文本分块大小 CHUNK_OVERLAP: int 50 # 块之间重叠字符数 # 检索配置 RETRIEVE_TOP_K: int 3 # 每次检索返回的记忆片段数量 # 文件监控配置 WATCH_PATTERNS: list [*.md, *.txt, *.py, *.js, *.java, *.json] class Config: env_file .env # 从.env文件加载环境变量 settings Settings()4.3 实现嵌入与向量存储模块创建memory_core/embedder.py和memory_core/vector_store.py。# memory_core/embedder.py from sentence_transformers import SentenceTransformer from langchain.embeddings import HuggingFaceEmbeddings import numpy as np from config import settings import logging logger logging.getLogger(__name__) class LocalEmbedder: 本地嵌入模型封装 def __init__(self): model_name settings.EMBEDDING_MODEL logger.info(f正在加载嵌入模型: {model_name}) # 使用LangChain封装的HuggingFace嵌入兼容性更好 self.embeddings HuggingFaceEmbeddings( model_namefsentence-transformers/{model_name}, model_kwargs{device: cpu}, # 有GPU可改为 cuda encode_kwargs{normalize_embeddings: True} # 归一化便于余弦相似度计算 ) logger.info(嵌入模型加载完毕。) def embed_documents(self, texts: list[str]) - list[list[float]]: 将一批文本转换为向量 return self.embeddings.embed_documents(texts) def embed_query(self, text: str) - list[float]: 将单个查询文本转换为向量 return self.embeddings.embed_query(text) # 全局嵌入器实例 embedder LocalEmbedder()# memory_core/vector_store.py import chromadb from chromadb.config import Settings as ChromaSettings from typing import List, Dict, Any from config import settings import logging from memory_core.embedder import embedder logger logging.getLogger(__name__) class VectorStoreManager: 向量数据库管理类 def __init__(self): self.client chromadb.PersistentClient( pathstr(settings.VECTOR_DB_PATH), settingsChromaSettings(anonymized_telemetryFalse) # 禁用匿名数据收集 ) self.collection self.client.get_or_create_collection( namesettings.COLLECTION_NAME, metadata{description: AI工作记忆存储} ) logger.info(f向量数据库连接成功集合: {settings.COLLECTION_NAME}) def add_memories(self, documents: List[str], metadatas: List[Dict], ids: List[str]): 添加记忆片段到向量数据库 if not documents: return # 生成嵌入向量 embeddings embedder.embed_documents(documents) # 添加到集合 self.collection.add( embeddingsembeddings, documentsdocuments, metadatasmetadatas, idsids ) logger.info(f成功添加 {len(documents)} 条记忆。) def search(self, query: str, filter_metadata: Dict None, top_k: int None) - List[Dict[str, Any]]: 检索与查询最相关的记忆 if top_k is None: top_k settings.RETRIEVE_TOP_K # 将查询文本向量化 query_embedding embedder.embed_query(query) # 执行搜索 results self.collection.query( query_embeddings[query_embedding], n_resultstop_k, wherefilter_metadata # 元数据过滤条件 ) # 格式化结果 memories [] if results[documents]: for i in range(len(results[documents][0])): memories.append({ content: results[documents][0][i], metadata: results[metadatas][0][i], distance: results[distances][0][i] # 距离越小越相似 }) return memories def list_all_collections(self): 列出所有集合用于调试 return self.client.list_collections() # 全局向量存储管理器实例 vector_store VectorStoreManager()4.4 实现文件收集器创建memory_core/collector.py实现一个监控指定目录的简易收集器。# memory_core/collector.py import hashlib import time from pathlib import Path from typing import List, Dict, Any from langchain.document_loaders import TextLoader, UnstructuredMarkdownLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from config import settings from memory_core.vector_store import vector_store import logging logger logging.getLogger(__name__) class FileCollector: 文件收集器 def __init__(self): self.text_splitter RecursiveCharacterTextSplitter( chunk_sizesettings.CHUNK_SIZE, chunk_overlapsettings.CHUNK_OVERLAP, length_functionlen, separators[\n\n, \n, 。, , , , , , ] ) def process_file(self, file_path: Path) - List[Dict[str, Any]]: 处理单个文件返回文本块和元数据列表 memories [] try: # 根据后缀选择加载器 if file_path.suffix.lower() in [.txt, .md, .py, .js, .java, .json]: if file_path.suffix .md: loader UnstructuredMarkdownLoader(str(file_path)) else: loader TextLoader(str(file_path), encodingutf-8) documents loader.load() for doc in documents: # 分割文本 chunks self.text_splitter.split_text(doc.page_content) for i, chunk in enumerate(chunks): if not chunk.strip(): continue # 为每个块生成唯一ID文件路径内容哈希 chunk_id f{file_path.stem}_{i}_{hashlib.md5(chunk.encode()).hexdigest()[:8]} # 构建元数据 metadata { source: str(file_path.relative_to(settings.BASE_DIR)), type: self._get_file_type(file_path), timestamp: time.time(), chunk_index: i } memories.append({ id: chunk_id, content: chunk, metadata: metadata }) logger.info(f处理文件 {file_path.name}生成 {len(chunks)} 个文本块。) else: logger.warning(f暂不支持的文件格式: {file_path.suffix}) except Exception as e: logger.error(f处理文件 {file_path} 时出错: {e}) return memories def _get_file_type(self, file_path: Path) - str: 根据文件后缀判断类型 suffix file_path.suffix.lower() if suffix in [.py, .js, .java, .cpp, .go]: return code elif suffix in [.md, .txt]: return note elif suffix in [.log, .err]: return log else: return other def process_directory(self, directory: Path) - int: 处理整个目录并存入向量数据库 all_memories [] if not directory.exists(): logger.warning(f目录不存在: {directory}) return 0 # 递归查找支持的文件 for pattern in settings.WATCH_PATTERNS: for file_path in directory.rglob(pattern): if file_path.is_file(): memories self.process_file(file_path) all_memories.extend(memories) # 批量添加到向量数据库 if all_memories: documents [m[content] for m in all_memories] metadatas [m[metadata] for m in all_memories] ids [m[id] for m in all_memories] vector_store.add_memories(documents, metadatas, ids) return len(all_memories) # 全局收集器实例 collector FileCollector()4.5 实现检索器与API接口创建memory_core/retriever.py和app.py提供检索API。# memory_core/retriever.py from typing import List, Dict, Any from config import settings from memory_core.vector_store import vector_store import logging logger logging.getLogger(__name__) class MemoryRetriever: 记忆检索器 def retrieve(self, query: str, source_filter: str None, type_filter: str None) - List[Dict[str, Any]]: 根据查询检索相关记忆。 Args: query: 查询文本 source_filter: 按来源过滤如 sources/notes/xxx.md type_filter: 按类型过滤如 code, note, log Returns: 相关记忆列表按相关性排序 # 构建元数据过滤条件 filter_metadata {} if source_filter: filter_metadata[source] {$contains: source_filter} if type_filter: filter_metadata[type] type_filter logger.info(f检索查询: {query}, 过滤器: {filter_metadata}) memories vector_store.search(query, filter_metadatafilter_metadata if filter_metadata else None) # 格式化输出 formatted_results [] for mem in memories: formatted_results.append({ content: mem[content][:200] ... if len(mem[content]) 200 else mem[content], source: mem[metadata].get(source, unknown), type: mem[metadata].get(type, unknown), relevance_score: round(1 - mem[distance], 4) # 将距离转换为相似度分数 }) return formatted_results def get_context_for_ai(self, query: str, max_chars: int 1500) - str: 为AI对话生成上下文提示。 将检索到的记忆片段拼接成一段连贯的上下文。 memories self.retrieve(query) if not memories: return 暂无相关历史上下文。\n context_parts [以下是你之前的相关工作记录供参考\n] total_chars 0 for i, mem in enumerate(memories, 1): mem_text f[{i}] 来源{mem[source]} (类型{mem[type]}, 相关度{mem[relevance_score]:.2%})\n{mem[content]}\n\n if total_chars len(mem_text) max_chars: break context_parts.append(mem_text) total_chars len(mem_text) context_parts.append(f\n--- 以上是基于你工作记忆的 {len(context_parts)-1} 条相关上下文 ---\n) return .join(context_parts) retriever MemoryRetriever()# app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional, List import uvicorn import logging from memory_core.retriever import retriever from memory_core.collector import collector from config import settings # 配置日志 logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) app FastAPI(titleAI记忆卡 API, description自动管理工作上下文并与AI集成的服务) class SearchRequest(BaseModel): query: str source_filter: Optional[str] None type_filter: Optional[str] None class SearchResponse(BaseModel): query: str memories: List[dict] count: int class IndexRequest(BaseModel): path: str # 相对于sources目录的路径或绝对路径 app.get(/) async def root(): return {message: AI记忆卡服务运行中, version: 1.0.0} app.post(/search, response_modelSearchResponse) async def search_memories(request: SearchRequest): 检索相关记忆 try: memories retriever.retrieve( queryrequest.query, source_filterrequest.source_filter, type_filterrequest.type_filter ) return SearchResponse( queryrequest.query, memoriesmemories, countlen(memories) ) except Exception as e: logger.error(f检索失败: {e}) raise HTTPException(status_code500, detailf检索失败: {str(e)}) app.post(/index) async def index_directory(request: IndexRequest): 手动索引一个目录或文件 import os from pathlib import Path target_path Path(request.path) if not target_path.is_absolute(): target_path settings.BASE_DIR / target_path if not target_path.exists(): raise HTTPException(status_code404, detailf路径不存在: {target_path}) try: if target_path.is_file(): # 索引单个文件 memories collector.process_file(target_path) if memories: documents [m[content] for m in memories] metadatas [m[metadata] for m in memories] ids [m[id] for m in memories] from memory_core.vector_store import vector_store vector_store.add_memories(documents, metadatas, ids) count len(memories) else: count 0 msg f文件已索引 else: # 索引目录 count collector.process_directory(target_path) msg f目录已索引 return {message: f{msg}新增 {count} 条记忆。} except Exception as e: logger.error(f索引失败: {e}) raise HTTPException(status_code500, detailf索引失败: {str(e)}) app.get(/generate-context) async def generate_context(query: str, max_chars: int 1500): 生成用于AI提示的上下文文本 try: context retriever.get_context_for_ai(query, max_chars) return {query: query, context: context} except Exception as e: logger.error(f生成上下文失败: {e}) raise HTTPException(status_code500, detailf生成上下文失败: {str(e)}) if __name__ __main__: logger.info(启动AI记忆卡服务...) uvicorn.run(app, host0.0.0.0, port8000)4.6 运行与验证现在让我们启动服务并进行测试。步骤1启动API服务# 在项目根目录下执行 python app.py看到日志INFO: Uvicorn running on http://0.0.0.0:8000表示启动成功。步骤2准备示例数据在sources/notes/目录下创建一个笔记文件project_plan.md。# 项目计划个人任务管理系统 **目标**开发一个CLI工具管理每日任务。 **技术栈**Python, Typer, SQLite。 **核心功能** 1. 添加任务addtaskman add “写周报” -p high 2. 列出任务list按优先级排序。 3. 完成任务done标记任务状态。 **当前进展**已完成数据库模型设计正在开发add命令。 **遇到的问题**Typer的回调函数中如何共享数据库连接步骤3手动索引数据使用curl或浏览器访问API端点索引我们刚创建的笔记目录。# 使用curl命令 curl -X POST http://localhost:8000/index \ -H Content-Type: application/json \ -d {path: sources/notes} # 预期返回 # {message:目录已索引新增 X 条记忆。}步骤4检索记忆现在我们可以模拟一个开发问题看看系统能否找到相关记忆。curl -X POST http://localhost:8000/search \ -H Content-Type: application/json \ -d {query: Typer回调函数里怎么共享数据库连接, type_filter: note} # 预期返回简化 # { # query: ..., # memories: [ # { # content: **遇到的问题**Typer的回调函数中如何共享数据库连接..., # source: sources/notes/project_plan.md, # type: note, # relevance_score: 0.92 # }, # ...其他相关片段 # ], # count: 2 # }步骤5生成AI对话上下文这是最关键的一步获取格式化后的上下文以便粘贴给AI助手。curl http://localhost:8000/generate-context?queryTyper共享数据库连接max_chars1000返回的context字段内容可以直接作为提示词前缀发给ChatGPT等AI例如以下是你之前的相关工作记录供参考 [1] 来源sources/notes/project_plan.md (类型note, 相关度92.00%) **遇到的问题**Typer的回调函数中如何共享数据库连接 --- 以上是基于你工作记忆的 1 条相关上下文 --- 我的问题是Typer共享数据库连接5. 自动化集成与高级用法基础系统搭建完成后我们可以让它更自动化、更智能。5.1 实现文件监控与自动索引创建scripts/watch_and_index.py实现后台自动监控。# scripts/watch_and_index.py import time from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler from pathlib import Path from memory_core.collector import collector from config import settings import logging logger logging.getLogger(__name__) class SourceFileHandler(FileSystemEventHandler): 处理文件系统事件 def on_modified(self, event): if not event.is_directory: self._process_event(event) def on_created(self, event): if not event.is_directory: self._process_event(event) def _process_event(self, event): file_path Path(event.src_path) logger.info(f检测到文件变动: {file_path}) # 避免频繁触发可加入防抖逻辑 time.sleep(0.5) # 简单防抖 try: memories collector.process_file(file_path) if memories: logger.info(f自动索引文件: {file_path.name}, 新增 {len(memories)} 个片段。) except Exception as e: logger.error(f自动索引失败: {e}) def start_watching(): 启动文件监控 event_handler SourceFileHandler() observer Observer() observer.schedule(event_handler, str(settings.SOURCES_DIR), recursiveTrue) observer.start() logger.info(f开始监控目录: {settings.SOURCES_DIR}) try: while True: time.sleep(1) except KeyboardInterrupt: observer.stop() observer.join() if __name__ __main__: start_watching()5.2 集成到AI聊天工具浏览器扩展思路完全自动化需要与AI界面深度集成。一个可行的方案是开发一个浏览器扩展如Chrome Extension。扩展核心逻辑content script监听ChatGPT等网页的输入框。当用户开始输入时提取输入框中的关键词或句子。向本地http://localhost:8000/generate-context发送请求获取相关上下文。将返回的上下文自动插入到输入框的最前面。由于实现浏览器扩展涉及特定API这里提供概念性伪代码// 伪代码Chrome扩展内容脚本概念 async function enhanceAIInput() { const inputBox document.querySelector(textarea[aria-label*message]); // 根据实际网页调整选择器 if (!inputBox) return; inputBox.addEventListener(input, _.debounce(async (e) { const query e.target.value; if (query.length 5) return; // 输入过短时不查询 try { const resp await fetch(http://localhost:8000/generate-context?query${encodeURIComponent(query)}); const data await resp.json(); if (data.context data.context.includes(相关上下文)) { // 将上下文智能地添加到输入内容前 const enhancedPrompt data.context \n\n query; // 注意直接设置value会触发循环需要巧妙处理 // 一种方法是提供一个“添加上下文”的按钮 } } catch (err) { console.error(获取记忆上下文失败:, err); } }, 500)); // 防抖500毫秒 }5.3 与Git集成捕获代码上下文创建.git/hooks/post-commit钩子需赋予可执行权限在每次提交后自动记录。#!/bin/bash # .git/hooks/post-commit REPO_ROOT$(git rev-parse --show-toplevel) HOOKS_DIR$REPO_ROOT/.git/hooks MEMORY_CARD_DIR$REPO_ROOT/../ai_memory_card # 假设记忆卡项目在仓库同级目录 # 获取本次提交信息 COMMIT_MSG$(git log -1 --pretty%B) LAST_COMMIT_HASH$(git rev-parse HEAD) AUTHOR$(git log -1 --prettyformat:%an) # 构建记忆文本 MEMORY_CONTENTGit提交$LAST_COMMIT_HASH 作者$AUTHOR 信息$COMMIT_MSG # 调用记忆卡API进行记录 curl -X POST http://localhost:8000/index \ -H Content-Type: application/json \ -d {\path\: \$REPO_ROOT\} \ --max-time 2 /dev/null 21 echo [AI记忆卡] 已尝试记录本次提交上下文。6. 常见问题与排查思路在搭建和使用过程中你可能会遇到以下问题。问题现象可能原因解决思路启动app.py时报ImportError依赖未安装或虚拟环境未激活1. 确认已激活虚拟环境。2. 运行pip install -r requirements.txt。访问http://localhost:8000无响应服务未启动或端口被占用1. 检查python app.py是否运行。2. 使用netstat -ano | findstr :8000(Win) 或lsof -i:8000(Mac/Linux) 查看端口占用并终止冲突进程。文件监控脚本不触发文件路径不正确或事件未捕获1. 确认sources/目录下有文件。2. 检查watchdog库是否安装正确。3. 尝试手动修改一个文件看日志输出。检索结果不相关嵌入模型不适合或文本分块过大1. 尝试更换嵌入模型如paraphrase-multilingual-MiniLM-L12-v2支持中文。2. 调整config.py中的CHUNK_SIZE如改为300和CHUNK_OVERLAP。3. 检查查询语句是否足够明确。向量数据库报错Collection not found数据库路径损坏或版本不兼容1. 删除storage/chroma_db目录重启服务会重建。2. 检查 ChromaDB 版本尝试降级到稳定版如pip install chromadb0.4.15。处理中文文本乱码或错误文件编码问题1. 确保源代码文件保存为 UTF-8 编码。2. 在TextLoader中明确指定encodingutf-8。内存占用过高嵌入模型加载或文件过大1. 对于大型文档先进行预处理和过滤。2. 考虑使用更轻量的嵌入模型。3. 定期清理不重要的记忆片段可通过API扩展删除功能。7. 最佳实践与工程建议将“AI记忆卡”投入日常使用以下建议能帮助你获得更好体验并避免陷阱。7.1 数据源管理分级分类在sources/下建立清晰的子目录如sources/project_a/code/,sources/project_b/docs/。通过元数据type和source进行高效过滤。敏感信息过滤切勿将包含密码、密钥、个人隐私信息的文件放入监控目录。可通过在collector.py的process_file方法中添加关键词过滤逻辑。文件类型限制只监控你真正关心的文件类型在config.py的WATCH_PATTERNS中配置避免处理二进制文件如图片、视频产生无意义内容。7.2 性能与可维护性增量索引当前示例是全量处理目录。生产环境应记录已索引文件的哈希值实现增量更新避免重复计算嵌入向量。批量操作向量数据库的add操作应批量进行而不是单条插入以提高效率。定期清理设计一个简单的TTL生存时间机制或手动审核界面定期清理过时、无效的记忆片段控制数据库大小。日志记录为关键操作如文件处理、向量添加、检索查询添加详细日志便于后期调试和审计。7.3 提示词工程优化上下文格式化get_context_for_ai方法生成的上下文格式直接影响AI的理解。可以优化模板使其更符合特定AI助手的提示风格。例如为ChatGPT设计专用模板。相关性阈值在retriever.py中可以为检索结果设置一个相似度分数阈值如relevance_score 0.7则丢弃避免注入不相关的低质量上下文。元数据利用在构建最终提示时除了内容还可以强调来源和时间如“这是你昨天写的关于XXX的代码”帮助AI更好地理解上下文的新旧和重要性。7.4 安全边界本地化部署本文方案的核心优势是数据完全本地化。嵌入模型、向量数据库均在本地运行确保了工作隐私的安全。网络访问控制app.py默认监听0.0.0.0:8000。如果仅在本地使用可改为127.0.0.1:8000避免局域网内其他设备访问。输入验证API接口如/index应对传入的路径参数进行严格校验防止目录遍历攻击Path Traversal。7.5 扩展方向支持更多数据源除了文件系统可以扩展收集器以支持从Notion、飞书文档、Jira Issue、Slack频道等平台同步数据。集成更多AI工具除了Web端可以开发VS Code插件、Obsidian插件在编码环境和笔记软件内直接调用记忆上下文。实现记忆“对话”引入轻量级LLM如通过Ollama运行本地模型对检索到的记忆进行总结、关联分析而不仅仅是简单罗列。这套“AI记忆卡”系统将一个理想化的概念变成了可运行的代码。它可能不是最完美的但提供了一个坚实、可扩展的起点。你可以从今天开始先将sources/notes目录用于记录工作日志和问题体验上下文自动关联带来的效率提升。然后逐步将代码库、项目文档纳入监控范围。随着记忆库的丰富你会发现向AI提问前不再需要费力组织背景信息它仿佛真的成为了你项目团队中的一员始终记得之前的每一次讨论和决策。技术的价值在于解决真实世界的摩擦希望这个项目能成为你高效工作的得力助手。 构建本地AI记忆卡系统：实现工作上下文智能管理与自动关联
在项目迭代和日常开发中我们常常需要向AI助手如ChatGPT、Claude、DeepSeek等反复提供项目背景、代码片段、API文档等上下文信息。每次开启新对话都像“从头开始”手动复制粘贴既低效又容易遗漏关键细节。你是否也渴望一个能自动记录、整理并智能关联工作上下文的“AI记忆卡”本文将分享一套完整的解决方案通过构建一个轻量级的本地“AI记忆卡”系统自动收集你的工作进度、代码变更、会议纪要和灵感碎片并在与AI交互时自动附上相关上下文。这套方案基于开源工具链无需复杂部署适合开发者、产品经理和任何需要频繁使用AI辅助工作的知识工作者。学完后你将掌握从环境搭建、数据收集、向量化存储到智能检索集成的全流程实战能力。1. 背景与核心概念为什么需要“AI记忆卡”1.1 当前AI协作的痛点现代AI大模型在代码生成、问题排查、方案设计等方面表现出色但其对话本质上是“无状态”的。这意味着上下文丢失每次新对话AI都不知道你之前讨论过什么、项目进展到哪一步。信息重复输入你需要反复解释项目结构、技术栈、业务规则。知识碎片化有价值的工作记录、决策思路散落在各个聊天记录和本地文件中难以系统化利用。1.2 “AI记忆卡”是什么“AI记忆卡”是一个比喻它指的是一套本地化、自动化的工作上下文管理与注入系统。其核心功能是自动收集监控你的工作环境如代码仓库、笔记文档、通讯工具自动抓取结构化信息。智能存储将收集的信息转化为向量Embedding存入向量数据库实现语义化检索。按需注入在你与AI助手对话时系统自动根据你的问题从记忆库中检索最相关的上下文片段并拼接到提示词Prompt中实现“有记忆”的对话。1.3 核心价值与应用场景对开发者自动关联当前Git提交、代码片段、错误日志让AI精准理解bug上下文。对团队共享项目文档、API设计稿、会议纪要让AI基于最新团队共识提供建议。对个人链接你的知识库、学习笔记、待办清单让AI成为你的个性化第二大脑。 与手动整理相比这套系统实现了从“被动喂养”到“主动关联”的转变显著提升AI协作的深度和效率。2. 环境准备与版本说明我们将使用Python作为主要开发语言结合一系列轻量级开源库。请确保你的环境满足以下要求。2.1 基础环境操作系统Windows 10/11, macOS 10.15, 或主流Linux发行版如Ubuntu 20.04。Python版本 3.8 - 3.11。推荐使用3.9或3.10以获得最佳兼容性。包管理工具pip通常随Python安装。版本控制Git用于监控代码变更。可选IDEVS Code, PyCharm等。2.2 核心依赖库及版本我们将创建一个requirements.txt文件来管理依赖。以下是核心库及其作用# 核心框架与异步 fastapi0.104.1 uvicorn[standard]0.24.0 pydantic2.5.0 # 文件监控与系统交互 watchfiles0.20.0 python-dotenv1.0.0 # 文本处理与加载 langchain0.0.340 langchain-community0.0.10 # 包含多种文档加载器 unstructured0.12.0 # 用于解析多种文档格式 # 向量数据库与嵌入模型 chromadb0.4.18 # 轻量级向量数据库 sentence-transformers2.2.2 # 本地运行嵌入模型 # 或者使用OpenAI嵌入需API Key # openai1.3.0 # 前端界面可选 streamlit1.28.0版本说明以上版本在撰写时经过测试能保证基本功能运行。由于开源库迭代迅速实际使用时若遇到兼容性问题可适当调整次要版本。重点在于理解各库的职责和配置方式。2.3 项目结构预览在开始前我们先规划项目目录这有助于理解后续的代码组织。ai_memory_card/ ├── .env # 环境变量如API密钥 ├── requirements.txt # 项目依赖 ├── app.py # FastAPI主应用入口 ├── config.py # 配置文件 ├── memory_core/ # 核心模块 │ ├── __init__.py │ ├── collector.py # 数据收集器 │ ├── embedder.py # 嵌入模型管理 │ ├── vector_store.py # 向量数据库操作 │ └── retriever.py # 检索器 ├── sources/ # 监控的数据源目录示例 │ ├── code/ │ ├── notes/ │ └── logs/ ├── storage/ # 向量数据库持久化目录 │ └── chroma_db/ └── scripts/ # 工具脚本 └── init_memory.py # 初始化记忆库脚本接下来我们开始搭建系统核心。3. 核心模块拆解与原理“AI记忆卡”系统主要由四个核心模块组成收集器Collector、嵌入器Embedder、向量存储Vector Store和检索器Retriever。3.1 收集器Collector自动捕获工作上下文收集器的职责是从不同源头抓取数据。我们采用基于事件如文件变化和基于轮询如定时读取Git日志的混合策略。关键设计文件系统监控使用watchfiles库监听sources/目录下的文件变动增、删、改实时触发处理流程。Git钩子集成在项目的.git/hooks/post-commit中注入脚本在每次提交后自动捕获提交信息、差异代码作为记忆片段。文档加载器利用LangChain的UnstructuredFileLoader、TextLoader等支持解析.txt,.md,.py,.pdf等多种格式。元数据附加为每一段文本记忆片段附加来源、时间戳、类型代码/笔记/日志等元数据便于后续筛选。3.2 嵌入器Embedder与向量存储这是实现语义检索的核心。我们将文本转换为计算机能理解的数值向量。工作流程文本分块长文档需要被切割成大小适中的片段如500字符同时保持语义连贯。LangChain的RecursiveCharacterTextSplitter是常用工具。向量化使用嵌入模型将文本块转换为固定维度的向量如384维或768维。我们优先选用本地模型如all-MiniLM-L6-v2它平衡了速度与精度且无需网络调用和付费。向量存储将向量及其对应的文本块、元数据存入ChromaDB。ChromaDB 是一个轻量级、可持久化的向量数据库支持按集合Collection组织数据非常适合个人或小团队使用。3.3 检索器Retriever智能关联问题与记忆当用户提出问题时检索器负责找到最相关的记忆片段。检索过程将用户问题同样转换为向量。在向量数据库中进行相似性搜索通常使用余弦相似度。返回相似度最高的前k个文本片段如前3个。混合检索除了向量检索还可以结合元数据过滤例如只检索“代码”类型的记忆或最近一周的记忆使结果更精准。4. 完整实战构建你的第一张“AI记忆卡”现在我们将一步步实现上述系统。请跟随操作并注意代码中的注释。4.1 初始化项目与环境首先创建项目目录并安装依赖。# 1. 创建项目目录 mkdir ai_memory_card cd ai_memory_card # 2. 创建虚拟环境推荐 python -m venv venv # Windows 激活: venv\Scripts\activate # macOS/Linux 激活: source venv/bin/activate # 3. 创建 requirements.txt 并写入上一节的依赖内容 # 可以使用编辑器创建或使用 echo 命令注意版本号 # 4. 安装依赖 pip install -r requirements.txt # 5. 创建项目结构所需目录 mkdir -p memory_core sources/{code,notes,logs} storage/chroma_db scripts4.2 编写核心配置文件创建config.py集中管理配置参数。# config.py import os from pathlib import Path from pydantic_settings import BaseSettings class Settings(BaseSettings): # 项目路径 BASE_DIR: Path Path(__file__).parent SOURCES_DIR: Path BASE_DIR / sources STORAGE_DIR: Path BASE_DIR / storage VECTOR_DB_PATH: Path STORAGE_DIR / chroma_db # 向量数据库配置 COLLECTION_NAME: str work_memory EMBEDDING_MODEL: str all-MiniLM-L6-v2 # 本地句子嵌入模型 # 如果使用OpenAI请取消注释并设置API Key # OPENAI_API_KEY: str # EMBEDDING_MODEL: str text-embedding-ada-002 # 文本处理配置 CHUNK_SIZE: int 500 # 文本分块大小 CHUNK_OVERLAP: int 50 # 块之间重叠字符数 # 检索配置 RETRIEVE_TOP_K: int 3 # 每次检索返回的记忆片段数量 # 文件监控配置 WATCH_PATTERNS: list [*.md, *.txt, *.py, *.js, *.java, *.json] class Config: env_file .env # 从.env文件加载环境变量 settings Settings()4.3 实现嵌入与向量存储模块创建memory_core/embedder.py和memory_core/vector_store.py。# memory_core/embedder.py from sentence_transformers import SentenceTransformer from langchain.embeddings import HuggingFaceEmbeddings import numpy as np from config import settings import logging logger logging.getLogger(__name__) class LocalEmbedder: 本地嵌入模型封装 def __init__(self): model_name settings.EMBEDDING_MODEL logger.info(f正在加载嵌入模型: {model_name}) # 使用LangChain封装的HuggingFace嵌入兼容性更好 self.embeddings HuggingFaceEmbeddings( model_namefsentence-transformers/{model_name}, model_kwargs{device: cpu}, # 有GPU可改为 cuda encode_kwargs{normalize_embeddings: True} # 归一化便于余弦相似度计算 ) logger.info(嵌入模型加载完毕。) def embed_documents(self, texts: list[str]) - list[list[float]]: 将一批文本转换为向量 return self.embeddings.embed_documents(texts) def embed_query(self, text: str) - list[float]: 将单个查询文本转换为向量 return self.embeddings.embed_query(text) # 全局嵌入器实例 embedder LocalEmbedder()# memory_core/vector_store.py import chromadb from chromadb.config import Settings as ChromaSettings from typing import List, Dict, Any from config import settings import logging from memory_core.embedder import embedder logger logging.getLogger(__name__) class VectorStoreManager: 向量数据库管理类 def __init__(self): self.client chromadb.PersistentClient( pathstr(settings.VECTOR_DB_PATH), settingsChromaSettings(anonymized_telemetryFalse) # 禁用匿名数据收集 ) self.collection self.client.get_or_create_collection( namesettings.COLLECTION_NAME, metadata{description: AI工作记忆存储} ) logger.info(f向量数据库连接成功集合: {settings.COLLECTION_NAME}) def add_memories(self, documents: List[str], metadatas: List[Dict], ids: List[str]): 添加记忆片段到向量数据库 if not documents: return # 生成嵌入向量 embeddings embedder.embed_documents(documents) # 添加到集合 self.collection.add( embeddingsembeddings, documentsdocuments, metadatasmetadatas, idsids ) logger.info(f成功添加 {len(documents)} 条记忆。) def search(self, query: str, filter_metadata: Dict None, top_k: int None) - List[Dict[str, Any]]: 检索与查询最相关的记忆 if top_k is None: top_k settings.RETRIEVE_TOP_K # 将查询文本向量化 query_embedding embedder.embed_query(query) # 执行搜索 results self.collection.query( query_embeddings[query_embedding], n_resultstop_k, wherefilter_metadata # 元数据过滤条件 ) # 格式化结果 memories [] if results[documents]: for i in range(len(results[documents][0])): memories.append({ content: results[documents][0][i], metadata: results[metadatas][0][i], distance: results[distances][0][i] # 距离越小越相似 }) return memories def list_all_collections(self): 列出所有集合用于调试 return self.client.list_collections() # 全局向量存储管理器实例 vector_store VectorStoreManager()4.4 实现文件收集器创建memory_core/collector.py实现一个监控指定目录的简易收集器。# memory_core/collector.py import hashlib import time from pathlib import Path from typing import List, Dict, Any from langchain.document_loaders import TextLoader, UnstructuredMarkdownLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from config import settings from memory_core.vector_store import vector_store import logging logger logging.getLogger(__name__) class FileCollector: 文件收集器 def __init__(self): self.text_splitter RecursiveCharacterTextSplitter( chunk_sizesettings.CHUNK_SIZE, chunk_overlapsettings.CHUNK_OVERLAP, length_functionlen, separators[\n\n, \n, 。, , , , , , ] ) def process_file(self, file_path: Path) - List[Dict[str, Any]]: 处理单个文件返回文本块和元数据列表 memories [] try: # 根据后缀选择加载器 if file_path.suffix.lower() in [.txt, .md, .py, .js, .java, .json]: if file_path.suffix .md: loader UnstructuredMarkdownLoader(str(file_path)) else: loader TextLoader(str(file_path), encodingutf-8) documents loader.load() for doc in documents: # 分割文本 chunks self.text_splitter.split_text(doc.page_content) for i, chunk in enumerate(chunks): if not chunk.strip(): continue # 为每个块生成唯一ID文件路径内容哈希 chunk_id f{file_path.stem}_{i}_{hashlib.md5(chunk.encode()).hexdigest()[:8]} # 构建元数据 metadata { source: str(file_path.relative_to(settings.BASE_DIR)), type: self._get_file_type(file_path), timestamp: time.time(), chunk_index: i } memories.append({ id: chunk_id, content: chunk, metadata: metadata }) logger.info(f处理文件 {file_path.name}生成 {len(chunks)} 个文本块。) else: logger.warning(f暂不支持的文件格式: {file_path.suffix}) except Exception as e: logger.error(f处理文件 {file_path} 时出错: {e}) return memories def _get_file_type(self, file_path: Path) - str: 根据文件后缀判断类型 suffix file_path.suffix.lower() if suffix in [.py, .js, .java, .cpp, .go]: return code elif suffix in [.md, .txt]: return note elif suffix in [.log, .err]: return log else: return other def process_directory(self, directory: Path) - int: 处理整个目录并存入向量数据库 all_memories [] if not directory.exists(): logger.warning(f目录不存在: {directory}) return 0 # 递归查找支持的文件 for pattern in settings.WATCH_PATTERNS: for file_path in directory.rglob(pattern): if file_path.is_file(): memories self.process_file(file_path) all_memories.extend(memories) # 批量添加到向量数据库 if all_memories: documents [m[content] for m in all_memories] metadatas [m[metadata] for m in all_memories] ids [m[id] for m in all_memories] vector_store.add_memories(documents, metadatas, ids) return len(all_memories) # 全局收集器实例 collector FileCollector()4.5 实现检索器与API接口创建memory_core/retriever.py和app.py提供检索API。# memory_core/retriever.py from typing import List, Dict, Any from config import settings from memory_core.vector_store import vector_store import logging logger logging.getLogger(__name__) class MemoryRetriever: 记忆检索器 def retrieve(self, query: str, source_filter: str None, type_filter: str None) - List[Dict[str, Any]]: 根据查询检索相关记忆。 Args: query: 查询文本 source_filter: 按来源过滤如 sources/notes/xxx.md type_filter: 按类型过滤如 code, note, log Returns: 相关记忆列表按相关性排序 # 构建元数据过滤条件 filter_metadata {} if source_filter: filter_metadata[source] {$contains: source_filter} if type_filter: filter_metadata[type] type_filter logger.info(f检索查询: {query}, 过滤器: {filter_metadata}) memories vector_store.search(query, filter_metadatafilter_metadata if filter_metadata else None) # 格式化输出 formatted_results [] for mem in memories: formatted_results.append({ content: mem[content][:200] ... if len(mem[content]) 200 else mem[content], source: mem[metadata].get(source, unknown), type: mem[metadata].get(type, unknown), relevance_score: round(1 - mem[distance], 4) # 将距离转换为相似度分数 }) return formatted_results def get_context_for_ai(self, query: str, max_chars: int 1500) - str: 为AI对话生成上下文提示。 将检索到的记忆片段拼接成一段连贯的上下文。 memories self.retrieve(query) if not memories: return 暂无相关历史上下文。\n context_parts [以下是你之前的相关工作记录供参考\n] total_chars 0 for i, mem in enumerate(memories, 1): mem_text f[{i}] 来源{mem[source]} (类型{mem[type]}, 相关度{mem[relevance_score]:.2%})\n{mem[content]}\n\n if total_chars len(mem_text) max_chars: break context_parts.append(mem_text) total_chars len(mem_text) context_parts.append(f\n--- 以上是基于你工作记忆的 {len(context_parts)-1} 条相关上下文 ---\n) return .join(context_parts) retriever MemoryRetriever()# app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional, List import uvicorn import logging from memory_core.retriever import retriever from memory_core.collector import collector from config import settings # 配置日志 logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) app FastAPI(titleAI记忆卡 API, description自动管理工作上下文并与AI集成的服务) class SearchRequest(BaseModel): query: str source_filter: Optional[str] None type_filter: Optional[str] None class SearchResponse(BaseModel): query: str memories: List[dict] count: int class IndexRequest(BaseModel): path: str # 相对于sources目录的路径或绝对路径 app.get(/) async def root(): return {message: AI记忆卡服务运行中, version: 1.0.0} app.post(/search, response_modelSearchResponse) async def search_memories(request: SearchRequest): 检索相关记忆 try: memories retriever.retrieve( queryrequest.query, source_filterrequest.source_filter, type_filterrequest.type_filter ) return SearchResponse( queryrequest.query, memoriesmemories, countlen(memories) ) except Exception as e: logger.error(f检索失败: {e}) raise HTTPException(status_code500, detailf检索失败: {str(e)}) app.post(/index) async def index_directory(request: IndexRequest): 手动索引一个目录或文件 import os from pathlib import Path target_path Path(request.path) if not target_path.is_absolute(): target_path settings.BASE_DIR / target_path if not target_path.exists(): raise HTTPException(status_code404, detailf路径不存在: {target_path}) try: if target_path.is_file(): # 索引单个文件 memories collector.process_file(target_path) if memories: documents [m[content] for m in memories] metadatas [m[metadata] for m in memories] ids [m[id] for m in memories] from memory_core.vector_store import vector_store vector_store.add_memories(documents, metadatas, ids) count len(memories) else: count 0 msg f文件已索引 else: # 索引目录 count collector.process_directory(target_path) msg f目录已索引 return {message: f{msg}新增 {count} 条记忆。} except Exception as e: logger.error(f索引失败: {e}) raise HTTPException(status_code500, detailf索引失败: {str(e)}) app.get(/generate-context) async def generate_context(query: str, max_chars: int 1500): 生成用于AI提示的上下文文本 try: context retriever.get_context_for_ai(query, max_chars) return {query: query, context: context} except Exception as e: logger.error(f生成上下文失败: {e}) raise HTTPException(status_code500, detailf生成上下文失败: {str(e)}) if __name__ __main__: logger.info(启动AI记忆卡服务...) uvicorn.run(app, host0.0.0.0, port8000)4.6 运行与验证现在让我们启动服务并进行测试。步骤1启动API服务# 在项目根目录下执行 python app.py看到日志INFO: Uvicorn running on http://0.0.0.0:8000表示启动成功。步骤2准备示例数据在sources/notes/目录下创建一个笔记文件project_plan.md。# 项目计划个人任务管理系统 **目标**开发一个CLI工具管理每日任务。 **技术栈**Python, Typer, SQLite。 **核心功能** 1. 添加任务addtaskman add “写周报” -p high 2. 列出任务list按优先级排序。 3. 完成任务done标记任务状态。 **当前进展**已完成数据库模型设计正在开发add命令。 **遇到的问题**Typer的回调函数中如何共享数据库连接步骤3手动索引数据使用curl或浏览器访问API端点索引我们刚创建的笔记目录。# 使用curl命令 curl -X POST http://localhost:8000/index \ -H Content-Type: application/json \ -d {path: sources/notes} # 预期返回 # {message:目录已索引新增 X 条记忆。}步骤4检索记忆现在我们可以模拟一个开发问题看看系统能否找到相关记忆。curl -X POST http://localhost:8000/search \ -H Content-Type: application/json \ -d {query: Typer回调函数里怎么共享数据库连接, type_filter: note} # 预期返回简化 # { # query: ..., # memories: [ # { # content: **遇到的问题**Typer的回调函数中如何共享数据库连接..., # source: sources/notes/project_plan.md, # type: note, # relevance_score: 0.92 # }, # ...其他相关片段 # ], # count: 2 # }步骤5生成AI对话上下文这是最关键的一步获取格式化后的上下文以便粘贴给AI助手。curl http://localhost:8000/generate-context?queryTyper共享数据库连接max_chars1000返回的context字段内容可以直接作为提示词前缀发给ChatGPT等AI例如以下是你之前的相关工作记录供参考 [1] 来源sources/notes/project_plan.md (类型note, 相关度92.00%) **遇到的问题**Typer的回调函数中如何共享数据库连接 --- 以上是基于你工作记忆的 1 条相关上下文 --- 我的问题是Typer共享数据库连接5. 自动化集成与高级用法基础系统搭建完成后我们可以让它更自动化、更智能。5.1 实现文件监控与自动索引创建scripts/watch_and_index.py实现后台自动监控。# scripts/watch_and_index.py import time from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler from pathlib import Path from memory_core.collector import collector from config import settings import logging logger logging.getLogger(__name__) class SourceFileHandler(FileSystemEventHandler): 处理文件系统事件 def on_modified(self, event): if not event.is_directory: self._process_event(event) def on_created(self, event): if not event.is_directory: self._process_event(event) def _process_event(self, event): file_path Path(event.src_path) logger.info(f检测到文件变动: {file_path}) # 避免频繁触发可加入防抖逻辑 time.sleep(0.5) # 简单防抖 try: memories collector.process_file(file_path) if memories: logger.info(f自动索引文件: {file_path.name}, 新增 {len(memories)} 个片段。) except Exception as e: logger.error(f自动索引失败: {e}) def start_watching(): 启动文件监控 event_handler SourceFileHandler() observer Observer() observer.schedule(event_handler, str(settings.SOURCES_DIR), recursiveTrue) observer.start() logger.info(f开始监控目录: {settings.SOURCES_DIR}) try: while True: time.sleep(1) except KeyboardInterrupt: observer.stop() observer.join() if __name__ __main__: start_watching()5.2 集成到AI聊天工具浏览器扩展思路完全自动化需要与AI界面深度集成。一个可行的方案是开发一个浏览器扩展如Chrome Extension。扩展核心逻辑content script监听ChatGPT等网页的输入框。当用户开始输入时提取输入框中的关键词或句子。向本地http://localhost:8000/generate-context发送请求获取相关上下文。将返回的上下文自动插入到输入框的最前面。由于实现浏览器扩展涉及特定API这里提供概念性伪代码// 伪代码Chrome扩展内容脚本概念 async function enhanceAIInput() { const inputBox document.querySelector(textarea[aria-label*message]); // 根据实际网页调整选择器 if (!inputBox) return; inputBox.addEventListener(input, _.debounce(async (e) { const query e.target.value; if (query.length 5) return; // 输入过短时不查询 try { const resp await fetch(http://localhost:8000/generate-context?query${encodeURIComponent(query)}); const data await resp.json(); if (data.context data.context.includes(相关上下文)) { // 将上下文智能地添加到输入内容前 const enhancedPrompt data.context \n\n query; // 注意直接设置value会触发循环需要巧妙处理 // 一种方法是提供一个“添加上下文”的按钮 } } catch (err) { console.error(获取记忆上下文失败:, err); } }, 500)); // 防抖500毫秒 }5.3 与Git集成捕获代码上下文创建.git/hooks/post-commit钩子需赋予可执行权限在每次提交后自动记录。#!/bin/bash # .git/hooks/post-commit REPO_ROOT$(git rev-parse --show-toplevel) HOOKS_DIR$REPO_ROOT/.git/hooks MEMORY_CARD_DIR$REPO_ROOT/../ai_memory_card # 假设记忆卡项目在仓库同级目录 # 获取本次提交信息 COMMIT_MSG$(git log -1 --pretty%B) LAST_COMMIT_HASH$(git rev-parse HEAD) AUTHOR$(git log -1 --prettyformat:%an) # 构建记忆文本 MEMORY_CONTENTGit提交$LAST_COMMIT_HASH 作者$AUTHOR 信息$COMMIT_MSG # 调用记忆卡API进行记录 curl -X POST http://localhost:8000/index \ -H Content-Type: application/json \ -d {\path\: \$REPO_ROOT\} \ --max-time 2 /dev/null 21 echo [AI记忆卡] 已尝试记录本次提交上下文。6. 常见问题与排查思路在搭建和使用过程中你可能会遇到以下问题。问题现象可能原因解决思路启动app.py时报ImportError依赖未安装或虚拟环境未激活1. 确认已激活虚拟环境。2. 运行pip install -r requirements.txt。访问http://localhost:8000无响应服务未启动或端口被占用1. 检查python app.py是否运行。2. 使用netstat -ano | findstr :8000(Win) 或lsof -i:8000(Mac/Linux) 查看端口占用并终止冲突进程。文件监控脚本不触发文件路径不正确或事件未捕获1. 确认sources/目录下有文件。2. 检查watchdog库是否安装正确。3. 尝试手动修改一个文件看日志输出。检索结果不相关嵌入模型不适合或文本分块过大1. 尝试更换嵌入模型如paraphrase-multilingual-MiniLM-L12-v2支持中文。2. 调整config.py中的CHUNK_SIZE如改为300和CHUNK_OVERLAP。3. 检查查询语句是否足够明确。向量数据库报错Collection not found数据库路径损坏或版本不兼容1. 删除storage/chroma_db目录重启服务会重建。2. 检查 ChromaDB 版本尝试降级到稳定版如pip install chromadb0.4.15。处理中文文本乱码或错误文件编码问题1. 确保源代码文件保存为 UTF-8 编码。2. 在TextLoader中明确指定encodingutf-8。内存占用过高嵌入模型加载或文件过大1. 对于大型文档先进行预处理和过滤。2. 考虑使用更轻量的嵌入模型。3. 定期清理不重要的记忆片段可通过API扩展删除功能。7. 最佳实践与工程建议将“AI记忆卡”投入日常使用以下建议能帮助你获得更好体验并避免陷阱。7.1 数据源管理分级分类在sources/下建立清晰的子目录如sources/project_a/code/,sources/project_b/docs/。通过元数据type和source进行高效过滤。敏感信息过滤切勿将包含密码、密钥、个人隐私信息的文件放入监控目录。可通过在collector.py的process_file方法中添加关键词过滤逻辑。文件类型限制只监控你真正关心的文件类型在config.py的WATCH_PATTERNS中配置避免处理二进制文件如图片、视频产生无意义内容。7.2 性能与可维护性增量索引当前示例是全量处理目录。生产环境应记录已索引文件的哈希值实现增量更新避免重复计算嵌入向量。批量操作向量数据库的add操作应批量进行而不是单条插入以提高效率。定期清理设计一个简单的TTL生存时间机制或手动审核界面定期清理过时、无效的记忆片段控制数据库大小。日志记录为关键操作如文件处理、向量添加、检索查询添加详细日志便于后期调试和审计。7.3 提示词工程优化上下文格式化get_context_for_ai方法生成的上下文格式直接影响AI的理解。可以优化模板使其更符合特定AI助手的提示风格。例如为ChatGPT设计专用模板。相关性阈值在retriever.py中可以为检索结果设置一个相似度分数阈值如relevance_score 0.7则丢弃避免注入不相关的低质量上下文。元数据利用在构建最终提示时除了内容还可以强调来源和时间如“这是你昨天写的关于XXX的代码”帮助AI更好地理解上下文的新旧和重要性。7.4 安全边界本地化部署本文方案的核心优势是数据完全本地化。嵌入模型、向量数据库均在本地运行确保了工作隐私的安全。网络访问控制app.py默认监听0.0.0.0:8000。如果仅在本地使用可改为127.0.0.1:8000避免局域网内其他设备访问。输入验证API接口如/index应对传入的路径参数进行严格校验防止目录遍历攻击Path Traversal。7.5 扩展方向支持更多数据源除了文件系统可以扩展收集器以支持从Notion、飞书文档、Jira Issue、Slack频道等平台同步数据。集成更多AI工具除了Web端可以开发VS Code插件、Obsidian插件在编码环境和笔记软件内直接调用记忆上下文。实现记忆“对话”引入轻量级LLM如通过Ollama运行本地模型对检索到的记忆进行总结、关联分析而不仅仅是简单罗列。这套“AI记忆卡”系统将一个理想化的概念变成了可运行的代码。它可能不是最完美的但提供了一个坚实、可扩展的起点。你可以从今天开始先将sources/notes目录用于记录工作日志和问题体验上下文自动关联带来的效率提升。然后逐步将代码库、项目文档纳入监控范围。随着记忆库的丰富你会发现向AI提问前不再需要费力组织背景信息它仿佛真的成为了你项目团队中的一员始终记得之前的每一次讨论和决策。技术的价值在于解决真实世界的摩擦希望这个项目能成为你高效工作的得力助手。