cocoindex

CocoIndex

概述

CocoIndex 是一个面向 AI 的超高性能实时数据转换框架，支持增量处理。此技能可用于构建索引流，从数据源提取数据，应用转换（分块、嵌入、LLM 提取），并将数据导出到目标位置（向量数据库、图数据库、关系数据库）。

核心能力：

1. 编写索引流 - 使用 Python 定义 ETL 管道
2. 创建自定义函数 - 构建可复用的转换逻辑
3. 运行流 - 使用 CLI 或 Python API 运行和管理流

主要功能：

• 增量处理（仅处理发生变化的数据）
• 实时更新（持续将数据源的更改同步到目标位置）
• 内置函数（文本分块、嵌入、LLM 提取）
• 多种数据源（本地文件、S3、Azure Blob、Google Drive、Postgres）
• 多种目标位置（Postgres+pgvector、Qdrant、LanceDB、Neo4j、Kuzu）

详细文档： https://cocoindex.io/docs/ 搜索文档： https://cocoindex.io/docs/search?q=url%20encoded%20keyword

何时使用此技能

在用户提出以下请求时使用：

• “为我的文档构建向量搜索索引”
• “为代码/PDF/图像创建嵌入管道”
• “使用 LLM 提取结构化信息”
• “根据文档构建知识图谱”
• “设置实时文档索引”
• “创建自定义转换函数”
• “运行/更新我的 CocoIndex 流”

编写流的工作流程

步骤 1：了解需求

提出澄清性问题，以了解：

数据源：

• 数据在哪里？（本地文件、S3、数据库等）
• 文件类型有哪些？（文本、PDF、JSON、图像、代码等）
• 数据多久变化一次？（一次性、定期、持续）

转换：

• 需要进行哪些处理？（分块、嵌入、提取等）
• 使用哪种嵌入模型？（SentenceTransformer、OpenAI、自定义）
• 是否需要自定义逻辑？（过滤、解析、丰富数据）

目标位置：

• 结果应存放在哪里？（Postgres、Qdrant、Neo4j 等）
• 需要什么架构？（字段、主键、索引）
• 是否需要向量搜索？（指定相似度指标）

步骤 2：设置依赖项

根据用户的需求，引导用户为其项目添加带有适当 extras 的 CocoIndex：

必需依赖项：

• cocoindex - 核心功能、CLI 和大多数内置函数

可选 extras（按需添加）：

• cocoindex[embeddings] - 用于 SentenceTransformer 嵌入（使用 SentenceTransformerEmbed 时）
• cocoindex[colpali] - 用于 ColPali 图像/文档嵌入（使用 ColPaliEmbedImage 或 ColPaliEmbedQuery 时）
• cocoindex[lancedb] - 用于 LanceDB 目标位置（导出到 LanceDB 时）
• cocoindex[embeddings,lancedb] - 可以组合多个 extras

包含的内容：

• 基础包：核心功能、CLI、大多数内置函数，以及 Postgres/Qdrant/Neo4j/Kuzu 目标位置
• embeddings extra：用于本地嵌入模型的 SentenceTransformers 库
• colpali extra：用于多模态文档/图像嵌入的 ColPali 引擎
• lancedb extra：用于支持 LanceDB 向量数据库的 LanceDB 客户端库

用户可以使用自己偏好的包管理器（pip、uv、poetry 等）安装，也可以将其添加到 pyproject.toml。

安装详情： https://cocoindex.io/docs/getting_started/installation

第 3 步：设置环境

先检查现有环境：

1. 检查环境变量中是否存在 COCOINDEX_DATABASE_URL

   • 如果不存在，则使用默认值：postgres://cocoindex:cocoindex@localhost/cocoindex

2. 对于需要 LLM API 的工作流（嵌入、提取）：

   • 询问用户想使用哪种 LLM 提供商：
     • OpenAI——用于生成和嵌入
     • Anthropic——仅用于生成
     • Gemini——用于生成和嵌入
     • Voyage——仅用于嵌入
     • Ollama——本地模型（生成和嵌入）
   • 检查环境变量中是否存在相应的 API 密钥
   • 如果不存在，请用户提供 API 密钥
   • 绝不要创建不使用 LLM 的简化示例——务必获取正确的 API 密钥，并使用真实的 LLM 函数

