拓冰网站定制 
1. 项目缘起我的信息宇宙需要一个擎天柱朋友前阵子问我一个问题你每天收藏那么多文章、写那么多笔记、存那么多PDF真正想用的时候你找得到吗我愣了一下因为答案很尴尬——大部分时候找不到。我用过Notion、Obsidian、印象笔记各有各的好但都缺一个关键能力把存进去的资料变成能直接调用的答案。收藏夹越堆越满认知负担越来越大工具反而成了压力来源。所以我决定从零搭一个自己的系统代号Atlas。命名为Atlas是因为希腊神话里阿特拉斯用肩膀扛起整个天穹——我想让这个系统帮我扛起个人知识宇宙的承重。它不是一个笔记软件而是一个AI 增强的个人知识与生产力系统把散落的网页、PDF、Markdown、碎片想法统一收进本地知识库用大模型做切分、向量化、语义检索再通过几个AI Agent完成问答、摘要、任务拆解、待办提取这些日常动作。整个过程完全自控数据在自己手里想怎么改就怎么改。这篇博文会把我整套搭建思路、技术选型依据、核心代码实现、踩过的坑一次性说清楚。不管你是有一定 Python 基础的程序员还是刚开始接触 RAG 和个人知识管理的爱好者只要你想要一个越用越聪明的第二大脑这篇文章都值得你花十分钟读完。我尽量说人话该给代码给代码该给参数给参数保证你能照着搭起来。在动手之前先明确一件事Atlas 不是要替代 Notion 或 Obsidian而是做一个AI 层叠加在它们之上。知识源文件仍然可以用你习惯的工具管理Atlas 负责把内容切碎、向量化、做语义检索再把检索结果喂给大模型生成答案或行动项。这样各司其职既保留手动整理的安全感又获得 AI 检索的爽快感。1.1 我到底要解决哪几类痛点很多人以为知识管理就是存得多其实真正的问题在取不出。我给自己列了一个需求清单每一项都对应一个具体的日常场景。第一检索效率太低。关键词搜索有一个死穴你记不住原文里的精确措辞。比如你看过一篇讲分布式事务 Seata 原理的文章三个月后想找AT 模式回滚时全局锁怎么处理用关键词搜AT 模式可能命中搜全局锁就悬了。语义检索可以用向量距离匹配意思相近的内容模糊记得的东西也能捞出来。这是我的刚需也是 Atlas 的基石功能。第二已有资料没有被榨干。我的硬盘里躺着几百个 PDF很多是从 Tech 社区下载的白皮书、开源项目文档甚至有些读过一遍就再也没打开。这些资料本身有很高价值但它们是非结构化的没法直接参与问答。Atlas 可以把它们切块、嵌入成一个可检索的向量索引之后就能直接问这个框架的故障转移策略是什么它会从对应 PDF 的对应段落里找答案。第三AI 对话和知识库脱节。ChatGPT 用多了就发现一个问题它不知道我自己的私有资料每次都要把上下文粘进去既费 token 又丢信息。Atlas 的 RAG检索增强生成架构专门解决这个先检索我的知识库再把检索到的块拼进提示词让大模型带着材料回答问题准确率和可追溯性都高很多。第四被动收藏和主动行动之间缺一座桥。我经常收藏一篇讲 OKR 的文章然后就没有然后了。Atlas 里我加了一个行动提取Agent喂给它任何一篇文章它能抽出里面的可执行项、时间节点、责任人整理成待办清单。这个功能在周报场景特别实用——把一周读过的内容统一跑一遍自动生成下周的行动计划。1.2 为什么不用现成的方案市面上已经有不少AI 笔记工具比如我有段时间重度使用某款海外工具的 AI 问答和自动标签功能。说实话很惊艳但几个点让我最终放弃了第一数据不在本地私密性存疑有些工作材料根本不敢传第二可定制性差想调整分块策略、自定义检索权重、接入自己的模型都很难第三这类工具的 AI 功能大多有配额限制用几次就提示你升级会员。Atlas 走的是本地优先、组件解耦路线。向量数据库、嵌入模型、大模型、Web 框架全部可以替换没有供应商锁定。你早期可以用 OpenAI 的接口快速验证效果后期想换成本地模型比如 ollama 部署 Qwen 或 Llama也只需改一处封装。这个自由度是商业产品给不了你的。另外还有一个非常现实的理由这是一个极佳的学习项目。搭建 Atlas 的过程中你会亲手接触 RAG、embedding、向量检索、Agent 编排、异步任务队列这些概念不是看过教程那种程度的了解而是写代码让它跑起来的真掌握。我在搭完这套系统后对 AI 应用层的理解明显上了一个台阶。2. 系统设计与技术选型每一个选择的背后逻辑Atlas 的整体架构我用一句话概括采集层收料 → 处理层切片嵌入 → 存储层双写元数据和向量 → 检索层语义召回 → 生成层整合回答 → 应用层对接用户。听起来有点长但每一层解决的都是具体问题。先给一张整体逻辑表后面逐个讲层次负责内容我用到的组件替代方案采集层网页剪藏、PDF导入、Markdown导入Python trafilatura pypdf浏览器插件、Readwise处理层清洗、切块、嵌入向量递归字符切块 OpenAI embeddingLangChain 的 RecursiveCharacterTextSplitter存储层元数据 向量双写SQLite QdrantPostgreSQL pgvector检索层语义检索、元数据过滤Qdrant APIMilvus、Chroma生成层问答、摘要、行动提取OpenAI GPT-4o-miniClaude、本地 Qwen应用层REST API、任务编排FastAPI asyncioGradio、Chainlit这套选型不是拍脑袋拍出来的我逐个说理由。2.1 向量库选型为什么我选了 Qdrant 而不是其他向量库是整个 RAG 系统的核心底座选型时我对比过 Chroma、Milvus、Weaviate、pgvector。最终选了Qdrant核心原因是部署简单但能力不缩水。Qdrant 支持单机 Docker 启动也能用嵌入式模式直接跑在 Python 进程里对个人项目极其友好同时它的过滤能力很强可以按元数据字段来源、日期、标签做精确过滤后再做向量检索这对知识库场景太重要了——我只想搜某个 PDF 而不是全库时这个能力是刚需。为什么不选 pgvector如果你已经有 PostgreSQL那 pgvector 确实省一个组件但它的索引构建和查询性能在小数据量下没问题一旦超过几万条向量、还希望做带过滤的混合检索时调优成本就上来了。Qdrant 的 HNSW 索引是开箱即用的配置几个参数就行而且它原生返回 payload不用再回表查原始文本省一层麻烦。为什么不选 ChromaChroma 确实最轻但功能也最轻元数据过滤、负载均衡这些能力相对弱。个人项目初期可能够用但等你想做多用户或者上云托管时就尴尬了。Qdrant 可以无缝从本地 Docker 迁移到云托管版这个成长空间值得提前留好。2.2 嵌入模型选型便宜大碗才是真道理嵌入模型决定语义相似度的质量但这个质量并不非得靠贵模型。我的原则是嵌入模型要便宜、稳定、矢量维度适中。目前主力用的是text-embedding-3-small1536 维每百万 token 价格极低对个人项目来说成本几乎可以忽略。它的效果在英文和中文上都够用尤其适合知识库这种不需要太精细语义分野的场景。如果你有数据隐私顾虑完全可以用本地模型替代。我在一台不带 GPU 的 MacBook 上测试过bge-small-zh-v1.5512 维速度很快中文语义效果和 OpenAI 的 small 模型差距不大。关键是在我的代码里 embedding 是被封装成一个类方法embed_texts()的换模型只需要改这一个方法。这里强烈建议大家也这么做因为嵌入模型迭代很快你不想每次换个模型就重构整个数据管道。2.3 大模型选型生成质量与成本的平衡点生成层我选了 GPT-4o-mini理由比很多人想的简单它够聪明而且便宜。知识库问答这个场景真正要拼的不是模型的推理天花板而是 RAG 检索质量。检索给的材料对小模型也能给出很好的答案检索给错材料再大的模型也爱莫能助。所以把预算花在好的切块和检索上比盲目上旗舰模型划算得多。在这里多说一句现在很多人赶时髦用 Claude Opus 或 GPT-4o 做所有事其实有点浪费。我的使用习惯是分级调用——简单问答走 4o-mini需要复杂推理或多步工具调用时再走更强的模型。这个逻辑我封装在 LLM 调用层里用一个model_selector()函数根据任务类型自动选模型。后面讲 Agent 编排时你会看到它的好处。3. 从零实操搭建 Atlas 核心管线的完整过程这一章是全文的重头戏。我会按照自己实操的顺序逐步展开每一步都给出能直接用的代码和配置。我给 Atlas 写的代码全部开源在我的 GitHub 上这里展示的是核心部分的精简版目标是一个人可以在一小时内从空目录跑到能聊天的状态。3.1 环境准备与项目结构先交代运行环境我用的是 macOS Python 3.11用venv做虚拟环境所有服务都跑在本机。如果你是 Windows 用户绝大多数步骤一样只有 Docker 启动命令和文件路径写法有细微差异。项目结构采用模块化布局这样后续加功能不会变乱atlas/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── config.py # 配置项API Key、模型名、路径 │ ├── schemas.py # Pydantic 数据模型 │ ├── dao/ │ │ ├── sqlite_dao.py # 元数据读写 │ │ └── qdrant_dao.py # 向量读写 │ ├── services/ │ │ ├── ingest.py # 文档接入 │ │ ├── splitter.py # 切块逻辑 │ │ ├── embedder.py # 嵌入封装 │ │ ├── retriever.py # 检索合并 │ │ └── generator.py # 大模型调用 │ └── agents/ │ ├── orchestrator.py # 主控 Agent │ ├── researcher.py # 研究型 Agent │ └── action_extractor.py # 行动提取 Agent ├── data/ # 本地数据目录 │ ├── raw/ # 原始文件 │ └── atlas.db # SQLite 数据库 ├── scripts/ │ └── setup.sh # 一键初始化脚本 └── requirements.txt安装依赖。我的原则是能少就少用到才装。核心依赖如下pip install fastapi uvicorn qdrant-client openai pypdf \ trafilatura sqlalchemy python-dotenv tiktoken如果你打算跑本地嵌入模型再加一句pip install sentence-transformers初始化 Qdrant。我是用 Docker 启动的一条命令搞定docker run -d \ --name qdrant \ -p 6333:6333 \ -v $(pwd)/data/qdrant_storage:/qdrant/storage \ qdrant/qdrant:latest解释一下为什么要挂载存储目录Qdrant 的向量数据默认存在容器内部不挂载出来的话容器一删数据全没。个人知识库最不能丢的就是数据所以这一步别省。3.2 元数据模型与向量集合设计先想清楚要存什么动手写代码之前先把数据模型设计好。一个常见的误区是只管向量不管元数据等想按日期、来源筛选时发现没法查。我的设计分两层SQLite 存文档源信息Qdrant 存分块向量加轻量 payload。SQLite 里两张表-- 文档源表 CREATE TABLE sources ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, source_type TEXT NOT NULL, -- web / pdf / md url TEXT, file_path TEXT, author TEXT, tags TEXT, -- 逗号分隔 created_at DATETIME DEFAULT CURRENT_TIMESTAMP, raw_text TEXT -- 清洗后的完整文本 ); -- 分块表可选主要用来做映射和人工抽查 CREATE TABLE chunks ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_id INTEGER REFERENCES sources(id), chunk_index INTEGER, chunk_content TEXT, token_count INTEGER );这里raw_text字段很多人会觉得冗余但我故意保留。原因有两点一是做数据一致性校验时能快速看某个 source 是否正常入库二是如果你想切换嵌入模型、重新向量化无需重新抓取和清洗直接从 SQLite 里读原文就行省大量网络请求。Qdrant 里的 collection 我命名为atlas_docspayload 字段设计如下{ source_id: 对应SQLite里的source id, chunk_index: 第几个分块, title: 源文档标题, url: 如果来自网页就有否则为空, date: 采集日期, tags: [标签1, 标签2] }向量本身由 embedding 模型生成维度取决于你选的模型1536 或 512。Payload 里放 source_id 和 chunk_index 是必要的这样检索后能回 SQLite 拿完整上下文。这里有个实践细节切块时不要只存块本身最好把相邻的上一块和下一块也存进去。我后面讲到检索优化时会展开先埋个伏笔。3.3 文档接入与清洗网页、PDF、Markdown 三路归一采集层的目标是把不同类型的输入统一转成干净的文本再进入切块流程。我不做实时爬虫而是提供导入接口——你在浏览器看到一篇好文复制 URL 调接口手头有 PDF直接上传写了篇 Markdown 笔记也传进来。Atlas 对它们一视同仁。网页抓取我用trafilatura而不是简单的requests BeautifulSoup。原因很直接trafilatura 是专门做正文抽取的库能自动去掉导航、脚注、侧边栏这些噪声对我这种新闻、文档、博客混合场景的清洗效果明显更好。代码很简单import trafilatura def fetch_web_article(url: str) - str: downloaded trafilatura.fetch_url(url) if downloaded is None: raise ValueError(f无法抓取该URL请检查地址或网络) text trafilatura.extract( downloaded, include_commentsFalse, include_tablesTrue, favor_precisionTrue ) return text说个小坑favor_precisionTrue这个参数是个双刃剑。它会让抽取结果更精确、噪声更少但有时候会把正文截断尤其遇到那种阅读全文式的分页文章。我的建议是默认开 precision如果发现某篇文章抓回的内容明显变短再用favor_recallTrue重抓一次做对比。PDF 处理我用pypdf但纯文本 PDF 好解决扫描版 PDF 就麻烦了。扫描版需要 OCR我在项目里留了一个可选的 OCR 分支调用pytesseract做识别。代码不复杂但依赖较重默认不启用。如果你经常处理扫描文档可以在config.py里加一个enable_ocr True开关。from pypdf import PdfReader def extract_pdf_text(path: str) - str: reader PdfReader(path) text_parts [] for i, page in enumerate(reader.pages): page_text page.extract_text() if page_text and page_text.strip(): text_parts.append(f--- 第{i1}页 ---\n{page_text}) return \n\n.join(text_parts)Markdown 导入最省事直接读文件然后用正则把代码块和普通段落分开标记就行。我这里特意做了一个处理保留代码块的语言标记和标题层级标记因为切块时我想让模型知道这部分是代码那部分是一级标题对后续问答准确度有帮助。3.4 切块策略让语义完整性和检索粒度达到平衡切块是整个 RAG 里最容易被低估的环节。块太大向量被稀释检索出来的东西不精准块太小语义不完整模型没法理解上下文。我试过固定大小切块比如每 500 字符硬切效果很差——经常一句话被从中间切断检索回来就是一堆残缺信息。最终我采用递归字符切块 重叠窗口的组合策略。思路是优先按 Markdown 标题切切出来的段落如果太长再按段落符\n\n切还是太长就按句子切。每一步都检查长度上限超出上限才继续往下拆。核心实现如下可以做简化版import tiktoken enc tiktoken.encoding_for_model(gpt-4o-mini) CHUNK_SIZE 800 # token 上限 CHUNK_OVERLAP 120 # 相邻块重叠 token 数 def count_tokens(text: str) - int: return len(enc.encode(text)) def split_text_recursive(text: str, chunk_sizeCHUNK_SIZE, chunk_overlapCHUNK_OVERLAP): # 先按标题分 sections re.split(r(?m)^(?#{1,3} ), text) chunks [] buffer for section in sections: # 如果单块超长按段落继续拆 if count_tokens(section) chunk_size: sub_parts re.split(r\n\n, section) for part in sub_parts: if count_tokens(part) chunk_size: # 按句子继续拆 sentences re.split(r(?[。.!?])\s*, part) local_buffer for sent in sentences: if count_tokens(local_buffer sent) chunk_size: local_buffer sent else: if local_buffer: chunks.append(buffer local_buffer) buffer local_buffer[-chunk_overlap:] # 重叠尾巴 local_buffer sent else: if count_tokens(buffer part) chunk_size: buffer part else: chunks.append(buffer part) buffer part[-chunk_overlap:] else: if count_tokens(buffer section) chunk_size: buffer section else: chunks.append(buffer section) buffer section[-chunk_overlap:] if buffer: chunks.append(buffer) return chunks这段代码我故意写得比较啰嗦目的是让你看清递归切块的真实流程。实际工程里可以直接用 LangChain 的RecursiveCharacterTextSplitter参数调好了一样工作。但理解原理比调库更重要——我后来调很多检索问题靠的都是对切块在哪一步破坏了语义的判断。关于 chunk size 的大小我的经验数据是中文场景 500~800 token 比较合适英文可以偏大到 1000。为什么中文要偏小因为中文每个 token 承载的信息密度比英文高同样 800 token 的中文块语义复杂度和英文 1200 token 差不多。重叠窗口设在 100~150 token能保证跨块语义的连贯性。3.5 向量化与入库写一个可替换的嵌入服务做完整条切块管线后就可以向量化入库了。我强烈推荐把嵌入逻辑封装成独立类原因前面讲过——模型会更新你的代码不应该跟着改。from openai import OpenAI class EmbeddingService: def __init__(self, provideropenai): self.provider provider if provider openai: self.client OpenAI() self.model text-embedding-3-small self.dim 1536 else: from sentence_transformers import SentenceTransformer self.model_local SentenceTransformer(BAAI/bge-small-zh-v1.5) self.dim 512 def embed_texts(self, texts: list[str]) - list[list[float]]: if self.provider openai: resp self.client.embeddings.create(modelself.model, inputtexts) return [item.embedding for item in resp.data] else: return self.model_local.encode(texts, normalize_embeddingsTrue).tolist() def embed_query(self, query: str) - list[float]: return self.embed_texts([query])[0]入库逻辑我写成批量处理每凑够 32 个块就做一次批量向量化再批量 upsert 到 Qdrant。用小批量而不是全量是为了在出问题时能快速定位是哪个文档的哪个块出了问题。from qdrant_client import QdrantClient from qdrant_client.http import models class VectorStore: def __init__(self, collection_nameatlas_docs): self.client QdrantClient(hostlocalhost, port6333) self.collection collection_name self.ensure_collection() def ensure_collection(self): existing self.client.get_collections().collections if not any(c.name self.collection for c in existing): self.client.create_collection( collection_nameself.collection, vectors_configmodels.VectorParams( size1536, # 如果换了嵌入模型注意同步改 distancemodels.Distance.COSINE ) ) def upsert_chunks(self, chunks_with_payloads): points [ models.PointStruct( idrow[uuid], vectorrow[vector], payloadrow[payload] ) for row in chunks_with_payloads ] self.client.upsert(collection_nameself.collection, pointspoints)有几个细节值得注意。距离度量我选COSINE而不是L2文本向量经过归一化后余弦相似度对语义方向一致但长度不同的情况更宽容实际检索效果也是 COSINE 更符合直觉。另外我给每条点指定的 id 是 uuidQdrant 允许用自增整数但多文档并行导入时整数冲突的排查成本更高uuid 虽然占用稍大但省心。3.6 检索与生成RAG 最核心的 20 行代码所有前面的努力最终都要落到用户提问 → 检索 → 生成答案这个闭环上。先说检索我的实现是query 经 embedding 查 Qdrant 取 top-k然后按 payload 里的 source_id 去 SQLite 取原始上下文拼入 prompt 喂给大模型。from qdrant_client import QdrantClient from qdrant_client.http import models class Retriever: def __init__(self, embedder, vector_store): self.embedder embedder self.store vector_store def retrieve(self, query: str, top_k: int 5, source_filter: int None): query_vector self.embedder.embed_query(query) query_filter None if source_filter is not None: query_filter models.Filter( must[models.FieldCondition( keysource_id, matchmodels.MatchValue(valuesource_filter) )] ) hits self.store.client.query_points( collection_nameself.store.collection, queryquery_vector, limittop_k, query_filterquery_filter ) return [ { score: hit.score, payload: hit.payload, text: hit.payload.get(text, ) } for hit in hits.points ]注意我这里把每个 chunk 的原文也放进了 Qdrant 的 payload 里。这看似冗余SQLite 里已经有 raw_text但能少一次回表查询对响应速度有帮助。代价是向量库空间占用多一点对个人项目完全可接受。生成 prompt 的设计上我的模板是你是 Atlas 知识库助手。请基于下面的资料回答用户问题。 资料引用时标注来源编号。如果资料不足以回答请明确说知识库中暂无相关信息。 资料 [1] {chunk_text_1} (来源{title}) [2] {chunk_text_2} (来源{title}) ... 问题{query}这里有两个小心机一是要求模型在信息不足时直接承认避免幻觉二是要求标注来源编号方便用户回查原文。我实际使用中这两个规则让 Atlas 的答案可信度高了很多尤其是涉及技术资料时我能直接点开原文核对而不是盲目相信大模型的转述。生成调用我用了一个简单的分级模型选择器def generate_answer(query, context_chunks, query_typesimple): model gpt-4o-mini if query_type complex: model gpt-4o messages [ {role: system, content: 你是Atlas知识库助手。}, {role: user, content: build_prompt(query, context_chunks)} ] resp client.chat.completions.create( modelmodel, messagesmessages, temperature0.3 ) return resp.choices[0].message.contenttemperature 设 0.3 而不是 0留一点点随机性防止措辞太死板但又能保证事实准确性。很多教程直接设 0我试过之后觉得回答太机械0.3 是比较舒服的区间。3.7 用 FastAPI 包一层让系统可以随时被调用为了让 Atlas 能被浏览器、命令行、甚至是未来的手机快捷指令调用我用 FastAPI 包了一层 REST API。设计成三个核心接口导入文档、查询问答、提取行动项。这样 Atlas 就是一个本地 AI 服务而不是一个孤立的脚本。from fastapi import FastAPI, UploadFile, File, Form from pydantic import BaseModel app FastAPI(titleAtlas) class QueryRequest(BaseModel): question: str source_id: int | None None top_k: int 5 class QueryResponse(BaseModel): answer: str references: list[dict] app.post(/api/query, response_modelQueryResponse) async def query_atlas(req: QueryRequest): hits retriever.retrieve(req.question, top_kreq.top_k, source_filterreq.source_id) context [{ chunk_text: h[text], title: h[payload].get(title, ), source_id: h[payload].get(source_id) } for h in hits] answer generate_answer(req.question, context) return QueryResponse(answeranswer, referencescontext) app.post(/api/ingest/web) async def ingest_web(url: str Form(...)): text fetch_web_article(url) source_id save_source(titleurl, source_typeweb, urlurl, raw_texttext) process_and_embed(source_id, text) return {status: ok, source_id: source_id} app.post(/api/ingest/pdf) async def ingest_pdf(file: UploadFile File(...)): content await file.read() path fdata/raw/{file.filename} with open(path, wb) as f: f.write(content) text extract_pdf_text(path) source_id save_source(titlefile.filename, source_typepdf, file_pathpath, raw_texttext) process_and_embed(source_id, text) return {status: ok, source_id: source_id} if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)启动uvicorn app.main:app --reloadAtlas 的第一个可用版本就跑起来了。此时你导入一篇文章然后问这篇文章的核心观点是什么它就能基于你的资料库作答。4. Agent 编排让 Atlas 从回答工具变成生产力助手RAG 问答只是第一步Atlas 更重要的价值在于多 AI 协作地完成生产力任务。我不满足于有问必答我还想要它能把读到的内容变成待办事项、自动生成摘要、为复杂问题拆解研究计划并执行。这就轮到 Agent 机制上场了。4.1 Orchestrator一个会点将的主控 Agent我的设计是轻量的Orchestrator 专用 Agent模式。Orchestrator 不直接生成答案它负责理解用户需求判断该调用哪个专用 Agent再把多个 Agent 的结果合并输出。这样做的好处是每个专用 Agent 逻辑简单、容易调试而 orchestration 层可以自由组合出复杂行为。我举一个实际例子。用户问帮我整理一下最近收藏的内容看看有没有适合做周末读书分享的材料。 Orchestrator 会把这个任务拆成几步调用 Researcher Agent检索最近一周入库的文档生成每篇的摘要调用 ActionExtractor Agent从检索结果中提取可以作为分享话题的候选由 Writer Agent 把这些候选整合成一个带推荐理由的清单。这三个步骤不是串行等待我用asyncio.gather让独立的检索任务并行执行整体耗时从十几秒降到五六秒。代码结构如下async def run_orchestration(user_request: str): plan planner_agent.create_plan(user_request) results await asyncio.gather( *[execute_agent_step(step) for step in plan] ) return writer_agent.combine_results(results, user_request)这个实现看起来简单但踩过一个大坑如果每个 Agent 都独立调用 LLM最后汇总时上下文会丢失。比如 Researcher 生成了摘要Writer 再要写清单时它不知道摘要对应的原文是什么。解决办法是在 Agent 之间传递结构化的中间结果JSON 对象而不是只传纯文本。我在AgentResult里统一加了source_id和confidence字段让下游 Agent 能追溯到原始材料。4.2 ActionExtractor从文章到待办事项的自动流转ActionExtractor 是我用得最频繁的一个 Agent。它的功能一句话说明把输入文本中的可执行内容和时间敏感信息提取成结构化待办。比如你导入一篇产品规划文章里面写了下周三前完成竞品分析整理成文档发给技术团队。普通收藏这件事就过去了但 ActionExtractor 能识别出这是一个任务竞品分析截止时间下周三负责动作完成并整理文档发送。实现上我用了一次小模型 强提示词的方案ACTION_PROMPT 请从以下文本中提取所有可执行任务。以JSON数组输出每个任务包含 - task: 任务描述尽量简洁 - due: 截止时间若不明确则为 null - priority: high/medium/low - related_source_id: {source_id} - context: 任务来源的简要上下文不超过20字 文本内容 {input_text} 这个方案的工程关键点在于JSON 解析容错。大模型偶尔会输出非标准 JSON比如带注释、带 markdown 代码块标记直接用json.loads会崩。我的处理是先用正则去掉json标记再做一轮简单的花括号平衡检查解析失败时再给模型一次修正机会。def safe_parse_json(raw: str): text re.sub(rjson|, , raw).strip() try: return json.loads(text) except json.JSONDecodeError: # 尝试从第一个 { 到最后一个 } 截取 start text.find({) end text.rfind(}) if start ! -1 and end ! -1: candidate text[start:end1] return json.loads(candidate) raise ValueError(无法解析模型输出为JSON)后来我把提取出来的任务直接同步到本地的一个 Todoist 测试账号用它们的 REST API 走了一轮自动化流转。效果比我预期的好现在每周五下午我会把一周收藏的文章统一喂给 ActionExtractor生成下周计划再人工过一遍。这个AI 辅助规划 人工确认的模式比全自动靠谱得多。4.3 多 Agent 协作时的上下文管理多 Agent 协作最头疼的问题是上下文管理。我的经验是不要让 Agent 记忆长对话而是让它带上必要的最小上下文。比如 Researcher Agent 负责检索它没必要知道用户之前的聊天记录ActionExtractor 也没必要知道用户目前看到第几页。每个 Agent 只接收它完成任务所需的字段查询文本、可用的检索结果、特定的指令。这就像团队协作时你给某个同事交代任务只需要告诉他这部分材料你负责而不是把整场会议纪要都发给他。我在代码里给每个 Agent 定义了明确的输入输出 schemadataclass class ResearcherInput: query: str top_k: int 5 dataclass class ResearcherOutput: summary: str references: list[dict] source_ids: list[int]这套设计让我后来调试为什么答案不对时非常舒服每个中间结果都能单独打印出来检查整个链路的可观测性很强。如果你用 LangGraph 这类框架逻辑类似但自己实现一遍会更加理解状态流转的本质。5. 常见问题与排查技巧实录踩过的坑全靠这些笔记兜底最后这部分是纯干货记录我在搭建和持续使用 Atlas 过程中遇到的高频问题。每一条都是我实际遇到并解决的看完应该能帮你少走很多弯路。5.1 切块结果不理想为什么检索总是答非所问遇到最多的问题是我明明导入了相关内容为什么问它它说不知道。排查步骤基本是固定的链路先看检索结果再看生成过程。第一步打印检索结果。我加了一个 debug 接口当检索分数低于 0.6 时就自动输出检索到的前三条块的内容。你会发现大概率是切块切碎了关键信息。比如一个表格被拆到两个块里模型只拿到其中一半自然答不上来。针对切块的修正方法先看问题是长文本还是结构化表格数据。表格数据我会在预处理阶段单独标记不让切块器把表格行拆散长文本则考虑把 chunk_size 调大一点。另一个很有效的技巧是在存储时把前一 block 的尾巴拼到当前块的开头也就是前面埋的伏笔。我把这个实现为recursive_overlap参数默认 120 token。这样即使语义跨块至少前后呼应的内容不会丢。还有一个被忽略的问题是query 和 chunk 的 embedding 维度不一致。如果你中途换过 embedding 模型Qdrant collection 的维度和你新 query 的维度对不上所有检索都会报错。检查方式很直接client.get_collection(atlas_docs)看 vector size再打印embed_query(test)的长度。5.2 多来源混合检索时的权重问题个人知识库里的来源五花八门网页收藏、PDF 论文、个人笔记。它们的可信度完全不同。我早期不做区分检索结果经常被低质量网页占满PDF 里的高质量内容反而排不上。解法是给每个来源类型设置一个权重系数在最终排序时对分数做加权。Qdrant 允许返回原始分数我在后端做排序修正SOURCE_WEIGHT { web: 0.9, pdf: 1.1, md: 1.0, } def rerank_hits(hits): for hit in hits: src_type hit.payload.get(source_type, web) hit.score * SOURCE_WEIGHT.get(src_type, 1.0) hits.sort(keylambda h: h.score, reverseTrue) return hits听着简单效果立竿见影。之后涉及技术论文的查询答案的引用来源明显偏向 PDF 资料。这个权重你也可以继续细化比如给高星博客高一点、给论坛帖子低一点逐步调成自己的偏好。5.3 Agent 任务编排时的超时与重试机制Agent 编排跑起来后你会遇到另一个问题某个中间步骤超时整个任务卡在那里。尤其当 Researcher Agent 需要连续调用多次 LLM 时一个慢响应就会拖累全局。我的解决方案是在每步 LLM 调用上设置超时和重试。OpenAI 的 Python SDK 支持超时参数client OpenAI(timeout30.0, max_retries2)但这只能保证单次调用不卡死。对 Agent 级别的任务我在 orchestrator 层做了一层熔断控制——每个子任务最多执行 3 次超过就标记失败并跳过而不是让整个任务失败。毕竟在个人知识系统里一次检索失败没那么严重重新问一次就行。实际开发中还有一个让我印象很深的坑asyncio 和同步库混用导致事件循环阻塞。因为在异步接口里去调 OpenAI 这种同步库一个慢请求会卡住整个 worker。解决方法是把耗时操作丢到线程池或者直接用asyncio.to_thread。如果你也遇到了一个请求慢其他请求全部排队的问题多半就是这个原因。5.4 成本与隐私控制别让知识库吃穷你个人项目也要考虑成本和隐私。我现在跑 Atlas 的月度成本大概在几美元量级大头是 GPT-4o-mini 的生成调用。为了控制开销我有几个经验一是缓存高频问题。同一个问题的嵌入向量和答案都可以缓存我用的是一个简单的 SQLite 缓存表7 天内命中直接返回。二是优先用便宜嵌入模型。嵌入调用虽然便宜但量大了也占成本。我用text-embedding-3-small而不是text-embedding-3-large在个人场景下效果差距可感知但不大成本却差 5 倍。三是本地化嵌入模型 云端生成模型。我在数据隐私敏感的项目里嵌入完全走本地 bge 模型只有最终的答案生成才调用云端 API。这样源文档内容不会上传传出去的只是向量虽然向量也能反推信息但风险比全文低得多。如果你的隐私要求极高可以把 LLM 也换成本地模型比如 ollama 跑 Qwen2.5牺牲一点生成质量换完全的数据主权。5.5 常用故障速查表遇到问题先查这里我把最常见的故障和排查动作整理成一张表方便你对照处理现象可能原因排查与解决检索结果为空embedding 维度不匹配检查 collection 的 vector size 与embed_query长度检索分数普遍很低查询和文档语言/领域差异大切换为中文专用 embedding 或换更强嵌入模型回答明显答非所问切块破坏了关键上下文增加 overlap 或改用按标题切块Agent 任务卡死同步库阻塞事件循环用asyncio.to_thread包裹 LLM 调用重复导入同一文档缺少去重逻辑入库前按 URL 标题哈希检查 sources 表导入 PDF 后文本乱码扫描版PDF未OCR开启 OCR 分支或先人工转文本向量库重启后数据丢失未挂载 Qdrant 存储目录检查 Docker 挂载配置改用持久化路径回答总说暂无相关信息RAG 检索没召回相关内容打印检索结果确认是切块问题还是嵌入模型不匹配这张表看起来简单但每一条背后都是我真金白银踩过的坑。尤其是重复导入这条我早期没有加去重导致同一篇文章在知识库里出现 5 份检索时同一个来源占掉一半的 top-k 名额。后来在save_source入口加了一个SELECT id FROM sources WHERE url ? AND title ?的唯一检查问题一次性解决。6. 扩展与未来规划Atlas 还能长成什么样我的习惯是搭一个能用的版本后先在实际使用中观察痛点再迭代下一个版本。Atlas 目前已经稳定服务我三个月但我已经在规划下一步的增强方向。第一个方向是浏览器剪藏插件。目前从网页导入需要复制 URL 调 API体验不够顺滑。我计划做一个浏览器扩展右键一键把当前页面标题、URL、正文摘要发给 Atlas 的/api/ingest/web接口并自动打上默认标签。这个插件的核心逻辑很简单难的是处理好正文提取这一步和 Atlas 的对接方式。第二个方向是双向同步 Obsidian。我有很多笔记写在 Obsidian 里目前 Atlas 也会导入这些 Markdown 文件但导入后如果笔记更新了Atlas 的向量化内容不会自动更新。我在ingest.py里加了一个文件哈希检查只在文件内容变化时重新向量化。同步的方案有两种一种是定时轮询 Obsidian 的 vault 目录另一种是 Obsidian 插件触发 webhook 通知 API。我倾向后者实时性好且不浪费计算资源。第三个方向是更聪明的知识图谱。目前 Atlas 的检索是纯向量语义检索但它不懂知识之间的关联。我想在向量检索之外叠加一层知识图谱——把每个 chunk 抽取出的实体人名、技术名词、项目名提取出来形成实体关系网。这样用户问讲一下我之前记录的 A 项目和 B 项目的关联系统可以走图谱路径找到直接相关的内容而不是完全依赖向量语义匹配。这块用到了命名实体识别和图数据库我准备用轻量的spaCy做 NER用Neo4j做图存储还在验证可行性。第四个方向是移动端入口。因为 FastAPI 已经暴露了完整的 REST API理论上一个最简单的手机网页就能调通。我想做一个极简的 PWA渐进式 Web 应用放在手机桌面上支持语音输入问题。语音转文字用系统自带的 API识别后的文字直接 POST 到 Atlas 的/api/query返回答案后再用系统的语音合成读出来。这就像随身带了一个你的知识库专属 AI 助手。7. 写在最后的实操体会Atlas 从构思到现在跑起来我最大的感受是知识管理的核心不是收集而是提取和连接。AI 增强让提取这件事变得几乎零成本但前提是你把管道搭对、把数据存好。整个搭建过程让我把 RAG、向量检索、Agent 编排这些概念从听说过变成了能落地。如果你也想搭一个自己的 Atlas我的建议是不要一上来就想搞个大而全的系统。先把导入文档 → 切块 → 向量化 → 检索 → 问答这条最小链路跑通用一周时间把自己的真实资料丢进去试再根据使用中的卡点去加功能。这比一开始就规划一堆 Agent、知识图谱、多端同步要靠谱得多——因为只有真实用起来你才知道哪些功能是刚需哪些只是想象。一个实践细节一定要人工抽查检索结果。不要只看生成的答案是否漂亮——RAG 系统的幻觉往往出在检索出来的材料本身与问题不符但模型把它包装得很顺。我每隔几天就会手动打开 debug 接口随机挑几个查询看 top-k 的原文片段一旦发现明显不相关的内容就去调切块策略或嵌入模型。这种人工质检对个人知识系统的长期可靠性至关重要。最后再分享一个让我很有成就感的小场景上周一个同事问我之前你发的那个关于 Postgres 分区表的文章是在哪看到的我打开 Atlas输入Postgres 分区表“为什么大表要分区”五秒钟给出了原文和出处链接。那一刻我知道这套系统不再只是一个技术玩具它已经成了我日常信息处理的基础设施。接下来的迭代方向很清楚但眼下它已经比任何我花钱买的商业笔记产品都更适合我了。 从零搭建AI增强的个人知识管理系统：RAG与向量检索实战
1. 项目缘起我的信息宇宙需要一个擎天柱朋友前阵子问我一个问题你每天收藏那么多文章、写那么多笔记、存那么多PDF真正想用的时候你找得到吗我愣了一下因为答案很尴尬——大部分时候找不到。我用过Notion、Obsidian、印象笔记各有各的好但都缺一个关键能力把存进去的资料变成能直接调用的答案。收藏夹越堆越满认知负担越来越大工具反而成了压力来源。所以我决定从零搭一个自己的系统代号Atlas。命名为Atlas是因为希腊神话里阿特拉斯用肩膀扛起整个天穹——我想让这个系统帮我扛起个人知识宇宙的承重。它不是一个笔记软件而是一个AI 增强的个人知识与生产力系统把散落的网页、PDF、Markdown、碎片想法统一收进本地知识库用大模型做切分、向量化、语义检索再通过几个AI Agent完成问答、摘要、任务拆解、待办提取这些日常动作。整个过程完全自控数据在自己手里想怎么改就怎么改。这篇博文会把我整套搭建思路、技术选型依据、核心代码实现、踩过的坑一次性说清楚。不管你是有一定 Python 基础的程序员还是刚开始接触 RAG 和个人知识管理的爱好者只要你想要一个越用越聪明的第二大脑这篇文章都值得你花十分钟读完。我尽量说人话该给代码给代码该给参数给参数保证你能照着搭起来。在动手之前先明确一件事Atlas 不是要替代 Notion 或 Obsidian而是做一个AI 层叠加在它们之上。知识源文件仍然可以用你习惯的工具管理Atlas 负责把内容切碎、向量化、做语义检索再把检索结果喂给大模型生成答案或行动项。这样各司其职既保留手动整理的安全感又获得 AI 检索的爽快感。1.1 我到底要解决哪几类痛点很多人以为知识管理就是存得多其实真正的问题在取不出。我给自己列了一个需求清单每一项都对应一个具体的日常场景。第一检索效率太低。关键词搜索有一个死穴你记不住原文里的精确措辞。比如你看过一篇讲分布式事务 Seata 原理的文章三个月后想找AT 模式回滚时全局锁怎么处理用关键词搜AT 模式可能命中搜全局锁就悬了。语义检索可以用向量距离匹配意思相近的内容模糊记得的东西也能捞出来。这是我的刚需也是 Atlas 的基石功能。第二已有资料没有被榨干。我的硬盘里躺着几百个 PDF很多是从 Tech 社区下载的白皮书、开源项目文档甚至有些读过一遍就再也没打开。这些资料本身有很高价值但它们是非结构化的没法直接参与问答。Atlas 可以把它们切块、嵌入成一个可检索的向量索引之后就能直接问这个框架的故障转移策略是什么它会从对应 PDF 的对应段落里找答案。第三AI 对话和知识库脱节。ChatGPT 用多了就发现一个问题它不知道我自己的私有资料每次都要把上下文粘进去既费 token 又丢信息。Atlas 的 RAG检索增强生成架构专门解决这个先检索我的知识库再把检索到的块拼进提示词让大模型带着材料回答问题准确率和可追溯性都高很多。第四被动收藏和主动行动之间缺一座桥。我经常收藏一篇讲 OKR 的文章然后就没有然后了。Atlas 里我加了一个行动提取Agent喂给它任何一篇文章它能抽出里面的可执行项、时间节点、责任人整理成待办清单。这个功能在周报场景特别实用——把一周读过的内容统一跑一遍自动生成下周的行动计划。1.2 为什么不用现成的方案市面上已经有不少AI 笔记工具比如我有段时间重度使用某款海外工具的 AI 问答和自动标签功能。说实话很惊艳但几个点让我最终放弃了第一数据不在本地私密性存疑有些工作材料根本不敢传第二可定制性差想调整分块策略、自定义检索权重、接入自己的模型都很难第三这类工具的 AI 功能大多有配额限制用几次就提示你升级会员。Atlas 走的是本地优先、组件解耦路线。向量数据库、嵌入模型、大模型、Web 框架全部可以替换没有供应商锁定。你早期可以用 OpenAI 的接口快速验证效果后期想换成本地模型比如 ollama 部署 Qwen 或 Llama也只需改一处封装。这个自由度是商业产品给不了你的。另外还有一个非常现实的理由这是一个极佳的学习项目。搭建 Atlas 的过程中你会亲手接触 RAG、embedding、向量检索、Agent 编排、异步任务队列这些概念不是看过教程那种程度的了解而是写代码让它跑起来的真掌握。我在搭完这套系统后对 AI 应用层的理解明显上了一个台阶。2. 系统设计与技术选型每一个选择的背后逻辑Atlas 的整体架构我用一句话概括采集层收料 → 处理层切片嵌入 → 存储层双写元数据和向量 → 检索层语义召回 → 生成层整合回答 → 应用层对接用户。听起来有点长但每一层解决的都是具体问题。先给一张整体逻辑表后面逐个讲层次负责内容我用到的组件替代方案采集层网页剪藏、PDF导入、Markdown导入Python trafilatura pypdf浏览器插件、Readwise处理层清洗、切块、嵌入向量递归字符切块 OpenAI embeddingLangChain 的 RecursiveCharacterTextSplitter存储层元数据 向量双写SQLite QdrantPostgreSQL pgvector检索层语义检索、元数据过滤Qdrant APIMilvus、Chroma生成层问答、摘要、行动提取OpenAI GPT-4o-miniClaude、本地 Qwen应用层REST API、任务编排FastAPI asyncioGradio、Chainlit这套选型不是拍脑袋拍出来的我逐个说理由。2.1 向量库选型为什么我选了 Qdrant 而不是其他向量库是整个 RAG 系统的核心底座选型时我对比过 Chroma、Milvus、Weaviate、pgvector。最终选了Qdrant核心原因是部署简单但能力不缩水。Qdrant 支持单机 Docker 启动也能用嵌入式模式直接跑在 Python 进程里对个人项目极其友好同时它的过滤能力很强可以按元数据字段来源、日期、标签做精确过滤后再做向量检索这对知识库场景太重要了——我只想搜某个 PDF 而不是全库时这个能力是刚需。为什么不选 pgvector如果你已经有 PostgreSQL那 pgvector 确实省一个组件但它的索引构建和查询性能在小数据量下没问题一旦超过几万条向量、还希望做带过滤的混合检索时调优成本就上来了。Qdrant 的 HNSW 索引是开箱即用的配置几个参数就行而且它原生返回 payload不用再回表查原始文本省一层麻烦。为什么不选 ChromaChroma 确实最轻但功能也最轻元数据过滤、负载均衡这些能力相对弱。个人项目初期可能够用但等你想做多用户或者上云托管时就尴尬了。Qdrant 可以无缝从本地 Docker 迁移到云托管版这个成长空间值得提前留好。2.2 嵌入模型选型便宜大碗才是真道理嵌入模型决定语义相似度的质量但这个质量并不非得靠贵模型。我的原则是嵌入模型要便宜、稳定、矢量维度适中。目前主力用的是text-embedding-3-small1536 维每百万 token 价格极低对个人项目来说成本几乎可以忽略。它的效果在英文和中文上都够用尤其适合知识库这种不需要太精细语义分野的场景。如果你有数据隐私顾虑完全可以用本地模型替代。我在一台不带 GPU 的 MacBook 上测试过bge-small-zh-v1.5512 维速度很快中文语义效果和 OpenAI 的 small 模型差距不大。关键是在我的代码里 embedding 是被封装成一个类方法embed_texts()的换模型只需要改这一个方法。这里强烈建议大家也这么做因为嵌入模型迭代很快你不想每次换个模型就重构整个数据管道。2.3 大模型选型生成质量与成本的平衡点生成层我选了 GPT-4o-mini理由比很多人想的简单它够聪明而且便宜。知识库问答这个场景真正要拼的不是模型的推理天花板而是 RAG 检索质量。检索给的材料对小模型也能给出很好的答案检索给错材料再大的模型也爱莫能助。所以把预算花在好的切块和检索上比盲目上旗舰模型划算得多。在这里多说一句现在很多人赶时髦用 Claude Opus 或 GPT-4o 做所有事其实有点浪费。我的使用习惯是分级调用——简单问答走 4o-mini需要复杂推理或多步工具调用时再走更强的模型。这个逻辑我封装在 LLM 调用层里用一个model_selector()函数根据任务类型自动选模型。后面讲 Agent 编排时你会看到它的好处。3. 从零实操搭建 Atlas 核心管线的完整过程这一章是全文的重头戏。我会按照自己实操的顺序逐步展开每一步都给出能直接用的代码和配置。我给 Atlas 写的代码全部开源在我的 GitHub 上这里展示的是核心部分的精简版目标是一个人可以在一小时内从空目录跑到能聊天的状态。3.1 环境准备与项目结构先交代运行环境我用的是 macOS Python 3.11用venv做虚拟环境所有服务都跑在本机。如果你是 Windows 用户绝大多数步骤一样只有 Docker 启动命令和文件路径写法有细微差异。项目结构采用模块化布局这样后续加功能不会变乱atlas/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── config.py # 配置项API Key、模型名、路径 │ ├── schemas.py # Pydantic 数据模型 │ ├── dao/ │ │ ├── sqlite_dao.py # 元数据读写 │ │ └── qdrant_dao.py # 向量读写 │ ├── services/ │ │ ├── ingest.py # 文档接入 │ │ ├── splitter.py # 切块逻辑 │ │ ├── embedder.py # 嵌入封装 │ │ ├── retriever.py # 检索合并 │ │ └── generator.py # 大模型调用 │ └── agents/ │ ├── orchestrator.py # 主控 Agent │ ├── researcher.py # 研究型 Agent │ └── action_extractor.py # 行动提取 Agent ├── data/ # 本地数据目录 │ ├── raw/ # 原始文件 │ └── atlas.db # SQLite 数据库 ├── scripts/ │ └── setup.sh # 一键初始化脚本 └── requirements.txt安装依赖。我的原则是能少就少用到才装。核心依赖如下pip install fastapi uvicorn qdrant-client openai pypdf \ trafilatura sqlalchemy python-dotenv tiktoken如果你打算跑本地嵌入模型再加一句pip install sentence-transformers初始化 Qdrant。我是用 Docker 启动的一条命令搞定docker run -d \ --name qdrant \ -p 6333:6333 \ -v $(pwd)/data/qdrant_storage:/qdrant/storage \ qdrant/qdrant:latest解释一下为什么要挂载存储目录Qdrant 的向量数据默认存在容器内部不挂载出来的话容器一删数据全没。个人知识库最不能丢的就是数据所以这一步别省。3.2 元数据模型与向量集合设计先想清楚要存什么动手写代码之前先把数据模型设计好。一个常见的误区是只管向量不管元数据等想按日期、来源筛选时发现没法查。我的设计分两层SQLite 存文档源信息Qdrant 存分块向量加轻量 payload。SQLite 里两张表-- 文档源表 CREATE TABLE sources ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, source_type TEXT NOT NULL, -- web / pdf / md url TEXT, file_path TEXT, author TEXT, tags TEXT, -- 逗号分隔 created_at DATETIME DEFAULT CURRENT_TIMESTAMP, raw_text TEXT -- 清洗后的完整文本 ); -- 分块表可选主要用来做映射和人工抽查 CREATE TABLE chunks ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_id INTEGER REFERENCES sources(id), chunk_index INTEGER, chunk_content TEXT, token_count INTEGER );这里raw_text字段很多人会觉得冗余但我故意保留。原因有两点一是做数据一致性校验时能快速看某个 source 是否正常入库二是如果你想切换嵌入模型、重新向量化无需重新抓取和清洗直接从 SQLite 里读原文就行省大量网络请求。Qdrant 里的 collection 我命名为atlas_docspayload 字段设计如下{ source_id: 对应SQLite里的source id, chunk_index: 第几个分块, title: 源文档标题, url: 如果来自网页就有否则为空, date: 采集日期, tags: [标签1, 标签2] }向量本身由 embedding 模型生成维度取决于你选的模型1536 或 512。Payload 里放 source_id 和 chunk_index 是必要的这样检索后能回 SQLite 拿完整上下文。这里有个实践细节切块时不要只存块本身最好把相邻的上一块和下一块也存进去。我后面讲到检索优化时会展开先埋个伏笔。3.3 文档接入与清洗网页、PDF、Markdown 三路归一采集层的目标是把不同类型的输入统一转成干净的文本再进入切块流程。我不做实时爬虫而是提供导入接口——你在浏览器看到一篇好文复制 URL 调接口手头有 PDF直接上传写了篇 Markdown 笔记也传进来。Atlas 对它们一视同仁。网页抓取我用trafilatura而不是简单的requests BeautifulSoup。原因很直接trafilatura 是专门做正文抽取的库能自动去掉导航、脚注、侧边栏这些噪声对我这种新闻、文档、博客混合场景的清洗效果明显更好。代码很简单import trafilatura def fetch_web_article(url: str) - str: downloaded trafilatura.fetch_url(url) if downloaded is None: raise ValueError(f无法抓取该URL请检查地址或网络) text trafilatura.extract( downloaded, include_commentsFalse, include_tablesTrue, favor_precisionTrue ) return text说个小坑favor_precisionTrue这个参数是个双刃剑。它会让抽取结果更精确、噪声更少但有时候会把正文截断尤其遇到那种阅读全文式的分页文章。我的建议是默认开 precision如果发现某篇文章抓回的内容明显变短再用favor_recallTrue重抓一次做对比。PDF 处理我用pypdf但纯文本 PDF 好解决扫描版 PDF 就麻烦了。扫描版需要 OCR我在项目里留了一个可选的 OCR 分支调用pytesseract做识别。代码不复杂但依赖较重默认不启用。如果你经常处理扫描文档可以在config.py里加一个enable_ocr True开关。from pypdf import PdfReader def extract_pdf_text(path: str) - str: reader PdfReader(path) text_parts [] for i, page in enumerate(reader.pages): page_text page.extract_text() if page_text and page_text.strip(): text_parts.append(f--- 第{i1}页 ---\n{page_text}) return \n\n.join(text_parts)Markdown 导入最省事直接读文件然后用正则把代码块和普通段落分开标记就行。我这里特意做了一个处理保留代码块的语言标记和标题层级标记因为切块时我想让模型知道这部分是代码那部分是一级标题对后续问答准确度有帮助。3.4 切块策略让语义完整性和检索粒度达到平衡切块是整个 RAG 里最容易被低估的环节。块太大向量被稀释检索出来的东西不精准块太小语义不完整模型没法理解上下文。我试过固定大小切块比如每 500 字符硬切效果很差——经常一句话被从中间切断检索回来就是一堆残缺信息。最终我采用递归字符切块 重叠窗口的组合策略。思路是优先按 Markdown 标题切切出来的段落如果太长再按段落符\n\n切还是太长就按句子切。每一步都检查长度上限超出上限才继续往下拆。核心实现如下可以做简化版import tiktoken enc tiktoken.encoding_for_model(gpt-4o-mini) CHUNK_SIZE 800 # token 上限 CHUNK_OVERLAP 120 # 相邻块重叠 token 数 def count_tokens(text: str) - int: return len(enc.encode(text)) def split_text_recursive(text: str, chunk_sizeCHUNK_SIZE, chunk_overlapCHUNK_OVERLAP): # 先按标题分 sections re.split(r(?m)^(?#{1,3} ), text) chunks [] buffer for section in sections: # 如果单块超长按段落继续拆 if count_tokens(section) chunk_size: sub_parts re.split(r\n\n, section) for part in sub_parts: if count_tokens(part) chunk_size: # 按句子继续拆 sentences re.split(r(?[。.!?])\s*, part) local_buffer for sent in sentences: if count_tokens(local_buffer sent) chunk_size: local_buffer sent else: if local_buffer: chunks.append(buffer local_buffer) buffer local_buffer[-chunk_overlap:] # 重叠尾巴 local_buffer sent else: if count_tokens(buffer part) chunk_size: buffer part else: chunks.append(buffer part) buffer part[-chunk_overlap:] else: if count_tokens(buffer section) chunk_size: buffer section else: chunks.append(buffer section) buffer section[-chunk_overlap:] if buffer: chunks.append(buffer) return chunks这段代码我故意写得比较啰嗦目的是让你看清递归切块的真实流程。实际工程里可以直接用 LangChain 的RecursiveCharacterTextSplitter参数调好了一样工作。但理解原理比调库更重要——我后来调很多检索问题靠的都是对切块在哪一步破坏了语义的判断。关于 chunk size 的大小我的经验数据是中文场景 500~800 token 比较合适英文可以偏大到 1000。为什么中文要偏小因为中文每个 token 承载的信息密度比英文高同样 800 token 的中文块语义复杂度和英文 1200 token 差不多。重叠窗口设在 100~150 token能保证跨块语义的连贯性。3.5 向量化与入库写一个可替换的嵌入服务做完整条切块管线后就可以向量化入库了。我强烈推荐把嵌入逻辑封装成独立类原因前面讲过——模型会更新你的代码不应该跟着改。from openai import OpenAI class EmbeddingService: def __init__(self, provideropenai): self.provider provider if provider openai: self.client OpenAI() self.model text-embedding-3-small self.dim 1536 else: from sentence_transformers import SentenceTransformer self.model_local SentenceTransformer(BAAI/bge-small-zh-v1.5) self.dim 512 def embed_texts(self, texts: list[str]) - list[list[float]]: if self.provider openai: resp self.client.embeddings.create(modelself.model, inputtexts) return [item.embedding for item in resp.data] else: return self.model_local.encode(texts, normalize_embeddingsTrue).tolist() def embed_query(self, query: str) - list[float]: return self.embed_texts([query])[0]入库逻辑我写成批量处理每凑够 32 个块就做一次批量向量化再批量 upsert 到 Qdrant。用小批量而不是全量是为了在出问题时能快速定位是哪个文档的哪个块出了问题。from qdrant_client import QdrantClient from qdrant_client.http import models class VectorStore: def __init__(self, collection_nameatlas_docs): self.client QdrantClient(hostlocalhost, port6333) self.collection collection_name self.ensure_collection() def ensure_collection(self): existing self.client.get_collections().collections if not any(c.name self.collection for c in existing): self.client.create_collection( collection_nameself.collection, vectors_configmodels.VectorParams( size1536, # 如果换了嵌入模型注意同步改 distancemodels.Distance.COSINE ) ) def upsert_chunks(self, chunks_with_payloads): points [ models.PointStruct( idrow[uuid], vectorrow[vector], payloadrow[payload] ) for row in chunks_with_payloads ] self.client.upsert(collection_nameself.collection, pointspoints)有几个细节值得注意。距离度量我选COSINE而不是L2文本向量经过归一化后余弦相似度对语义方向一致但长度不同的情况更宽容实际检索效果也是 COSINE 更符合直觉。另外我给每条点指定的 id 是 uuidQdrant 允许用自增整数但多文档并行导入时整数冲突的排查成本更高uuid 虽然占用稍大但省心。3.6 检索与生成RAG 最核心的 20 行代码所有前面的努力最终都要落到用户提问 → 检索 → 生成答案这个闭环上。先说检索我的实现是query 经 embedding 查 Qdrant 取 top-k然后按 payload 里的 source_id 去 SQLite 取原始上下文拼入 prompt 喂给大模型。from qdrant_client import QdrantClient from qdrant_client.http import models class Retriever: def __init__(self, embedder, vector_store): self.embedder embedder self.store vector_store def retrieve(self, query: str, top_k: int 5, source_filter: int None): query_vector self.embedder.embed_query(query) query_filter None if source_filter is not None: query_filter models.Filter( must[models.FieldCondition( keysource_id, matchmodels.MatchValue(valuesource_filter) )] ) hits self.store.client.query_points( collection_nameself.store.collection, queryquery_vector, limittop_k, query_filterquery_filter ) return [ { score: hit.score, payload: hit.payload, text: hit.payload.get(text, ) } for hit in hits.points ]注意我这里把每个 chunk 的原文也放进了 Qdrant 的 payload 里。这看似冗余SQLite 里已经有 raw_text但能少一次回表查询对响应速度有帮助。代价是向量库空间占用多一点对个人项目完全可接受。生成 prompt 的设计上我的模板是你是 Atlas 知识库助手。请基于下面的资料回答用户问题。 资料引用时标注来源编号。如果资料不足以回答请明确说知识库中暂无相关信息。 资料 [1] {chunk_text_1} (来源{title}) [2] {chunk_text_2} (来源{title}) ... 问题{query}这里有两个小心机一是要求模型在信息不足时直接承认避免幻觉二是要求标注来源编号方便用户回查原文。我实际使用中这两个规则让 Atlas 的答案可信度高了很多尤其是涉及技术资料时我能直接点开原文核对而不是盲目相信大模型的转述。生成调用我用了一个简单的分级模型选择器def generate_answer(query, context_chunks, query_typesimple): model gpt-4o-mini if query_type complex: model gpt-4o messages [ {role: system, content: 你是Atlas知识库助手。}, {role: user, content: build_prompt(query, context_chunks)} ] resp client.chat.completions.create( modelmodel, messagesmessages, temperature0.3 ) return resp.choices[0].message.contenttemperature 设 0.3 而不是 0留一点点随机性防止措辞太死板但又能保证事实准确性。很多教程直接设 0我试过之后觉得回答太机械0.3 是比较舒服的区间。3.7 用 FastAPI 包一层让系统可以随时被调用为了让 Atlas 能被浏览器、命令行、甚至是未来的手机快捷指令调用我用 FastAPI 包了一层 REST API。设计成三个核心接口导入文档、查询问答、提取行动项。这样 Atlas 就是一个本地 AI 服务而不是一个孤立的脚本。from fastapi import FastAPI, UploadFile, File, Form from pydantic import BaseModel app FastAPI(titleAtlas) class QueryRequest(BaseModel): question: str source_id: int | None None top_k: int 5 class QueryResponse(BaseModel): answer: str references: list[dict] app.post(/api/query, response_modelQueryResponse) async def query_atlas(req: QueryRequest): hits retriever.retrieve(req.question, top_kreq.top_k, source_filterreq.source_id) context [{ chunk_text: h[text], title: h[payload].get(title, ), source_id: h[payload].get(source_id) } for h in hits] answer generate_answer(req.question, context) return QueryResponse(answeranswer, referencescontext) app.post(/api/ingest/web) async def ingest_web(url: str Form(...)): text fetch_web_article(url) source_id save_source(titleurl, source_typeweb, urlurl, raw_texttext) process_and_embed(source_id, text) return {status: ok, source_id: source_id} app.post(/api/ingest/pdf) async def ingest_pdf(file: UploadFile File(...)): content await file.read() path fdata/raw/{file.filename} with open(path, wb) as f: f.write(content) text extract_pdf_text(path) source_id save_source(titlefile.filename, source_typepdf, file_pathpath, raw_texttext) process_and_embed(source_id, text) return {status: ok, source_id: source_id} if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)启动uvicorn app.main:app --reloadAtlas 的第一个可用版本就跑起来了。此时你导入一篇文章然后问这篇文章的核心观点是什么它就能基于你的资料库作答。4. Agent 编排让 Atlas 从回答工具变成生产力助手RAG 问答只是第一步Atlas 更重要的价值在于多 AI 协作地完成生产力任务。我不满足于有问必答我还想要它能把读到的内容变成待办事项、自动生成摘要、为复杂问题拆解研究计划并执行。这就轮到 Agent 机制上场了。4.1 Orchestrator一个会点将的主控 Agent我的设计是轻量的Orchestrator 专用 Agent模式。Orchestrator 不直接生成答案它负责理解用户需求判断该调用哪个专用 Agent再把多个 Agent 的结果合并输出。这样做的好处是每个专用 Agent 逻辑简单、容易调试而 orchestration 层可以自由组合出复杂行为。我举一个实际例子。用户问帮我整理一下最近收藏的内容看看有没有适合做周末读书分享的材料。 Orchestrator 会把这个任务拆成几步调用 Researcher Agent检索最近一周入库的文档生成每篇的摘要调用 ActionExtractor Agent从检索结果中提取可以作为分享话题的候选由 Writer Agent 把这些候选整合成一个带推荐理由的清单。这三个步骤不是串行等待我用asyncio.gather让独立的检索任务并行执行整体耗时从十几秒降到五六秒。代码结构如下async def run_orchestration(user_request: str): plan planner_agent.create_plan(user_request) results await asyncio.gather( *[execute_agent_step(step) for step in plan] ) return writer_agent.combine_results(results, user_request)这个实现看起来简单但踩过一个大坑如果每个 Agent 都独立调用 LLM最后汇总时上下文会丢失。比如 Researcher 生成了摘要Writer 再要写清单时它不知道摘要对应的原文是什么。解决办法是在 Agent 之间传递结构化的中间结果JSON 对象而不是只传纯文本。我在AgentResult里统一加了source_id和confidence字段让下游 Agent 能追溯到原始材料。4.2 ActionExtractor从文章到待办事项的自动流转ActionExtractor 是我用得最频繁的一个 Agent。它的功能一句话说明把输入文本中的可执行内容和时间敏感信息提取成结构化待办。比如你导入一篇产品规划文章里面写了下周三前完成竞品分析整理成文档发给技术团队。普通收藏这件事就过去了但 ActionExtractor 能识别出这是一个任务竞品分析截止时间下周三负责动作完成并整理文档发送。实现上我用了一次小模型 强提示词的方案ACTION_PROMPT 请从以下文本中提取所有可执行任务。以JSON数组输出每个任务包含 - task: 任务描述尽量简洁 - due: 截止时间若不明确则为 null - priority: high/medium/low - related_source_id: {source_id} - context: 任务来源的简要上下文不超过20字 文本内容 {input_text} 这个方案的工程关键点在于JSON 解析容错。大模型偶尔会输出非标准 JSON比如带注释、带 markdown 代码块标记直接用json.loads会崩。我的处理是先用正则去掉json标记再做一轮简单的花括号平衡检查解析失败时再给模型一次修正机会。def safe_parse_json(raw: str): text re.sub(rjson|, , raw).strip() try: return json.loads(text) except json.JSONDecodeError: # 尝试从第一个 { 到最后一个 } 截取 start text.find({) end text.rfind(}) if start ! -1 and end ! -1: candidate text[start:end1] return json.loads(candidate) raise ValueError(无法解析模型输出为JSON)后来我把提取出来的任务直接同步到本地的一个 Todoist 测试账号用它们的 REST API 走了一轮自动化流转。效果比我预期的好现在每周五下午我会把一周收藏的文章统一喂给 ActionExtractor生成下周计划再人工过一遍。这个AI 辅助规划 人工确认的模式比全自动靠谱得多。4.3 多 Agent 协作时的上下文管理多 Agent 协作最头疼的问题是上下文管理。我的经验是不要让 Agent 记忆长对话而是让它带上必要的最小上下文。比如 Researcher Agent 负责检索它没必要知道用户之前的聊天记录ActionExtractor 也没必要知道用户目前看到第几页。每个 Agent 只接收它完成任务所需的字段查询文本、可用的检索结果、特定的指令。这就像团队协作时你给某个同事交代任务只需要告诉他这部分材料你负责而不是把整场会议纪要都发给他。我在代码里给每个 Agent 定义了明确的输入输出 schemadataclass class ResearcherInput: query: str top_k: int 5 dataclass class ResearcherOutput: summary: str references: list[dict] source_ids: list[int]这套设计让我后来调试为什么答案不对时非常舒服每个中间结果都能单独打印出来检查整个链路的可观测性很强。如果你用 LangGraph 这类框架逻辑类似但自己实现一遍会更加理解状态流转的本质。5. 常见问题与排查技巧实录踩过的坑全靠这些笔记兜底最后这部分是纯干货记录我在搭建和持续使用 Atlas 过程中遇到的高频问题。每一条都是我实际遇到并解决的看完应该能帮你少走很多弯路。5.1 切块结果不理想为什么检索总是答非所问遇到最多的问题是我明明导入了相关内容为什么问它它说不知道。排查步骤基本是固定的链路先看检索结果再看生成过程。第一步打印检索结果。我加了一个 debug 接口当检索分数低于 0.6 时就自动输出检索到的前三条块的内容。你会发现大概率是切块切碎了关键信息。比如一个表格被拆到两个块里模型只拿到其中一半自然答不上来。针对切块的修正方法先看问题是长文本还是结构化表格数据。表格数据我会在预处理阶段单独标记不让切块器把表格行拆散长文本则考虑把 chunk_size 调大一点。另一个很有效的技巧是在存储时把前一 block 的尾巴拼到当前块的开头也就是前面埋的伏笔。我把这个实现为recursive_overlap参数默认 120 token。这样即使语义跨块至少前后呼应的内容不会丢。还有一个被忽略的问题是query 和 chunk 的 embedding 维度不一致。如果你中途换过 embedding 模型Qdrant collection 的维度和你新 query 的维度对不上所有检索都会报错。检查方式很直接client.get_collection(atlas_docs)看 vector size再打印embed_query(test)的长度。5.2 多来源混合检索时的权重问题个人知识库里的来源五花八门网页收藏、PDF 论文、个人笔记。它们的可信度完全不同。我早期不做区分检索结果经常被低质量网页占满PDF 里的高质量内容反而排不上。解法是给每个来源类型设置一个权重系数在最终排序时对分数做加权。Qdrant 允许返回原始分数我在后端做排序修正SOURCE_WEIGHT { web: 0.9, pdf: 1.1, md: 1.0, } def rerank_hits(hits): for hit in hits: src_type hit.payload.get(source_type, web) hit.score * SOURCE_WEIGHT.get(src_type, 1.0) hits.sort(keylambda h: h.score, reverseTrue) return hits听着简单效果立竿见影。之后涉及技术论文的查询答案的引用来源明显偏向 PDF 资料。这个权重你也可以继续细化比如给高星博客高一点、给论坛帖子低一点逐步调成自己的偏好。5.3 Agent 任务编排时的超时与重试机制Agent 编排跑起来后你会遇到另一个问题某个中间步骤超时整个任务卡在那里。尤其当 Researcher Agent 需要连续调用多次 LLM 时一个慢响应就会拖累全局。我的解决方案是在每步 LLM 调用上设置超时和重试。OpenAI 的 Python SDK 支持超时参数client OpenAI(timeout30.0, max_retries2)但这只能保证单次调用不卡死。对 Agent 级别的任务我在 orchestrator 层做了一层熔断控制——每个子任务最多执行 3 次超过就标记失败并跳过而不是让整个任务失败。毕竟在个人知识系统里一次检索失败没那么严重重新问一次就行。实际开发中还有一个让我印象很深的坑asyncio 和同步库混用导致事件循环阻塞。因为在异步接口里去调 OpenAI 这种同步库一个慢请求会卡住整个 worker。解决方法是把耗时操作丢到线程池或者直接用asyncio.to_thread。如果你也遇到了一个请求慢其他请求全部排队的问题多半就是这个原因。5.4 成本与隐私控制别让知识库吃穷你个人项目也要考虑成本和隐私。我现在跑 Atlas 的月度成本大概在几美元量级大头是 GPT-4o-mini 的生成调用。为了控制开销我有几个经验一是缓存高频问题。同一个问题的嵌入向量和答案都可以缓存我用的是一个简单的 SQLite 缓存表7 天内命中直接返回。二是优先用便宜嵌入模型。嵌入调用虽然便宜但量大了也占成本。我用text-embedding-3-small而不是text-embedding-3-large在个人场景下效果差距可感知但不大成本却差 5 倍。三是本地化嵌入模型 云端生成模型。我在数据隐私敏感的项目里嵌入完全走本地 bge 模型只有最终的答案生成才调用云端 API。这样源文档内容不会上传传出去的只是向量虽然向量也能反推信息但风险比全文低得多。如果你的隐私要求极高可以把 LLM 也换成本地模型比如 ollama 跑 Qwen2.5牺牲一点生成质量换完全的数据主权。5.5 常用故障速查表遇到问题先查这里我把最常见的故障和排查动作整理成一张表方便你对照处理现象可能原因排查与解决检索结果为空embedding 维度不匹配检查 collection 的 vector size 与embed_query长度检索分数普遍很低查询和文档语言/领域差异大切换为中文专用 embedding 或换更强嵌入模型回答明显答非所问切块破坏了关键上下文增加 overlap 或改用按标题切块Agent 任务卡死同步库阻塞事件循环用asyncio.to_thread包裹 LLM 调用重复导入同一文档缺少去重逻辑入库前按 URL 标题哈希检查 sources 表导入 PDF 后文本乱码扫描版PDF未OCR开启 OCR 分支或先人工转文本向量库重启后数据丢失未挂载 Qdrant 存储目录检查 Docker 挂载配置改用持久化路径回答总说暂无相关信息RAG 检索没召回相关内容打印检索结果确认是切块问题还是嵌入模型不匹配这张表看起来简单但每一条背后都是我真金白银踩过的坑。尤其是重复导入这条我早期没有加去重导致同一篇文章在知识库里出现 5 份检索时同一个来源占掉一半的 top-k 名额。后来在save_source入口加了一个SELECT id FROM sources WHERE url ? AND title ?的唯一检查问题一次性解决。6. 扩展与未来规划Atlas 还能长成什么样我的习惯是搭一个能用的版本后先在实际使用中观察痛点再迭代下一个版本。Atlas 目前已经稳定服务我三个月但我已经在规划下一步的增强方向。第一个方向是浏览器剪藏插件。目前从网页导入需要复制 URL 调 API体验不够顺滑。我计划做一个浏览器扩展右键一键把当前页面标题、URL、正文摘要发给 Atlas 的/api/ingest/web接口并自动打上默认标签。这个插件的核心逻辑很简单难的是处理好正文提取这一步和 Atlas 的对接方式。第二个方向是双向同步 Obsidian。我有很多笔记写在 Obsidian 里目前 Atlas 也会导入这些 Markdown 文件但导入后如果笔记更新了Atlas 的向量化内容不会自动更新。我在ingest.py里加了一个文件哈希检查只在文件内容变化时重新向量化。同步的方案有两种一种是定时轮询 Obsidian 的 vault 目录另一种是 Obsidian 插件触发 webhook 通知 API。我倾向后者实时性好且不浪费计算资源。第三个方向是更聪明的知识图谱。目前 Atlas 的检索是纯向量语义检索但它不懂知识之间的关联。我想在向量检索之外叠加一层知识图谱——把每个 chunk 抽取出的实体人名、技术名词、项目名提取出来形成实体关系网。这样用户问讲一下我之前记录的 A 项目和 B 项目的关联系统可以走图谱路径找到直接相关的内容而不是完全依赖向量语义匹配。这块用到了命名实体识别和图数据库我准备用轻量的spaCy做 NER用Neo4j做图存储还在验证可行性。第四个方向是移动端入口。因为 FastAPI 已经暴露了完整的 REST API理论上一个最简单的手机网页就能调通。我想做一个极简的 PWA渐进式 Web 应用放在手机桌面上支持语音输入问题。语音转文字用系统自带的 API识别后的文字直接 POST 到 Atlas 的/api/query返回答案后再用系统的语音合成读出来。这就像随身带了一个你的知识库专属 AI 助手。7. 写在最后的实操体会Atlas 从构思到现在跑起来我最大的感受是知识管理的核心不是收集而是提取和连接。AI 增强让提取这件事变得几乎零成本但前提是你把管道搭对、把数据存好。整个搭建过程让我把 RAG、向量检索、Agent 编排这些概念从听说过变成了能落地。如果你也想搭一个自己的 Atlas我的建议是不要一上来就想搞个大而全的系统。先把导入文档 → 切块 → 向量化 → 检索 → 问答这条最小链路跑通用一周时间把自己的真实资料丢进去试再根据使用中的卡点去加功能。这比一开始就规划一堆 Agent、知识图谱、多端同步要靠谱得多——因为只有真实用起来你才知道哪些功能是刚需哪些只是想象。一个实践细节一定要人工抽查检索结果。不要只看生成的答案是否漂亮——RAG 系统的幻觉往往出在检索出来的材料本身与问题不符但模型把它包装得很顺。我每隔几天就会手动打开 debug 接口随机挑几个查询看 top-k 的原文片段一旦发现明显不相关的内容就去调切块策略或嵌入模型。这种人工质检对个人知识系统的长期可靠性至关重要。最后再分享一个让我很有成就感的小场景上周一个同事问我之前你发的那个关于 Postgres 分区表的文章是在哪看到的我打开 Atlas输入Postgres 分区表“为什么大表要分区”五秒钟给出了原文和出处链接。那一刻我知道这套系统不再只是一个技术玩具它已经成了我日常信息处理的基础设施。接下来的迭代方向很清楚但眼下它已经比任何我花钱买的商业笔记产品都更适合我了。