cocoindex

CocoIndex

CocoIndex 是一个 Python 库，用于通过声明式目标状态构建增量数据处理流水线。可以把它想成数据流水线领域的电子表格或 React：根据当前输入声明输出应是什么样子，CocoIndex 会自动处理增量更新、变更检测以及与外部系统的同步。

概览

CocoIndex 支持构建具备以下特性的数据流水线：

• 自动处理增量更新：仅重新处理发生变化的数据
• 使用声明式目标状态：声明应该存在什么，而不是如何更新
• 支持任意 Python 类型：无需自定义 DSL——可使用 dataclass、Pydantic、NamedTuple
• 提供函数记忆化：当输入或代码未变化时，跳过耗时操作
• 同步到多个目标：PostgreSQL、SQLite、LanceDB、Qdrant、SurrealDB、Apache Doris、文件系统、Kafka

核心原则：TargetState = Transform(SourceState)

何时使用此技能

在构建涉及以下内容的流水线时使用此技能：

• 文档处理：PDF/Markdown 转换、文本提取、分块
• 向量嵌入：为文档/代码生成嵌入，用于语义搜索
• 数据库转换：从源数据库到目标数据库的 ETL
• 知识图谱：从数据中提取实体和关系
• 基于 LLM 的提取：使用 LLM 提取结构化数据
• 基于文件的流水线：将文件从一种格式转换为另一种格式
• 增量索引：使搜索索引随源数据变化保持最新
• 流式流水线：基于 Kafka 的实时数据处理

快速开始：创建新项目

初始化项目

cocoindex init my-project
cd my-project

此命令会创建：main.py、pyproject.toml、README.md。生成的 main.py 会在其生命周期中通过 builder.settings.db_path = pathlib.Path("./cocoindex.db") 设置数据库位置。

添加依赖项

# For vector embeddings with PostgreSQL
dependencies = ["cocoindex>=1.0.0", "sentence-transformers", "asyncpg"]

# For LLM extraction
dependencies = ["cocoindex>=1.0.0", "litellm", "instructor", "pydantic>=2.0"]

完整示例请参阅 references/setup_project.md。

运行流水线

uv run cocoindex update main.py   # installs dependencies, then runs

核心概念

1. 应用

应用是顶层可执行对象，用于将主函数与参数绑定：

import cocoindex as coco

@coco.fn
async def app_main(sourcedir: pathlib.Path) -> None:
    ...

app = coco.App(
    coco.AppConfig(name="MyApp"),
    app_main,
    sourcedir=pathlib.Path("./data"),
)

2. 函数（@coco.fn）

@coco.fn 装饰器用于将函数标记为 CocoIndex 处理函数。添加 memo=True，可在输入/代码未变化时跳过重新执行：

@coco.fn(memo=True)
async def expensive_operation(data: str) -> Result:
    # LLM call, embedding generation, heavy computation
    return await expensive_transform(data)

关键参数：

• memo=True——启用记忆化（输入/代码未变化时跳过执行）
• version=1——显式提升版本，以强制重新执行
• batching=True——自动批处理并发调用（仅限异步）
• runner=coco.GPU——串行执行 GPU 密集型任务

3. 处理组件

处理组件将一个项的处理逻辑与其目标状态组合在一起。

使用 mount_each()（列表场景首选）或 mount() 挂载组件：

# One component per item (preferred for lists)
await coco.mount_each(process_file, files.items(), target_table)

# Single component (subpath auto-derived from fn.__name__)
await coco.mount(setup_fn, arg1)

# Dependent component (blocks until result returned)
result = await coco.use_mount(init_fn)

# Explicit subpath (when you need a specific path, e.g. in loops)
await coco.mount(coco.component_subpath("item", item_id), process_item, item)

要点：

• 所有挂载 API 都是 async
• mount()、use_mount() 和 mount_each() 会根据 fn.__name__ 自动推导子路径；也可以将显式子路径作为第一个参数传入
• 需要返回值时使用 use_mount()
• 使用稳定的组件路径，以实现正确的记忆化和清理

