CocoIndex 实战：将论文 PDF 文件夹转化为结构化元数据与向量索引（Paper Metadata 示例详解）

CocoIndex 实战：将论文 PDF 文件夹转化为结构化元数据与向量索引（Paper Metadata 示例详解）

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

本篇技术指南围绕 CocoIndex 仓库中的 paper_metadata 示例展开：它读取一个本地 PDF 文件夹中每篇论文的第一页，交给 LLM 以严格 Schema 抽取title、authors、abstract三类结构化字段，再对标题与摘要分块做向量化嵌入，最终写入 Postgres（pgvector）的三张表中，并支持按语义而非关键词检索。读完本文，你将掌握@coco.fn(memo=True)增量管线、mount_table_target三表联动、RecursiveSplitter分块与SentenceTransformerEmbedder嵌入的完整用法，并能在自己的论文/报告/文档集上直接复刻这套"PDF → 结构化 + 可搜索"流水线。

从一页 PDF 到三张 Postgres 表：流程总览

论文的第一页几乎包含了你想查询的全部信息——标题、作者、摘要，但它们被锁在 PDF 排版文本里。该示例的思路非常克制：只读第一页，把该页文本交给 LLM 按严格 Schema 抽取，得到干净的强类型 JSON；随后把同样的元数据向量化，从而支持"按含义搜索"而非"按词精确匹配"。

从 main.py 的代码结构看，整条流水线分为四步：

1. 读取：从本地目录（live 模式支持文件变更监听）读入 PDF 文件；
2. 抽取：用 pypdf 从每份 PDF 中切出第一页并提取文本，交给 LLM 返回title、authors、abstract的结构化 JSON；
3. 嵌入：对标题和摘要分块分别做 embedding；
4. 落库：把元数据行、作者索引行、嵌入行声明为三个 Postgres target state，由引擎自动同步。

整条管线的控制流是原生 async Python，你只声明转换逻辑：target_state = transformation(source_state)。增量处理、变更追踪、target 管理这些重活全部由底层的 Rust 引擎承担，因此只有发生变更的 PDF才会被重新抽取和重新嵌入。一份 PDF 展开成三张表（论文元数据、作者-论文索引、嵌入向量），CocoIndex 负责让三者始终保持一致。

环境准备

示例的运行依赖如下：

• 带 pgvector 扩展的 Postgres。仓库自带 compose 文件 dev/postgres.yaml，使用pgvector/pgvector:pg17镜像，默认账号/密码/库均为cocoindex：

  docker compose -f dev/postgres.yaml up -d export POSTGRES_URL="postgres://cocoindex:cocoindex@localhost/cocoindex"

• OpenAI API Key，用于抽取步骤：

  export OPENAI_API_KEY="your_key"

• 安装依赖（CocoIndex 及其 postgres、sentence_transformers 附加组件，外加 asyncpg、pgvector 等）：

  pip install -U "cocoindex[postgres,sentence_transformers]" asyncpg pgvector numpy pypdf openai pydantic python-dotenv

  示例的 pyproject.toml 给出了完整依赖清单：cocoindex[postgres,sentence_transformers]>=1.0.7、asyncpg>=0.29.0、pgvector>=0.4.1、pypdf>=5.7.0、openai>=1.0.0、pydantic>=2.12.5、python-dotenv>=1.0.1，Python 版本要求>=3.11。也可以直接pip install -e .安装整个示例。

• 若干 PDF 论文。示例自带 papers/ 目录，内含 4 篇知名论文（1706.03762v7.pdf（Attention Is All You Need）、1810.04805v2.pdf、2502.06786v3.pdf、2502.20346v1.pdf），也可以放入你自己的 PDF。

另外，示例提供 .env.example，复制为.env后填入POSTGRES_URL与OPENAI_API_KEY，运行时会通过python-dotenv自动加载（load_dotenv()）。

定义目标 Schema：用 Pydantic 锁死抽取结果

在写管线之前，先明确元数据的形状。示例用 models.py 中的两个 Pydantic 模型约束 LLM 的输出：

class AuthorModel(BaseModel): name: str email: str | None = None affiliation: str | None = None class PaperMetadataModel(BaseModel): title: str authors: list[AuthorModel] = Field(default_factory=list) abstract: str

AuthorModel中email、affiliation可缺省（None），PaperMetadataModel.authors默认空列表。这两个模型的关键作用在抽取函数中体现：PaperMetadataModel.model_validate_json(content)会对 LLM 返回的 JSON 做严格校验，任何不符合 Schema 的输出都会直接抛错——坏数据在进入数据库之前就被拦下，绝不会污染 Postgres。

