构建本地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(用于监控代码变更)。
- (可选)IDE:VS Code, PyCharm等。
2.2 核心依赖库及版本
我们将创建一个requirements.txt文件来管理依赖。以下是核心库及其作用:
# 核心框架与异步 fastapi==0.104.1 uvicorn[standard]==0.24.0 pydantic==2.5.0 # 文件监控与系统交互 watchfiles==0.20.0 python-dotenv==1.0.0 # 文本处理与加载 langchain==0.0.340 langchain-community==0.0.10 # 包含多种文档加载器 unstructured==0.12.0 # 用于解析多种文档格式 # 向量数据库与嵌入模型 chromadb==0.4.18 # 轻量级向量数据库 sentence-transformers==2.2.2 # 本地运行嵌入模型 # 或者使用OpenAI嵌入(需API Key) # openai==1.3.0 # 前端界面(可选) streamlit==1.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_name=f"sentence-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( path=str(settings.VECTOR_DB_PATH), settings=ChromaSettings(anonymized_telemetry=False) # 禁用匿名数据收集 ) self.collection = self.client.get_or_create_collection( name=settings.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( embeddings=embeddings, documents=documents, metadatas=metadatas, ids=ids ) 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_results=top_k, where=filter_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_size=settings.CHUNK_SIZE, chunk_overlap=settings.CHUNK_OVERLAP, length_function=len, 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), encoding='utf-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_metadata=filter_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(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) app = FastAPI(title="AI记忆卡 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_model=SearchResponse) async def search_memories(request: SearchRequest): """检索相关记忆""" try: memories = retriever.retrieve( query=request.query, source_filter=request.source_filter, type_filter=request.type_filter ) return SearchResponse( query=request.query, memories=memories, count=len(memories) ) except Exception as e: logger.error(f"检索失败: {e}") raise HTTPException(status_code=500, detail=f"检索失败: {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_code=404, detail=f"路径不存在: {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_code=500, detail=f"索引失败: {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_code=500, detail=f"生成上下文失败: {str(e)}") if __name__ == "__main__": logger.info("启动AI记忆卡服务...") uvicorn.run(app, host="0.0.0.0", port=8000)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. 添加任务(add):`taskman 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?query=Typer共享数据库连接&max_chars=1000"返回的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), recursive=True) 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 --pretty=format:'%an') # 构建记忆文本 MEMORY_CONTENT="Git提交:$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 2>&1 & 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 chromadb==0.4.15)。 |
| 处理中文文本乱码或错误 | 文件编码问题 | 1. 确保源代码文件保存为 UTF-8 编码。 2. 在 TextLoader中明确指定encoding='utf-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提问前不再需要费力组织背景信息,它仿佛真的成为了你项目团队中的一员,始终记得之前的每一次讨论和决策。技术的价值在于解决真实世界的摩擦,希望这个项目能成为你高效工作的得力助手。