4. 目标状态

声明应该存在什么——CocoIndex 会处理创建/更新/删除：

# Database row target
table.declare_row(row=MyRecord(id=1, name="example"))

# File target
localfs.declare_file(outdir / "output.txt", content, create_parent_dirs=True)

# Kafka message target
topic_target.declare_target_state(key="msg-1", value=json.dumps(data))

5. 共享资源的上下文

使用 ContextKey 在多个组件间共享开销较大的资源（数据库连接、模型）：

EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder")

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    builder.provide(EMBEDDER, SentenceTransformerEmbedder("all-MiniLM-L6-v2"))
    yield

# In processing functions:
embedder = coco.use_context(EMBEDDER)

@coco.lifespan 装饰器会将函数注册到默认 CocoIndex 环境。默认情况下，所有应用共享该环境。ContextKey 也用作源/目标的稳定标识——key 字符串在各次运行之间必须保持稳定。

6. ID 生成

生成在增量更新期间保持不变的稳定唯一标识符：

from cocoindex.resources.id import generate_id, IdGenerator

# Deterministic: same dep -> same ID
chunk_id = await generate_id(chunk.text)

# Always distinct: each call -> new ID, even with same dep
id_gen = IdGenerator()
for chunk in chunks:
    chunk_id = await id_gen.next_id(chunk.text)

7. 追赶模式与实时模式

默认情况下，app.update() 会以追赶模式运行：扫描所有来源，处理自上次运行以来发生的更改（跳过已记忆的组件），同步目标状态，然后返回。每次调用仍需扫描来源以发现更改。

实时模式会在追赶完成后让应用继续运行，并允许组件持续从其来源流式传输更改（例如，文件监视器、Kafka 消费者），以极低延迟应用这些更改。

# Enable live mode
app.update_blocking(live=True)
# Or: cocoindex update main.py -L

需要完成两件事：（1）在应用上启用实时模式；（2）使用支持实时更新的来源。

• LiveMapView 来源（例如，localfs.walk_dir(..., live=True)）会先扫描当前状态，然后监视更改。它们也可在追赶模式下使用——只需编写一次管道，并在运行时选择模式。
• LiveMapFeed 来源（例如，kafka.topic_as_map()）只会流式传输更改，不提供初始快照。mount_each() 会自动检测这类来源，并在内部创建实时组件。

# LocalFS with live watching
files = localfs.walk_dir(sourcedir, live=True, ...)
await coco.mount_each(process_file, files.items(), target)

# Kafka -- inherently live
items = kafka.topic_as_map(consumer, "my-topic")
await coco.mount_each(process_message, items, target)

常见管道模式

模式 1：文件转换

import pathlib
import cocoindex as coco
from cocoindex.connectors import localfs
from cocoindex.resources.file import FileLike, PatternFilePathMatcher

@coco.fn(memo=True)
async def process_file(file: FileLike, outdir: pathlib.Path) -> None:
    content = await file.read_text()
    transformed = transform_content(content)
    outname = file.file_path.path.stem + ".out"
    localfs.declare_file(outdir / outname, transformed, create_parent_dirs=True)

@coco.fn
async def app_main(sourcedir: pathlib.Path, outdir: pathlib.Path) -> None:
    files = localfs.walk_dir(
        sourcedir,
        recursive=True,
        path_matcher=PatternFilePathMatcher(included_patterns=["**/*.md"]),
    )
    await coco.mount_each(process_file, files.items(), outdir)

app = coco.App(
    coco.AppConfig(name="Transform"),
    app_main,
    sourcedir=pathlib.Path("./data"),
    outdir=pathlib.Path("./out"),
)

模式 2：向量嵌入管道

import pathlib
from dataclasses import dataclass
from typing import AsyncIterator, Annotated

import asyncpg
from numpy.typing import NDArray

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

DATABASE_URL = "postgres://cocoindex:cocoindex@localhost/cocoindex"
PG_DB = coco.ContextKey[asyncpg.Pool]("pg_db")
EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder")

_splitter = RecursiveSplitter()