指导用户创建 .env 文件：

# Database connection (required - internal storage)
COCOINDEX_DATABASE_URL=postgres://cocoindex:cocoindex@localhost/cocoindex

# LLM API keys (add the ones you need)
OPENAI_API_KEY=sk-...          # For OpenAI (generation + embeddings)
ANTHROPIC_API_KEY=sk-ant-...   # For Anthropic (generation only)
GOOGLE_API_KEY=...             # For Gemini (generation + embeddings)
VOYAGE_API_KEY=pa-...          # For Voyage (embeddings only)
# Ollama requires no API key (local)

了解更多 LLM 选项： https://cocoindex.io/docs/ai/llm

创建基本项目结构：

# main.py
from dotenv import load_dotenv
import cocoindex

@cocoindex.flow_def(name="FlowName")
def my_flow(flow_builder: cocoindex.FlowBuilder, data_scope: cocoindex.DataScope):
    # Flow definition here
    pass

if __name__ == "__main__":
    load_dotenv()
    cocoindex.init()
    my_flow.update()

第 4 步：编写工作流

遵循以下结构：

@cocoindex.flow_def(name="DescriptiveName")
def flow_name(flow_builder: cocoindex.FlowBuilder, data_scope: cocoindex.DataScope):
    # 1. Import source data
    data_scope["source_name"] = flow_builder.add_source(
        cocoindex.sources.SourceType(...)
    )

    # 2. Create collector(s) for outputs
    collector = data_scope.add_collector()

    # 3. Transform data (iterate through rows)
    with data_scope["source_name"].row() as item:
        # Apply transformations
        item["new_field"] = item["existing_field"].transform(
            cocoindex.functions.FunctionName(...)
        )

        ...

        # Nested iteration (e.g., chunks within documents)
        with item["nested_table"].row() as nested_item:
            # More transformations
            nested_item["embedding"] = nested_item["text"].transform(...)

            # Collect data for export
            collector.collect(
                field1=nested_item["field1"],
                field2=item["field2"],
                generated_id=cocoindex.GeneratedField.UUID
            )

    # 4. Export to target
    collector.export(
        "target_name",
        cocoindex.targets.TargetType(...),
        primary_key_fields=["field1"],
        vector_indexes=[...]  # If needed
    )

关键原则：

• 每个源都会在顶层数据作用域中创建一个字段
• 使用 .row() 遍历表格数据
• 重要：始终将转换后的数据赋给行字段——使用 item["new_field"] = item["existing_field"].transform(...)，不要使用 new_field = item["existing_field"].transform(...) 这样的局部变量
• 转换会创建新字段，不会修改现有数据
• 收集器可以从任意作用域级别收集数据
• 导出必须在顶层进行（不能在行迭代中进行）

应避免的常见错误：

❌ 错误： 使用局部变量进行转换

with data_scope["files"].row() as file:
    summary = file["content"].transform(...)  # ❌ Local variable
    summaries_collector.collect(filename=file["filename"], summary=summary)

✅ 正确： 将值赋给行字段

with data_scope["files"].row() as file:
    file["summary"] = file["content"].transform(...)  # ✅ Field assignment
    summaries_collector.collect(filename=file["filename"], summary=file["summary"])

❌ 错误： 创建不必要的数据类来映射工作流字段

from dataclasses import dataclass

@dataclass
class FileSummary:  # ❌ Unnecessary - CocoIndex manages fields automatically
    filename: str
    summary: str
    embedding: list[float]

# This dataclass is never used in the flow!

第 5 步：设计工作流解决方案

重要： 以下模式是常见的起点，但无法穷举所有可能的场景。当用户需求与现有模式不匹配时：

