Declaring functions with @coco.fn

The @coco.fn decorator makes Python functions participate in CocoIndex's change detection, memoization, and execution pipeline — with async adapters, batching, and GPU-aware runners included.

Version

v 1.0.14

Last reviewed

Jul 4, 2026

It’s common to factor work into helper functions (for parsing, chunking, embedding, formatting, etc.). In CocoIndex, you can decorate any Python function with @coco.fn when you want to add incremental capabilities to it. The decorated function is still a normal Python function: its signature stays the same, and you can call it normally.

@coco.fn
async def process_file(file: FileLike) -> str:
    return await file.read_text()

# Can be called like any normal function
result = await process_file(file)

@coco.fn preserves the sync/async nature of the underlying function. Decorating a sync function yields a sync function; decorating an async function yields an async function.

How to think about @coco.fn

Decorating a function tells CocoIndex that calls to it are part of the incremental update engine. You still write normal Python, but CocoIndex can now:

• Detect when inputs or code have changed (change detection)
• Skip work when nothing has changed (memoization)

This is what lets CocoIndex avoid rerunning expensive steps on every app.update(). See Processing Component for how decorated functions are mounted at component paths.

If you don’t need any of the above for a helper, keep it as a plain Python function.

Change detection and memoization

Every @coco.fn function participates in CocoIndex’s change detection system. With memo=True, the function’s results are cached and reused when nothing has changed. These two mechanisms — detecting changes and acting on them — work together to enable incremental updates. For the big-picture view — how memoization skips unchanged work across both input and code changes — see the Function memoization section in Core Concepts.

Change detection

CocoIndex detects three kinds of changes:

Logic changes — a function’s logic is its own source code (or, when you set version, that integer instead of the source), plus any deps values. Tracked by @coco.fn. Logic fingerprints propagate transitively up the call chain, following runtime invocations rather than the static call graph: every @coco.fn function reached during a memoized call — directly, or through plain undecorated helpers — contributes its logic to that call. So if foo (memoized) calls bar (also @coco.fn-decorated, with or without memo=True) and bar’s logic changes, foo’s memo is invalidated; undecorated helpers in between don’t break the chain. What the decorator buys is having a function’s own logic tracked: an undecorated helper is invisible to change detection, so editing it invalidates nothing. This is why @coco.fn matters for any function in the call chain, not just memoized ones — see Common patterns and the FAQ. To stop propagation deliberately, set logic_tracking to "self" or None.

Input changes — the function’s arguments. Tracked by @coco.fn. When you call a function with different arguments, the fingerprints change. Input fingerprints do not propagate transitively.

Context changes — context values with detect_change=True. Tracked by use_context() at the call site — independent of @coco.fn. Context reads propagate transitively: if a memoized foo calls bar and bar reads a change-detected context value, then changing that value invalidates foo’s memo too — even though foo itself never called use_context(). What matters is whether the key was read anywhere during the memoized call, not where in the call chain.

Memoization

With memo=True, the function’s result is cached. On subsequent calls, if no logic, input, or read context values have changed, the cached result is reused without executing the function body — it carries over target states declared during the function’s previous invocation and returns its previous return value.

@coco.fn(memo=True)
def process_chunk(chunk: Chunk) -> list[float]:
    # This computation is skipped if chunk, logic, and context are unchanged
    return embed(chunk.text)

Cache-hit semantics

When a memoized function’s cache hits, its body does not run — and neither do any nested @coco.fn calls inside it. The cached output is replayed directly: the previous return value is returned, and any target states the function declared on its previous run are carried over.

@coco.fn(memo=True)
async def inner(text: str) -> str:
    print("inner ran")
    return await call_llm(text)

@coco.fn(memo=True)
async def outer(text: str) -> str:
    print("outer ran")
    return await inner(text) + "!"

# First call: both print, both bodies execute
result = await outer("hello")

# Second call with same inputs: nothing prints — outer's cached value is returned
# directly, and inner is not invoked at all.
result = await outer("hello")

Propagation (logic, deps, context) is recorded during the previous invocation and replayed on cache lookup. You don’t need to model the internals — change anything that affects behavior and the cache invalidates correctly.

When to memoize

