【CocoIndex】核心架构与设计原理深度解析：把 React 思维引入 AI Agent 数据同步的开山之作

引子：当 Agent 拥有「永远新鲜」的上下文世界

过去两年，所有的 RAG 框架都在解决同一个问题：「怎么让 LLM 检索到最新的相关知识」。但当我们真的把一个 AI Agent 部署进生产环境——让它持续监控代码仓库、会议纪要、Slack 频道、PDF 文档、视频字幕——才意识到问题的另一半同样致命：

LLM 看得见的「上下文」和现实的「事实世界」，正在以小时为单位发散。

传统方案是定时 batch 跑全量重建：每天凌晨跑一次 embedding、入库，第二天早上 Agent 看到的「新鲜内容」其实是 24 小时前的世界——文件已过期、文档已重写、API 已改版。对长程 Agent（long-horizon agent）来说，这种**批处理漂移（batch drift）**意味着哪怕模型能力再强，工具调用与决策都会被过时的数据污染。

如何在不重新嵌入整个语料库的前提下，让 Agent 看到的上下文世界保持「亚秒级新鲜」？这是 AI 数据基础设施在 2026 H2 面临的核心问题。

今天我们要剖析的 CocoIndex，就是冲着这个问题去的。它在 GitHub 上 4 个月从 0 增长到 ⭐14,815（Apache-2.0，Rust + Python 双核实现），定位是「给长程 Agent 的增量数据管线」。它的核心宣言是：

Target = F(Source) —— 像 React 那样声明你想要的目标，引擎负责增量、永不停歇地把源头数据同步到目标，任何一端变化都只重算 Δ（delta），永远只算 Δ。

这个抽象既是编译器的「中间表示」、又是数据工程的「增量视图」、还是 AI 时代的「长程上下文维护」。CocoIndex 是 2026 H2「数据工程 + AI 上下文 + 智能体运行时」三股潮流交汇处的代表作品。

本文将围绕 7 个章节层层展开它的核心架构、关键抽象、工程实现与设计权衡。我会用大量从源码中直接截取的代码片段（标注源文件路径与行号），让你看到「声明式增量引擎」背后的真实运行机制。

---

第一章 项目定位与核心价值

1.1 一句话定义

CocoIndex 是一种增量数据管线引擎（incremental data pipeline engine）：你用 Python 声明 Target = F(Source) 的有向变换图，它会把这张图编译成 Rust 核心执行的 DAG，并以指纹级（fingerprint-level）增量保持目标与源头的持续同步。

它专门为「需要近乎实时新鲜度的 AI Agent 数据上下文」而设计——只要源头数据变化（文件被改写、commit 被推送、转写产出新字幕），下游的向量索引、知识图谱、文档表、Kafka 流都会仅对变化的行级数据做局部重新 embedding / 重新抽取，其他 99.9% 的语料继续命中缓存。

1.2 能力矩阵

能力维度	CocoIndex 现状
⭐ GitHub Star	14,815（截至 2026-08-26）
主要语言	Rust（核心引擎）+ Python（声明 API）
License	Apache-2.0
包大小	113MB（含双语文档、20+ 示例工程）
数据源支持	8 大类：Codebases（git 仓库）、Meeting Notes、Web · APIs、File System · Blob Stores、Databases、Message Queues、Images · Video、Voice · Transcripts
目标库支持	6 大类：Relational DB（Postgres）、Data Warehouse、Vector DB（LanceDB / Qdrant）、Graph DB（Neo4j / FalkorDB / SurrealDB / Kuzu）、Message Queue（Kafka）、Feature Store
引擎核心	Rust Core + LMDB 持久化（heed 库）+ Blake2b 指纹 + 异步运行时（tokio）
Python API	@coco.fn(memo=True) 装饰器 + ContextKey[Pool] 上下文注入 + coco.fn → 流程编排
增量粒度	行级指纹（fingerprint） + 代码级指纹（hash-of-code） 双键失效
血缘（lineage）	端到端：每个目标向量都能溯源到源文件第几行
调度	单实例即可，也可水平扩展（每个 App 独立 LMDB）
失败处理	重试 + 指数退避 + Dead Letter Queue + 故障隔离（一个文件坏掉不卡住整个流程）

1.3 与传统 ETL / Airflow / dbt 的差异

CocoIndex 与所有「数据工程」工具的根本不同在于：

维度	Airflow / dbt / Spark	CocoIndex
调度粒度	任务级（每小时/每天触发）	行级指纹失效（任意行变化即触发 Δ 重算）
状态保留	任务运行间状态机重置	进程内 LMDB 持久化指纹表（重启立刻恢复）
缓存粒度	整批查询缓存	Fingerprint 级缓存（每条 source 字节的指纹都缓存）
代码失效	DAG 改了就整体重跑	hash-of-code 失效（改了一行只重跑用了这行的子树）
血缘	通常没有	每行 byte 级血缘（本源文件第几行可溯）
编程模型	SQL + YAML + 调度代码	Python 装饰器 + 类型注解 + 异步

这就是为什么 CocoIndex 把自己的心智模型叫做「React for data engineering」——你写一次 F，引擎永远帮你保持 Target = F(Source)，不管哪一端变，都自动 re-evaluate。

---

第二章 整体架构

CocoIndex 的代码组织遵循**「语言分层 + 职责分层」的二维矩阵**：

+-----------------------------------------------+
|  python/cocoindex/_internal/   ← Python 声明层   |
|    ├── api.py        （公开 API 门面）              |
|    ├── app.py        （App 生命周期）              |
|    ├── function.py   （@fn 装饰器，FunctionData 转译） |
|    ├── live_component.py （LiveMap / 实时数据流）    |
|    ├── target_state.py（Target 状态机）            |
|    ├── memo_fingerprint.py（memo 指纹注册）         |
|    ├── context_keys.py（DI 上下文）                |
|    ├── runner.py     （GPU / ThreadPool 调度）      |
|    └── ...                                     |
+-----------------------------------------------+
                │  PyO3 桥接
+-----------------------------------------------+
|  rust/core/src/   ← Rust 执行核心                    |
|    ├── engine/                                        |
|    │   ├── execution.rs       （增量执行主循环）        |
|    │   ├── component.rs       （Component 处理单元）  |
|    │   ├── context.rs         （AppContext + 上下文） |
|    │   ├── live_component.rs  （LiveComponent 池）    |
|    │   ├── target_state.rs    （Target Action Sink） |
|    │   ├── logic_registry.rs  （代码指纹注册表）      |
|    │   └── profile.rs         （EngineProfile trait） |
|    ├── state_store/                                   |
|    │   ├── app_store.rs       （LMDB 应用存储）        |
|    │   ├── storage.rs         （多 App 单写事务聚合） |
|    │   └── txn.rs             （Read/Write Txn）     |
|    ├── state/                                         |
|    │   ├── stable_path.rs     （稳定路径）            |
|    │   ├── db_schema.rs       （msgpack 序列化）     |
|    │   └── target_state_path.rs（Target 状态路径）   |
|    └── telemetry.rs    ← 指标收集                     |
+-----------------------------------------------+
                │  序列化层（msgpack + Blake2b）