1. 组合多种模式中的元素——灵活组合不同的源、转换和目标
2. 查看更多示例——访问 https://github.com/cocoindex-io/cocoindex?tab=readme-ov-file#-examples-and-demo，了解各种真实场景（人脸识别、多模态搜索、产品推荐、患者表单提取等）
3. 从基本原理出发思考——使用核心 API（源、转换、收集器、导出），并运用常识解决新问题
4. 发挥创造力——CocoIndex 灵活多变；独特的组件组合可以解决独特的问题

常见起始模式（详细示例请参考相关资料）：

文本嵌入： 加载 references/flow_patterns.md，并参考“模式 1：简单文本嵌入”

代码嵌入： 加载 references/flow_patterns.md，并参考“模式 2：带语言检测的代码嵌入”

LLM 提取 + 知识图谱： 加载 references/flow_patterns.md，并参考“模式 3：基于 LLM 的知识图谱提取”

实时更新： 加载 references/flow_patterns.md，并参考“模式 4：使用刷新间隔的实时更新”

自定义函数： 加载 references/flow_patterns.md，并参考“模式 5：自定义转换函数”

可复用的查询逻辑： 加载 references/flow_patterns.md，并参考“模式 6：用于可复用逻辑的转换工作流”

并发控制： 加载 references/flow_patterns.md，并参考“模式 7：并发控制”

模式组合示例：

如果用户要求“从 S3 为图像建立索引，使用视觉 API 生成说明文字，并存储到 Qdrant”，则组合：

• AmazonS3 源（来自 S3 示例）
• 调用视觉 API 的自定义函数（来自自定义函数模式）
• 使用 EmbedText 对说明文字进行嵌入（来自嵌入模式）
• Qdrant 目标（来自目标示例）

没有任何单一模式能完全覆盖这一场景，但各个构建块可以组合使用。

第 6 步：测试并运行

引导用户进行测试：

# 1. Run with setup
cocoindex update --setup -f main   # -f force setup without confirmation prompts


# 2. Start a server and redirect users to CocoInsight
cocoindex server -ci main
# Then open CocoInsight at https://cocoindex.io/cocoinsight

数据类型

CocoIndex 拥有独立于编程语言的类型系统。所有数据类型都在流定义时确定，因此架构清晰且可预测。

重要：何时定义类型：

• 自定义函数：返回值必须添加类型注解（这是类型推断的唯一依据）
• 流字段：无需添加类型注解——CocoIndex 会根据来源、函数和转换自动推断类型
• 数据类/Pydantic 模型：仅在实际使用时才创建（作为函数参数/返回值或 ExtractByLlm 的 output_type），而不是为了与流字段架构保持一致

类型注解要求：

• 自定义函数的返回值：必须使用具体类型注解——这是类型推断的唯一依据
• 自定义函数的参数：要求较宽松——可以使用 Any、dict[str, Any]，也可以省略注解；引擎已知类型
• 流定义：无需显式添加类型注解——CocoIndex 会根据来源和函数自动推断类型

**具体返回类型为何重要：**自定义函数的返回类型可让 CocoIndex 无需处理真实数据，就能推断整个流中的字段类型。这使其能够创建适当的目标架构（例如，具有固定维度的向量索引）。

常见类型类别：

1. 基本类型：str、int、float、bool、bytes、datetime.date、datetime.datetime、uuid.UUID

2. 向量类型（嵌入）：如果计划将向量导出到目标，请在返回类型中指定维度，因为大多数目标都要求向量维度固定

   • cocoindex.Vector[cocoindex.Float32, typing.Literal[768]] - 768 维 float32 向量（推荐）
   • 不指定维度的 list[float] 也可以

3. 结构体类型：数据类、NamedTuple 或 Pydantic 模型

   • 返回类型：必须使用具体类（例如，Person）
   • 参数：可以使用 dict[str, Any] 或 Any

4. 表类型：

   • KTable（带键）：dict[K, V]，其中 K = 键类型（基本类型或不可变结构体），V = 结构体类型
   • LTable（有序）：list[R]，其中 R = 结构体类型
   • 参数：可以使用 dict[Any, Any] 或 list[Any]

5. Json 类型：cocoindex.Json，用于非结构化/动态数据

6. 可选类型：T | None，用于可空值

示例：

from dataclasses import dataclass
from typing import Literal
import cocoindex

