Create basic project structure:

# 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()

Step 4: Write the Flow

Follow this structure:

@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
    )

• Each source creates a field in the top-level data scope
• Use .row() to iterate through table data
• CRITICAL: Always assign transformed data to row fields - Use item["new_field"] = item["existing_field"].transform(...), NOT local variables like new_field = item["existing_field"].transform(...)
• Transformations create new fields without mutating existing data
• Collectors gather data from any scope level
• Export must happen at top level (not within row iterations)

❌ Wrong: Using local variables for transformations

with data_scope["files"].row() as file:
    summary = file["content"].transform(...)  # ❌ Local variable
    summaries_collector.collect(filename=file["filename"], summary=summary)

✅ Correct: Assigning to row fields

with data_scope["files"].row() as file:
    file["summary"] = file["content"].transform(...)  # ✅ Field assignment
    summaries_collector.collect(filename=file["filename"], summary=file["summary"])

❌ Wrong: Creating unnecessary dataclasses to mirror flow fields

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!

Step 5: Design the Flow Solution

• Combine elements from multiple patterns - Mix and match sources, transformations, and targets creatively
• Review additional examples - See for diverse real-world use cases (face recognition, multimodal search, product recommendations, patient form extraction, etc.)
• Think from first principles - Use the core APIs (sources, transforms, collectors, exports) and apply common sense to solve novel problems
• Be creative - CocoIndex is flexible; unique combinations of components can solve unique problems

If a user asks to "index images from S3, generate captions with a vision API, and store in Qdrant", combine:

• AmazonS3 source (from S3 examples)
• Custom function for vision API calls (from custom functions pattern)
• EmbedText to embed the captions (from embedding patterns)
• Qdrant target (from target examples)

No single pattern covers this exact scenario, but the building blocks are composable.

Step 6: Test and Run

Guide user through testing:

# 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

Data Types

CocoIndex has a type system independent of programming languages. All data types are determined at flow definition time, making schemas clear and predictable.

• Custom functions: Type annotations are required for return values (these are the source of truth for type inference)
• Flow fields: Type annotations are NOT needed - CocoIndex automatically infers types from sources, functions, and transformations
• Dataclasses/Pydantic models: Only create them when they're actually used (as function parameters/returns or ExtractByLlm output_type), NOT to mirror flow field schemas

• Return values of custom functions: Must use specific type annotations - these are the source of truth for type inference
• Arguments of custom functions: Relaxed - can use Any, dict[str, Any], or omit annotations; engine already knows the types
• Flow definitions: No explicit type annotations needed - CocoIndex automatically infers types from sources and functions

• Primitive types: str, int, float, bool, bytes, datetime.date, datetime.datetime, uuid.UUID

• Vector types (embeddings): Specify dimension in return type if you plan to export as vectors to targets, as most targets require a fixed vector dimension

- list[float] wi