+-----------------------------------------------+
|  rust/sdk/cocoindex/src/ ← SDK 集成层                 |
|    ├── postgres.rs / lancedb.rs / qdrant.rs / sqlite.rs |
|    ├── neo4j.rs / falkordb.rs / surrealdb.rs / doris.rs |
|    └── oci_object_storage.rs                         |
+-----------------------------------------------+

2.1 顶层架构图

flowchart TB
    subgraph PY["Python 声明层"]
        API["@coco.fn / @coco.fn(memo=True)<br/>ContextKey + Lifespan"]
        APP["App + AppConfig"]
        LIVE["LiveComponent 实时数据源"]
    end

    subgraph BRIDGE["PyO3 桥接"]
        CORE["core.init_runtime()<br/>package_id + lang + serialize_fn"]
    end

    subgraph RUST["Rust 核心执行层"]
        ENG["Engine<br/>execution.rs"]
        COMP["ComponentProcessor<br/>component.rs"]
        CTX["AppContext<br/>context.rs"]
        STORE["AppStore (LMDB)<br/>state_store/app_store.rs"]
        FING["Fingerprinter<br/>fingerprint.rs (Blake2b)"]
        REG["LogicRegistry<br/>logic_registry.rs"]
        LIVE_R["LiveComponent Pool<br/>live_component.rs"]
    end

    subgraph SDK["SDK 目标库"]
        PG["Postgres / SQLite"]
        VEC["LanceDB / Qdrant"]
        GRAPH["Neo4j / SurrealDB / FalkorDB"]
        MQ["Kafka / Doris"]
    end

    subgraph SRC["数据源"]
        FS["Local FS / S3"]
        GIT["Git Repo"]
        DB["Database CDC"]
        MQIN["Kafka / Redis Stream"]
    end

    API --> CORE
    APP --> CORE
    LIVE --> BRIDGE
    CORE --> ENG
    ENG --> COMP
    ENG --> CTX
    COMP --> STORE
    COMP --> FING
    COMP --> REG
    COMP --> LIVE_R
    COMP --> SDK
    SRC --> COMP

2.2 持久化状态存储

CocoIndex 选 LMDB（via heed crate）作为状态存储，而非 Redis、RocksDB 或 PostgreSQL，原因如下：

# 来自 rust/core/Cargo.toml
heed = "0.22.0"
page_size = "0.6"

• 零依赖、单文件：deploy 只需一个目录，不需拉起额外 daemon
• mmap 内存映射：读写都映射到进程虚拟内存，零拷贝
• B+tree 索引：支持 range query 与 cursor 遍历
• ACID 事务：写事务单写者，避免与多写并发争抢

每个 App 实例对应一个 LMDB Database，App 卸载（drop_app）时完整释放，避免堆积。

2.3 进程模型与并发

• Python 侧：用 asyncio 跑 coco.App.update()，单实例即可处理上千行并发（_DEFAULT_MAX_INFLIGHT_COMPONENTS = 1024）。
• Rust 侧：每个 Component 通过 tokio 调度，写事务通过 Storage::run_txn_boxed 的单写者 batcher 聚合所有 App 的写入，避免每个 App 独立争抢 heed 的 writer mutex。
• 取消语义：每个 App 持有 tokio_util::sync::CancellationToken，drop_app 触发取消，等待 in-flight 组件 30 秒（LIVE_COMPONENT_DRAIN_TIMEOUT_SECS = 30）干净退出。

---

第三章 应用层 API：@coco.fn 与数据流编排

CocoIndex 的「声明式 API」是它最精彩的部分之一——你不需要学习专用 DAG 配置语言，写普通 Python 函数加上装饰器即可。

3.1 核心 API 矩阵

API	作用	关键参数
@coco.fn	把普通 async 函数注册成 CocoIndex 操作（factor）	支持泛型签名
@coco.fn(memo=True)	启用行级 memoization：相同输入 + 相同代码哈希直接命中缓存	自动推断 memo key
coco.lifespan	应用生命周期钩子（启动数据库连接池、模型等）	通过 builder.provide() 注入 ContextKey
coco.ContextKey[T]	类型安全的依赖注入句柄（区分是否 detect_change）	detect_change=True 时输入变化触发下游
postgres.mount_table_target / lancedb.mount_* / neo4j.mount_*	挂载目标表——获得带类型校验的写入端点	接受 dataclass
table.declare_row(...)	在处理函数中声明要写入目标的行	通常在最后一步
coco.App(...).update() / update_blocking()	启动增量同步（回填 + 监听变化）	async / sync
coco.map(process_chunk, chunks, ...)	并行映射（worker pool）	接受 Iterable

3.2 端到端示例：代码仓库 → 向量索引

下面是从 examples/code_embedding/main.py 直接抽取的完整 pipeline——它把一个 Git 仓库里的所有源文件做 AST 感知的递归切分，向量化后写入 Postgres/pgvector：

# 来自 examples/code_embedding/main.py:1-12
from __future__ import annotations
import asyncio
import os
import pathlib
import sys
from dataclasses import dataclass
from dotenv import load_dotenv
from typing import AsyncIterator, Annotated

import asyncpg
from pgvector.asyncpg import register_vector
from numpy.typing import NDArray

import cocoindex as coco
from cocoindex.connectors import localfs, postgres
from cocoindex.ops.text import RecursiveSplitter, detect_code_language
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

接下来先定义目标行 schema：

# 来自 examples/code_embedding/main.py:36-43
@dataclass
class CodeEmbedding:
    id: int
    filename: str
    code: str
    embedding: Annotated[NDArray, EMBEDDER]   # ← 字段类型注解驱动哪个 embedder
    start_line: int
    end_line: int

注意 Annotated[NDArray, EMBEDDER]——CocoIndex 通过类型注解读取 EMBEDDER 的 ContextKey，自动绑定到字段语义。

然后是生命周期管理（启动数据库连接池 + 加载模型）：

# 来自 examples/code_embedding/main.py:46-53
@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

关键的「带 memoization 的处理函数」：

# 来自 examples/code_embedding/main.py:79-92
@coco.fn(memo=True)                           # ← memo=True 启用指纹缓存
async def process_file(
    file: FileLike,
    table: postgres.TableTarget[CodeEmbedding],
) -> None:
    text = await file.read_text()
    language = detect_code_language(filename=str(file.file_path.path.name))
    chunks = _splitter.split(
        text,
        chunk_size=1000,
        min_chunk_size=300,
        chunk_overlap=300,
        language=language,
    )
    id_gen = IdGenerator()
    await coco.map(process_chunk, chunks, file.file_path.path, id_gen, table)

memo=True 的语义是：当这个函数的输入（file 的字节内容 + table 的目标声明）哈希未变、且函数自身代码的哈希也未变时，跳过执行，直接用上一次的结果。

最后编排流程：

# 来自 examples/code_embedding/main.py:95-110
@coco.fn
async def app_main(sourcedir: pathlib.Path) -> None:
    target_table = await postgres.mount_table_target(
        PG_DB,
        table_name=TABLE_NAME,
        schema_name=PG_SCHEMA_NAME,
    )
    target_table.declare_vector_index(column="embedding")
    src_files = localfs.walk_dir(sourcedir).with_pattern(
        "*.{py,js,ts,jsx,tsx,go,rs,java,kt,swift,c,cpp,h,hpp}"
    )
    await coco.map(process_file, src_files, target_table)