@dataclass
class DocEmbedding:
    id: int
    filename: str
    text: str
    embedding: Annotated[NDArray, EMBEDDER]  # Dimensions inferred from ContextKey
    chunk_start: int
    chunk_end: int

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    async with await asyncpg.create_pool(DATABASE_URL) as pool:
        builder.provide(PG_DB, pool)
        builder.provide(EMBEDDER, SentenceTransformerEmbedder("all-MiniLM-L6-v2"))
        yield

@coco.fn
async def process_chunk(
    chunk: Chunk, filename: pathlib.PurePath,
    id_gen: IdGenerator, table: postgres.TableTarget[DocEmbedding],
) -> None:
    table.declare_row(row=DocEmbedding(
        id=await id_gen.next_id(chunk.text),
        filename=str(filename),
        text=chunk.text,
        embedding=await coco.use_context(EMBEDDER).embed(chunk.text),
        chunk_start=chunk.start.char_offset,
        chunk_end=chunk.end.char_offset,
    ))

@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)
    id_gen = IdGenerator()
    await coco.map(process_chunk, chunks, file.file_path.path, id_gen, table)

@coco.fn
async def app_main(sourcedir: pathlib.Path) -> None:
    target_table = await postgres.mount_table_target(
        PG_DB,
        table_name="embeddings",
        table_schema=await postgres.TableSchema.from_class(DocEmbedding, primary_key=["id"]),
    )
    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="Embedding"), app_main, sourcedir=pathlib.Path("./data"))

模式 3：基于大模型的提取

import instructor
from pydantic import BaseModel
from litellm import acompletion

_instructor_client = instructor.from_litellm(acompletion, mode=instructor.Mode.JSON)

class ExtractionResult(BaseModel):
    title: str
    topics: list[str]

@coco.fn(memo=True)  # Memo avoids re-calling LLM
async def extract_and_store(content: str, message_id: int, table) -> None:
    result = await _instructor_client.chat.completions.create(
        model="gpt-4",
        response_model=ExtractionResult,
        messages=[{"role": "user", "content": f"Extract topics: {content}"}],
    )
    table.declare_row(row=Message(id=message_id, title=result.title, content=content))

连接器和操作

CocoIndex 提供用于从外部系统读取数据和向其写入数据的连接器：

连接器	来源	目标	向量	用途
PostgreSQL	Y	Y	pgvector	生产环境中的 SQL + 向量
SQLite	-	Y	sqlite-vec	本地 SQL + 向量
LanceDB	-	Y	Y	云原生向量数据库
Qdrant	-	Y	Y	专用向量数据库
SurrealDB	-	Y	Y	图数据库 + 文档数据库
Apache Doris	-	Y	Y	分析型数据库 + 向量
LocalFS	Y	Y	N/A	基于文件的管道
Amazon S3	Y	-	N/A	云对象存储
Kafka	Y	Y	N/A	流式管道
Google Drive	Y	-	N/A	云文件来源

如需查看详细的连接器文档，请参阅 references/connectors.md。

文本和嵌入操作

文本拆分

from cocoindex.ops.text import RecursiveSplitter, detect_code_language

splitter = RecursiveSplitter()
language = detect_code_language(filename="example.py")
chunks = splitter.split(text, chunk_size=1000, chunk_overlap=200, language=language)

嵌入

from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder

embedder = SentenceTransformerEmbedder("sentence-transformers/all-MiniLM-L6-v2")
embedding = await embedder.embed(text)  # Returns NDArray

CLI 命令

cocoindex init my-project              # Create new project
cocoindex update main.py               # Run app
cocoindex update main.py:my_app        # Run specific app
cocoindex update main.py -L            # Run in live mode (continuous)
cocoindex update main.py --full-reprocess  # Reprocess everything
cocoindex drop main.py [-f]            # Drop and reset all state
cocoindex ls [main.py]                 # List apps
cocoindex show main.py [--tree]        # Show component paths

最佳实践

1. 为所有处理函数使用 @coco.fn

