10分钟从0到1：用CocoIndex把一堆Markdown变成可增量更新的向量索引

10分钟从0到1：用CocoIndex把一堆Markdown变成可增量更新的向量索引

【免费下载链接】cocoindexIncremental engine for long horizon agents 🌟 Star if you like it!项目地址: https://gitcode.com/GitHub_Trending/co/cocoindex

手头有一堆 Markdown 文档，想让 AI 按语义检索，而不是只能做关键词匹配？CocoIndex 是一个增量索引引擎：你用原生 Python 声明"目标状态 = 源状态的转换"，底下的 Rust 引擎负责跟踪变化、只对变化的部分重算。读完本文你将拥有：一个可运行的向量索引 + Postgres 里一张可以直接查询的向量表。

先跑起来：最小可运行环境 🔧

克隆仓库，进入示例目录

git clone https://gitcode.com/GitHub_Trending/co/cocoindex cd cocoindex/examples/text_embedding

本文所有代码都对应这个示例：扫目录里的 Markdown → 分块 → 嵌入 → 存入 Postgres，是整条链路里最短的一条。

安装依赖

pip install -e .

pyproject.toml 已声明cocoindex[postgres,sentence_transformers]、asyncpg、pgvector 等全部依赖，一条命令装完。

起一个带 pgvector 的 Postgres

docker compose -f ../../dev/postgres.yaml up -d

仓库里现成的配置：pgvector/pgvector:pg17镜像，账号密码都是 cocoindex，端口 5432。不展开 Docker 原理，先跑起来再说。

示例数据已就位

markdown_files/目录下有三个 Markdown 文件（两篇论文笔记 + RFC 8259），不需要自己准备。

配置连接串

export POSTGRES_URL="postgres://cocoindex:cocoindex@localhost:5432/cocoindex"

代码里就是这个默认值，本地 Docker 部署不加也能跑；数据库在别的机器上时改这里。

写一份索引定义：核心代码按数据流向拆解 🧩

完整实现在 examples/text_embedding/main.py，先看骨架：

import os from dataclasses import dataclass from typing import Annotated import asyncpg import cocoindex as coco from cocoindex.connectors import localfs, postgres from cocoindex.ops.text import RecursiveSplitter from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder from cocoindex.resources.file import FileLike, PatternFilePathMatcher from cocoindex.resources.id import IdGenerator from numpy.typing import NDArray DB_URL = os.getenv("POSTGRES_URL", "postgres://cocoindex:cocoindex@localhost/cocoindex") PG_DB = coco.ContextKeyasyncpg.Pool EMBEDDER = coco.ContextKeySentenceTransformerEmbedder _splitter = RecursiveSplitter() @dataclass class DocEmbedding: # 表里一行 = 一个 chunk id: int filename: str text: str embedding: Annotated[NDArray, EMBEDDER] @coco.fn(memo=True) async def process_file(file: FileLike, table): text = await file.read_text() chunks = _splitter.split(text, chunk_size=2000, chunk_overlap=500, language="markdown") id_gen = IdGenerator() await coco.map(process_chunk, chunks, file.file_path.path, id_gen, table)

下面不按行号讲，按数据流走一遍：数据从哪来 → 怎么切 → 怎么变成向量 → 往哪存。

数据从哪来：walk_dir

files = localfs.walk_dir( sourcedir, recursive=True, path_matcher=PatternFilePathMatcher(included_patterns=["**/*.md"]), live=True, )

localfs.walk_dir递归扫描目录，只挑**/*.md。live=True让它具备监听能力——配合cocoindex update -L main就能常驻盯着目录，文件一变就重算，而不是每次手动跑批。

怎么切：RecursiveSplitter

分块参数就两个词：chunk_size=2000（每块约 2000 字符）、chunk_overlap=500（相邻块重叠 500 字符）。重叠不是摆设：如果一个概念正好横跨块边界，重叠区能保证它至少在一个块里是完整的。language="markdown"让分块器优先按段落、标题这类 Markdown 结构切，而不是硬截断句子。

怎么变成向量：embed

每个 chunk 的嵌入在一个@coco.fn里完成，完整版里是这样：

@coco.fn async def process_chunk(chunk, filename, id_gen, table): table.declare_row(row=DocEmbedding( id=await id_gen.next_id(chunk.text), # id 由 chunk 文本推导 filename=str(filename), text=chunk.text, embedding=await coco.use_context(EMBEDDER).embed(chunk.text), ))

注意id_gen.next_id(chunk.text)：id 从内容推导，重跑时同一 chunk 落到同一行，天然的 upsert 语义，删除逻辑一行不用写。EMBEDDER是ContextKey（上下文共享对象），整个 pipeline 和查询端复用同一个嵌入器——索引用哪个模型，查询就必须用哪个，否则向量空间对不上。

往哪存：mount_table_target

@coco.fn async def app_main(sourcedir: pathlib.Path): table = await postgres.mount_table_target( PG_DB, table_name="doc_embeddings", table_schema=await postgres.TableSchema.from_class( DocEmbedding, primary_key=["id"]), ) table.declare_vector_index(column="embedding") files = localfs.walk_dir( sourcedir, recursive=True, path_matcher=PatternFilePathMatcher(included_patterns=["**/*.md"]), live=True, ) await coco.mount_each(process_file, files.items(), table) app = coco.App( coco.AppConfig(name="TextEmbedding"), app_main, sourcedir=pathlib.Path("./markdown_files"), )