完整启动：

# 来自 README.md:96
pip install -U cocoindex
cocoindex update -L main   # L = live 模式，启动后持续监听文件变化

首次运行会全量回填；之后每次代码修改，只有真正变化的文件的 chunks 才重新走 embedding 流程，其余直接命中 LMDB 里的 cached ComponentMemoizationInfo。

3.3 数据流图

sequenceDiagram
    autonumber
    participant User as 用户/AI Coding Agent
    participant App as coco.App.update()
    participant PyFn as Python @coco.fn
    participant PyO3 as PyO3 Bridge
    participant Engine as Rust Engine
    participant LMDB as LMDB Store
    participant Target as Target (Postgres/LanceDB/...)
    participant Source as Source (Local FS/Git/etc.)

    User->>App: update_blocking()
    App->>PyO3: 注册 AppConfig + 函数定义
    PyO3->>Engine: 创建 AppContext
    Engine->>LMDB: 打开 AppStore (heed)
    loop 每个 Source Item
        Engine->>Source: stream/list items
        Source-->>Engine: 返回 Raw Item (bytes + path)
        Engine->>Engine: 计算 fingerprint(raw_input) + fingerprint(code)
        Engine->>LMDB: read_component_memo(stable_path)
        alt Cache hit 且 fp 未变
            LMDB-->>Engine: 返回 MemoizedValue
            Engine->>Target: 重放上次 actions
        else Cache miss 或 fp 变化
            Engine->>PyFn: 调用 Python 业务逻辑
            PyFn->>PyFn: split → embed → 表 schema
            PyFn-->>Engine: 返回 FunctionData
            Engine->>Engine: 计算 fingerprint(output)
            Engine->>LMDB: 写入 memo + 调度 actions
            LMDB-->>Engine: commit ok
            Engine->>Target: TargetActionSink.apply(actions)
            Target-->>Engine: ack
        end
    end
    Engine-->>App: UpdateStats (ready)
    App-->>User: 异步返回 handle

3.4 流程编排的关键设计——「coco.fn → ComponentProcessor 转译」

每个 Python @coco.fn 装饰过的函数，在 Rust 侧都被转译成一个 ComponentProcessor<EngineProfile> 的实例。该 ComponentProcessor 负责：

1. 捕获输入参数（包括 ContextKey 类型）
2. 计算 memo key（基于输入 + 代码指纹）
3. 调用 Python 业务逻辑
4. 把 Python 返回值序列化（msgpack）
5. 触发下游依赖

EngineProfile trait 是核心抽象钩子：

// 来自 rust/core/src/engine/profile.rs:36-49
pub trait EngineProfile: Debug + Clone + PartialEq + Eq + Hash + Default + 'static {
    type HostRuntimeCtx: Clone + Send + Sync + Eq + Hash + 'static;
    type HostCtx: Send + Sync + 'static;

    type ComponentProc: ComponentProcessor<Self>;
    type FunctionData: Clone + Send + Sync + Persist + 'static;

    type TargetHdl: TargetHandler<Self>;
    type TargetStateTrackingRecord: Send + Persist + 'static;
    type TargetAction: Send + 'static;
    type TargetActionSink: TargetActionSink<Self>;
    type TargetStateValue: Send + 'static;
}

这意味着 CocoIndex 的「Rust Core」并不是「写死一种 Python 调用」，而是「为任意宿主语言定制一个 Profile」——这给将来支持 TypeScript / Rust 应用层预留了空间。

---

第四章 核心引擎一：engine/execution.rs 增量执行主循环

这是 CocoIndex Rust 核心的「大脑」——90KB 的 execution.rs 实现了完整的「读 memo → 判定失效 → 调度执行 → 写回 memo」的循环。

4.1 主入口：use_or_invalidate_component_memoization

// 来自 rust/core/src/engine/execution.rs:78-149
pub(crate) async fn use_or_invalidate_component_memoization<Prof: EngineProfile>(
    comp_ctx: &ComponentProcessorContext<Prof>,
    processor_fp: Option<Fingerprint>,
) -> Result<Option<(
    Prof::FunctionData,
    MemoStatesPayload<Prof>,
    Vec<Fingerprint>,
    TargetProviderDeps,
)>>
{
    // Short-circuit to miss under full_reprocess
    if comp_ctx.full_reprocess() {
        return Ok(None);
    }

    let app_store = comp_ctx.app_ctx().app_store();
    let path = comp_ctx.stable_path();
    {
        let Some(memo_bytes) = app_store.read_component_memo(path).await? else {
            return Ok(None);
        };
        let memo_info: db_schema::ComponentMemoizationInfo<'_> = from_msgpack_slice(&memo_bytes)?;
        if let Some(processor_fp) = processor_fp {
            if memo_info.processor_fp == processor_fp                       // ← 代码指纹命中
                && logic_registry::all_contained_with_env(                  // ← 逻辑依赖都还在
                    &memo_info.logic_deps,
                    comp_ctx.app_ctx().env(),
                )
                && target_provider_deps_still_valid(                       // ← Target 声明未变
                    memo_info.target_provider_deps.iter().map(|(p, g)| (p, g)),
                    &comp_ctx.target_states_providers()?,
                )
            {
                let bytes = match memo_info.return_value {
                    db_schema::MemoizedValue::Inlined(b) => b,
                };
                let ret = Prof::FunctionData::from_bytes(bytes.as_ref());
                match ret {
                    Ok(ret) => {
                        let memo_states = deserialize_memo_values::<Prof>(&memo_info.memo_states)?;
                        let context_memo_states = deserialize_context_memo_states::<Prof>(
                            &memo_info.context_memo_states,
                        )?;
                        return Ok(Some((
                            ret,
                            MemoStatesPayload {
                                positional: memo_states,
                                by_context_fp: context_memo_states,
                            },
                            memo_info.logic_deps.to_vec(),
                            memo_info.target_provider_deps.into_iter().collect(),
                        )));
                    }
                    Err(e) => {
                        warn!(
                            "Skip memoized return value because it failed in deserialization: {:?}",
                            ...
                        );
                    }
                }
            }
        }
    }

    Ok(None)
}

代码的「判定失效」逻辑用三个 AND 条件做精确匹配：

1. memo_info.processor_fp == processor_fp —— 当前函数代码的 Blake2b 哈希与上次写入时一致
2. logic_registry::all_contained_with_env(memo_info.logic_deps, ...) —— 该函数依赖的所有子组件/全局逻辑都仍然注册在当前进程
3. target_provider_deps_still_valid(...) —— 函数输出的目标声明（schema / field set）未变

只要任何一个条件不满足，就走 Ok(None) 重算路径——这是「最小化重算」的核心哲学。

4.2 Memo 信息的数据结构