Cost: A directly-called memoized function’s cache entry must store everything needed to serve a hit — the return value, plus the path of every target state it declared during the call (the state values themselves stay in the enclosing component’s tracking; the recorded paths are what keep those states alive on a hit). Large return values, or a call that declares many target states, mean large entries. (A function mounted as a component avoids the per-state paths entirely: a memoized component’s entry holds fingerprints plus its return value — None for a typical mounted declarer.)

Benefit: Memoization saves more when:

• The computation is expensive
• The function’s caller is reprocessed frequently (due to logic or input changes)

Placement — one rule: memoize at a boundary only if its key can hit while every enclosing memo boundary misses. Stacked levels pay off exactly when the outer boundary is keyed more coarsely than the inner one’s inputs — a per-file component plus a per-chunk embedding memo composes well, because an edited file re-runs its component while its unchanged chunks still skip embedding. But if an enclosing memoized component’s identity already commits to this function’s inputs — say, the component path is keyed by a hash of the very content the function transforms — the inner memo can never hit on a content-driven rerun: the component memo shadows all of those. What remains is insurance against reruns of the enclosing component itself (its code edited, its version bumped): each entry permanently stores a value — often one your target also stores — that only such reruns can read. The arithmetic bites at scale: a 1536-dim float32 vector is ~6 KB, so an embedding memo under content-hash-keyed components costs ~600 MB per 100k chunks to insure against those rare reruns — usually a bad trade.

Examples:

• ✅ Embedding functions — good to memoize when callers can re-run with the same text (e.g., per-file components re-running on edits). Computation is heavy; return value is fixed-size and not too large.
• ❌ Splitting text into fixed-size chunks — usually not worth memoizing. Computation is light; return value can be large.
• ✅ Processing component for files that are mostly stable between runs — very beneficial to memoize, since unchanged files are skipped entirely. We can save the cost of reading file content and processing them when they haven’t changed.
• 🤔 Chunk embedding when file-level memoization is already enabled — still beneficial, but less so for stable files. The benefit increases for files that change frequently, or when your code evolves (e.g., adding more features per file triggers file-level reprocessing, but unchanged chunks can still skip embedding).
• ❌ Chunk embedding inside a memoized component keyed by the chunk’s content hash — usually redundant by the placement rule: the component memo already commits to the text, so the inner memo serves only reruns caused by the component’s own code or version changes — at the cost of storing every vector twice.

Controlling change detection scope

Three parameters on @coco.fn let you customize how logic changes are detected:

• logic_tracking — controls the scope of automatic logic change detection
• version — provides explicit manual control over when dependent memos are invalidated
• deps — declares external values (e.g. a module-level prompt string) as part of the function’s logic, so changing them invalidates dependent memos

These parameters control logic fingerprinting. Data fingerprinting (for arguments, deps values, and context values) is controlled by the objects themselves (see Memoization Keys & States).

logic_tracking

The logic_tracking parameter controls whether and how logic changes are detected:

• "full" (default): Track this function’s logic AND all transitively called @coco.fn functions’ logic. A change anywhere in the call chain invalidates dependent memos.
• "self": Track only this function’s own logic — its fingerprint still propagates to its callers, but the logic of functions it calls does not propagate through it.
• None: Don’t track logic through this function at all — neither its own logic nor its callees’ reaches its callers.

These settings govern nested function calls. A mounted child component is tracked through the component tree instead, so its logic reaches the parent regardless of the parent’s logic_tracking.

version

The version parameter lets you explicitly invalidate dependent memos by bumping an integer:

@coco.fn(version=2)
def process_chunk(chunk: Chunk) -> list[float]:
    # Bumping version invalidates all memoized callers, even if code looks the same
    return embed(chunk.text)

deps

The deps parameter declares external value(s) the function logic depends on but that aren’t visible in its body — for example a prompt string or a model identifier defined at module scope. When the value changes, the function’s logic fingerprint changes and dependent memos are invalidated, exactly as if the function body had been edited.

SYSTEM_PROMPT = "You are a helpful assistant. Be concise."