声明数据行与共享资源：dataclass + ContextKey + lifespan

每个输出表对应一个 dataclass，定义在 main.py 中：

EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2" PG_DB = coco.ContextKeyasyncpg.Pool EMBEDDER = coco.ContextKeySentenceTransformerEmbedder @dataclass class PaperMetadataRow: filename: str title: str authors: list[dict[str, str | None]] abstract: str num_pages: int @dataclass class AuthorPaperRow: author_name: str filename: str @dataclass class MetadataEmbeddingRow: id: uuid.UUID filename: str location: str text: str embedding: Annotated[NDArray, EMBEDDER] @coco.lifespan async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]: async with asyncpg.create_pool(os.environ["POSTGRES_URL"]) as pool: builder.provide(PG_DB, pool) builder.provide(EMBEDDER, SentenceTransformerEmbedder(EMBED_MODEL)) yield

三种行的职责各有侧重：

• PaperMetadataRow：每篇论文一行，主键filename；
• AuthorPaperRow：每个（作者，论文）一对一行，构成可 join 的作者索引；
• MetadataEmbeddingRow：每段被嵌入的文本一行，location字段区分这段文本来自title还是abstract。

@coco.lifespan在启动时一次性创建共享资源——Postgres 连接池和嵌入模型，通过builder.provide注册到ContextKey上，之后在任何@coco.fn里都能用coco.use_context(EMBEDDER)取用。

最值得注意的一行是embedding: Annotated[NDArray, EMBEDDER]：它把向量列和嵌入器绑定，向量维度由嵌入器自动推断（all-MiniLM-L6-v2输出 384 维）。如果日后更换嵌入模型，由于EMBEDDER声明了detect_change=True，CocoIndex 会自动感知并触发全量重新嵌入，无需手动清缓存。

抽取三步曲：切首页、提文本、LLM 抽字段

三个@coco.fn完成抽取，见 main.py：

@coco.fn def extract_basic_info(content: bytes) -> PaperBasicInfo: reader = PdfReader(io.BytesIO(content)) output = io.BytesIO() writer = PdfWriter() writer.add_page(reader.pages[0]) writer.write(output) return PaperBasicInfo(num_pages=len(reader.pages), first_page=output.getvalue()) @coco.fn def pdf_to_markdown(content: bytes) -> str: reader = PdfReader(io.BytesIO(content)) return (reader.pages[0].extract_text() if reader.pages else "") or "" @coco.fn def extract_metadata(markdown: str) -> PaperMetadataModel: response = openai_client().chat.completions.create( model=LLM_MODEL, messages=[ {"role": "system", "content": ( "You extract metadata from academic paper first pages. " "Return only JSON with keys: title, authors, abstract. " "authors is a list of {name, email, affiliation}. " "Use null for missing fields." )}, {"role": "user", "content": markdown[:4000]}, ], response_format={"type": "json_object"}, temperature=0, ) content = response.choices[0].message.content if not content: raise RuntimeError("LLM returned empty content.") return PaperMetadataModel.model_validate_json(content)

• extract_basic_info：用PdfReader打开 PDF 字节流，PdfWriter.add_page(reader.pages[0])只保留第一页，同时返回num_pages总页数与first_page单页字节；
• pdf_to_markdown：对第一页执行extract_text()提取文本，空页兜底为空字符串；
• extract_metadata：把文本交给gpt-4o（LLM_MODEL常量），response_format={"type": "json_object"}配合temperature=0保证输出为确定性 JSON，再经model_validate_json解析成强类型PaperMetadataModel；若 LLM 返回空内容则直接抛RuntimeError。

注意两个成本控制细节：只读第一页，且 prompt 内容截断为markdown[:4000]字符——这对标题块加摘要几乎总是足够，同时让 token 成本与论文长度无关。

单文件处理：process_file与三表扇出

process_file对每份 PDF 执行一次，负责串联全部步骤并声明行，见 main.py：

