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：LLM ベースの抽出

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	クラウドネイティブなベクトル DB
Qdrant	-	Y	Y	ベクトル専用 DB
SurrealDB	-	Y	Y	グラフ＋ドキュメント DB
Apache Doris	-	Y	Y	分析 DB ＋ベクトル
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
• LLM 抽出 → 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) 向けです。v0 とは完全に異なる API を使用します。

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(...)); 状態はローカルの db パスに保存されます
CLI cocoindex setup	セットアップ手順は不要 — cocoindex update を実行するだけです (ライブモードには -L/--live)

これらの v0 シンボルに言及しているサードパーティのチュートリアルやモデルの記憶を読むときは、それらを無視し、代わりにこのスキルと references/api_reference.md のパターンを使用してください。