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、画像、コードなど）
• どのくらいの頻度で変更されるか？（1 回限り、定期的、継続的）

変換：

• どのような処理が必要か？（チャンキング、埋め込み、抽出など）
• どの埋め込みモデル？（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 ファイルを作成するようガイドします：

# データベース接続（必須 - 内部ストレージ）
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 オプション： 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):
    # フロー定義をここに記述
    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. ソースデータをインポート
    data_scope["source_name"] = flow_builder.add_source(
        cocoindex.sources.SourceType(...)
    )

    # 2. 出力用のコレクターを作成
    collector = data_scope.add_collector()

    # 3. データを変換（行を反復処理）
    with data_scope["source_name"].row() as item:
        # 変換を適用
        item["new_field"] = item["existing_field"].transform(
            cocoindex.functions.FunctionName(...)
        )

        ...

        # ネストされた反復処理（例：ドキュメント内のチャンク）
        with item["nested_table"].row() as nested_item:
            # さらに変換
            nested_item["embedding"] = nested_item["text"].transform(...)

            # エクスポート用のデータを収集
            collector.collect(
                field1=nested_item["field1"],
                field2=item["field2"],
                generated_id=cocoindex.GeneratedField.UUID
            )

    # 4. ターゲットにエクスポート
    collector.export(
        "target_name",
        cocoindex.targets.TargetType(...),
        primary_key_fields=["field1"],
        vector_indexes=[...]  # 必要に応じて
    )

主要な原則：

• 各ソースはトップレベルのデータスコープにフィールドを作成
• .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(...)  # ❌ ローカル変数
    summaries_collector.collect(filename=file["filename"], summary=summary)

✅ 正しい： 行フィールドに割り当て

with data_scope["files"].row() as file:
    file["summary"] = file["content"].transform(...)  # ✅ フィールド割り当て
    summaries_collector.collect(filename=file["filename"], summary=file["summary"])

❌ 間違い： フローフィールドをミラーする不要なデータクラスを作成

from dataclasses import dataclass

@dataclass
class FileSummary:  # ❌ 不要 - CocoIndex が自動的にフィールドを管理
    filename: str
    summary: str
    embedding: list[float]

# このデータクラスはフローで使用されません！

ステップ 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. セットアップで実行
cocoindex update --setup -f main   # -f 確認プロンプトなくセットアップを強制


# 2. サーバーを起動し、ユーザーを CocoInsight にリダイレクト
cocoindex server -ci main
# その後 https://cocoindex.io/cocoinsight で 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 null 許容値用

例：

from dataclasses import dataclass
from typing import Literal
import cocoindex

@dataclass
class Person:
    name: str
    age: int

# ✅ 次元付きベクトル（ベクトル検索に推奨）
@cocoindex.op.function(behavior_version=1)
def embed_text(text: str) -> cocoindex.Vector[cocoindex.Float32, Literal[768]]:
    """768 次元埋め込みを生成 - ベクトルインデックスに次元が必要。"""
    # ... 埋め込みロジック ...
    return embedding  # 768 個の浮動小数点数の numpy 配列またはリスト

# ✅ 構造体戻り値の型、緩和された引数
@cocoindex.op.function(behavior_version=1)
def process_person(person: dict[str, Any]) -> Person:
    """引数は dict[str, Any] になります、戻り値は特定の構造体である必要があります。"""
    return Person(name=person["name"], age=person["age"])

# ✅ LTable 戻り値の型
@cocoindex.op.function(behavior_version=1)
def filter_people(people: list[Any]) -> list[Person]:
    """戻り値の型が特定の構造体のリストを指定します。"""
    return [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}

包括的なデータ型ドキュメント： 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:
    """
    関数の説明。

    Args:
        input_arg: 説明
        optional_arg: オプション説明
    """
    # 変換ロジック
    return {"result": f"processed-{input_arg}"}

要件：

• デコレーター：@cocoindex.op.function()
• すべての引数と戻り値に型注釈を付ける
• オプションパラメータ：高コストの操作には cache=True、behavior_version が必須（キャッシュ使用時）

仕様+エグゼキューター関数の作成

# 1. 構成仕様を定義
class MyFunction(cocoindex.op.FunctionSpec):
    """MyFunction の構成。"""
    model_name: str
    threshold: float = 0.5

# 2. エグゼキューターを定義
@cocoindex.op.executor_class(cache=True, behavior_version=1)
class MyFunctionExecutor:
    spec: MyFunction  # 必須：仕様へのリンク
    model = None      # 状態用インスタンス変数

    def prepare(self) -> None:
        """オプション：実行前に 1 回実行。"""
        # モデル読み込み、接続セットアップなど
        self.model = load_model(self.spec.model_name)

    def __call__(self, text: str) -> dict:
        """必須：各データ行に対して実行。"""
        # 構成に self.spec を使用
        # 読み込まれたリソースに self.model を使用
        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

1 回限りの更新：

cocoindex update main

# 自動セットアップ付き
cocoindex update --setup main

# セットアップと更新の前にすべてをリセットするのを強制
cocoindex update --reset main

ライブ更新（継続的監視）：

cocoindex update main.py -L

# ソースのリフレッシュ間隔またはソース固有の変更キャプチャが必要

フローを削除（すべてのリソースを削除）：

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):
    # ... フロー定義 ...
    pass

1 回限りの更新：

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

# 非同期
stats = await my_flow.update_async()

ライブ更新：

# コンテキストマネージャーとして
with cocoindex.FlowLiveUpdater(my_flow) as updater:
    # アップデーターはバックグラウンドで実行
    # アプリケーションロジックはここに
    pass

# 手動制御
updater = cocoindex.FlowLiveUpdater(
    my_flow,
    cocoindex.FlowLiveUpdaterOptions(
        live_mode=True,
        print_stats=True
    )
)
updater.start()
# ... アプリケーションロジック ...
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="...")
    )

# インデックスフローで使用
doc["embedding"] = text_to_embedding(doc["content"])

# クエリに使用
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",  # または "python"、"javascript" など
    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 - ローカル埋め込みモデル

# 必須：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 - マルチモーダル画像埋め込み

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

LLM 抽出

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

これは抽出とサマリー化タスクに LLM を使用する推奨される方法です。構造化出力（データクラス、Pydantic モデル）と単純なテキスト出力（str）の両方をサポートします。

import dataclasses

# 構造化抽出用
@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"
    )
)

# テキストサマリー/生成用
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：

# 必須：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 - カスタム関数の詳細なガイドと例（スタンドアロン関数、仕様+エグゼキューターパターン、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