CocoIndex 上手指南：10分钟搭建增量向量索引-育师

CocoIndex 上手指南：10分钟搭建增量向量索引

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

本地文档散落一堆，问问题时关键词搜索帮不上忙。CocoIndex 就是为这事而生的增量索引引擎：你用 Python 写一次转换逻辑，它替你维护一份向量索引——新增文件只处理新文件，删掉文件只删对应向量。跟着做完，10 分钟内你会拥有一个能自然语言提问的本地文档向量索引。

先看成品：一个能自然语言提问的本地文档库

动手前先交代目标。做完本文，你手上会有两样东西：

Postgres 里一张doc_embeddings表：每行存一个文本片段加 384 维嵌入向量，并建有 pgvector 向量索引；
一条命令行：输入"增量处理是怎么工作的？"，它返回语义上最相关的段落——即使和文档没有一个共同词。

"磁盘上的文档"和"Postgres 里的表"之间只隔一层转换：读取 → 分块 → 嵌入 → 入库。CocoIndex 的心智模型是一个等式target_state = transformation(source_state)：你只声明"输出该长什么样"，"哪里变了、要重算多少"由引擎负责。下面这张图就是这套分工：左边是源数据，右边是你想要的索引，中间的引擎按增量方式执行你的转换逻辑（这里是向量嵌入）。

搭好环境：克隆仓库、拉起 Postgres 🛠

先克隆仓库，后面要复用它的三样东西

仓库自带一份预置 pgvector 的 Postgres 配置、3 篇示例 Markdown 文档，还有一个可对照的完整参考实现。这三样都用得上，先拉下来：

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

拉起数据库，装好 Python 依赖

向量索引要存在 Postgres 里（靠 pgvector 扩展）。仓库的dev/postgres.yaml是现成配置（账号cocoindex、数据库cocoindex、端口 5432），一条命令启动：

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

Python 侧装 CocoIndex 及 Postgres、本地嵌入两个插件，外加查询用的驱动依赖：

pip install -U "cocoindex[postgres,sentence_transformers]" asyncpg pgvector numpy

嵌入模型用all-MiniLM-L6-v2，它在本机跑、不需要 API key，首次运行时自动下载，不用额外配置。

写流水线：把 Markdown 文件夹变成向量

把 3 篇示例文档拷进工作目录，再建一个空的main.py：

mkdir -p cocoindex-quickstart && cd cocoindex-quickstart cp -r ../examples/text_embedding/markdown_files . touch main.py

先声明"一行输出长什么样"

第一段代码干三件事：导入依赖；定义两个"共享资源"——Postgres 连接池PG_DB和嵌入模型EMBEDDER（detect_change=True表示换模型时缓存自动失效、全量重算）；用 dataclass 声明目标表的行结构。注意embedding字段用Annotated绑到EMBEDDER，384 维这个 schema 由模型自动推断，不用手写维度。

import asyncio, os, pathlib, sys from dataclasses import dataclass from typing import Annotated, AsyncIterator 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.chunk import Chunk from cocoindex.resources.file import FileLike, PatternFilePathMatcher from cocoindex.resources.id import IdGenerator from numpy.typing import NDArray DATABASE_URL = os.getenv( "POSTGRES_URL", "postgres://cocoindex:cocoindex@localhost/cocoindex" ) EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2" _splitter = RecursiveSplitter() # 两个共享资源：Postgres 连接池和本地嵌入模型 PG_DB = coco.ContextKeyasyncpg.Pool EMBEDDER = coco.ContextKeySentenceTransformerEmbedder @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)) yield @dataclass class DocEmbedding: id: int filename: str chunk_start: int chunk_end: int text: str embedding: Annotated[NDArray, EMBEDDER]

再写"分块 + 嵌入 + 入库"，组装成 App

第二段是主体。process_file负责拆一个文件：RecursiveSplitter切出 2000 字一块、重叠 500 字，保证跨边界的句子不断成两半；coco.map把分块并行派发给process_chunk，后者给每块算嵌入，再用declare_row声明一行入库。id由IdGenerator从分块文本派生——内容相同就落到同一行，这是"重跑不重复、删文件自动删行"的基础。app_main里mount_table_target接管这张表（schema、向量索引、增量 upsert 都由它管理），localfs.walk_dir扫描markdown_files/，每个文件挂一个处理组件，最后coco.App把整个绑定到一个具体目录。process_file上的@coco.fn(memo=True)是增量关键：下次运行文件没变就整体跳过。