@coco.fn(memo=True) async def process_file( file: FileLike, metadata_table: postgres.TableTarget[PaperMetadataRow], author_table: postgres.TableTarget[AuthorPaperRow], embedding_table: postgres.TableTarget[MetadataEmbeddingRow], ) -> None: content = await file.read() basic_info = extract_basic_info(content) first_page_md = pdf_to_markdown(basic_info.first_page) metadata = extract_metadata(first_page_md) metadata_table.declare_row( row=PaperMetadataRow( filename=str(file.file_path.path), title=metadata.title, authors=[a.model_dump() for a in metadata.authors], abstract=metadata.abstract, num_pages=basic_info.num_pages, ), ) for author in metadata.authors: if author.name: author_table.declare_row( row=AuthorPaperRow( author_name=author.name, filename=str(file.file_path.path), ), ) title_embedding = await coco.use_context(EMBEDDER).embed(metadata.title) embedding_table.declare_row( row=MetadataEmbeddingRow( id=uuid.uuid4(), filename=str(file.file_path.path), location="title", text=metadata.title, embedding=title_embedding, ), ) abstract_chunks = _abstract_splitter.split( metadata.abstract, chunk_size=500, min_chunk_size=200, chunk_overlap=150, language="abstract", ) for chunk in abstract_chunks: embedding_table.declare_row( row=MetadataEmbeddingRow( id=uuid.uuid4(), filename=str(file.file_path.path), location="abstract", text=chunk.text, embedding=await coco.use_context(EMBEDDER).embed(chunk.text), ), )

它声明三种行：

1. 一行元数据：PaperMetadataRow，authors用model_dump()转成 dict 列表后由引擎序列化（Postgres 侧对应 jsonb 列，见下文类型映射说明）；
2. 每作者一行索引：仅当author.name非空时声明AuthorPaperRow；
3. 标题一行嵌入 + 摘要每 chunk 一行嵌入：标题整体作为单行嵌入；摘要用_abstract_splitter按句切分，location标记命中来源。

增量与 memo 机制

@coco.fn(memo=True)是整个示例增量能力的核心：当某 PDF 的字节内容与该函数代码均未变化时，整个文件在下一次运行中会被直接跳过——LLM 调用和嵌入计算都不会为已处理的 PDF 重复付费。这正是"长程 Agent / 长任务"场景下 CocoIndex 的价值所在：不需要你写任何 diff 逻辑。

摘要分块：RecursiveSplitter 与自定义语言

_abstract_splitter是 python/cocoindex/ops/text.py 中RecursiveSplitter的实例，通过CustomLanguageConfig为摘要文本定义了分层分隔符：

_abstract_splitter = RecursiveSplitter( custom_languages=[ CustomLanguageConfig( language_name="abstract", separators_regex=[r"[.?!]+\s+", r"[:;]\s+", r",\s+", r"\s+"], ) ] )

分隔符按优先级排列：句子结束符（.?!）最优先，其次是冒号分号、逗号、空白。split()的调用参数chunk_size=500, min_chunk_size=200, chunk_overlap=150表示：目标块约 500 字符、低于 200 字符的碎片会与相邻块合并、块间保留 150 字符重叠以缓解切句导致的信息割裂。从 text.py 的源码可以看到，RecursiveSplitter是可复用实例，一次构造可对多篇文本反复split，且支持按语言配置做语法感知切分。

主函数：装配数据源与三个表目标

app_main负责把 source 接到 targets，见 main.py：