@dataclass
class Person:
    name: str
    age: int

# ✅ Vector with dimension (recommended for vector search)
@cocoindex.op.function(behavior_version=1)
def embed_text(text: str) -> cocoindex.Vector[cocoindex.Float32, Literal[768]]:
    """Generate 768-dim embedding - dimension needed for vector index."""
    # ... embedding logic ...
    return embedding  # numpy array or list of 768 floats

# ✅ Struct return type, relaxed argument
@cocoindex.op.function(behavior_version=1)
def process_person(person: dict[str, Any]) -> Person:
    """Argument can be dict[str, Any], return must be specific Struct."""
    return Person(name=person["name"], age=person["age"])

# ✅ LTable return type
@cocoindex.op.function(behavior_version=1)
def filter_people(people: list[Any]) -> list[Person]:
    """Return type specifies list of specific Struct."""
    return [p for p in people if p.age >= 18]

# ❌ Wrong: dict[str, str] is not a valid specific CocoIndex type
# @cocoindex.op.function(...)
# def bad_example(person: Person) -> dict[str, str]:
#     return {"name": person.name}

完整的数据类型文档： https://cocoindex.io/docs/core/data_types

自定义函数

当用户需要自定义转换逻辑时，创建自定义函数。

选择：独立函数还是规范+执行器

在以下情况下使用独立函数：

• 简单转换
• 无需配置
• 无需设置/初始化

在以下情况下使用规范+执行器：

• 需要配置（模型名称、API 端点、参数）
• 需要进行设置（加载模型、建立连接）
• 多步骤处理较复杂

创建独立函数

@cocoindex.op.function(behavior_version=1)
def my_function(input_arg: str, optional_arg: int | None = None) -> dict:
    """
    Function description.

    Args:
        input_arg: Description
        optional_arg: Optional description
    """
    # Transformation logic
    return {"result": f"processed-{input_arg}"}

要求：

• 装饰器：@cocoindex.op.function()
• 所有参数和返回值都必须添加类型注解
• 可选参数：对于开销较大的操作使用 cache=True，启用缓存时必须指定 behavior_version

创建规范+执行器函数

# 1. Define configuration spec
class MyFunction(cocoindex.op.FunctionSpec):
    """Configuration for MyFunction."""
    model_name: str
    threshold: float = 0.5

# 2. Define executor
@cocoindex.op.executor_class(cache=True, behavior_version=1)
class MyFunctionExecutor:
    spec: MyFunction  # Required: link to spec
    model = None      # Instance variables for state

    def prepare(self) -> None:
        """Optional: run once before execution."""
        # Load model, setup connections, etc.
        self.model = load_model(self.spec.model_name)

    def __call__(self, text: str) -> dict:
        """Required: execute for each data row."""
        # Use self.spec for configuration
        # Use self.model for loaded resources
        result = self.model.process(text)
        return {"result": result}

适合启用缓存的场景：

• LLM API 调用
• 模型推理
• 外部 API 调用
• 计算开销较大的操作

**重要：**函数逻辑变更时，递增 behavior_version 以使缓存失效。

如需查看详细示例和模式，请加载 references/custom_functions.md。

有关自定义函数的更多信息： https://cocoindex.io/docs/custom_ops/custom_functions

流操作

CLI 操作

设置流程（创建资源）：

cocoindex setup main

单次更新：

cocoindex update main

# With auto-setup
cocoindex update --setup main

# Force reset everything before setup and update
cocoindex update --reset main

实时更新（持续监控）：

cocoindex update main.py -L

# Requires refresh_interval on source or source-specific change capture

删除流程（移除所有资源）：

cocoindex drop main.py

检查流程：

cocoindex show main.py:FlowName

无副作用地测试：

cocoindex evaluate main.py:FlowName --output-dir ./test_output

完整的 CLI 参考，请加载 references/cli_operations.md。

CLI 文档： https://cocoindex.io/docs/core/cli

API 操作

基本设置：

from dotenv import load_dotenv
import cocoindex

load_dotenv()
cocoindex.init()