@coco.fn async def process_chunk( chunk: Chunk, filename: pathlib.PurePath, id_gen: IdGenerator, table: postgres.TableTarget[DocEmbedding], ) -> None: # id 由分块文本派生：内容相同永远落同一行 await table.declare_row(row=DocEmbedding( id=await id_gen.next_id(chunk.text), filename=str(filename), chunk_start=chunk.start.char_offset, chunk_end=chunk.end.char_offset, text=chunk.text, embedding=await coco.use_context(EMBEDDER).embed(chunk.text), )) @coco.fn(memo=True) async def process_file( file: FileLike, table: postgres.TableTarget[DocEmbedding], ) -> None: text = await file.read_text() chunks = _splitter.split( text, chunk_size=2000, chunk_overlap=500, language="markdown" ) await coco.map(process_chunk, chunks, file.file_path.path, IdGenerator(), table) @coco.fn async def app_main(sourcedir: pathlib.Path) -> None: target_table = await postgres.mount_table_target( PG_DB, table_name="doc_embeddings", table_schema=await postgres.TableSchema.from_class( DocEmbedding, primary_key=["id"] ), pg_schema_name="coco_examples", ) target_table.declare_vector_index(column="embedding") files = localfs.walk_dir( sourcedir, recursive=True, path_matcher=PatternFilePathMatcher(included_patterns=["**/*.md"]), ) await coco.mount_each(process_file, files.items(), target_table) app = coco.App( coco.AppConfig(name="TextEmbedding"), app_main, sourcedir=pathlib.Path("./markdown_files"), )

最后补一个自然语言查询入口：它用同一个模型给你的问题算向量，再用 pgvector 的余弦距离取最相似的 5 行。索引和查询用同一个模型，结果才可比。

from pgvector.asyncpg import register_vector async def query(query_text: str, top_k: int = 5) -> None: async with asyncpg.create_pool(DATABASE_URL, init=register_vector) as pool: vec = await SentenceTransformerEmbedder(EMBED_MODEL).embed(query_text) async with pool.acquire() as conn: rows = await conn.fetch( """ SELECT filename, text, embedding <=> $1 AS distance FROM "coco_examples"."doc_embeddings" ORDER BY distance ASC LIMIT $2 """, vec, top_k, ) for r in rows: print(f"[{1.0 - float(r['distance']):.3f}] {r['filename']}") print(f" {r['text']}\n---") if __name__ == "__main__": query_text = " ".join(sys.argv[1:]) or "增量处理是怎么工作的？" asyncio.run(query(query_text))

跑起来：一条命令建索引，一句话查询

`cocoindex update`是唯一执行入口

cocoindex update main.py会告诉引擎对齐源状态和目标状态，把缺口补上：建表、处理 3 篇 Markdown 文档、写入向量。首次运行因下载模型稍慢，第二次运行 3 个文件全部命中缓存，一两秒就结束。

cocoindex update main.py

输入一个问题，验证"语义命中"

现在输入一个和文档不共享关键词的问题：

python main.py "增量处理是怎么工作的？"

你会看到一组按相似度排序的段落，每行前面是 0 到 1 的分数。要点在于：你不必复述文档原话——嵌入向量把措辞变成了空间中的位置，"意思相近"就是"距离很近"。

验证增量：只处理新文件 🔍

这是 CocoIndex 真正想让你看见的部分。依次做三个操作，观察"全量重算"和"增量"的差别：

加文件：往markdown_files/丢一篇新 Markdown，再跑一次cocoindex update main.py。只有新文件的分块被处理，已有 3 篇原样跳过。

删文件：删掉其中一篇再运行。Postgres 里对应的向量自动清掉，你不用写一行删除逻辑。

监听模式：cocoindex update -L main.py会常驻进程，边改文件边处理新变化，文档和索引始终保持一致。

文本检索只是第一步。图里这只长颈鹿来自仓库的图片搜索示例：用 CLIP 模型直接对图片做嵌入，拿一张照片去找视觉上相似的——你刚跑通的流水线结构原样复用，只是换了一个嵌入模型。

再动手：换数据源，或升级到图片搜索

两个马上可以做的方向：

换数据源：把sourcedir指向你真实的文档目录，或者把localfs.walk_dir换成 Google Drive、Amazon S3 连接器，app_main里只改一行。
弄懂底层模型：读一遍核心概念，理解@coco.fn(memo=True)背后的状态驱动编程——你会明白"只处理新文件"为什么成立，而不只是它成立。 </输出文章

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

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

CocoIndex 上手指南：10分钟搭建增量向量索引