@coco.fn(memo=True, deps=SYSTEM_PROMPT)
def summarize(text: str) -> str:
    # Editing SYSTEM_PROMPT invalidates this function's memo
    # (and propagates to memoized callers) just like a logic change would.
    return call_llm(SYSTEM_PROMPT, text)

For multiple dependencies, pass a tuple or dict:

SYSTEM_PROMPT = "..."
MODEL = "claude-haiku-4-5"

@coco.fn(memo=True, deps={"prompt": SYSTEM_PROMPT, "model": MODEL})
def summarize(text: str) -> str:
    return call_llm(SYSTEM_PROMPT, text, model=MODEL)

The value is canonicalized through the memoization-key pipeline, which honors __coco_memo_key__(), registered memo key functions, and the standard handling for primitives, dataclasses, and Pydantic models.

deps requires logic_tracking to be enabled; combining deps=<value> with logic_tracking=None raises ValueError.

Common patterns

These parameters can be set on any @coco.fn function — not just memoized ones.

Fully automatic (default) — use logic_tracking="full" (or omit it) without setting version. Any logic change in the function or its callees invalidates dependent memos. This always just works.

@coco.fn
async def process_file(file: FileLike) -> list[Chunk]:
    # Any change here or in called @coco.fn functions invalidates dependent memos
    text = await file.read_text()
    return split_and_embed(text)

Manual, precise control — use logic_tracking="self" with version. You decide what counts as a behavior change by bumping version, without being affected by implementation detail changes (performance optimizations, logging tweaks, refactoring, etc.).

@coco.fn(logic_tracking="self", version=3)
async def process(data: str) -> str:
    # Bump version when behavior changes (e.g., new output format).
    # Internal refactors or logging changes won't trigger reprocessing.
    return await transform(data)

Opt out of tracking — use logic_tracking=None for functions with a stable contract (where logic changes don’t affect output), or functions whose changes don’t affect behavior (e.g., logging, performance hints). This prevents unnecessary reprocessing when only internals change.

@coco.fn(logic_tracking=None)
def embed(text: str) -> list[float]:
    # Contract is stable: same input always produces the same embedding.
    # Internal changes (e.g., switching backends) are handled by version bumps.
    return model.encode(text)

Debugging unexpected re-runs

If a memoized function is re-running when you expect a cache hit, walk through the inputs in order:

1. Logic — did the function’s source code change? Did any @coco.fn it transitively calls change? Was a version bumped? Was a deps value edited?
2. Inputs — are the arguments byte-identical to the previous call (after canonicalization)? Custom types with unstable equality are a common source of spurious invalidation — see Memoization Keys & States.
3. Context — did any detect_change=True context value read during the previous invocation change? Remember this propagates transitively through nested @coco.fn calls.

If none of those changed and the function still re-runs, the most common cause is a non-stable fingerprint on a custom-typed argument or context value — define __coco_memo_key__ to make it deterministic.

Conversely, if a memo is hitting when you expect invalidation, common causes are:

• A logic change in a helper that is not decorated with @coco.fn. Add @coco.fn to it (no memo= needed) so its logic participates in propagation — see Common patterns.
• A deps value that you changed at runtime: deps is snapshotted at decoration time, so per-instance or per-request values must be passed as regular arguments instead.
• The function uses logic_tracking=None, which opts it out of code-change detection entirely.

Customizing data fingerprinting

By default, CocoIndex fingerprints function arguments, deps values, and context values automatically for most types — primitives, containers, dataclasses, Pydantic models, and picklable objects. For custom types, or when you need multi-level validation (e.g., check mtime first, then content hash), see Memoization Keys & States.

For per-function overrides — excluding an argument from the memo key, or transforming it just for this function — pass memo_key={...} on @coco.fn / @coco.fn.as_async; see Override at the call site with memo_key=.

Execution capabilities

The following capabilities control how the function executes, independent of change detection and memoization.

Async adapter

Use @coco.fn.as_async when you need an async interface for a function that has a sync underlying implementation. This is useful for compute-intensive leaf functions, and is required for features like batching and runner.

@coco.fn.as_async
def embed(text: str) -> list[float]:
    return model.encode([text])[0]

# External usage: always async, even though the function body is sync
embedding = await embed("hello world")