@cocoindex.flow_def(name="MyFlow")
def my_flow(flow_builder, data_scope):
    # ... flow definition ...
    pass

单次更新：

stats = my_flow.update()
print(f"Processed {stats.total_rows} rows")

# Async
stats = await my_flow.update_async()

实时更新：

# As context manager
with cocoindex.FlowLiveUpdater(my_flow) as updater:
    # Updater runs in background
    # Your application logic here
    pass

# Manual control
updater = cocoindex.FlowLiveUpdater(
    my_flow,
    cocoindex.FlowLiveUpdaterOptions(
        live_mode=True,
        print_stats=True
    )
)
updater.start()
# ... application logic ...
updater.wait()

设置/删除：

my_flow.setup(report_to_stdout=True)
my_flow.drop(report_to_stdout=True)
cocoindex.setup_all_flows()
cocoindex.drop_all_flows()

使用转换流程进行查询：

@cocoindex.transform_flow()
def text_to_embedding(text: cocoindex.DataSlice[str]) -> cocoindex.DataSlice[list[float]]:
    return text.transform(
        cocoindex.functions.SentenceTransformerEmbed(model="...")
    )

# Use in flow for indexing
doc["embedding"] = text_to_embedding(doc["content"])

# Use for querying
query_embedding = text_to_embedding.eval("search query")

完整的 API 参考和模式，请加载 references/api_operations.md。

API 文档： https://cocoindex.io/docs/core/flow_methods

内置函数

文本处理

SplitRecursively - 智能拆分文本

doc["chunks"] = doc["content"].transform(
    cocoindex.functions.SplitRecursively(),
    language="markdown",  # or "python", "javascript", etc.
    chunk_size=2000,
    chunk_overlap=500
)

ParseJson - 解析 JSON 字符串

data = json_string.transform(cocoindex.functions.ParseJson())

DetectProgrammingLanguage - 根据文件名检测语言

file["language"] = file["filename"].transform(
    cocoindex.functions.DetectProgrammingLanguage()
)

嵌入

SentenceTransformerEmbed - 本地嵌入模型

# Requires: cocoindex[embeddings]
chunk["embedding"] = chunk["text"].transform(
    cocoindex.functions.SentenceTransformerEmbed(
        model="sentence-transformers/all-MiniLM-L6-v2"
    )
)

EmbedText - 大模型 API 嵌入

这是使用大模型 API（OpenAI、Voyage 等）生成嵌入的推荐方式。

chunk["embedding"] = chunk["text"].transform(
    cocoindex.functions.EmbedText(
        api_type=cocoindex.LlmApiType.OPENAI,
        model="text-embedding-3-small",
    )
)

ColPaliEmbedImage - 多模态图像嵌入

# Requires: cocoindex[colpali]
image["embedding"] = image["img_bytes"].transform(
    cocoindex.functions.ColPaliEmbedImage(model="vidore/colpali-v1.2")
)

大模型提取

ExtractByLlm - 使用大模型提取结构化数据

这是使用大模型执行提取和摘要任务的推荐方式。它同时支持结构化输出（dataclass、Pydantic 模型）和简单文本输出（str）。

import dataclasses

# For structured extraction
@dataclasses.dataclass
class ProductInfo:
    name: str
    price: float
    category: str

item["product_info"] = item["text"].transform(
    cocoindex.functions.ExtractByLlm(
        llm_spec=cocoindex.LlmSpec(
            api_type=cocoindex.LlmApiType.OPENAI,
            model="gpt-4o-mini"
        ),
        output_type=ProductInfo,
        instruction="Extract product information"
    )
)

# For text summarization/generation
file["summary"] = file["content"].transform(
    cocoindex.functions.ExtractByLlm(
        llm_spec=cocoindex.LlmSpec(
            api_type=cocoindex.LlmApiType.OPENAI,
            model="gpt-4o-mini"
        ),
        output_type=str,
        instruction="Summarize this document in one paragraph"
    )
)

常见来源和目标

浏览所有来源： https://cocoindex.io/docs/sources/ 浏览所有目标： https://cocoindex.io/docs/targets/

来源

LocalFile：

