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: 依存関係の設定

必要に応じて、適切なエクストラを付けて 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 エクストラ: ローカルの埋め込みモデル用の SentenceTransformers ライブラリ
• colpali エクストラ: マルチモーダルなドキュメント/画像の埋め込み用の ColPali エンジン
• lancedb エクストラ: 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() を使用します
• 重要: 変換結果は必ず行のフィールドに代入してください - new_field = item["existing_field"].transform(...) のようなローカル変数ではなく、item["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 を読み込み、「Pattern 1: Simple Text Embedding」を参照してください

コード埋め込みの場合: references/flow_patterns.md を読み込み、「Pattern 2: Code Embedding with Language Detection」を参照してください

LLM による抽出 + ナレッジグラフの場合: references/flow_patterns.md を読み込み、「Pattern 3: LLM-based Extraction to Knowledge Graph」を参照してください

ライブ更新の場合: references/flow_patterns.md を読み込み、「Pattern 4: Live Updates with Refresh Interval」を参照してください

カスタム関数の場合: references/flow_patterns.md を読み込み、「Pattern 5: Custom Transform Function」を参照してください

再利用可能なクエリロジックの場合: references/flow_patterns.md を読み込み、「Pattern 6: Transform Flow for Reusable Logic」を参照してください

同時実行制御の場合: references/flow_patterns.md を読み込み、「Pattern 7: Concurrency Control」を参照してください

パターン組み合わせの例:

ユーザーが「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 がソース、関数、変換処理から型を自動的に推論します
• dataclass / 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. 構造体型: dataclass、NamedTuple、または Pydantic モデル

   • 戻り値の型: 具体的なクラスを使用する必要があります（例: Person）
   • 引数: dict[str, Any] または Any を使用できます

4. テーブル型:

   • KTable（キー付き）: dict[K, V]。K はキーの型（プリミティブ型または frozen 構造体）、V は構造体型です
   • LTable（順序付き）: list[R]。R は構造体型です
   • 引数: dict[Any, Any] または list[Any] を使用できます

5. Json 型: 非構造化・動的データには cocoindex.Json を使用します

6. オプショナル型: NULL を許容する値には 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

カスタム関数

ユーザーが独自の変換ロジックを必要とする場合は、カスタム関数を作成します。

判断: 単独関数か Spec+Executor か

単独関数を使用する場合:

• 単純な変換である
• 設定が不要である
• セットアップや初期化が不要である

Spec+Executor を使用する場合:

• 設定が必要である（モデル名、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（キャッシュを使用する場合は必須）

Spec+Executor 関数の作成

# 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 - LLM API によるエンベディング

LLM 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")
)

LLM による抽出

ExtractByLlm - LLM を使って構造化データを抽出

抽出や要約のタスクで LLM を使用する推奨方法です。構造化された出力（データクラス、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"]
)

よくある問題と解決策

「Flow not found」

• APP_TARGET の形式を確認してください: cocoindex show main.py
• プロジェクトのルートにいない場合は --app-dir を使用してください
• フロー名がデコレーターの名前と一致しているか確認してください

「Database connection failed」

• .env に COCOINDEX_DATABASE_URL が設定されているか確認してください
• 接続をテストしてください: psql $COCOINDEX_DATABASE_URL
• カスタムの場所を指定するには --env-file を使用してください

「Schema mismatch」

• セットアップを再実行してください: cocoindex setup main.py
• ドロップして再作成してください: cocoindex drop main.py && cocoindex setup main.py

「Live update exits immediately」

• ソースに refresh_interval を追加してください
• または、ソース固有の変更キャプチャ（Postgres の通知、S3 イベント）を使用してください

「Out of memory」

• ソースに同時実行数の上限を設定してください: max_inflight_rows、max_inflight_bytes
• .env でグローバルな上限を設定してください: COCOINDEX_SOURCE_MAX_INFLIGHT_ROWS

リファレンスドキュメント

このスキルには、一般的なパターンと操作に関する包括的なリファレンスドキュメントが含まれています:

• references/flow_patterns.md - 一般的なフローパターンの完全な例（テキストのエンベディング、コードのエンベディング、ナレッジグラフ、ライブ更新、同時実行制御など）
• references/custom_functions.md - カスタム関数を作成するための詳細なガイド（スタンドアロン関数、spec+executor パターン、LLM 呼び出し、外部 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