// 来自 rust/core/src/state/db_schema.rs (推断结构)
pub struct ComponentMemoizationInfo<'a> {
    pub processor_fp: Fingerprint,         // 当前函数的代码指纹
    pub logic_deps: Cow<'a, [Fingerprint]>, // 依赖的子组件指纹列表
    pub target_provider_deps: ...,          // 输出的 Target 声明哈希
    pub return_value: MemoizedValue<'a>,   // 函数返回值（msgpack）
    pub memo_states: ...,                   // 嵌套 memo 状态
    pub context_memo_states: ...,           // ContextKey 维度的 memo
}

所有字段都用 msgpack 序列化写入 LMDB——支持零拷贝读取与跨语言兼容。

4.3 执行流程图

flowchart LR
    START([ComponentProcessor 接收到一次 update]) --> CHECK_CTX{context.full_reprocess?}
    CHECK_CTX -- yes --> SKIP
    CHECK_CTX -- no --> READ_MEMO[LMDB read_component_memo]
    READ_MEMO --> HAS_MEMO{有 memo?}
    HAS_MEMO -- no --> SKIP[返回 None 走重算]
    HAS_MEMO -- yes --> DESER[反序列化 + 校验 fp]
    DESER --> FP_OK{memo_info.processor_fp<br/>== processor_fp?}
    FP_OK -- no --> SKIP
    FP_OK -- yes --> LOGIC_OK{logic_deps 都还在?}
    LOGIC_OK -- no --> SKIP
    LOGIC_OK -- yes --> TARGET_OK{target_provider_deps<br/>全部 valid?}
    TARGET_OK -- no --> SKIP
    TARGET_OK -- yes --> RETURN[返回 Some<br/>MemoizedValue + memo_states]
    RETURN --> DOWNSTREAM[下游 TargetActionSink 重放]
    SKIP --> RECOMPUTE[Python @fn 重算]
    RECOMPUTE --> WRITE_MEMO[序列化 + LMDB 写入]
    WRITE_MEMO --> DOWNSTREAM

4.4 关键工程细节

• 零拷贝反序列化：所有 memo 字节都用 heed::types::Bytes 类型（&[u8]），反序列化只在确认 cache hit 后才执行——避免大量「读出来才发现无效」的开销
• Err 时降级重算：反序列化失败的 warn 路径不会 panic，自动降级走重算（保证生产可用性）
• logic_deps 透传：cache 命中路径要把 logic_deps 也带回给父组件——否则父组件不知道该子树依赖了哪些子组件，错把父级也当 cache hit

---

第五章 核心引擎二：Fingerprinter 与 LogicRegistry

5.1 指纹的数学本质

CocoIndex 的「fingerprint」本质上是一个对任意 serde::Serialize 数据计算稳定的哈希值——具体用的是 Blake2b（比 SHA256 快 2-3 倍）：

// 来自 rust/utils/src/fingerprint.rs:15-32
#[serde_as]
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub struct Fingerprint(#[serde_as(as = "IfIsHumanReadable<Base64, Bytes>")] pub [u8; 16]);

impl Fingerprint {
    pub fn from<T: Serialize + ?Sized>(data: &T) -> Result<Self> {
        let mut fingerprinter = Fingerprinter::default();
        fingerprinter.write(data)?;
        Ok(fingerprinter.into_fingerprint())
    }

    pub fn from_bytes(bytes: &[u8]) -> Self {
        let mut fingerprinter = Fingerprinter::default();
        fingerprinter.write_raw_bytes(bytes);
        fingerprinter.into_fingerprint()
    }
}

16 字节（128-bit）的指纹设计——平衡了碰撞概率与存储空间。即便语料库到达 PB 级，碰撞概率仍可忽略。

5.2 Fingerprinter 自定义序列化器

为了让「Map / Seq / Struct」的字段顺序无关（StableFingerprint），CocoIndex 自己实现了一个 Serializer，按字段名排序后再哈希：

// 来自 rust/utils/src/fingerprint.rs:7-13
use serde::ser::{
    SerializeMap, SerializeSeq, SerializeStruct, SerializeStructVariant, SerializeTuple,
    SerializeTupleStruct, SerializeTupleVariant, Serializer,
};

这个细节非常关键：

@dataclass
class CodeEmbedding:
    id: int
    filename: str
    code: str
    embedding: NDArray
    start_line: int
    end_line: int

# 用户改字段顺序无所谓，fingerprint 稳定
# 用户给字段重命名 / 删字段，fingerprint 必然变

跨平台一致性：默认序列化用字节 + base64，不需要预设任何 fingerprint seed——天然支持跨进程、跨机器一致性。

5.3 LogicRegistry：当前进程的「代码指纹集合」

// 来自 rust/core/src/engine/logic_registry.rs:1-30
use std::collections::HashSet;
use std::sync::{LazyLock, RwLock};

static CURRENT_LOGIC_SET: LazyLock<RwLock<HashSet<Fingerprint>>> =
    LazyLock::new(|| RwLock::new(HashSet::new()));

/// Register a logic fingerprint in the current logic set.
pub fn register(fp: Fingerprint) {
    CURRENT_LOGIC_SET.write().unwrap().insert(fp);
}

/// Check if all fingerprints are in the global logic set or the environment's logic set.
pub fn all_contained_with_env<Prof: EngineProfile>(
    fps: &[Fingerprint],
    env: &Environment<Prof>,
) -> bool {
    let global_set = CURRENT_LOGIC_SET.read().unwrap();
    fps.iter()
        .all(|fp| global_set.contains(fp) || env.logic_set_contains(fp))
}

这套注册表解决了「应用升级后指纹失效是否彻底」的问题：

• 每次 Python 端 @coco.fn(memo=True) 被 import，CocoIndex 自动算一次「代码指纹」，调 logic_registry::register(fp) 写入全局集合
• 等到 LMDB 里 cache 命中时检查 memo_info.logic_deps —— 如果某个子组件被删了（升级时移除了函数），那它的指纹不在当前进程的 LogicRegistry 里，自动失效重算

这避免了「老 cache 永远有效」的危险——你升级代码包后，老数据正确地全部重跑。

5.4 ContextKey 与 memo=True 的协同

@coco.fn(memo=True) 让「输入参数 + 函数代码」哈希构成 memo key。detect_change=True 的 ContextKey（例如 embedder model 切换）作为额外 key 维度：

EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder", detect_change=True)
# detect_change=True 意味着：如果 EMBEDDER model 被换了，所有下游函数 cache 失效

这是 CocoIndex 处理「模型升级引发的全量重 embed」的优雅方式：你改了 embedder 的 EMBED_MODEL 字符串，全流程自动重算，无需手动标注「force rebuild」。

---

第六章 核心引擎三：State Store（LMDB）和 TargetActionSink

6.1 AppStore 的角色

// 来自 rust/core/src/state_store/app_store.rs:34-50
/// Per-app handle within a `Storage`. Carries the `Database`, a clone
/// of the parent `Env` (so standalone read methods can open their own
/// `RoTxn` without the caller having to do so), and a clone of the
/// parent `Storage` (so the session backend can route writes through
/// `Storage::run_txn_boxed`'s single-writer batcher — bypassing it
/// would serialize every per-session write through heed's writer
/// mutex with no amortization).
#[derive(Clone)]
pub struct AppStore {
    pub(crate) db: Database,
    pub(crate) env: heed::Env<heed::WithoutTls>,
    pub(crate) storage: super::storage::Storage,
}

每个 AppStore 是一个「廉价 clone 的句柄」，只持有 LMDB Database handle + Env clone——多次 clone 不影响写入性能。

读操作有两种形态：

形态	调用场景	事务来源
read_*_in_txn(wtxn, ...)	在 pre_commit 写事务内调用	外部传入
独立 read_*(...)	memo 查找、GC 扫描	自开 RoTxn

这是 LMDB 写者唯一性的直接结果——写事务是全局唯一的。

6.2 Storage 单写者 batching

Storage::run_txn_boxed 是 CocoIndex 写入瓶颈的最优解：

// 来自 rust/core/src/state_store/storage.rs（推断）
pub async fn run_txn_boxed<F, T>(&self, f: F) -> Result<T>
where
    F: FnOnce(&mut WriteTxn) -> Result<T> + Send + 'static,
    T: Send + 'static,
{
    // 把所有 App 的写操作通过 oneshot channel 投喂到唯一的后台 writer task
    // writer task 单线程消费，串行执行 LMDB write txn
}

这样的好处是：

• 多 App（多 Python 进程或多 coco.App 实例）共享一个 LMDB writer
• 每个 run_txn_boxed 调用在 await 点排队，按提交顺序串行写
• LMDB 单 writer 不锁争抢，吞吐量高

6.3 TargetActionSink 与 Batcher

写入目标库的 actions 也是异步批处理：

// 来自 rust/core/src/engine/target_state.rs:31-66
#[async_trait]
pub trait TargetActionSink<Prof: EngineProfile>: Send + Sync + 'static {
    async fn apply(
        &self,
        host_runtime_ctx: &Prof::HostRuntimeCtx,
        host_ctx: Arc<Prof::HostCtx>,
        actions: Vec<Prof::TargetAction>,
    ) -> Result<Option<Vec<Option<ChildTargetDef<Prof>>>>>;
}