@coco.fn async def app_main(sourcedir: pathlib.Path) -> None: metadata_table = await postgres.mount_table_target( PG_DB, table_name=TABLE_METADATA, table_schema=await postgres.TableSchema.from_class( PaperMetadataRow, primary_key=["filename"], ), pg_schema_name=PG_SCHEMA_NAME, # "coco_examples_v1" ) author_table = await postgres.mount_table_target( PG_DB, table_name=TABLE_AUTHOR_PAPERS, table_schema=await postgres.TableSchema.from_class( AuthorPaperRow, primary_key=["author_name", "filename"], ), pg_schema_name=PG_SCHEMA_NAME, ) embedding_table = await postgres.mount_table_target( PG_DB, table_name=TABLE_EMBEDDINGS, table_schema=await postgres.TableSchema.from_class( MetadataEmbeddingRow, primary_key=["id"], ), pg_schema_name=PG_SCHEMA_NAME, ) files = localfs.walk_dir( sourcedir, recursive=True, path_matcher=PatternFilePathMatcher(included_patterns=["**/*.pdf"]), live=True, # watch for changes; pass -L to `cocoindex update` to run live ) await coco.mount_each( process_file, files.items(), metadata_table, author_table, embedding_table ) app = coco.App( coco.AppConfig(name="PaperMetadataV1"), app_main, sourcedir=pathlib.Path("./papers"), )

几个要点：

• mount_table_target替你建表。源码位于 python/cocoindex/connectors/postgres/_target.py，其本质是table_target()+coco.mount_target()的语法糖，负责建表、幂等 upsert，以及当某个 PDF 消失时自动清理孤儿行。
• TableSchema.from_class自动做类型映射。源码在 同一文件 L364-L391，支持 dataclass/NamedTuple/Pydantic 模型三种行类型，根据 asyncpg 的类型转换把 Python 类型映射为 Postgres 类型；list[dict]这类复杂类型自动映射为 jsonb，Annotated[NDArray, EMBEDDER]通过VectorSchemaProvider映射为vector(384)。
• 三种不同的主键：论文元数据按filename唯一；作者索引按(author_name, filename)联合主键；嵌入表用生成的id。主键决定了 upsert 与去重语义。
• localfs.walk_dir匹配**/*.pdf，recursive=True递归子目录，live=True表示该 source 支持监听（真正跑 live 需在 CLI 上加-L）。
• mount_each每文件挂一个组件，让引擎能独立跟踪和更新每个 PDF，同时写入全部三张表。

关于向量索引的说明：为了保持示例最小化，本示例未声明向量索引，查询走顺序扫描——对几十篇论文完全够用。若语料规模变大，只需加一行embedding_table.declare_vector_index(column="embedding")（参考仓库中的 text_embedding 示例），pgvector 便会改用近似最近邻（ANN）查询。对应 opclass 定义（vector_cosine_ops等）可参见 postgres/_target.py。

运行管线：catch-up 与 live 两种模式

使用cocoindexCLI 构建并更新索引，两种模式任选：

# Catch-up run：扫描、同步、退出 cocoindex update main # Live run：先追平，再持续监听文件变化 cocoindex update -L main

其中main是app = coco.App(...)暴露的应用名（PaperMetadataV1）。catch-up 模式适合一次性全量构建；live 模式适合长期运行的场景——例如把papers/目录挂载在持续运行的索引服务上，新论文放入即被索引。

查询索引：复用同一个嵌入器做语义检索

查询直接用 SQL 即可，关键是复用索引流程中的同一个嵌入器，保证索引与查询的向量空间一致：

async def query_once(pool, embedder, query: str, *, top_k: int = 5) -> None: query_vec = await embedder.embed(query) async with pool.acquire() as conn: rows = await conn.fetch( f""" SELECT filename, location, text, embedding <=> $1 AS distance FROM "{PG_SCHEMA_NAME}"."{TABLE_EMBEDDINGS}" 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']} ({r['location']})") print(f" {r['text']}") print("---")

<=>是 pgvector 的余弦距离算子，1.0 - distance即相似度分数。输出会打印命中文件、命中位置（title或abstract）以及匹配文本。命令行直接查询：

python main.py "graph neural networks"

示例代码里main.py的__main__分支支持两种模式：传入命令行参数则查询一次后退出；不传参则进入交互式循环，反复输入查询语句（query函数中while True+input()），直到空输入退出。连接池创建时通过init=register_vector注册 pgvector 类型编解码。当示例论文被索引后，即使查询词与论文标题/摘要没有任何重合词，语义最相近的标题和摘要也会被排到前面——这正是"嵌入元数据"的意义。

增量更新机制：最小工作量，三种变更场景

CocoIndex 让三张表与 PDF 文件夹保持同步，并只做达到该状态所需的最小工作量。你从不手写 diff 或更新逻辑，两处机制分工明确：

• @coco.fn(memo=True)决定"重算什么"：某 PDF 的字节与函数代码均未变化时直接跳过，LLM 与嵌入器都不运行；
• mount_table_target决定"写什么"：只 upsert 实际变化的行，删除 source 已消失的行，且三张表同时生效。

具体到三种变更场景：

• 新增 PDF：只处理该文件——读取、抽取、嵌入，插入它的元数据、作者、嵌入行，其余文件不受影响；
• 替换 PDF：重新抽取；元数据行更新，作者行按新的作者列表对账（新增的作者插入、消失的作者删除），嵌入全部重算；
• 删除 PDF：三张表中该文件的所有行被自动清理。

同一套机制同样覆盖逻辑变更：修改 prompt、把gpt-4o换成其他模型、或更换嵌入模型后，CocoIndex 会把新输出与 Postgres 中已有数据比对，只应用差异。cocoindex update main一次追平后退出；cocoindex update -L main保持监听，以低延迟逐项应用变更。

完整代码与延伸方向

完整可运行代码位于 examples/paper_metadata/：入口 main.py（约 340 行，含管线与查询演示）、Schema models.py、依赖 pyproject.toml、环境模板 .env.example，示例论文放在 papers/。

如果你只需要"按语义搜索 PDF 全文"而不做结构化抽取，可以参考同仓库的 pdf_embedding 示例（对全文分块嵌入）；如果希望把 PDF 转成 Markdown 文本本身作为输出，可参考 pdf_to_markdown 示例。而本文的 paper_metadata 方案，则适用于更细粒度的场景：把论文/报告/申报文件批量变成结构化的、可 join、可语义检索的数据行，为 RAG、文献管理、知识库等上层应用直接供数。

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