用 CocoIndex 把 Markdown 文件夹变成可语义搜索的向量索引分块、嵌入、pgvector 存储全流程【免费下载链接】cocoindexIncremental engine for long horizon agents Star if you like it!项目地址: https://gitcode.com/GitHub_Trending/co/cocoindex导读本指南以examples/text_embedding示例为核心讲解如何用 CocoIndex 在纯异步 Python 中搭建一条「读取 Markdown → 递归分块 → 本地模型嵌入 → 写入 Postgres pgvector」的端到端语义搜索管线无需 API Key、无需手写增量逻辑编辑一个文件只会重嵌一个文件。读完本文你将掌握RecursiveSplitter分块、SentenceTransformerEmbedder嵌入、mount_table_target托管目标表与向量索引的完整用法并能用一句自然语言查询召回「不含任何相同关键词」的正确段落。背景为什么需要向量索引一堆 Markdown 文档里往往藏着答案但传统的关键词检索LIKE、全文索引只能命中字面匹配——查询 How does incremental processing work? 无法召回一篇只讨论增量计算、措辞完全不同的文章段落。语义搜索的思路是把文档切分成小块每块用嵌入模型映射为稠密向量查询时把问题嵌入成同一向量空间中的点按余弦距离返回最近的块。这正是 RAG 与语义检索系统的共同地基。CocoIndex 把这条流水线收敛为一行核心抽象target_state transformation(source_state)。数据变换用原生 Python 和自定义类型声明而增量处理、变更追踪、托管目标managed targets等重活由底层 Rust 引擎承担因此「改一个文件只重嵌一个文件而不是重嵌整个文件夹」。流水线总览Walk → Chunk → Embed → Store整条管线在 examples/text_embedding/main.py 中呈现逻辑分四步Walk用localfs.walk_dir递归扫描本地目录liveTrue开启变更监听能力Chunk用RecursiveSplitter把每个文件切成带重叠的片段——小而聚焦重叠区保证跨边界的思想不会断成两截Embed用all-MiniLM-L6-v2嵌入每个片段——模型小巧快速、完全本地运行无需 API KeyStore每个片段在 Postgres 中写入一行并为 embedding 列声明 pgvector 向量索引。其中process_file对每个文件运行一次memoTrue让它具备增量能力只要文件内容与函数代码均未变化下次运行会整体跳过该文件。示例自带三个样例文档markdown_files/1706.03762v7.mdAttention 论文、1810.04805v2.mdBERT 论文与rfc8259.mdJSON 规范开箱即可体验。核心代码逐段拆解行类型自己的 dataclass 就是表结构dataclass class DocEmbedding: id: int filename: str chunk_start: int chunk_end: int text: str embedding: Annotated[NDArray, EMBEDDER] # dimension inferred from the embedderDocEmbedding既是 Python 行对象也是 Postgres 目标表的 schema 来源。embedding字段用Annotated[NDArray, EMBEDDER]标注——EMBEDDER是声明了detect_changeTrue的上下文键向量维度由嵌入器自动推导SentenceTransformerEmbedder实现VectorSchemaProvider见 python/cocoindex/ops/sentence_transformers.py 中的__coco_vector_schema__返回VectorSchema(dtypefloat32, sizedim)TableSchema.from_class据此生成 DDL。生命周期提供数据库连接池与嵌入器coco.lifespan async def coco_lifespan( builder: coco.EnvironmentBuilder, ) - AsyncIterator[None]: async with asyncpg.create_pool(DATABASE_URL) as pool: builder.provide(PG_DB, pool) builder.provide(EMBEDDER, SentenceTransformerEmbedder(EMBED_MODEL)) yieldcoco.lifespan在应用启动/关闭时执行向环境注入两个全局资源asyncpg 连接池PG_DB以及嵌入器EMBEDDER。注意EMBEDDER coco.ContextKeySentenceTransformerEmbedder——detect_changeTrue意味着嵌入器的「身份」变化如更换模型名会被引擎识别为逻辑变更从而自动对全量数据重嵌入无需手动清缓存。分块与嵌入两个函数两层分工coco.fn async def process_chunk( chunk: Chunk, filename: pathlib.PurePath, id_gen: IdGenerator, table: postgres.TableTarget[DocEmbedding], ) - None: table.declare_row( rowDocEmbedding( idawait id_gen.next_id(chunk.text), filenamestr(filename), chunk_startchunk.start.char_offset, chunk_endchunk.end.char_offset, textchunk.text, embeddingawait coco.use_context(EMBEDDER).embed(chunk.text), ), ) coco.fn(memoTrue) async def process_file( file: FileLike, table: postgres.TableTarget[DocEmbedding], ) - None: text await file.read_text() chunks _splitter.split( text, chunk_size2000, chunk_overlap500, languagemarkdown ) id_gen IdGenerator() await coco.map(process_chunk, chunks, file.file_path.path, id_gen, table)process_file负责读文件与分块process_chunk负责逐块嵌入并声明行id_gen.next_id(chunk.text)让id由块文本内容派生——这是增量更新的关键见下文「变更如何被追踪」chunk.start/end.char_offset记录了块在原文中的字符偏移来自Chunk的位置信息python/cocoindex/resources/chunk.py便于回溯来源。RecursiveSplitter 参数说明RecursiveSplitter定义于 python/cocoindex/ops/text.py支持语法感知的递归切分markdown 走段落/句子边界代码语言可走 tree-sitter。核心参数参数作用示例取值chunk_size目标块大小字节非字符数2000chunk_overlap相邻块之间重叠的字节数防止跨边界语义断裂500min_chunk_size最小块大小默认chunk_size / 2不传则取1000language语法感知语言名或扩展名如markdown、python、.rs有 tree-sitter 支持时启用语法感知markdown返回的每个Chunk包含text、start、end三个字段其中TextPosition携带byte_offset、char_offset、line、column四类位置信息可用于精确回溯原文。组装应用挂载目标表与文件源coco.fn async def app_main(sourcedir: pathlib.Path) - None: target_table await postgres.mount_table_target( PG_DB, table_nameTABLE_NAME, table_schemaawait postgres.TableSchema.from_class(DocEmbedding, primary_key[id]), pg_schema_namePG_SCHEMA_NAME, ) target_table.declare_vector_index(columnembedding) files localfs.walk_dir(sourcedir, recursiveTrue, path_matcherPatternFilePathMatcher(included_patterns[**/*.md]), liveTrue) await coco.mount_each(process_file, files.items(), target_table) app coco.App( coco.AppConfig(nameTextEmbeddingV1), app_main, sourcedirpathlib.Path(./markdown_files), )mount_table_targetmount_table_target(db, table_name, table_schema, pg_schema_name)是「创建表目标 coco.mount_target()挂载 包装」的组合糖见 python/cocoindex/connectors/postgres/_target.py。它全权托管表结构、幂等 upsert、以及源文件消失时的删除——你永远不需要写 diff 逻辑declare_vector_index声明 pgvector 索引实际索引命名为{table_name}__vector__{name}。参数包括column索引列、metriccosine/l2/ip默认cosine、methodivfflat/hnsw默认ivfflat以及 ivfflat 的lists和 hnsw 的m/ef_constructionwalk_dirwalk_dir(path, live, recursive, path_matcher, rescan_interval)返回可异步迭代的DirWalkerpython/cocoindex/connectors/localfs/_source.py。liveTrue表示源支持实时监听但真正进入 live 模式还需在 CLI 加-Lpath_matcher用PatternFilePathMatcher(included_patterns[**/*.md])过滤只收录 Markdownmount_each把process_file应用到文件流的每一项并接到target_table上。增量更新机制为什么改一个文件只重嵌一个文件coco.fn(memoTrue)为每个文件建立缓存若文件内容与该函数的代码均未变化下一轮运行直接跳过整个文件。而每行的id由块文本派生因此重跑时内容未变的块 →id不变 → 引擎幂等 upsert结果不变内容变化的块 →id变化 → 更新对应行源文件被删 → 相关块随之消失 → 引擎自动清理孤儿行。此外EMBEDDER声明了detect_changeTrue其 memo 键由(model_name_or_path, device, trust_remote_code)构成见__coco_memo_key__所以一旦更换嵌入模型引擎会识别到缓存失效并对全量数据重新嵌入——这是「诚实的缓存失效」无需手动清库。查询同一模型嵌入余弦距离排序示例查询代码同样位于 main.pyasync def query_once(pool, embedder, query, *, top_kTOP_K) - None: query_vec await embedder.embed(query) async with pool.acquire() as conn: rows await conn.fetch( f SELECT filename, text, embedding $1 AS distance FROM {PG_SCHEMA_NAME}.{TABLE_NAME} ORDER BY distance ASC LIMIT $2 , query_vec, top_k, ) for r in rows: score 1.0 - float(r[distance]) print(f[{score:.3f}] {r[filename]}) print(f {r[text]})查询时复用同一个SentenceTransformerEmbedder同一模型保证索引与检索向量空间一致是 pgvector 的余弦距离运算符score 1 - distance换算为相似度越接近 1 越相关TOP_K 5控制返回条数交互模式支持反复输入查询空行退出也可一次传入参数直接查询。运行指南1. 启动 Postgres pgvector仓库提供了现成的 compose 配置 dev/postgres.yaml镜像pgvector/pgvector:pg17账号/密码/库名均为cocoindex端口5432docker compose -f ../../dev/postgres.yaml up -d2. 配置与安装cp .env.example .env # set POSTGRES_URL (defaults to the local docker one) pip install -e ..env.example 中可配置的环境变量变量说明默认值POSTGRES_URLPostgres 连接串postgres://cocoindex:cocoindexlocalhost/cocoindexCOCOINDEX_DBCocoIndex 本地元数据库路径./cocoindex.dbPYTORCH_ENABLE_MPS_FALLBACKMac 上 MPS 不支持的算子回退 CPU其他平台无副作用1依赖声明于 pyproject.tomlcocoindex[postgres,sentence_transformers]1.0.7、asyncpg0.29.0、pgvector0.4.1、numpy、python-dotenv1.0.1要求 Python ≥ 3.11。3. 构建索引cocoindex update main # catch-up: scan, sync, exit cocoindex update -L main # live: keep watching for file changes第一条为一次性追平扫描、同步、退出第二条为 live 模式持续监听文件变化并自动增量同步。4. 语义搜索python main.py what is self-attention?查询被同一模型嵌入后按余弦距离召回最相似的块并排序——即使它们与查询不含任何相同单词。这就是向量索引的意义。从源码看底层保障嵌入批处理与 OOM 保护SentenceTransformerEmbedder._embed以coco.fn.as_async(batchingTrue, runnercoco.GPU, max_batch_size64)声明并发单文本调用会被引擎自动合批GPU 显存不足时抛出coco.RetryWithSmallerBatch()并清空加速器缓存后减半重试见 python/cocoindex/ops/sentence_transformers.py目标表声明语义TableTarget.declare_row提取主键列值并声明目标状态upsert 语义declare_vector_index通过 attachment 机制管理索引生命周期——表或索引变更时由引擎统一协调建/删/重建python/cocoindex/connectors/postgres/_target.py文件监听的自愈walk_dir(liveTrue)默认每小时全量重扫一次并重建 OS 级 watcherrescan_interval可调、可禁用防御平台级 watcher 失效如 macOS FSEvents 静默停止。小结examples/text_embedding是 CocoIndex 语义检索场景的「最小完整闭环」一个main.py覆盖 walk → chunk → embed → store → search 全部环节增量处理、变更追踪与托管目标由 Rust 引擎透明承担。要扩展为真正的 RAG 应用只需替换嵌入模型HuggingFace 上任意 sentence-transformers 模型均可、调整分块参数或把目标换成其他受支持的存储即可。【免费下载链接】cocoindexIncremental engine for long horizon agents Star if you like it!项目地址: https://gitcode.com/GitHub_Trending/co/cocoindex创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 用 CocoIndex 把 Markdown 文件夹变成可语义搜索的向量索引：分块、嵌入、pgvector 存储全流程
用 CocoIndex 把 Markdown 文件夹变成可语义搜索的向量索引分块、嵌入、pgvector 存储全流程【免费下载链接】cocoindexIncremental engine for long horizon agents Star if you like it!项目地址: https://gitcode.com/GitHub_Trending/co/cocoindex导读本指南以examples/text_embedding示例为核心讲解如何用 CocoIndex 在纯异步 Python 中搭建一条「读取 Markdown → 递归分块 → 本地模型嵌入 → 写入 Postgres pgvector」的端到端语义搜索管线无需 API Key、无需手写增量逻辑编辑一个文件只会重嵌一个文件。读完本文你将掌握RecursiveSplitter分块、SentenceTransformerEmbedder嵌入、mount_table_target托管目标表与向量索引的完整用法并能用一句自然语言查询召回「不含任何相同关键词」的正确段落。背景为什么需要向量索引一堆 Markdown 文档里往往藏着答案但传统的关键词检索LIKE、全文索引只能命中字面匹配——查询 How does incremental processing work? 无法召回一篇只讨论增量计算、措辞完全不同的文章段落。语义搜索的思路是把文档切分成小块每块用嵌入模型映射为稠密向量查询时把问题嵌入成同一向量空间中的点按余弦距离返回最近的块。这正是 RAG 与语义检索系统的共同地基。CocoIndex 把这条流水线收敛为一行核心抽象target_state transformation(source_state)。数据变换用原生 Python 和自定义类型声明而增量处理、变更追踪、托管目标managed targets等重活由底层 Rust 引擎承担因此「改一个文件只重嵌一个文件而不是重嵌整个文件夹」。流水线总览Walk → Chunk → Embed → Store整条管线在 examples/text_embedding/main.py 中呈现逻辑分四步Walk用localfs.walk_dir递归扫描本地目录liveTrue开启变更监听能力Chunk用RecursiveSplitter把每个文件切成带重叠的片段——小而聚焦重叠区保证跨边界的思想不会断成两截Embed用all-MiniLM-L6-v2嵌入每个片段——模型小巧快速、完全本地运行无需 API KeyStore每个片段在 Postgres 中写入一行并为 embedding 列声明 pgvector 向量索引。其中process_file对每个文件运行一次memoTrue让它具备增量能力只要文件内容与函数代码均未变化下次运行会整体跳过该文件。示例自带三个样例文档markdown_files/1706.03762v7.mdAttention 论文、1810.04805v2.mdBERT 论文与rfc8259.mdJSON 规范开箱即可体验。核心代码逐段拆解行类型自己的 dataclass 就是表结构dataclass class DocEmbedding: id: int filename: str chunk_start: int chunk_end: int text: str embedding: Annotated[NDArray, EMBEDDER] # dimension inferred from the embedderDocEmbedding既是 Python 行对象也是 Postgres 目标表的 schema 来源。embedding字段用Annotated[NDArray, EMBEDDER]标注——EMBEDDER是声明了detect_changeTrue的上下文键向量维度由嵌入器自动推导SentenceTransformerEmbedder实现VectorSchemaProvider见 python/cocoindex/ops/sentence_transformers.py 中的__coco_vector_schema__返回VectorSchema(dtypefloat32, sizedim)TableSchema.from_class据此生成 DDL。生命周期提供数据库连接池与嵌入器coco.lifespan async def coco_lifespan( builder: coco.EnvironmentBuilder, ) - AsyncIterator[None]: async with asyncpg.create_pool(DATABASE_URL) as pool: builder.provide(PG_DB, pool) builder.provide(EMBEDDER, SentenceTransformerEmbedder(EMBED_MODEL)) yieldcoco.lifespan在应用启动/关闭时执行向环境注入两个全局资源asyncpg 连接池PG_DB以及嵌入器EMBEDDER。注意EMBEDDER coco.ContextKeySentenceTransformerEmbedder——detect_changeTrue意味着嵌入器的「身份」变化如更换模型名会被引擎识别为逻辑变更从而自动对全量数据重嵌入无需手动清缓存。分块与嵌入两个函数两层分工coco.fn async def process_chunk( chunk: Chunk, filename: pathlib.PurePath, id_gen: IdGenerator, table: postgres.TableTarget[DocEmbedding], ) - None: table.declare_row( rowDocEmbedding( idawait id_gen.next_id(chunk.text), filenamestr(filename), chunk_startchunk.start.char_offset, chunk_endchunk.end.char_offset, textchunk.text, embeddingawait coco.use_context(EMBEDDER).embed(chunk.text), ), ) coco.fn(memoTrue) async def process_file( file: FileLike, table: postgres.TableTarget[DocEmbedding], ) - None: text await file.read_text() chunks _splitter.split( text, chunk_size2000, chunk_overlap500, languagemarkdown ) id_gen IdGenerator() await coco.map(process_chunk, chunks, file.file_path.path, id_gen, table)process_file负责读文件与分块process_chunk负责逐块嵌入并声明行id_gen.next_id(chunk.text)让id由块文本内容派生——这是增量更新的关键见下文「变更如何被追踪」chunk.start/end.char_offset记录了块在原文中的字符偏移来自Chunk的位置信息python/cocoindex/resources/chunk.py便于回溯来源。RecursiveSplitter 参数说明RecursiveSplitter定义于 python/cocoindex/ops/text.py支持语法感知的递归切分markdown 走段落/句子边界代码语言可走 tree-sitter。核心参数参数作用示例取值chunk_size目标块大小字节非字符数2000chunk_overlap相邻块之间重叠的字节数防止跨边界语义断裂500min_chunk_size最小块大小默认chunk_size / 2不传则取1000language语法感知语言名或扩展名如markdown、python、.rs有 tree-sitter 支持时启用语法感知markdown返回的每个Chunk包含text、start、end三个字段其中TextPosition携带byte_offset、char_offset、line、column四类位置信息可用于精确回溯原文。组装应用挂载目标表与文件源coco.fn async def app_main(sourcedir: pathlib.Path) - None: target_table await postgres.mount_table_target( PG_DB, table_nameTABLE_NAME, table_schemaawait postgres.TableSchema.from_class(DocEmbedding, primary_key[id]), pg_schema_namePG_SCHEMA_NAME, ) target_table.declare_vector_index(columnembedding) files localfs.walk_dir(sourcedir, recursiveTrue, path_matcherPatternFilePathMatcher(included_patterns[**/*.md]), liveTrue) await coco.mount_each(process_file, files.items(), target_table) app coco.App( coco.AppConfig(nameTextEmbeddingV1), app_main, sourcedirpathlib.Path(./markdown_files), )mount_table_targetmount_table_target(db, table_name, table_schema, pg_schema_name)是「创建表目标 coco.mount_target()挂载 包装」的组合糖见 python/cocoindex/connectors/postgres/_target.py。它全权托管表结构、幂等 upsert、以及源文件消失时的删除——你永远不需要写 diff 逻辑declare_vector_index声明 pgvector 索引实际索引命名为{table_name}__vector__{name}。参数包括column索引列、metriccosine/l2/ip默认cosine、methodivfflat/hnsw默认ivfflat以及 ivfflat 的lists和 hnsw 的m/ef_constructionwalk_dirwalk_dir(path, live, recursive, path_matcher, rescan_interval)返回可异步迭代的DirWalkerpython/cocoindex/connectors/localfs/_source.py。liveTrue表示源支持实时监听但真正进入 live 模式还需在 CLI 加-Lpath_matcher用PatternFilePathMatcher(included_patterns[**/*.md])过滤只收录 Markdownmount_each把process_file应用到文件流的每一项并接到target_table上。增量更新机制为什么改一个文件只重嵌一个文件coco.fn(memoTrue)为每个文件建立缓存若文件内容与该函数的代码均未变化下一轮运行直接跳过整个文件。而每行的id由块文本派生因此重跑时内容未变的块 →id不变 → 引擎幂等 upsert结果不变内容变化的块 →id变化 → 更新对应行源文件被删 → 相关块随之消失 → 引擎自动清理孤儿行。此外EMBEDDER声明了detect_changeTrue其 memo 键由(model_name_or_path, device, trust_remote_code)构成见__coco_memo_key__所以一旦更换嵌入模型引擎会识别到缓存失效并对全量数据重新嵌入——这是「诚实的缓存失效」无需手动清库。查询同一模型嵌入余弦距离排序示例查询代码同样位于 main.pyasync def query_once(pool, embedder, query, *, top_kTOP_K) - None: query_vec await embedder.embed(query) async with pool.acquire() as conn: rows await conn.fetch( f SELECT filename, text, embedding $1 AS distance FROM {PG_SCHEMA_NAME}.{TABLE_NAME} ORDER BY distance ASC LIMIT $2 , query_vec, top_k, ) for r in rows: score 1.0 - float(r[distance]) print(f[{score:.3f}] {r[filename]}) print(f {r[text]})查询时复用同一个SentenceTransformerEmbedder同一模型保证索引与检索向量空间一致是 pgvector 的余弦距离运算符score 1 - distance换算为相似度越接近 1 越相关TOP_K 5控制返回条数交互模式支持反复输入查询空行退出也可一次传入参数直接查询。运行指南1. 启动 Postgres pgvector仓库提供了现成的 compose 配置 dev/postgres.yaml镜像pgvector/pgvector:pg17账号/密码/库名均为cocoindex端口5432docker compose -f ../../dev/postgres.yaml up -d2. 配置与安装cp .env.example .env # set POSTGRES_URL (defaults to the local docker one) pip install -e ..env.example 中可配置的环境变量变量说明默认值POSTGRES_URLPostgres 连接串postgres://cocoindex:cocoindexlocalhost/cocoindexCOCOINDEX_DBCocoIndex 本地元数据库路径./cocoindex.dbPYTORCH_ENABLE_MPS_FALLBACKMac 上 MPS 不支持的算子回退 CPU其他平台无副作用1依赖声明于 pyproject.tomlcocoindex[postgres,sentence_transformers]1.0.7、asyncpg0.29.0、pgvector0.4.1、numpy、python-dotenv1.0.1要求 Python ≥ 3.11。3. 构建索引cocoindex update main # catch-up: scan, sync, exit cocoindex update -L main # live: keep watching for file changes第一条为一次性追平扫描、同步、退出第二条为 live 模式持续监听文件变化并自动增量同步。4. 语义搜索python main.py what is self-attention?查询被同一模型嵌入后按余弦距离召回最相似的块并排序——即使它们与查询不含任何相同单词。这就是向量索引的意义。从源码看底层保障嵌入批处理与 OOM 保护SentenceTransformerEmbedder._embed以coco.fn.as_async(batchingTrue, runnercoco.GPU, max_batch_size64)声明并发单文本调用会被引擎自动合批GPU 显存不足时抛出coco.RetryWithSmallerBatch()并清空加速器缓存后减半重试见 python/cocoindex/ops/sentence_transformers.py目标表声明语义TableTarget.declare_row提取主键列值并声明目标状态upsert 语义declare_vector_index通过 attachment 机制管理索引生命周期——表或索引变更时由引擎统一协调建/删/重建python/cocoindex/connectors/postgres/_target.py文件监听的自愈walk_dir(liveTrue)默认每小时全量重扫一次并重建 OS 级 watcherrescan_interval可调、可禁用防御平台级 watcher 失效如 macOS FSEvents 静默停止。小结examples/text_embedding是 CocoIndex 语义检索场景的「最小完整闭环」一个main.py覆盖 walk → chunk → embed → store → search 全部环节增量处理、变更追踪与托管目标由 Rust 引擎透明承担。要扩展为真正的 RAG 应用只需替换嵌入模型HuggingFace 上任意 sentence-transformers 模型均可、调整分块参数或把目标换成其他受支持的存储即可。【免费下载链接】cocoindexIncremental engine for long horizon agents Star if you like it!项目地址: https://gitcode.com/GitHub_Trending/co/cocoindex创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考