1. 从一次召回失败说起为什么你的 Memory 向量记忆系统总答非所问如果你正在给本地知识库或代码仓库搭一套 Memory 向量记忆系统大概率遇到过这种场景明明文档里写了答案检索回来的却是隔壁章节的碎片代码里搜一个函数名召回的是注释里提过一嘴的无关文件。问题往往不在 embedding 模型本身而是出在文本向量化之前的切分环节——切得太碎、切断了语义边界、或者代码被按固定长度硬切成了半截函数。这篇是 Memory 向量记忆系统系列的第四篇聚焦文本向量化这条链路怎么把原始文本和代码切成适合 embedding 的 chunk怎么用 AST 做代码分块怎么把向量存进本地库并支持增量更新以及怎么用 TaoToken 的统一 Key 把 embedding 请求接进来。适合正在做本地 RAG、代码检索、个人知识库的开发者尤其是已经跑通 demo 但召回质量不稳定的同学。我会给出可直接复制的settings.json和config.toml骨架一套 AST 分块的 Python 实现以及向量化结果和召回效果的验证动作。全程不依赖任何特殊网络环境TaoToken 的 API 地址直接填https://taotoken.net/api即可。2. TaoToken 前置统一 Key 接入 embedding 与对话模型在动手切分之前先把模型接入这层理清楚。Memory 向量记忆系统通常需要两类模型一类是 embedding 模型负责把 chunk 转成向量另一类是对话模型负责基于召回结果生成回答。如果每个模型都单独配一套 Key 和 base_url配置会很快失控。TaoToken 的做法是提供一个统一的 API 入口embedding 和对话模型共用同一个 Keybase_url 统一为https://taotoken.net/api。你只需要在控制台创建一个 API Key然后在配置里指定模型名即可。对于本地知识库场景我一般用 BGE-M3 做 embedding对话侧按需选模型。先去控制台拿 Key访问https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite创建后复制保存。注意 Key 只在创建时完整显示一次丢了就重新建一个。拿到 Key 之后建议先确认模型列表和可用性。可以打开模型对话页面https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite手动发一条消息确认 Key 生效。这一步能省掉后面很多「到底是 Key 错了还是代码错了」的排查时间。如果你后续要做长期编码或 Agent 类应用可以了解下 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite。本篇主要走 API 接入路线Key 的管理入口在https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite接入细节文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite。注意TaoToken 是模型 API 的统一接入层不是编辑器替代品也不做任何灰色中转。你的代码、文档、向量数据始终留在本地只有 embedding 请求会发出去。3. 可复制配置settings.json 与 config.toml 骨架配置分两块一块是模型接入配置一块是分块与存储配置。我习惯把模型相关的放settings.json把分块参数和存储路径放config.toml这样换模型不用动分块逻辑。3.1 settings.json模型接入与 Key 管理{ provider: { name: taotoken, base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, timeout_seconds: 60, max_retries: 3 }, embedding: { model: BGE-M3, dimension: 1024, batch_size: 16, normalize: true }, chat: { model: default-chat, temperature: 0.2, max_tokens: 2048 }, memory: { store_path: ./data/memory.db, cache_path: ./data/embedding_cache.db } }Key 不要写进文件用环境变量注入export TAOTOKEN_API_KEY你的Keydimension必须和 embedding 模型输出维度一致。BGE-M3 是 1024 维如果你换成 GTE-large 就是 1024换成 E5-small 就是 384。维度填错写入向量库时会直接报错或者更糟——静默写入错误长度的向量导致检索全乱。3.2 config.toml分块参数与存储结构[chunking.text] strategy recursive chunk_size 500 chunk_overlap 80 separators [\n\n, \n, 。, , , ., , ] keep_metadata true [chunking.code] strategy ast max_chunk_tokens 500 include_leading_comments true fallback_to_lines true fallback_window 40 [storage] backend sqlite enable_fts true enable_vec true vec_dimension 1024 [retrieval] top_k 8 hybrid true dense_weight 0.7 sparse_weight 0.3文本侧用递归字符切分分隔符优先级从段落降到句子再到词尽量在语义边界断开。代码侧走 AST按函数/方法切超过max_chunk_tokens的函数再按逻辑块细分。fallback_to_lines是给解析失败的语言兜底避免整个文件被跳过。4. AST 分块实战把代码切成完整的函数单元AST 分块的核心思路是用语法解析器把代码变成树遍历树找到函数、类、方法这些节点用节点的起止行号切出完整代码块。这样切出来的 chunk 不会出现「函数头在上一块、函数体在下一块」的情况。4.1 Python 原生 ast 模块实现Python 自带ast模块零依赖就能做函数级切分import ast import hashlib from pathlib import Path def chunk_python_by_ast(source: str, file_path: str, max_lines: int 80): tree ast.parse(source) lines source.splitlines() chunks [] for node in ast.walk(tree): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): start node.lineno - 1 end node.end_lineno body \n.join(lines[start:end]) if end - start max_lines: chunks.extend(split_large_node(lines, node, file_path)) else: chunks.append(build_chunk(body, file_path, start 1, end, node.name)) return chunks def build_chunk(content, path, start_line, end_line, symbol): content_hash hashlib.sha256(content.encode()).hexdigest()[:16] chunk_id hashlib.sha256( fcode:{path}:{start_line}:{end_line}:{content_hash}.encode() ).hexdigest() return { id: chunk_id, file_path: path, content: content, start_line: start_line, end_line: end_line, symbol: symbol, hash: content_hash, }ast.walk会递归遍历所有节点FunctionDef、AsyncFunctionDef、ClassDef分别对应普通函数、异步函数和类。end_lineno是 Python 3.8 才有的属性低版本需要自己算。4.2 多语言场景用 tree-sitterPython 项目用原生 ast 就够了但如果你要处理 JavaScript、Go、Java建议上 tree-sitter。它支持 40 多种语言API 统一from tree_sitter import Language, Parser import tree_sitter_python as tspython PY_LANGUAGE Language(tspython.language()) parser Parser(PY_LANGUAGE) def chunk_by_tree_sitter(source: str, file_path: str): tree parser.parse(source.encode()) root tree.root_node chunks [] def visit(node): if node.type in (function_definition, class_definition): content source[node.start_byte:node.end_byte] chunks.append(build_chunk( content, file_path, node.start_point[0] 1, node.end_point[0] 1, node.child_by_field_name(name).text.decode() )) for child in node.children: visit(child) visit(root) return chunkstree-sitter 的节点类型名因语言而异Python 是function_definitionJavaScript 是function_declaration和method_definition。实际项目里我会维护一张语言到节点类型的映射表。4.3 大函数怎么处理一个函数超过 500 token 是常事直接整块 embedding 会稀释语义。我的做法是按函数体内的语句块再切def split_large_node(lines, node, file_path): sub_chunks [] for child in node.body: if isinstance(child, (ast.If, ast.For, ast.While, ast.Try)): start child.lineno - 1 end child.end_lineno body \n.join(lines[start:end]) sub_chunks.append(build_chunk( body, file_path, start 1, end, f{node.name}:block )) return sub_chunks这样切出来的子块仍然带父函数名作为元数据检索时能通过symbol字段回溯到完整函数。5. 向量存储与增量更新SQLite sqlite-vec 方案切完块就要存。本地场景我推荐 SQLite 加 sqlite-vec 扩展单文件、零运维、支持向量检索和全文检索混合。5.1 表结构设计CREATE TABLE IF NOT EXISTS files ( path TEXT PRIMARY KEY, hash TEXT NOT NULL, mtime INTEGER NOT NULL, size INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS chunks ( id TEXT PRIMARY KEY, file_path TEXT NOT NULL, content TEXT NOT NULL, start_line INTEGER, end_line INTEGER, symbol TEXT, hash TEXT NOT NULL, FOREIGN KEY (file_path) REFERENCES files(path) ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS embedding_cache ( content_hash TEXT NOT NULL, model TEXT NOT NULL, embedding TEXT NOT NULL, created_at INTEGER DEFAULT (unixepoch()), PRIMARY KEY (content_hash, model) ); CREATE VIRTUAL TABLE IF NOT EXISTS chunks_vec USING vec0( id TEXT PRIMARY KEY, embedding FLOAT[1024] ); CREATE VIRTUAL TABLE IF NOT EXISTS chunks_fts USING fts5( id, content, contentchunks, content_rowidrowid );embedding_cache用content_hash model做联合主键同一个 chunk 换模型会重新算不换模型直接命中缓存。这一层能省掉大量重复的 embedding 调用。5.2 增量更新逻辑更新时先比对文件 hash只处理变化的文件def sync_file(conn, file_path: Path, embed_fn): content file_path.read_text(encodingutf-8) file_hash hashlib.sha256(content.encode()).hexdigest() row conn.execute( SELECT hash FROM files WHERE path ?, (str(file_path),) ).fetchone() if row and row[0] file_hash: return 0 conn.execute(DELETE FROM chunks WHERE file_path ?, (str(file_path),)) chunks chunk_python_by_ast(content, str(file_path)) for chunk in chunks: cached conn.execute( SELECT embedding FROM embedding_cache WHERE content_hash ? AND model ?, (chunk[hash], BGE-M3) ).fetchone() if cached: vector json.loads(cached[0]) else: vector embed_fn(chunk[content]) conn.execute( INSERT OR REPLACE INTO embedding_cache VALUES (?, ?, ?, unixepoch()), (chunk[hash], BGE-M3, json.dumps(vector)) ) conn.execute( INSERT INTO chunks VALUES (?, ?, ?, ?, ?, ?, ?), (chunk[id], chunk[file_path], chunk[content], chunk[start_line], chunk[end_line], chunk[symbol], chunk[hash]) ) conn.execute( INSERT INTO chunks_vec (id, embedding) VALUES (?, ?), (chunk[id], json.dumps(vector)) ) conn.execute( INSERT OR REPLACE INTO files VALUES (?, ?, ?, ?), (str(file_path), file_hash, int(file_path.stat().st_mtime), len(content)) ) conn.commit() return len(chunks)删除文件时files表的级联删除会自动清掉对应 chunks但chunks_vec和chunks_fts是虚拟表需要手动同步删除。6. 验证请求与召回效果三步确认链路通了配置写完不算完得验证。我一般分三步先确认 embedding 接口通再确认向量写入正确最后确认检索召回合理。6.1 验证 embedding 接口import os, requests resp requests.post( https://taotoken.net/api/embeddings, headers{ Authorization: fBearer {os.environ[TAOTOKEN_API_KEY]}, Content-Type: application/json }, json{model: BGE-M3, input: [测试文本向量化]} ) data resp.json() print(维度:, len(data[data][0][embedding])) print(前5维:, data[data][0][embedding][:5])返回维度应该是 1024。如果报 401检查 Key 和环境变量如果报模型不存在去模型对话页面确认模型名。6.2 验证向量写入count conn.execute(SELECT COUNT(*) FROM chunks).fetchone()[0] vec_count conn.execute(SELECT COUNT(*) FROM chunks_vec).fetchone()[0] print(fchunks: {count}, vectors: {vec_count}) assert count vec_count, 向量数量与 chunk 数量不一致两个数必须相等。不相等通常是 embedding 请求失败但没抛异常或者维度不匹配导致写入被静默拒绝。6.3 验证召回效果def search(conn, query: str, top_k: int 5): q_vec embed_fn(query) rows conn.execute( SELECT c.id, c.content, c.file_path, c.symbol, v.distance FROM chunks_vec v JOIN chunks c ON c.id v.id WHERE v.embedding MATCH ? AND k ? ORDER BY v.distance , (json.dumps(q_vec), top_k)).fetchall() return rows for r in search(conn, add 函数怎么实现的): print(f[{r[4]:.4f}] {r[2]} :: {r[3]}) print(r[1][:120]) print(---)好的召回结果应该满足距离分数在合理区间归一化向量通常 0.3 以下算相关返回的 chunk 是完整函数而不是半截代码symbol字段能对上你搜的函数名。如果召回全是无关内容先检查 chunk 是否切得太碎再检查 embedding 模型是否和写入时一致。7. 本篇常见错排查报错一sqlite3.OperationalError: no such module: vec0sqlite-vec 是扩展需要单独加载。Python 里用sqlite_vec包import sqlite3, sqlite_vec conn sqlite3.connect(memory.db) conn.enable_load_extension(True) sqlite_vec.load(conn) conn.enable_load_extension(False)报错二Dimension mismatch: expected 1024, got 768embedding 模型换了但config.toml里的vec_dimension没改。改配置后需要重建chunks_vec表因为虚拟表的维度是建表时固定的。报错三AST 解析报SyntaxError文件里有语法错误或者你用了 Python 3.12 的新语法但解析器版本低。加个 try-except 兜底到行级切分try: chunks chunk_python_by_ast(source, path) except SyntaxError: chunks chunk_by_lines(source, path, window40)报错四embedding 请求 429批量 embedding 时并发太高。把batch_size降到 8 或 16加个time.sleep(0.1)在批次之间。TaoToken 侧有速率限制具体阈值看文档。报错五检索结果重复同一个函数被多个 chunk 覆盖通常是 AST 遍历时父节点和子节点都生成了 chunk。在visit函数里找到函数节点后不要再递归它的子节点。报错六中文文本切分后 chunk 全是单字分隔符列表里把空字符串放太前面了。separators的顺序是从粗到细必须放最后否则递归切分会一路切到单字符。8. 下一步把 Memory 接进你的工作流到这里文本向量化的完整链路就通了AST 分块保证代码语义完整递归字符切分处理文档SQLite 加 sqlite-vec 做本地存储TaoToken 统一 Key 管住模型接入。你可以先把这套跑在单个代码仓库上观察召回质量再逐步扩展到知识库。接入相关的 Key 管理和文档在这里API Keys 入口https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite。想先手动验证模型效果去模型对话页面https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite发几条测试。长期做编码 Agent 的话Coding Plan 在https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite。一个实用技巧每次改完分块参数别急着全量重建先拿 10 个典型查询跑一遍召回看 top-5 里有没有正确答案。召回对了再全量跑能省不少时间。