@coco.fn.as_async is equivalent to wrapping the function in asyncio.to_thread() — the sync function runs on a thread pool and doesn’t block the event loop.

You can also call any @coco.fn-decorated function asynchronously via the .as_async() method, without changing its primary signature:

@coco.fn
def expensive_fn(data: bytes) -> bytes:
    return process(data)

# Primary call is sync:
result = expensive_fn(data)

# Async call via .as_async():
result = await expensive_fn.as_async(data)

Batching

With batching=True, multiple concurrent calls to the function are automatically batched together. This is useful for operations that are more efficient when processing multiple inputs at once, such as embedding models.

Batching requires an async interface. If the underlying function is sync, use @coco.fn.as_async(batching=True). If the underlying function is already async def, @coco.fn(batching=True) works directly.

When batching is enabled:

• The function implementation receives a list[T] and returns a list[R]
• The external signature becomes async T -> R (single input, single output)
• Concurrent calls are collected and processed together

@coco.fn.as_async(batching=True, max_batch_size=32)
def embed(texts: list[str]) -> list[list[float]]:
    # Called with a batch of texts, returns a batch of embeddings
    return model.encode(texts)

# External usage: async, single input, single output
embedding = await embed("hello world")  # Returns list[float]

# Concurrent calls are automatically batched using asyncio.gather
embeddings = await asyncio.gather(
    embed("text1"),
    embed("text2"),
    embed("text3"),
)

The max_batch_size parameter limits how many inputs can be processed in a single batch.

Retrying a failed batch as smaller batches

Some batch failures depend on the batch’s composition rather than any single input — e.g. a provider’s per-request token or payload limit, or one input the provider rejects. No fixed max_batch_size can rule these out. When the function body hits such a failure, raise coco.RetryWithSmallerBatch from it: CocoIndex halves the batch and retries each half, recursing down to single items. At size 1, the wrapped original error is raised as that item’s own failure — so one bad input fails only its own caller, while the rest of the batch still succeeds. There is no need to check the batch size before raising: at size 1 the engine simply unwraps the signal and raises the original error.

@coco.fn(batching=True, max_batch_size=64)
async def embed(texts: list[str]) -> list[list[float]]:
    try:
        return await provider.embed(texts)
    except ProviderError as e:
        if is_transient(e) or is_auth_error(e):
            raise  # retrying smaller can't help these
        raise coco.RetryWithSmallerBatch() from e

Raising it is safe for any deterministic batch failure, even when you can’t tell whether the batch size was the cause: if the error was actually global (it would fail at any size), the recursion terminates at single items and every caller receives the underlying error, at the bounded cost of extra attempts. Only propagate directly for errors where splitting provably can’t help — global errors like invalid credentials, and transient errors (rate limits, network blips) that haven’t yet been retried at the same size. Once same-size retries are exhausted, splitting is a reasonable last resort: a smaller request may pass where the large one timed out.

Runner

The runner parameter allows functions to execute in a specific context, such as a dedicated GPU runner that serializes GPU workloads.

Like batching, a runner requires an async interface. If the underlying function is sync, use @coco.fn.as_async(runner=...) to make it async. If the underlying function is already async def, @coco.fn(runner=...) works directly.

@coco.fn.as_async(runner=coco.GPU)
def gpu_inference(data: bytes) -> bytes:
    # Runs with GPU serialization
    return model.predict(data)

# External usage: async
result = await gpu_inference(data)

The coco.GPU runner:

• By default, runs in-process with all functions sharing a queue for serial execution
• Sync functions run on a dedicated GPU thread to avoid blocking the event loop
• Set the environment variable COCOINDEX_RUN_GPU_IN_SUBPROCESS=1 to run in a subprocess for GPU memory isolation

You can combine batching with a runner:

@coco.fn.as_async(batching=True, max_batch_size=16, runner=coco.GPU)
def batch_gpu_embed(texts: list[str]) -> list[list[float]]:
    # Batched execution with GPU serialization
    return gpu_model.encode(texts)

# External usage: async
embedding = await batch_gpu_embed("hello world")

# Concurrent calls
embeddings = await asyncio.gather(
    batch_gpu_embed("text1"),
    batch_gpu_embed("text2"),
    batch_gpu_embed("text3"),
)