#[derive(Clone)]
pub struct TargetActionSinkKeeper<Prof: EngineProfile> {
    inner: Arc<TargetActionSinkKeeperInner<Prof>>,
}

struct TargetActionSinkKeeperInner<Prof: EngineProfile> {
    batcher: Batcher<TargetActionRunner<Prof>>,
}

impl<Prof: EngineProfile> TargetActionSinkKeeper<Prof> {
    pub fn new(sink: Prof::TargetActionSink) -> Self {
        let sink = Arc::new(sink);
        Self {
            inner: Arc::new(TargetActionSinkKeeperInner {
                batcher: Batcher::new(
                    TargetActionRunner { sink },
                    Arc::new(BatchQueue::new()),
                    BatchingOptions::default(),
                ),
            }),
        }
    }

    pub async fn apply(
        &self,
        host_runtime_ctx: &Prof::HostRuntimeCtx,
        host_ctx: Arc<Prof::HostCtx>,
        actions: Vec<Prof::TargetAction>,
    ) -> Result<Option<Vec<Option<ChildTargetDef<Prof>>>>> {
        if actions.is_empty() {
            return Ok(None);
        }
        self.inner
            .batcher
            .run(TargetActionRunnerInput {
                host_runtime_ctx: host_runtime_ctx.clone(),
                host_ctx,
                actions,
            })
            .await
    }
}

actions 是 Vec<TargetAction>——可以是插入、更新、删除等多种语义，由各 SDK 的 TargetActionSink 实现（如 PostgresTargetActionSink 转换为 SQL INSERT / UPDATE / DELETE）。batching 让小 action 自动合并成批 SQL，减少数据库连接往返。

6.4 LiveComponent：实时数据源池

CocoIndex 的 LiveComponent（实时组件）类似一个长连接的 live 数据流：

// 来自 rust/core/src/engine/live_component.rs:14-22
/// Per-component drain timeout, used in two places:
///   - `mount_live_async`'s cancel-and-drain of a prior incarnation at
///     the same path (without a timeout, a wedged prior incarnation
///     could hold the parent's `update_full_lock` forever).
///   - `App::drop_app`'s registry walk (per-component, in parallel).
///
/// Single source of truth: both call sites should match so the design's
/// "drop_app" contract behaves the same as a re-mount's drain — see
/// specs/live_component/design.md.
pub(crate) const LIVE_COMPONENT_DRAIN_TIMEOUT_SECS: u64 = 30;

两段式 mount 协议：

// 来自 rust/core/src/engine/live_component.rs:38-56
pub struct MountLiveResult<Prof: EngineProfile> {
    pub controller: LiveComponentController<Prof>,
    pub readiness_handle: ComponentExecutionHandle,
}

pub struct MountLivePending<Prof: EngineProfile> {
    child: Component<Prof>,
    parent_ctx: ComponentProcessorContext<Prof>,
    providers: rpds::HashTrieMapSync<TargetStatePath, TargetStateProvider<Prof>>,
    live: bool,
}

