# 可可索引 ## 概述 CocoIndex 是一个用于人工智能的超高性能实时数据转换框架，具有增量处理功能。此技能可以构建**索引流**，从源中提取数据、应用转换（分块、嵌入、LLM 提取）并导出到目标（矢量数据库、图形数据库、关系数据库）。 **核心能力：** 1. **编写索引流程** \- 使用 Python 定义 ETL 管道 2. **创建自定义函数** \- 构建可重用的转换逻辑 3. **操作流程** \- 使用 CLI 或 Python API 运行和管理流程 **主要特点：** * 增量处理（只处理变化的数据） * 实时更新（持续同步源更改到目标） * 内置函数（文本分块、嵌入、LLM 提取） * 多个数据源（本地文件、S3、Azure Blob、Google Drive、Postgres） * 多个目标（Postgres+pgvector、Qdrant、LanceDB、Neo4j、Kuzu） **详细文档：** **搜索文档：** ## 何时使用此技能当用户请求时使用： *“为我的文档构建矢量搜索索引” *“为代码/PDF/图像创建嵌入管道” *“使用法学硕士提取结构化信息” *“从文档构建知识图谱” *“设置实时文档索引” *“创建自定义转换函数” *“运行/更新我的 CocoIndex 流程” ## 流程写作流程 ### 第 1 步：了解需求提出澄清问题以理解： **数据来源：** * 数据在哪里？（本地文件、S3、数据库等） * 什么文件类型？（文本、PDF、JSON、图像、代码等） * 多久改变一次？（一次性、周期性、连续） **转换：** * 需要什么处理？（分块、嵌入、提取等） * 哪种嵌入模型？（SentenceTransformer、OpenAI、自定义） * 有自定义逻辑吗？（过滤、解析、丰富） **目标：** * 结果应该去哪里？（Postgres、Qdrant、Neo4j 等） * 什么模式？（字段、主键、索引） * 需要矢量搜索吗？（指定相似度度量） ### 第 2 步：设置依赖关系指导用户根据自己的需求向项目添加 CocoIndex 和适当的附加功能： **所需的依赖项：** * `cocoindex` \- 核心功能、CLI 和大多数内置功能 **可选附加功能（根据需要添加）：** * `cocoindex[embeddings]` \- 对于 SentenceTransformer 嵌入（使用 `SentenceTransformerEmbed` 时） * `cocoindex[colpali]` \- 用于 ColPali 图像/文档嵌入（当使用 `ColPaliEmbedImage` 或 `ColPaliEmbedQuery` 时） * `cocoindex[lancedb]` \- 对于 LanceDB 目标（导出到 LanceDB 时） * `cocoindex[embeddings,lancedb]` \- 可以组合多个额外内容 **内含内容：** * 基础包：核心功能、CLI、大多数内置功能、Postgres/Qdrant/Neo4j/Kuzu 目标 * `embeddings` extra：用于本地嵌入模型的 SentenceTransformers 库 * `colpali` extra：用于多模式文档/图像嵌入的 ColPali 引擎 * `lancedb` 额外：用于 LanceDB 矢量数据库支持的 LanceDB 客户端库用户可以使用自己喜欢的包管理器（pip、uv、poetry 等）进行安装或添加到“pyproject.toml”。 **有关安装详细信息：** ### 步骤 3：设置环境 **先检查现有环境：** 1.检查环境变量中是否存在`COCOINDEX_DATABASE_URL` * 如果没有找到，则使用默认值：`postgres://cocoindex:cocoindex@localhost/cocoindex` 2. **对于需要 LLM API 的流程**（嵌入、提取）： * 询问用户他们想要使用哪个 LLM 提供商： * **OpenAI** \- 生成和嵌入 * **人择** \- 仅一代 * **Gemini** \- 生成和嵌入 * **Voyage** \- 仅嵌入 * **Ollama** \- 本地模型（生成和嵌入） * 检查环境变量中是否存在对应的API key * 如果没有找到，**要求用户提供API密钥值** * **切勿在没有 LLM 的情况下创建简化示例** \- 始终获取正确的 API 密钥并使用真正的 LLM 函数 **引导用户创建`.env`文件：** # 数据库连接（必需-内部存储） COCOINDEX_DATABASE_URL=postgres://cocoindex:cocoindex@localhost/cocoindex # LLM API 密钥（添加您需要的密钥） OPENAI_API_KEY=sk-... # 对于 OpenAI（生成 + 嵌入） ANTHROPIC_API_KEY=sk-ant-... # 对于 Anthropic（仅限一代） GOOGLE_API_KEY=... # 对于 Gemini（生成 + 嵌入） VOYAGE_API_KEY=pa-... # 对于 Voyage（仅限嵌入） # Ollama 不需要 API 密钥（本地） **更多LLM选项：** 创建基本项目结构： # main.py 从 dotenv 导入 load_dotenv 导入可可指数 @cocoindex.flow_def(name="FlowName") def my_flow(flow_builder: cocoindex.FlowBuilder, data_scope: cocoindex.DataScope): # 此处定义流通过如果 __name__ == "__main__": 加载_dotenv() cocoindex.init() my_flow.update() ### 第 4 步：编写流程遵循这个结构： @cocoindex.flow_def(name="DescriptiveName") def flow_name(flow_builder: cocoindex.FlowBuilder, data_scope: cocoindex.DataScope): # 1.导入源数据 data_scope["source_name"] = flow_builder.add_source( cocoindex.sources.SourceType(...) ） # 2. 为输出创建收集器收集器 = data_scope.add_collector() # 3. 转换数据（遍历行）以 data_scope["source_name"].row() 作为项目： # 应用转换 item["new_field"] = item["existing_field"].transform( cocoindex.functions.FunctionName(...) ） ... # 嵌套迭代（例如文档中的块）将 item["nested_table"].row() 作为nested_item： # 更多转换 nested_item["嵌入"] =nested_item["文本"].transform(...) # 收集数据用于导出收集器.收集( 字段1=nested_item["字段1"], 字段2=项目["字段2"], generated_id=cocoindex.GenerateField.UUID ） # 4. 导出到目标收集器.导出（ “目标名称”， cocoindex.targets.TargetType(...), Primary_key_fields=["字段1"], vector_indexes=[...] # 如果需要） **关键原则：** * 每个源在顶级数据范围内创建一个字段 * 使用`.row()`迭代表数据 * **关键：始终将转换后的数据分配给行字段** \- 使用 `item["new_field"] = item["existing_field"].transform(...)`，而不是像 `new_field = item["existing_field"].transform(...)` 这样的局部变量 * 转换创建新字段而不改变现有数据 * 收集器从任何范围级别收集数据 * 导出必须发生在顶层（不在行迭代内） **要避免的常见错误：** ❌ **错误：** 使用局部变量进行转换将 data_scope["files"].row() 作为文件：摘要 = file["content"].transform(...) # ❌ 局部变量 summaries_collector.collect（文件名=文件[“文件名”]，摘要=摘要） ✅ **正确：** 分配给行字段将 data_scope["files"].row() 作为文件： file["summary"] = file["content"].transform(...) # ✅ 字段赋值 summaries_collector.collect(文件名=文件["文件名"], 摘要=文件["摘要"]) ❌ **错误：** 创建不必要的数据类来镜像流场从数据类导入数据类 @数据类 class FileSummary: # ❌ 不必要 - CocoIndex 自动管理字段文件名：str 摘要：str 嵌入：列表[float] # 该数据类从未在流程中使用！ ### 步骤 5：设计流程解决方案 **重要提示：** 下面列出的模式是常见的起点，但是 **您无法详尽列举所有可能的场景**。当用户需求与现有模式不匹配时： 1. **组合来自多个模式的元素** \- 创造性地混合和匹配源、转换和目标 2. **查看其他示例** \- 请参阅了解各种实际用例（人脸识别、多模式搜索、产品推荐、患者表格提取等） 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. 运行设置 cocoindex update --setup -f main # -f 强制安装而不提示确认 # 2. 启动服务器并将用户重定向到 CocoInsight cocoindex服务器-ci主 # 然后打开 CocoInsight https://cocoindex.io/cocoinsight ## 数据类型 CocoIndex 有一个独立于编程语言的类型系统。所有数据类型都是在流程定义时确定的，使架构清晰且可预测。 **重要：何时定义类型：** * **自定义函数** ：返回值**需要**类型注释（这些是类型推断的真相来源） * **流字段**：**不需要类型注释** \- CocoIndex 自动从源、函数和转换中推断类型 * **数据类/Pydantic模型**：仅在**实际使用**时创建它们（作为函数参数/返回或ExtractByLlm输出类型），而不是镜像流场模式 **类型注释要求：** * **自定义函数的返回值** ：必须使用**特定类型注释** \-这些是类型推断的真相来源 * **自定义函数的参数** ：宽松 - 可以使用 `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-dim float32 向量（推荐） * 没有维度的`list[float]`也可以工作 3. **结构类型**：Dataclass、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` 对于可为空的值 **示例：** 从数据类导入数据类从输入 import Literal 导入可可指数 @数据类类人：名称：str 年龄：整数 # ✅ 带维度的向量（推荐用于向量搜索） @cocoindex.op.function(behavior_version=1) def embed_text(text: str) -> cocoindex.Vector[cocoindex.Float32, Literal[768]]: """生成 768 维嵌入 - 向量索引所需的维度。""" # ...嵌入逻辑... return embedding # numpy 数组或 768 个浮点数的列表 # ✅ 结构体返回类型，宽松的参数 @cocoindex.op.function(behavior_version=1) def process_person(person: dict[str, Any]) -> 人: """参数可以是 dict[str, Any]，返回必须是具体的 Struct。""" 返回人（姓名=人[“姓名”]，年龄=人[“年龄”]） # ✅ LTable 返回类型 @cocoindex.op.function(behavior_version=1) def filter_people(人员: 列表[任意]) -> 列表[人员]: """返回类型指定特定结构的列表。""" 返回 [p for p in people if p.age >= 18] # ❌错误：dict[str, str] 不是有效的特定 CocoIndex 类型 # @cocoindex.op.function(...) # def bad_example(person: Person) -> dict[str, str]: # return {"name": person.name} **有关全面的数据类型文档：** ## 自定义函数当用户需要自定义转换逻辑时，创建自定义函数。 ### 决策：Standalone vs Spec+Executor **在以下情况下使用独立功能：** * 简单变换 * 无需配置 * 无需设置/初始化 **在以下情况下使用规范+执行器：** * 需要配置（模型名称、API端点、参数） * 需要设置（加载模型、建立连接） * 复杂的多步骤处理 ### 创建独立函数 @cocoindex.op.function(behavior_version=1) def my_function(input_arg: str, optional_arg: int | None = None) -> dict: ”“” 功能说明。参数：输入参数：描述可选_arg：可选描述 ”“” # 转换逻辑返回 {“结果”：f“已处理-{input_arg}”} **要求：** * 装饰器：`@cocoindex.op.function()` * 对所有参数和返回值进行类型注释 * 可选参数：“cache=True”用于昂贵的操作，“behavior_version”（缓存需要） ### 创建规范+执行器函数 # 1. 定义配置规范类 MyFunction(cocoindex.op.FunctionSpec): """MyFunction 的配置。""" 模型名称：str 阈值：浮动= 0.5 # 2.定义执行器 @cocoindex.op.executor_class（缓存= True，behavior_version = 1） MyFunctionExecutor 类：规范：MyFunction # 必需：链接到规范 model = None # 状态的实例变量 def 准备(self) -> 无: """可选：执行前运行一次。""" # 加载模型、设置连接等。 self.model = load_model(self.spec.model_name) def __call__(self, text: str) -> 字典： """必需：对每个数据行执行。""" # 使用 self.spec 进行配置 # 使用 self.model 加载资源结果 = self.model.process(text) 返回{“结果”：结果} **何时启用缓存：** * LLM API调用 * 模型推理 * 外部API调用 * 计算成本较高的操作 **重要：** 当函数逻辑更改以使缓存无效时，增加 `behavior_version`。有关详细示例和模式，请加载“references/custom_functions.md”。 **有关自定义函数的更多信息：** ## 操作流程 ### CLI 操作 **设置流程（创建资源）：** cocoindex设置主要 **一次性更新：** cocoindex主要更新 # 自动设置 cocoindex 更新 --setup main # 在安装和更新之前强制重置所有内容 cocoindex更新--重置主要 **实时更新（持续监控）：** cocoindex更新main.py -L # 需要对源或特定于源的更改捕获进行刷新间隔 **丢弃流量（删除所有资源）：** cocoindex 下降 main.py **检查流量：** cocoindex show main.py:FlowName **测试无副作用：** cocoindex 评估 main.py:FlowName --output-dir ./test_output 如需完整的 CLI 参考，请加载 `references/cli_operations.md`。 **对于 CLI 文档：** ### API操作 **基本设置：** 从 dotenv 导入 load_dotenv 导入可可指数加载_dotenv() cocoindex.init() @cocoindex.flow_def(名称=“MyFlow”) def my_flow(flow_builder, data_scope): # ...流定义... 通过 **一次性更新：** 统计数据 = my_flow.update() print(f"已处理 {stats.total_rows} 行") # 异步 stats = 等待 my_flow.update_async() **实时更新：** # 作为上下文管理器使用 cocoindex.FlowLiveUpdater(my_flow) 作为更新程序： # 更新程序在后台运行 # 这里是你的应用程序逻辑通过 # 手动控制更新器 = cocoindex.FlowLiveUpdater( 我的流， cocoindex.FlowLiveUpdaterOptions( live_mode=真， print_stats=真））更新程序.start() # ...应用程序逻辑... 更新程序.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]]: 返回文本.transform( cocoindex.functions.SentenceTransformerEmbed(model="...") ） # 在流中用于索引 doc["嵌入"] = text_to_embedding(doc["内容"]) # 用于查询 query_embedding = text_to_embedding.eval("搜索查询") 如需完整的 API 参考和模式，请加载“references/api_operations.md”。 **API 文档：** ## 内置函数 ### 文本处理 **递归分割** \- 智能地分块文本 doc["块"] = doc["内容"].transform( cocoindex.functions.SplitRecursively(), language="markdown"、# 或“python”、“javascript”等。块大小=2000，块重叠=500 ） **ParseJson** \- 解析 JSON 字符串数据 = json_string.transform(cocoindex.functions.ParseJson()) **DetectProgrammingLanguage** \- 从文件名中检测语言文件["语言"] = 文件["文件名"].transform( cocoindex.functions.DetectProgrammingLanguage() ） ### 嵌入 **SentenceTransformerEmbed** \- 本地嵌入模型 # 需要：cocoindex[嵌入] 块[“嵌入”] = 块[“文本”].transform( cocoindex.functions.SentenceTransformerEmbed( 模型=“句子变压器/all-MiniLM-L6-v2” ）） **EmbedText** \- LLM API 嵌入这是使用 LLM API（OpenAI、Voyage 等）生成嵌入的**推荐方式**。块[“嵌入”] = 块[“文本”].transform( cocoindex.functions.EmbedText( api_type=cocoindex.LlmApiType.OPENAI, 模型=“文本嵌入-3-小”，）） **ColPaliEmbedImage** \- 多模态图像嵌入 # 需要：cocoindex[colpali] 图像["嵌入"] = 图像["img_bytes"].transform( cocoindex.functions.ColPaliEmbedImage(model="vidore/colpali-v1.2") ） ### 法学硕士提取 **ExtractByLlm** \- 使用 LLM 提取结构化数据这是使用法学硕士进行提取和总结任务的**推荐方式**。它支持结构化输出（数据类、Pydantic 模型）和简单文本输出 (str)。导入数据类 # 用于结构化提取 @dataclasses.dataclass 产品信息类：名称：str 价格：浮动类别： STR 项目[“产品信息”] = 项目[“文本”].transform( cocoindex.functions.ExtractByLlm( llm_spec=cocoindex.LlmSpec( api_type=cocoindex.LlmApiType.OPENAI, 型号=“gpt-4o-mini” ), 输出类型=产品信息，指令=“提取产品信息” ）） # 用于文本摘要/生成文件[“摘要”] = 文件[“内容”].transform( cocoindex.functions.ExtractByLlm( llm_spec=cocoindex.LlmSpec( api_type=cocoindex.LlmApiType.OPENAI, 型号=“gpt-4o-mini” ), 输出类型=str， instructions="用一个段落总结本文档" ）） ## 常见来源和目标 **浏览所有源：** **浏览所有目标：** ### 来源 **本地文件：** cocoindex.sources.LocalFile( 路径=“文档”， include_patterns=["*.md", "*.txt"], excepted_patterns=["**/.*", "node_modules"] ） **亚马逊S3：** cocoindex.sources.AmazonS3( 桶=“我的桶”，前缀=“文档/”， aws_access_key_id=cocoindex.add_transient_auth_entry("..."), aws_secret_access_key=cocoindex.add_transient_auth_entry("...") ） **Postgres：** cocoindex.sources.Postgres( 连接 = cocoindex.add_auth_entry("conn", cocoindex.sources.PostgresConnection(...)), 查询=“选择id，来自文档的内容” ） ### 目标 **Postgres（具有向量支持）：** 收集器.导出（ “目标名称”， cocoindex.targets.Postgres(), Primary_key_fields=["id"], 矢量索引=[ cocoindex.VectorIndexDef( field_name="嵌入", 指标=cocoindex.VectorSimilarityMetric.COSINE_SIMILARITY ） ] ） **Qdrant：** 收集器.导出（ “目标名称”， cocoindex.targets.Qdrant(collection_name="my_collection"), Primary_key_fields=["id"] ） **兰斯数据库：** # 需要：cocoindex[lancedb] 收集器.导出（ “目标名称”， cocoindex.targets.LanceDB（uri =“lancedb_data”，table_name =“my_table”）， Primary_key_fields=["id"] ） **Neo4j（节点）：** 收集器.导出（ “节点”， cocoindex.targets.Neo4j( 连接=neo4j_conn，映射=cocoindex.targets.Nodes（标签=“实体”） ), Primary_key_fields=["id"] ） **Neo4j（关系）：** 收集器.导出（ “关系”， cocoindex.targets.Neo4j( 连接=neo4j_conn，映射=cocoindex.targets.Relationships( rel_type =“RELATES_TO”，源=cocoindex.targets.NodeFromFields( 标签=“实体”，字段=[cocoindex.targets.TargetFieldMapping（源=“source_id”，目标=“id”）] ), 目标=cocoindex.targets.NodeFromFields( 标签=“实体”， 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** \- 使用示例创建自定义函数的详细指南（独立函数、规范+执行器模式、LLM 调用、外部 API、缓存） * **references/cli_operations.md** \- 包含所有命令、选项和工作流程的完整 CLI 参考 * **references/api_operations.md** \- Python API 参考，包含编程流程控制、实时更新、查询和应用程序集成模式的示例当用户需要时加载这些参考： * 具体模式的详细示例 * 完整的API文档 * 高级使用场景 * 故障排除指导 **完整文档：** **搜索特定主题：**

可可指数

包含技能