mount_table_target是托管目标（managed target）：表结构从你的 dataclass 推导、pgvector 向量索引由declare_vector_index声明、行级 upsert 和孤儿行清理都由引擎代管。连接池和嵌入器通过@coco.lifespan注入，完整版里十几行，思路是"启动时建好，随 pipeline 生命周期销毁"。

增量是这里最值钱的部分：process_file标了memo=True，文件内容和处理代码都没变就整体跳过。改一个文件，就只重嵌入一个文件。

执行、验证与你会看到的输出 📦

构建索引

cocoindex update main

跑完后你会看到同步统计输出（各版本措辞略有差异，格式以你本机为准）：

documents: 3 added, 0 removed, 0 updated

看到3 added就对了——三个 Markdown 文件全部入表。

语义查询验证

示例自带查询入口，用同一个模型把你的问题嵌成向量再按余弦距离取 top5：

python main.py "what is self-attention?"

输出形如（分数因数据略有差异）：

[0.631] 1706.03762v7.md …（命中的 chunk 原文） ---

第一行是"注意力"论文的 chunk，哪怕你的提问和原文一个词都不重合——这就是向量索引存在的意义。

SQL 快速验证

不想走 Python，直接看表也行：

docker compose -f ../../dev/postgres.yaml exec postgres psql -U cocoindex -d cocoindex

SELECT filename, left(text, 40) FROM coco_examples.doc_embeddings LIMIT 5;

有行返回，说明向量已经落库。想验证增量：往markdown_files/扔一个新的 .md 再跑一次cocoindex update main，统计里只会出现新文件的那一行。

首次执行报错速查：避坑与调参 ⚠️

现象：update 报 connection refused→ 原因：Postgres 容器没起，或本机 5432 端口已被别的实例占用 → 修复：先docker compose -f ../../dev/postgres.yaml up -d；端口冲突就换一行起容器：docker run -d -p 5433:5432 -e POSTGRES_PASSWORD=cocoindex -e POSTGRES_USER=cocoindex -e POSTGRES_DB=cocoindex pgvector/pgvector:pg17，然后把POSTGRES_URL指到 5433。

现象：首次运行卡住几分钟没输出→ 原因：在从 Hugging Face 下载 all-MiniLM-L6-v2 模型，不是挂了 → 修复：耐心等；网络不畅时先export HF_ENDPOINT=https://hf-mirror.com再跑（镜像地址参考官方文档确认）。

现象：换了嵌入模型，检索结果却没变→ 原因：memo 缓存按内容判重，模型变化没被感知 → 修复：完整版里EMBEDDER声明时带了detect_change=True，换模型会自动触发全量重嵌入，照抄即可，不用手动清缓存。

调参方面，chunk_size和chunk_overlap是最常用的两个旋钮：块太小，行数和嵌入调用翻倍，上下文还被切碎；块太大，检索命中范围变宽、定位变糊。默认 2000/500 对论文类长文偏稳。

下一步：往哪个方向延伸 🧭

• 常驻监听：cocoindex update -L main让 pipeline 一直活着，保存文件即重嵌入，用法见 examples/text_embedding/README.md。
• 换更宽的文档类型：图片、PDF 混在一起也能索引，这个示例的数据源里就有这种图：

实现参考 examples/multi_format_indexing/。

• 换目标存储、看更多场景：LanceDB、Qdrant、Kafka、知识图谱等 20+ 示例都在 examples/，引擎原理从 docs/src/content/docs/getting_started/overview.mdx 读起。

把 chunk_size 改小一点，再跑一遍对比下检索效果。

【免费下载链接】cocoindexIncremental engine for long horizon agents 🌟 Star if you like it!项目地址: https://gitcode.com/GitHub_Trending/co/cocoindex