管道中的每个函数（声明目标状态、调用挂载 API 等）都必须使用 @coco.fn 装饰。

2. 为开销较大的操作添加记忆化

@coco.fn(memo=True)  # Skip re-execution when inputs/code unchanged
async def process_chunk(chunk, table):
    embedding = await embedder.embed(chunk.text)  # Expensive!
    table.declare_row(...)

3. 使用稳定的组件路径

# Good: Stable identifiers
coco.component_subpath("file", str(file.file_path.path))
coco.component_subpath("record", record.id)

# Bad: Unstable identifiers
coco.component_subpath("file", file)      # Object reference
coco.component_subpath("idx", idx)        # Index changes

4. 使用上下文共享资源

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    async with await asyncpg.create_pool(DATABASE_URL) as pool:
        builder.provide(PG_DB, pool)
        yield

5. 对向量使用 Annotated[NDArray, CONTEXT_KEY]

EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder")

@dataclass
class Record:
    vector: Annotated[NDArray, EMBEDDER]  # Auto-infer dimensions from ContextKey

6. 使用目标便捷 API

# Mount table target -- subpath is automatic
table = await postgres.mount_table_target(
    PG_DB,
    table_name="my_table",
    table_schema=await postgres.TableSchema.from_class(MyRecord, primary_key=["id"]),
)

故障排除

所有内容都在重新处理

为开销较大的函数添加 memo=True：

@coco.fn(memo=True)  # Add this
async def process_item(item):
    ...

记忆化不起作用

检查组件路径是否稳定。使用稳定 ID，而不是对象引用。

资源

references/

• api_reference.md：快速 API 参考
• connectors.md：完整的连接器参考
• patterns.md：详细的管道模式
• setup_project.md：项目设置指南
• setup_database.md：数据库设置指南

可运行示例

上文介绍的每种模式都在 examples/ 下有完整、可运行的应用——从最接近任务需求的示例开始并加以改造。每个示例都有自己的 README.md，大多数 Python 示例还提供 .env.example；完整的示例目录、凭据和各示例的运行命令，请参阅 examples/AGENTS.md。推荐的入门示例：

• 向量搜索 → text_embedding（Postgres）、text_embedding_qdrant / _lancedb（其他存储）
• 代码搜索 → code_embedding
• 大模型提取 → hn_trending_topics、patient_intake_extraction_baml
• 知识图谱 → conversation_to_knowledge、meeting_notes_graph_neo4j
• 自定义转换 → files_transform、pdf_to_markdown

外部资源

• CocoIndex 文档 — 完整文本见 llms-full.txt
• GitHub 示例

版本说明

此技能适用于 CocoIndex >=1.0.0（v1）。其 API 与 v0 完全不同。

你可能从训练数据中学到的代码是 v0 代码——请勿生成这类代码。 如果你发现自己正在编写以下任何符号，就说明你用的是已移除的 v0 API：

v0（已移除）	v1 对应项
@cocoindex.flow_def、FlowBuilder、Flow、open_flow	coco.App + 一个 @coco.fn 主函数
DataScope、DataSlice、add_collector()、collect()、export()	在挂载的组件内通过 Target API（declare_row、declare_file）声明目标状态
cocoindex.sources.LocalFile、cocoindex.sources.*	连接器 API，例如 localfs.walk_dir(...)
cocoindex.functions.SplitRecursively、cocoindex.functions.*	cocoindex.ops.*，例如 RecursiveSplitter
cocoindex.targets.Postgres、cocoindex.targets.* / storages.*	连接器目标，例如 postgres.declare_table_target(...)
transform_flow、cocoindex.op.function()	普通的 @coco.fn 函数
cocoindex.init()、settings、COCOINDEX_DATABASE_URL	coco.App(coco.AppConfig(...))；状态存储在本地数据库路径中
CLI cocoindex setup	无需设置步骤——只需运行 cocoindex update（实时模式使用 -L/--live）

阅读提及这些 v0 符号的第三方教程或模型记忆时，请忽略它们，并改用本技能及 references/api_reference.md 中的模式。