cocoindex.sources.LocalFile(
    path="documents",
    included_patterns=["*.md", "*.txt"],
    excluded_patterns=["**/.*", "node_modules"]
)

AmazonS3：

cocoindex.sources.AmazonS3(
    bucket="my-bucket",
    prefix="documents/",
    aws_access_key_id=cocoindex.add_transient_auth_entry("..."),
    aws_secret_access_key=cocoindex.add_transient_auth_entry("...")
)

Postgres：

cocoindex.sources.Postgres(
    connection=cocoindex.add_auth_entry("conn", cocoindex.sources.PostgresConnection(...)),
    query="SELECT id, content FROM documents"
)

目标

Postgres（支持向量）：

collector.export(
    "target_name",
    cocoindex.targets.Postgres(),
    primary_key_fields=["id"],
    vector_indexes=[
        cocoindex.VectorIndexDef(
            field_name="embedding",
            metric=cocoindex.VectorSimilarityMetric.COSINE_SIMILARITY
        )
    ]
)

Qdrant：

collector.export(
    "target_name",
    cocoindex.targets.Qdrant(collection_name="my_collection"),
    primary_key_fields=["id"]
)

LanceDB：

# Requires: cocoindex[lancedb]
collector.export(
    "target_name",
    cocoindex.targets.LanceDB(uri="lancedb_data", table_name="my_table"),
    primary_key_fields=["id"]
)

Neo4j（节点）：

collector.export(
    "nodes",
    cocoindex.targets.Neo4j(
        connection=neo4j_conn,
        mapping=cocoindex.targets.Nodes(label="Entity")
    ),
    primary_key_fields=["id"]
)

Neo4j（关系）：

collector.export(
    "relationships",
    cocoindex.targets.Neo4j(
        connection=neo4j_conn,
        mapping=cocoindex.targets.Relationships(
            rel_type="RELATES_TO",
            source=cocoindex.targets.NodeFromFields(
                label="Entity",
                fields=[cocoindex.targets.TargetFieldMapping(source="source_id", target="id")]
            ),
            target=cocoindex.targets.NodeFromFields(
                label="Entity",
                fields=[cocoindex.targets.TargetFieldMapping(source="target_id", target="id")]
            )
        )
    ),
    primary_key_fields=["id"]
)

常见问题与解决方案

“未找到流程”

• 检查 APP_TARGET 格式：cocoindex show main.py
• 如果当前目录不是项目根目录，请使用 --app-dir
• 确认流程名称与装饰器匹配

“数据库连接失败”

• 检查 .env 中是否包含 COCOINDEX_DATABASE_URL
• 测试连接：psql $COCOINDEX_DATABASE_URL
• 使用 --env-file 指定自定义位置

“架构不匹配”

• 重新运行设置：cocoindex setup main.py
• 删除并重新创建：cocoindex drop main.py && cocoindex setup main.py

“实时更新立即退出”

• 为来源添加 refresh_interval
• 或使用特定于来源的变更捕获机制（Postgres 通知、S3 事件）

“内存不足”

• 为来源设置并发限制：max_inflight_rows、max_inflight_bytes
• 在 .env 中设置全局限制：COCOINDEX_SOURCE_MAX_INFLIGHT_ROWS

参考文档

此技能包含常见模式和操作的全面参考文档：

• references/flow_patterns.md - 常见流程模式的完整示例（文本嵌入、代码嵌入、知识图谱、实时更新、并发控制等）
• references/custom_functions.md - 创建自定义函数的详细指南和示例（独立函数、spec+executor 模式、大模型调用、外部 API、缓存）
• references/cli_operations.md - 完整的 CLI 参考，包含所有命令、选项和工作流
• references/api_operations.md - Python API 参考，包含程序化流程控制、实时更新、查询和应用集成模式的示例

当用户需要以下内容时，请加载这些参考文档：

• 特定模式的详细示例
• 完整的 API 文档
• 高级用法场景
• 故障排除指南

完整文档： https://cocoindex.io/docs/ 搜索特定主题： https://cocoindex.io/docs/search?q=url%20encoded%20keyword