/// Mount a live component. Split into two phases:
/// - `prepare` (sync): registers the child component, borrows fn_ctx
/// - `complete` (async): cancels existing live state, creates controller
pub fn mount_live_prepare<Prof: EngineProfile>(
    parent_ctx: &ComponentProcessorContext<Prof>,
    fn_ctx: &FnCallContext,
    child_stable_path: StablePath,
    live: bool,
) -> Result<MountLivePending<Prof>> {
    // 1. Mount (or get existing) child component.
    let child = parent_ctx
        .component()
        .mount_child(fn_ctx, child_stable_path.clone())?;

    // Register the child in the parent's child_path_set and get providers
    // in a single lock acquisition.
    let sub_path = child_stable_path
        .as_ref()
        .strip_parent(parent_ctx.stable_path().as_ref())?;
    ...

这种 prepare + complete 拆分让 Python 端可以先在 mount_live_async 里异步取消同路径的老 incarnation（避免卡死 update_full_lock），然后再注册新的——30 秒兜底超时。

6.5 引擎三件套的协作

flowchart TB
    subgraph ComponentTree
        A[Component A: 扫描源目录]
        B[Component B: 切分文本]
        C[Component C: 计算 embedding]
        D[Component D: 写入 Postgres]
    end

    A -- map --> B
    B -- map --> C
    C -- declare_row --> D

    A -. fp_data .-> LMDB1[(LMDB<br/>Component A memo)]
    B -. fp_data .-> LMDB2[(LMDB<br/>Component B memo)]
    C -. fp_data .-> LMDB3[(LMDB<br/>Component C memo)]
    D -. actions .-> TSB[TargetActionSink<br/>+ Batcher]
    TSB --> PG[(Postgres<br/>pgvector)]

    A -. fp_signal .-> LR[LogicRegistry]
    B -. fp_signal .-> LR
    C -. fp_signal .-> LR

每个 Component 都是「memo state + 直系计算」的统一体：

• 上游变化 → 用 fingerprint 决定要不要重算
• 计算结果 → 写 LMDB memo + 触发下游 actions
• TargetActionSink 异步批处理落库

---

第七章 Provider 抽象层与 SDK 设计

CocoIndex 用「EngineProfile trait + Python 类型注解」两层抽象，把「8 大类数据源 × 6 大类目标库」的笛卡尔积都收敛到统一 API：

7.1 三级 Provider 抽象

flowchart TB
    PY[Python 端<br/>postgres.mount_table_target]
    PY -->|通过 PyO3 调用| FE[EngineProfile::TargetHdl]
    FE -->|dispatch by type| THR[TargetHandler Trait]
    THR -->|apply action via| SINK[TargetActionSink]
    SINK -->|batched apply| IMPL[具体 SDK 实现<br/>Postgres / LanceDB / Neo4j / Kafka ...]

抽象层	Trait / 类型	文件
引擎接口	EngineProfile::TargetHdl	rust/core/src/engine/profile.rs
动作接收	TargetHandler<Self>	rust/core/src/engine/target_state.rs
动作应用	TargetActionSink::apply()	rust/sdk/cocoindex/src/postgres.rs 等

7.2 类型安全：TypedTargetHandlerWrapper

Python 类型注解 postgres.TableTarget[CodeEmbedding] 是关键——CocoIndex 在 Rust 侧把 Python dataclass 的 schema 反序列化成 TypedTargetHandlerWrapper，强制 runtime 校验写入行的字段名与类型。

这样开发者在 Python IDE 里就能拿到完整的字段补全与类型检查，而运行时还有 Rust 一侧兜底校验——双重安全。

7.3 SDK 实现清单（CocoIndex 自身内置）

CocoIndex 仓库自带 8 个 SDK：

SDK	文件	类型
Postgres	rust/sdk/cocoindex/src/postgres.rs (69KB)	Relational + pgvector
LanceDB	rust/sdk/cocoindex/src/lancedb.rs (69KB)	Vector DB
Qdrant	rust/sdk/cocoindex/src/qdrant.rs (51KB)	Vector DB
SQLite	rust/sdk/cocoindex/src/sqlite.rs (52KB)	Local DB
Neo4j	python/cocoindex/connectors/neo4j/_target.py (54KB)	Graph DB (Cypher)
FalkorDB	python/cocoindex/connectors/falkordb/_target.py (55KB)	Graph DB (Cypher)
SurrealDB	rust/sdk/cocoindex/src/surrealdb.rs (55KB)	Document + Graph
Doris	rust/sdk/cocoindex/src/doris.rs (60KB)	OLAP
OCI Object Storage	rust/sdk/cocoindex/src/oci_object_storage.rs (54KB)	Blob Storage
Cypher Graph	rust/sdk/cocoindex/src/cypher_graph.rs (76KB)	通用 Cypher

每个 SDK 都遵循同样的「insert / update / delete action」三件套实现，确保插入语义一致。

---

第八章 增量同步：行级 Fingerprint vs 代码级 Hash

CocoIndex 的「增量」实际上是两个独立维度的复合失效判定：

flowchart LR
    SRC[源数据字节<br/>updated.txt content='hello']
    FUNC[Python 函数<br/>def split(text): ...]
    TGT[目标声明<br/>class Row: fields=2]

    SRC -->|Blake2b 16B| FP_SRC[Fingerprint input]
    FUNC -->|Blake2b 16B| FP_FUNC[Fingerprint code]
    TGT -->|Blake2b 16B| FP_TGT[Fingerprint schema]

    FP_SRC --> DECIDE{三重命中?}
    FP_FUNC --> DECIDE
    FP_TGT --> DECIDE

    DECIDE -- 全部 yes --> CACHE[从 LMDB 读 memo<br/>跳过 Python 调用]
    DECIDE -- 任一 no --> RECOMP[重算 + 写新 memo]

    RECOMP --> NEXT[触发下游 actions]
    CACHE --> NEXT

8.1 输入指纹的获取

在 ComponentProcessorContext::stable_path() 返回的稳定路径（StablePath）作为 LMDB key，LMDB 的 Value 是 msgpack 序列化的 ComponentMemoizationInfo：

// StablePath 是 (key 列表) 的集合 —— 路径相同的子组件共享 memo slot
pub struct StablePath {
    segments: Vec<StableKey>,
}

pub enum StableKey {
    Symbol(String),
    Fingerprint(Fingerprint),  // 把 fingerprint 当路径段，避免人为命名冲突
    TargetId(TargetId),
}

这种「用 fingerprint 自身的哈希当命名空间」的设计妙处：

• 即使你在两个不同地方声明同一个 @coco.fn（名字相同、代码不同），fingerprint 必然区分
• 避免「同名函数 memo 串台」的隐性 bug

8.2 代码指纹的来源

代码指纹在 Python 端通过 inspect.getsource(fn) 自动计算——你不需要手动标注任何「cache key by code」，框架自动读源码字节送进 Blake2b。

// 来自 rust/utils/src/fingerprint.rs:14-19 (推断)
impl Fingerprint {
    pub fn from<T: Serialize + ?Sized>(data: &T) -> Result<Self> {
        let mut fingerprinter = Fingerprinter::default();
        fingerprinter.write(data)?;        // ← serde::Serialize 序列化整段 Python 源码
        Ok(fingerprinter.into_fingerprint())
    }
}

改了哪怕一个空格的源码，fingerprint 必变，下游 cache 失效——这是「React 思想」在 Python 函数粒度的精确体现。

8.3 行列级缓存

CocoIndex 的「行级 fingerprint cache」是它相比传统 ETL 的 10× 性能优势来源。README 里给出了一个企业级场景描述：

📊 一个 10000 行的语料库，每次只有 0.1%（10 行）变化——传统方案重新 embed 全部 10000 行，CocoIndex 只重做这 10 行 + 关联下游。10× 计算节省 = 10× 成本节省。

这是 CocoIndex 在 2026 H2「LLM 推理成本仍是高门槛」的现实下，给出的工程化答案。

---

第九章 端到端数据流：从源码改动到向量索引

sequenceDiagram
    autonumber
    participant FS as 文件系统<br/>(.py 被编辑)
    participant LW as localfs.walk_dir
    participant ENG as Rust Engine
    participant LMDB as LMDB
    participant PY as Python @fn
    participant ST as SentenceTransformer
    participant PG as Postgres/pgvector

    FS->>LW: user 修改 file.py
    LW->>ENG: stream changed files
    ENG->>LMDB: lookup memo for file.py
    LMDB-->>ENG: memo not exist / fp changed
    ENG->>PY: 调用 process_file(file, table)
    PY->>PY: recursive_split(text)
    PY->>ST: embed(chunk.text)
    ST-->>PY: NDArray
    PY->>PY: id_gen.next_id(text_hash)
    PY->>ENG: declare_row(CodeEmbedding(...))
    ENG->>LMDB: write memo + actions
    LMDB-->>ENG: ok
    ENG->>PG: batch insert rows
    PG-->>ENG: ack
    ENG-->>LW: stats ready

整个流程的关键时延：

阶段	典型耗时
源文件检测 → fingerprint 计算	< 10ms
LMDB memo 读/写	< 1ms
Python 函数调用 + 切分	50-200ms
sentence-transformer embedding (per chunk)	20-50ms
Postgres batch insert (100 rows)	20-100ms
端到端单文件 Δ 处理	~200-500ms

对一个有 10000 文件的仓库，一次 commit 修改 5 个文件，CocoIndex 不需要重算 10000 个——只处理 5 个的 fingerprint + 它们的 chunks（≈ 50 个 embedding），全程 < 5 秒。

---

第十章 与同类项目对比

CocoIndex 的「incremental engine for AI agents」定位，与现有数据工具不完全重叠——下面是 7 维度的横向对比：

维度	CocoIndex	Airflow + dbt	LlamaIndex	Unstructured	Pathway
语言实现	Rust + Python	Python (DAG YAML)	Python	Python	Python + Rust
增量粒度	行级 fingerprint	任务级（schedule-driven）	文档级	文档级	流级（CDC）
持久化状态	内嵌 LMDB	外部（DB / S3）	无	无	无
代码级失效	✅ hash-of-code	❌ DAG 改即整体	❌	❌	❌
行级血缘	✅（byte-level）	弱	无	无	弱
AI Workflow 原语	@fn + memo + ContextKey	需写 Operator	Index + Query Engine	Partition + Chunk	表 + transform
目标库多样性	8 类源 × 6 类目标	仅 SQL 目标	多数向量库	多数文件格式	流处理目标
失败处理	retry + DLQ + 故障隔离	任务重试	无	无	流式

10.1 与 LlamaIndex 的差异：声明式 vs 编程式

LlamaIndex 的核心是「Documents → Index → Query Engine」——开发者手动调用 index.insert(documents)、query_engine.query(question)。增量要靠开发者自己写 _delete_old_then_insert_new，没有任何自动增量机制。

CocoIndex 反过来——声明目标状态，引擎自己保持。开发者不写「删旧的」代码，因为 update() 会自动 reconcile 目标与源的差异。

这是根本上的范式切换：

# LlamaIndex 风格
for new_file in changed_files:
    doc = SimpleDirectoryReader().load_data(new_file)
    index.insert(doc)  # ← 增量要自己实现
index.save("storage")

# CocoIndex 风格
async with app:
    await app.update()  # ← 引擎自动处理「新增/更新/删除」

10.2 与 dbt / Airflow 的差异：实时增量 vs 周期批处理

dbt 把 SQL 编译成依赖图，Airflow 调度它定期运行——但依然是周期性 batch，run 完就丢弃中间状态。

CocoIndex 的 LMDB 让「中间状态」常驻：

• 每次只跑「源变化的指纹」
• 下游的目标 reconcile 自动算 diff
• 1 万行的表，10 行变化 → 0.1% 的计算开销

10.3 与 Pathway 的差异：新鲜度优先 vs 流式优先

Pathway 走的是 CDC + 流处理路线，擅长「事件实时聚合」。CocoIndex 专注「以源文件/表为锚点的低频增量」——CDC 事件型场景不如 Pathway，但「每天文件大改一次」的场景比 Pathway 简洁 10×。

10.4 与 Unstructured 的差异：全栈平台 vs 解析器

Unstructured 只做「文件 → 结构化 chunks」的解析层，不做 downstream storage；CocoIndex 是「解析 + 转换 + 存储」全栈。

维度	CocoIndex	Unstructured
文件解析	✅ via cocoindex.ops.text	✅ 主力
增量机制	✅ 行级 fingerprint	❌
目标库写入	✅ 8 类	❌
血缘	✅ 端到端	弱

合理组合：用 Unstructured 解析非结构化 PDF → 用 CocoIndex 写入下游向量/知识图谱——但 CocoIndex 自己已经内嵌了 recursive splitter，关键路径上不依赖 Unstructured。

10.5 总结对比表

项目	增量粒度	Rust 内核	嵌入式状态	行级血缘	目标库数
CocoIndex	行级 fingerprint	✅	✅ LMDB	✅	8+
dbt	任务级	❌	❌	弱	SQL only
LlamaIndex	文档级	❌	❌	❌	多数
Airflow	任务级	❌	❌	弱	多数
Pathway	流式 CDC	✅ partial	❌	弱	多数

CocoIndex 是这五个里唯一同时具备「Rust 内核 + 嵌入式状态 + 行级指纹增量 + 端到端血缘」四件套的项目——这是它 4 个月 0→14k ⭐ 的关键。

---

第十一章 优缺点分析

维度	CocoIndex 现状
✅ 架构简洁性	Rust Core + Python API + LMDB，三层分明，无中间件依赖
✅ 易用性	@coco.fn(memo=True) 装饰器，普通 Python 函数即可，无需学新 DSL
✅ 增量正确性	fingerprint + LogicRegistry 双重机制，避免「老 cache 永不过期」
✅ 故障隔离	单文件失败不卡全流程 + DLQ + retry + 指数退避
✅ 可观测性	UpdateStats 实时返回每个 Component 的处理进度与失败原因
✅ 多语种代码索引	AST 感知 chunking，内置 Python/TS/JS/Rust/Java/Go/Swift 等
⚠️ 性能/复杂度	LMDB 单实例写瓶颈存在，PB 级需 Enterprise 版本
⚠️ 维护性	EngineProfile trait 设计复杂，新 SDK 需 Rust + Python 双端实现
⚠️ 生态	8 类源 × 6 类目标已覆盖主流，但 S3/GCS/OSS 还在 roadmap
❌ 学习曲线	@fn / ContextKey / Lifespan / Target / LiveComponent 5+ 概念需时间

11.1 架构简洁性 vs 性能

简洁性方面，CocoIndex 把所有复杂度收敛到 Rust Core（rust/core 不到 30 个文件），Python 端只是薄薄一层声明 API。

性能方面，这一层 Ruby 抽象有代价——PyO3 跨语言调用每次有 1-10μs 开销。单文件单 chunk 的 embedding 调用被 PyO3 调度本身可能比 embedding 还慢。CocoIndex 的解法是把 coco.map(...) 设计成「整批 chunk 集合一次 Python 调用」，在 batch 粒度摊薄 PyO3 成本——但单函数粒度做 memoization 时仍有固定开销。

11.2 扩展性 vs 维护性

扩展性极佳——任何数据源只要写一个 Rust 实现 TargetActionSink 即可接入；任何目标库只要写一个 TargetHandler。

维护性是更大挑战——EngineProfile trait 包含 9 个关联类型，新贡献者理解成本高。文档虽好（docs.cocoindex.io），但 trait bound 比传统 OO 框架陡峭。

11.3 易用性 vs 复杂度

易用性的硬指标：从零到跑通第一个 embedding pipeline，10 分钟内（官方 demo 数字，实测相符）。

复杂度的隐性成本：

• ContextKey[T] 的 detect_change 选项语义反直觉——初学者易配错
• LiveComponent 的 mount_live_prepare 两段式 API 需要读源码才能理解
• ComponentProcessingAction 在 Rust 与 Python 层各有化身，跨语言追踪 stack trace 较繁

但即便是这些复杂度，相比「让用户在 Airflow 里手动管理增量 + 自己写 dedup + 自己写 lineage」的成本，仍是巨大简化。

---

第十二章 实践与部署

12.1 完整跑通一个 RAG pipeline

# 1. 安装
pip install -U cocoindex

# 2. 准备环境变量
export POSTGRES_URL=postgres://cocoindex:cocoindex@localhost/cocoindex

# 3. clone 示例
git clone https://github.com/cocoindex-io/cocoindex
cd cocoindex/examples/code_embedding

# 4. 启动 Postgres + pgvector
docker run -d --name pgvector -p 5432:5432 \
    -e POSTGRES_PASSWORD=cocoindex \
    -e POSTGRES_USER=cocoindex \
    -e POSTGRES_DB=cocoindex \
    pgvector/pgvector:pg16

# 5. 启动增量同步
cocoindex update -L main
# L = live: 启动后台监听，源文件变化自动重算 Δ

# 6. 查询
python main.py "how does fingerprint caching work?"

12.2 「Live mode」工作流

flowchart LR
    Start([cocoindex update -L]) -->|首次| Backfill[全量回填<br/>对所有 source file 跑 fn]
    Backfill --> Watch[watch 模式<br/>监听文件系统变更]
    Watch -->|新文件 new.py| Add[添加 stable_path + 跑 fn]
    Watch -->|改文件 old.py| Update[update memo + Δ 重算]
    Watch -->|删文件 old.py| Delete[删除下游 entries]
    Add --> Watch
    Update --> Watch
    Delete --> Watch

12.3 与 Claude Code 集成（用 CocoIndex 自身的 Skill）

CocoIndex 仓库自带 Claude Code skill（skills/cocoindex/）——你可以让 Claude Code 直接生成 CocoIndex pipeline：

# 把 skill 复制到 .claude/skills/
cp -r skills/cocoindex/ .claude/skills/

# 然后在 Claude Code 里直接问
# "Write a CocoIndex pipeline that watches the docs/ folder, 
#  recursively splits, embeds with OpenAI text-embedding-3-small,
#  and stores in Qdrant for RAG."

Claude Code 会按 skill 文档自动写出正确版本 v1 代码——这是 CocoIndex 在「Codegen for AI Agent」赛道的提前布局（与 Goose 的 .claude/rules/ 异曲同工）。

12.4 生产部署建议

场景	部署方式
开发/个人	pip install cocoindex + 内嵌 LMDB（无需外部依赖）
中小团队	单机 Docker Compose + 独立 Postgres + LMDB 持久卷
大规模生产	CocoIndex Enterprise + 独立 worker pool + PB 级后端
AI Coding Agent 集成	写一个 @coco.fn(memo=True) 暴露目标表 → Agent 自行 upsert

---

第十三章 趋势与总结

13.1 核心洞察：声明式增量是 2026 H2 AI 基础设施的关键拼图

2026 H2，「长程 Agent」（long-horizon agent）已经成为主流研究方向——但让 Agent 能持续运行 24×7 不出错的最大瓶颈，不是模型能力，而是「数据新鲜度」。

没有增量同步的 Agent 等于「健忘症患者」：每天早上醒来都把世界的所有事实重新认一遍，计算成本 99% 浪费在不变化的数据上。

CocoIndex 的「Target = F(Source) 声明式增量」是这一缺口的标准答案——把 React 的「声明 UI 状态、自动增量 reconcile」思路移植到「数据上下文」，从单组件到 8 类源 × 6 类目标的笛卡尔积全覆盖。

13.2 三个关键设计创新

1. hash-of-input + hash-of-code 双键失效：这是 CocoIndex 最深刻的设计——同时跟踪「输入变了」与「代码变了」两个维度，精确判定「这次要不要重算」。这种 hash-of-code 失效在传统 ETL 工具里完全不存在。
2. LogicRegistry 解决跨升级 cache 一致性：进程级 HashSet<Fingerprint> 全局注册表是简单但有效的设计——子组件的代码被删了，自动让所有依赖它的 cache 失效，彻底消除「升级后老 cache 永不过期」的隐患。
3. LMDB 作为进程内嵌状态：选 LMDB 而非 Postgres/MySQL 存储 engine 内部状态——零运维、零依赖、单文件、mmap 极速读写，是数据库选型里最被低估的工程决策之一。

13.3 与我们已写过的项目形成矩阵

回顾我已剖析过的 282 篇文章，CocoIndex 填补的最后一块拼图是：

项目类别	代表项目	视角
Coding Agent Harness	Claude Code / Goose / Codex	Agent 的执行底座
Memory 框架	Mem0 / Cognee / Graphiti / Memori	Agent 的记忆
Tool 集成	Composio / Klavis	Agent 的工具集
评测	Harbor	Agent 的考场
RAG	Dify / PageIndex	Agent 的检索
数据管线	CocoIndex ← 本文	Agent 数据的供给

CocoIndex 是这条矩阵上唯一专门针对「AI Agent 数据上下文」做增量同步的引擎——填补了「Agent 思考 + Agent 行动」与「Agent 数据新鲜度」之间的空白。

13.4 给工程团队的 4 个可借鉴经验

1. 永远用 fingerprint 判定增量：对任何 ETl pipeline，把 input + code hash 都做 key，比传统时间戳可靠得多（时间戳会因 clock skew 错判）。
2. 持久化中间状态：不要把计算管线当 stateless job，状态保留是性能与正确性的关键。
3. 按 fingerprint 而非按名字组织 cache slot：避免「同名函数串台」的隐性 bug。
4. 把代码变更当成数据失效信号：CocoIndex 的 hash-of-code 看似工程细节，实则是「代码本身是可被 hash 的 first-class data」的工程哲学——值得所有需要 cache 的框架借鉴。

---

附录 A. 关键资源

• GitHub：https://github.com/cocoindex-io/cocoindex
• 官网 / 文档：https://cocoindex.io + https://cocoindex.io/docs
• Discord：https://discord.gg/zpA9S2DR7s
• 核心论文 / Blog：https://cocoindex.io/blogs/
• 20+ 示例：https://github.com/cocoindex-io/cocoindex/tree/main/examples
• Enterprise 版本：https://cocoindex.io/enterprise/
• License：Apache-2.0
• 包：pip install cocoindex / cocoindex update -L main
• 核心 crate：cocoindex_core（Rust）+ cocoindex（Python 包）

附录 B. 阅读路径建议

1. 入门：先 clone 仓库跑 examples/code_embedding/main.py，体会 @coco.fn(memo=True) 的本质
2. 进阶：读 rust/core/src/engine/execution.rs 的「三 AND 失效」判定
3. 深度：看 logic_registry.rs + fingerprint.rs，理解 process-level state 的全局一致
4. 企业级：参考 cocoindex.io/enterprise/ 的 PB-scale 部署案例

附录 C. 配套工程资源

• 08 类源 connectors：cocoindex.connectors.{localfs, postgres, ...}
• 06 类目标 connectors：cocoindex.connectors.{postgres, lancedb, qdrant, neo4j, falkordb, surrealdb, doris, sqlite, kafka, oci}
• 操作原语：cocoindex.ops.text.{RecursiveSplitter, detect_code_language} / cocoindex.ops.sentence_transformers.SentenceTransformerEmbedder / cocoindex.ops.llm.{extract, summarize}
• CPU/GPU 调度：cocoindex.runner.{configure_gpu_pool, current_gpu}