Comprehensive toolkit for developing with the CocoIndex library. Use when users need to create data transformation pipelines (flows), write custom functions, or operate flows via CLI or API. Covers building ETL workflows for AI data processing, including embedding documents into vector databases, building knowledge graphs, creating search indexes, or processing data streams with incremental updates.
Instalar com Codex ou Claude Copie este prompt, cole no Codex, Claude ou outro assistente e deixe que ele revise a página da skill e instale para você.
Um comando direto ignora o prompt de revisão. Verifique a origem antes de executá-lo.
Comprehensive toolkit for developing with the CocoIndex library. Use when users need to create data transformation pipelines (flows), write custom functions, or operate flows via CLI or API. Covers building ETL workflows for AI data processing, including embedding documents into vector databases, building knowledge graphs, creating search indexes, or processing data streams with incremental updates.
CocoIndex
Overview
CocoIndex is an ultra-performant real-time data transformation framework for AI with incremental processing. This skill enables building indexing flows that extract data from sources, apply transformations (chunking, embedding, LLM extraction), and export to targets (vector databases, graph databases, relational databases).
Core capabilities:
Write indexing flows - Define ETL pipelines using Python
@cocoindex.flow_def(name="DescriptiveName")defflow_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
)
Key principles:
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)
Common mistakes to avoid:
❌ 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
@dataclassclassFileSummary: # ❌ 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
IMPORTANT: The patterns listed below are common starting points, but you cannot exhaustively enumerate all possible scenarios. When user requirements don't match existing patterns:
Combine elements from multiple patterns - Mix and match sources, transformations, and targets creatively
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
Common starting patterns (use references for detailed examples):
For text embedding: Load references/flow_patterns.md and refer to "Pattern 1: Simple Text Embedding"
For code embedding: Load references/flow_patterns.md and refer to "Pattern 2: Code Embedding with Language Detection"
For LLM extraction + knowledge graph: Load references/flow_patterns.md and refer to "Pattern 3: LLM-based Extraction to Knowledge Graph"
For live updates: Load references/flow_patterns.md and refer to "Pattern 4: Live Updates with Refresh Interval"
For custom functions: Load references/flow_patterns.md and refer to "Pattern 5: Custom Transform Function"
For reusable query logic: Load references/flow_patterns.md and refer to "Pattern 6: Transform Flow for Reusable Logic"
For concurrency control: Load references/flow_patterns.md and refer to "Pattern 7: Concurrency Control"
Example of pattern composition:
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.
IMPORTANT: When to define types:
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
Type annotation requirements:
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
Why specific return types matter: Custom function return types let CocoIndex infer field types throughout the flow without processing real data. This enables creating proper target schemas (e.g., vector indexes with fixed dimensions).
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
Struct types: Dataclass, NamedTuple, or Pydantic model
Return type: Must use specific class (e.g., Person)
Argument: Can use dict[str, Any] or Any
Table types:
KTable (keyed): dict[K, V] where K = key type (primitive or frozen struct), V = Struct type
LTable (ordered): list[R] where R = Struct type
Arguments: Can use dict[Any, Any] or list[Any]
Json type: cocoindex.Json for unstructured/dynamic data
Optional types: T | None for nullable values
Examples:
from dataclasses import dataclass
from typing importLiteralimport cocoindex
@dataclassclassPerson:
name: str
age: int# ✅ Vector with dimension (recommended for vector search)@cocoindex.op.function(behavior_version=1)defembed_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)defprocess_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)deffilter_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}
Type annotations on all arguments and return value
Optional parameters: cache=True for expensive ops, behavior_version (required with cache)
Creating Spec+Executor Functions
# 1. Define configuration specclassMyFunction(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)classMyFunctionExecutor:
spec: MyFunction # Required: link to spec
model = None# Instance variables for statedefprepare(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}
When to enable cache:
LLM API calls
Model inference
External API calls
Computationally expensive operations
Important: Increment behavior_version when function logic changes to invalidate cache.
For detailed examples and patterns, load references/custom_functions.md.
@cocoindex.transform_flow()deftext_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")
For complete API reference and patterns, load references/api_operations.md.
This is the recommended way to use LLMs for extraction and summarization tasks. It supports both structured outputs (dataclasses, Pydantic models) and simple text outputs (str).
import dataclasses
# For structured extraction@dataclasses.dataclassclassProductInfo:
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"
)
)
Drop and recreate: cocoindex drop main.py && cocoindex setup main.py
"Live update exits immediately"
Add refresh_interval to source
Or use source-specific change capture (Postgres notifications, S3 events)
"Out of memory"
Add concurrency limits on sources: max_inflight_rows, max_inflight_bytes
Set global limits in .env: COCOINDEX_SOURCE_MAX_INFLIGHT_ROWS
Reference Documentation
This skill includes comprehensive reference documentation for common patterns and operations:
references/flow_patterns.md - Complete examples of common flow patterns (text embedding, code embedding, knowledge graphs, live updates, concurrency control, etc.)
references/custom_functions.md - Detailed guide for creating custom functions with examples (standalone functions, spec+executor pattern, LLM calls, external APIs, caching)
references/cli_operations.md - Complete CLI reference with all commands, options, and workflows
references/api_operations.md - Python API reference with examples for programmatic flow control, live updates, queries, and application integration patterns
