# Functions
source: https://docs.chalk.ai/docs/compute/functions

## Define and call remote functions that run on Chalk Compute.

### Overview

A Chalk Function is a Python callable that you deploy onto a
Scaling Group and register in the
Chalk Catalog. Decorate a function
with @chalkcompute.function and it becomes reachable from anywhere Chalk
runs: local scripts and notebooks, other functions, and (because of the
Catalog registration) the feature engine, including Chalk SQL, resolvers, the
query server, and DataFrames via F.catalog_call(). Arguments and results are
serialized automatically; you write ordinary Python and Chalk handles
scheduling, scaling, and delivery.

On top of what a bare Scaling Group provides, functions add async / streaming
invocation, nested calls, automatic retries, concurrency caps, rate limits,
and request-queue-driven autoscaling. Reach for a function when you want any
of those features, or when you want the callable to be visible to the rest
of Chalk.

Functions support three return styles:

- Single value: return a result and the caller receives it.
- Async generator: yield results incrementally and the caller receives them as
they're produced. See Streaming results
below for an end-to-end example.
- Nested calls: a function can call other functions, composing pipelines without
manual orchestration. See Nested function calls below
for a pipeline example.

### Defining a function

Use the @chalkcompute.function decorator:

```
import chalkcompute

@chalkcompute.function
def add(x: int, y: int) -> int:
    return x + y

if __name__ == "__main__":
    add.wait_ready()
    result = add.remote(3, 5)
    print(result)  # 8
```

Deploy by running the script: the decorator registers the function with
Chalk as soon as it runs. wait_ready() blocks until the deployed function
is available, and .remote(...) invokes it. add(3, 5) runs the Python body
locally instead; use .remote(...) to call the deployed version.

To call an already-deployed function from a different script — without
redeploying — attach to it by name with RemoteFunction.from_name:

```
import chalkcompute

fn = chalkcompute.RemoteFunction.from_name("add")
result = fn.remote(3, 5)
print(result)  # 8
```

### Deleting a function

Delete a function and all of its versions through its handle:

```
import chalkcompute

chalkcompute.RemoteFunction.from_name("my_function").delete()
```

This tears down the Scaling Group behind each version, including
that group's Kubernetes Deployment, Service, and HTTPRoute, along with the function's schedule and
its Chalk Catalog entry. The underlying records are soft-deleted, so the function still appears in
history. Deleting a function that is already gone succeeds without raising.

### Deferred calls with defer() and get()

.remote(...) blocks for the entire round trip. Use
.defer(...) instead when you want to submit a call and collect its result
later; it enqueues the call and returns a FunctionCallHandle immediately,
without waiting for the function to run:

```
handle = add.defer(3, 5)
# ...do other work while the call runs...
result = handle.get()
print(result)  # 8
```

handle.get(timeout=None) blocks until the result is ready and returns the
same value .remote(...) would have returned.

Each handle exposes the server-assigned call_id it was enqueued under:

```
handle = add.defer(3, 5)
print(handle.call_id)
```

### Fan-out and gather()

Because .defer(...) returns immediately, you can enqueue many calls up
front and collect their results afterward. Use chalkcompute.gather(...) to
wait for a batch of handles concurrently and get their results back in order:

```
import chalkcompute

handles = [classify.defer(text) for text in texts]
results = chalkcompute.gather(handles)
```

gather(...) waits for the handles concurrently rather than one at a time,
so it returns as soon as the slowest call finishes. Pass timeout to bound the whole wait:

```
results = chalkcompute.gather(handles, timeout=30.0)
```

Inside a running event loop, gather(...) returns an awaitable instead of
blocking: await chalkcompute.gather(handles).

Each deferred call is still subject to the function's own
concurrency and rate limit
policies, so fanning out this way doesn't bypass those caps.

### Timeouts

Pass timeout (seconds) to bound how long get() waits:

```
try:
    result = handle.get(timeout=5.0)
except TimeoutError:
    print("still running")
```

A timeout only stops waiting locally. It does not cancel the underlying
call, which keeps running and can still be collected with another get().

### Error handling

If the function raised an exception, .get() raises
chalkcompute.FunctionError describing the failure. Transient
infrastructure failures, such as a dropped connection between the queue and
the worker, are retried automatically before get() ever sees them; only
failures from the function body itself, or infra failures that exhaust
retries, are raised to the caller.

### Function dependencies in external files

A deployed function often calls helpers defined in other files. Whether a
helper resolves remotely depends on where it lives relative to the decorated
function (the function under @chalkcompute.function):

- Helper in the decorated function's own package: the function's entire
top-level package is shipped automatically, so imports of sibling modules
resolve remotely with no extra configuration.
- Helper anywhere else (a standalone module, or a different package):
must be declared on the image with
Image.add_local_python_source.
The same applies to all file-external helpers when the decorated function is
defined in a standalone script rather than a package.

### Same package: automatic

If the decorated function's module runs as part of a package (imported from
an entrypoint, or executed directly with python -m), Chalk ships the whole
top-level package in the deploy image:

```
# mypkg/helpers.py
def normalize(text: str) -> str:
    return text.strip().lower()
```

```
# mypkg/pipeline.py
import chalkcompute

from mypkg.helpers import normalize

@chalkcompute.function
def clean(text: str) -> str:
    return normalize(text)
```

```
# main.py  (outside mypkg/)
from mypkg.pipeline import clean

print(clean.remote("  Hello "))
```

```
uv run python main.py
```

Running the entrypoint (or the function's own module with
uv run python -m mypkg.pipeline) deploys clean with all of mypkg, so
normalize is importable remotely exactly as it is locally.

Running the file by path (uv run python mypkg/pipeline.py) ships only that one file
because Python doesn't know that it belongs to a package. In this scenario, Chalk will
raise an error at deploy time.

Every run of an entrypoint that imports the decorated function deploys a new
function version. To call an existing deployment without redeploying, attach
to it by name with RemoteFunction.from_name instead.

```
# call_clean.py: call the deployed function without triggering a redeploy
import chalkcompute

fn = chalkcompute.RemoteFunction.from_name("clean")
print(fn.remote("  Hello "))
```

```
uv run python call_clean.py
```

### Everything else: add_local_python_source

When the decorated function is defined in a standalone script, or imports helpers
from outside its own top-level package, list those local modules or packages on the
image:

```
# helpers.py  (next to deploy.py)
def normalize(text: str) -> str:
    return text.strip().lower()
```

```
# deploy.py
import chalkcompute

from helpers import normalize

@chalkcompute.function(
    image=chalkcompute.Image.debian_slim().add_local_python_source("helpers"),
)
def clean(text: str) -> str:
    return normalize(text)
```

```
uv run python deploy.py
```

add_local_python_source accepts any number of top-level module or package
names. Each must be importable on your local Python path; packages are copied
whole (excluding __pycache__). Dotted names, namespace packages spanning
multiple directories, and compiled extensions are rejected with a
ValueError.

This ships your own source code only. Third-party dependencies still need to
be installed in the image with pip_install or run_commands.

### Resource configuration

The decorator accepts a wide set of arguments that control the compute
environment a function runs in: image, CPU/memory/GPU, secrets, volumes,
batching, retries, rate limits, concurrency, autoscaling, and tracing.
A realistic decorator usually combines several of them:

```
@chalkcompute.function(
    image=chalkcompute.Image.base("python:3.12-slim").run_commands(
        "pip install sentence-transformers",
    ),
    cpu="2",
    memory="4Gi",
    gpu="nvidia-l4",
    secrets=[chalkcompute.Secret.from_chalk_env("HF_TOKEN")],
    max_batching_size=25,
    max_buffer_duration=2000,
    retries=3,
)
def embed(texts: list[str]) -> list[list[float]]:
    from sentence_transformers import SentenceTransformer
    model = SentenceTransformer("all-MiniLM-L6-v2")
    return model.encode(texts).tolist()
```

This page walks through the higher-level concepts behind each argument in its
own section below: batching, retries, rate limits, concurrency, autoscaling,
volumes, and tracing. For the full per-argument reference, types, and
defaults, see @chalkcompute.function
in the API reference.

### Secrets

Inject secrets into a function using the secrets parameter. Secrets are
resolved at deploy time and made available as environment variables inside
the container:

```
from chalkcompute import Secret

@chalkcompute.function(
    secrets=[
        Secret.from_chalk_env("OPENAI_API_KEY"),
        Secret.from_chalk_integration("prod_postgres"),
    ],
)
def call_api(prompt: str) -> str:
    import os
    openai_api_key = os.environ["OPENAI_API_KEY"]
    database_url = os.environ["DATABASE_URL"]
    ...
```

See Secrets for the full Secret API.

### Batching

When your handler is more efficient processing multiple items at once (e.g.
GPU batch inference), enable batching with max_buffer_duration and/or
max_batching_size. Chalk will accumulate incoming items and invoke the
handler in chunks:

```
@chalkcompute.function(
    max_buffer_duration=1000,  # buffer up to 1000 ms
    max_batching_size=32,      # or up to 32 items
    gpu="nvidia-l4",
)
def batch_embed(texts: list[str]) -> list[list[float]]:
    from sentence_transformers import SentenceTransformer
    model = SentenceTransformer("all-MiniLM-L6-v2")
    return model.encode(texts).tolist()
```

When batching is enabled, handler arguments are delivered as lists or a single
pyarrow table. Defaults apply when either parameter is set:

| Parameter             | Default (when batching enabled) |
| --------------------- | ------------------------------- |
| `max_buffer_duration` | `1000` ms                       |
| `max_batching_size`   | `10` items                      |

### Retry policies

Functions can automatically retry on failure. Pass a simple integer for
max-attempts with default exponential backoff, or use RetryPolicy for full
control:

```
import chalkcompute
from chalkcompute import RetryPolicy

# Simple: retry up to 3 times with exponential backoff
@chalkcompute.function(retries=3)
def flaky_api(query: str) -> str:
    ...

# Exponential backoff with custom parameters
@chalkcompute.function(
    retries=RetryPolicy.exponential(
        attempts=5,
        initial=0.5,
        max_wait=30.0,
    ),
)
def resilient_call(payload: str) -> str:
    ...

# Fixed delay, only retry on specific exceptions
@chalkcompute.function(
    retries=RetryPolicy.linear(
        attempts=3,
        delay=2.0,
        retry_on=(TimeoutError, ConnectionError),
    ),
)
def network_call(url: str) -> str:
    ...
```

### RetryPolicy.exponential()

Wait times grow geometrically: initial, initial * multiplier,
initial * multiplier², ..., capped at max_wait:

| Parameter    | Type               | Default        | Description                        |
| ------------ | ------------------ | -------------- | ---------------------------------- |
| `attempts`   | `int`              | `3`            | Max retry attempts                 |
| `initial`    | `float`            | `1.0`          | Base wait time in seconds          |
| `multiplier` | `float`            | `2.0`          | Backoff multiplier                 |
| `max_wait`   | `float`            | `60.0`         | Upper bound on wait time           |
| `jitter`     | `bool`             | `True`         | Add random jitter                  |
| `retry_on`   | `tuple[type, ...]` | `(Exception,)` | Exception types that trigger retry |

### RetryPolicy.linear()

Fixed delay between retries:

| Parameter  | Type               | Default        | Description                        |
| ---------- | ------------------ | -------------- | ---------------------------------- |
| `attempts` | `int`              | `3`            | Max retry attempts                 |
| `delay`    | `float`            | `1.0`          | Fixed wait in seconds              |
| `jitter`   | `bool`             | `False`        | Add random jitter                  |
| `retry_on` | `tuple[type, ...]` | `(Exception,)` | Exception types that trigger retry |

### Exception filtering

By default, all Exception subclasses trigger a retry. Pass a tuple of
exception classes to retry_on to restrict which failures are retried:

```
RetryPolicy.exponential(
    attempts=4,
    retry_on=(ConnectionError, TimeoutError, OSError),
)
```

Non-matching exceptions propagate immediately without consuming retry budget.

### Rate limiting

Cap the outbound request rate of a function using rate_limit. Pass an
integer for a simple "N per second" cap, or a RateLimitPolicy for full
control:

```
import chalkcompute
from chalkcompute import RateLimitPolicy

# Simple: up to 100 requests per second
@chalkcompute.function(rate_limit=100)
def call_provider(query: str) -> str:
    ...

# Full control: rate + time unit + shared-bucket key
@chalkcompute.function(
    rate_limit=RateLimitPolicy(rate=1000, per="minute", key="provider-api"),
)
def call_other_endpoint(query: str) -> str:
    ...
```

### RateLimitPolicy

| Parameter | Type  | Default       | Description                                    |
| --------- | ----- | ------------- | ---------------------------------------------- |
| `rate`    | `int` | *required*    | Maximum requests per `per` unit                |
| `per`     | `str` | `"second"`    | Time unit: `"second"`, `"minute"`, or `"hour"` |
| `key`     | `str` | function name | Shared-bucket identifier (see below)           |

### Sharing a bucket across functions

Two functions declaring the same key share a single rate-limit bucket. This
is useful when multiple functions call the same downstream service and the
service imposes a limit across all callers:

```
shared = RateLimitPolicy(rate=1000, per="second", key="provider-api")

@chalkcompute.function(rate_limit=shared)
def lookup(query: str) -> str: ...

@chalkcompute.function(rate_limit=shared)
def enrich(record: dict) -> dict: ...
# Calls to both functions count against the same 1000/s budget.
```

If key is omitted, each function gets its own bucket keyed by the function
name, so rate limits are per-function by default.

### Concurrency limits

Cap the in-flight count of a function using concurrency. This is distinct
from rate_limit: rate limits gate requests-over-time, concurrency gates how
many calls are executing simultaneously:

```
import chalkcompute
from chalkcompute import ConcurrencyPolicy

# Simple: at most 10 calls in flight at once
@chalkcompute.function(concurrency=10)
def gpu_infer(text: str) -> list[float]:
    ...

# Shared gate across multiple functions hitting the same resource
gpu_pool = ConcurrencyPolicy(max_concurrent=4, key="gpu-pool")

@chalkcompute.function(concurrency=gpu_pool, gpu="nvidia-l4")
def embed(text: str) -> list[float]: ...

@chalkcompute.function(concurrency=gpu_pool, gpu="nvidia-l4")
def rerank(query: str, docs: list[str]) -> list[int]: ...
# Only 4 calls across both functions run simultaneously.
```

### ConcurrencyPolicy

| Parameter        | Type  | Default       | Description                           |
| ---------------- | ----- | ------------- | ------------------------------------- |
| `max_concurrent` | `int` | *required*    | Maximum in-flight handler invocations |
| `key`            | `str` | function name | Shared-bucket identifier              |

When concurrency is not specified, the function queue consumer applies a
default of max_concurrent=1 per function, so calls to a single function are
serialized. Set concurrency=N explicitly to allow parallel execution.

### Rate limit vs concurrency

Both can be set independently and they interact:

| Policy        | Controls                 | Example use case                     |
| ------------- | ------------------------ | ------------------------------------ |
| `rate_limit`  | Requests per unit time   | "1000 calls/sec across all replicas" |
| `concurrency` | Parallel in-flight count | "4 GPU slots, don't exceed memory"   |

### Autoscaling

Functions accept four autoscaling hints that shape how replicas scale up and
down in response to load:

```
@chalkcompute.function(
    min_replicas=1,
    max_replicas=8,
    scaling_interval_seconds=30,
    target_cpu_utilization_percentage=70,
    shutdown_delay_seconds=120,
    gpu="nvidia-l4",
)
def infer(text: str) -> str:
    ...
```

| Parameter                           | Description                                                                                                                                           |
| ----------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- |
| `min_replicas`                      | Lower bound: replicas never scale below this. Use `0` for scale-to-zero.                                                                              |
| `max_replicas`                      | Upper bound: replicas never scale above this.                                                                                                         |
| `scaling_interval_seconds`          | How often the autoscaler scrapes metrics and makes scaling decisions. Defaults to 60s on the server.                                                  |
| `target_cpu_utilization_percentage` | Target average CPU utilization (0–100). Autoscaler adds replicas when the average exceeds this, removes them when below.                              |
| `target_queue_depth`                | Target pending-request queue depth per replica. When set, the autoscaler scales on the function's own queue depth instead of (or in addition to) CPU. |
| `shutdown_delay_seconds`            | Grace period on scale-down before replicas are forcibly killed. Lets in-flight requests finish cleanly. Defaults to 30s on the server.                |

### CPU-driven vs queue-driven scaling

- CPU (target_cpu_utilization_percentage): good for compute-bound
functions, such as model inference or data transformation, where CPU is the
bottleneck.
- Queue depth (target_queue_depth): good for I/O-bound or variable-latency
functions, or when you want to scale on demand rather than resource
saturation. Scales up when calls pile up regardless of CPU.

Both can be set together; the autoscaler takes the higher signal.

### Scheduling

Some jobs run on a clock instead of in response to a request: refreshing a
table of exchange rates, recomputing a daily rollup, warming a cache,
syncing an external source.
A scheduled function is a Function that Chalk invokes for you on a recurring
schedule: declare a schedule in the decorator, and Chalk runs it at the
cadence you specify, with the image, resources, secrets, and retry behavior
you configured. There is no separate cron server to operate.

```
import chalkcompute

@chalkcompute.function(
    schedule="0 * * * *",                      # every hour, on the hour
    secrets=[chalkcompute.Secret.from_chalk_env("EXCHANGE_API_KEY")],
    cpu="1",
    memory="1Gi",
    retries=3,
)
def refresh_exchange_rates() -> None:
    import os
    import requests

    resp = requests.get(
        "https://api.example.com/v1/rates",
        params={"base": "USD"},
        headers={"Authorization": f"Bearer {os.environ['EXCHANGE_API_KEY']}"},
    )
    rates = resp.json()["rates"]
    # write the latest rates to your store
    ...
```

schedule and retries govern how and when the platform invokes the
function; everything else configures the Scaling Group that hosts it, exactly
as it would for an on-demand function. Scheduled jobs often do heavier work,
such as nightly batch inference or periodic re-scoring of a dataset, and can
request a GPU the same way:

```
@chalkcompute.function(
    schedule="0 3 * * *",      # daily at 03:00 UTC
    gpu="nvidia-l4",
    cpu="4",
    memory="16Gi",
)
def nightly_batch_inference() -> None:
    ...
```

### Schedule syntax

schedule accepts either a crontab expression or a Chalk duration string. Use
whichever reads more clearly for your cadence. All schedules are evaluated
in UTC.

Crontab: five space-separated fields (minute hour day-of-month month
day-of-week). Standard syntax is supported: *, lists (1,15), ranges
(1-5), and steps (*/10).

| Expression       | Runs                      |
| ---------------- | ------------------------- |
| `"*/10 * * * *"` | every 10 minutes          |
| `"0 * * * *"`    | every hour, on the hour   |
| `"0 2 * * *"`    | every day at 02:00 UTC    |
| `"0 0 * * 0"`    | every Sunday at 00:00 UTC |

Duration shorthand: a single interval, for simple cadences. Supported units
are minutes (≤ 60), hours (≤ 24), and days (= 1), written as one value:

| Shorthand | Runs             | Equivalent crontab |
| --------- | ---------------- | ------------------ |
| `"10m"`   | every 10 minutes | `*/10 * * * *`     |
| `"6h"`    | every 6 hours    | `0 */6 * * *`      |
| `"1d"`    | every day        | `0 0 * * *`        |

Shorthand intervals are aligned to clock boundaries, not to your deployment
time: "6h" runs at 00:00, 06:00, 12:00, and 18:00 UTC. Anything longer or
compound ("90m", "1h30m", "2d") must be written as a crontab expression.

```
@chalkcompute.function(schedule="0 2 * * *")   # nightly at 02:00 UTC
def nightly_rollup() -> None:
    ...

@chalkcompute.function(schedule="15m")         # every 15 minutes
def warm_cache() -> None:
    ...
```

### Rules for scheduled functions

Because a scheduled function is invoked by the platform rather than a caller,
two constraints apply and are enforced when the function is defined:

- No inputs. A scheduled function must take no arguments. Have it read its
own inputs from your sources.
- Not a generator. A scheduled run has no caller to consume yielded
values, so a scheduled function cannot yield. Return None (or a value)
instead.

Everything else about a Compute Function still applies: custom images,
resource requests, secrets, and retries all work exactly as they do for
on-demand functions.

### Observing runs

Each invocation appears in the Chalk dashboard with its status, logs, and
resource usage, just like a manually invoked function. Emit logs from the
body to make each run observable:

```
from chalk import chalk_logger

@chalkcompute.function(schedule="1h")
def refresh_exchange_rates() -> None:
    chalk_logger.info("refresh started")
    rates = ...  # fetch the latest rates and write them to your store
    chalk_logger.info("refreshed %s rates", len(rates))
```

### Volumes

Attach persistent storage to a function with the volumes parameter. Each
entry is a (name, mount_path) string tuple. Inside the function, read and
write the volume as an ordinary filesystem path:

```
import chalkcompute

@chalkcompute.function(
    volumes=[("model-cache", "/models")],
    gpu="nvidia-l4",
)
def predict(text: str) -> list[float]:
    import os
    import json
    from sentence_transformers import SentenceTransformer

    cache_path = "/models/all-MiniLM-L6-v2"
    if os.path.exists(cache_path):
        model = SentenceTransformer(cache_path)
    else:
        model = SentenceTransformer("all-MiniLM-L6-v2")
        model.save(cache_path)

    return model.encode(text).tolist()
```

Volumes persist across invocations: data written by one call is available to
subsequent calls. This is useful for caching large model weights, storing
intermediate results, or sharing data between functions that mount the same
volume.

### Calling from a DataFrame

You can invoke a function across every row of a dataset using
F.catalog_call() inside a static resolver's
DataFrame pipeline. Chalk handles parallelism, batching, and delivery
automatically.

Unlike .remote() or .defer(), F.catalog_call() only takes effect once
the resolver is applied to your environment with chalk apply. It's
evaluated during chalk query (or an offline query), so there's no way to
call it from a plain script.

For example, define a function that extracts named entities from text:

```
import chalkcompute

@chalkcompute.function(
    image=chalkcompute.Image.base("python:3.12-slim").run_commands(
        "pip install spacy && python -m spacy download en_core_web_sm",
    ),
    cpu="2",
    memory="4Gi",
)
def extract_entities(memo: str) -> str:
    import json
    import spacy
    nlp = spacy.load("en_core_web_sm")
    doc = nlp(memo)
    return json.dumps([
        {"text": ent.text, "label": ent.label_}
        for ent in doc.ents
    ])
```

Then apply it across a parquet file of transaction memos with an
@offline(static=True) resolver:

```
import chalkdf
from chalk import offline
from chalk.features import DataFrame, features, _
import chalk.functions as F

@features
class Transaction:
    id: int
    memo: str
    entities: str

@offline(static=True)
def ingest_transaction_entities() -> DataFrame[
    Transaction.id,
    Transaction.memo,
    Transaction.entities,
]:
    return (
        chalkdf.DataFrame.scan(["transactions.parquet"])
        .with_columns({
            "transaction.entities": F.catalog_call("function.extract_entities", _.memo),
        })
        .select("transaction.id", "transaction.memo", "transaction.entities")
    )
```

Each row's memo column is passed to extract_entities, and the result is
written as the entities feature alongside it. .with_columns() keeps every
existing column and adds the ones you name; use
.project() instead when you want
only the columns you specify. Note the qualified name passed to
catalog_call: it's function.extract_entities, not a bare function name.
model.* and function.* are the two valid namespaces.

### Tracing

Tracing helps you understand where time is spent when a function runs. Use it
to debug slow calls, follow nested function calls, and see which parts of your
application are doing work.

After a traced call runs, view it from the function's Scaling Group page under
Remote Call Traces.

### Enabling tracing

Set tracing=True on a function:

```
import chalkcompute

@chalkcompute.function(
    cpu="1",
    memory="1Gi",
    tracing=True,
)
def score_transaction(transaction_id: str) -> float:
    return 0.42
```

Nested calls appear in the same trace when tracing is enabled on each function:

```
import chalkcompute

@chalkcompute.function(tracing=True)
def fetch_profile(user_id: str) -> dict:
    return {"user_id": user_id, "segment": "active"}


@chalkcompute.function(tracing=True)
def score_user(user_id: str) -> float:
    with chalkcompute.span("score_user.fetch_profile"):
        profile = fetch_profile.remote(user_id)

    return 0.9 if profile["segment"] == "active" else 0.1


with chalkcompute.span("demo.score_user"):
    score = score_user.remote("user_123")
```

In the trace you can see the outer function, the call to fetch_profile, and
the inner function's execution together. Add manual spans like
score_user.fetch_profile to label important parts of your own code.

### Manual spans

Use chalkcompute.span() to label meaningful application work:

```
with chalkcompute.span("checkout.score_transaction"):
    score = score_transaction.remote("txn_123")
```

For a whole caller function, use @chalkcompute.traced:

```
@chalkcompute.traced("batch.score_transactions")
def run_batch(transaction_ids: list[str]) -> list[float]:
    return [score_transaction.remote(txn_id) for txn_id in transaction_ids]
```

Manual spans are useful around model calls, retrieval steps, preprocessing, and
other code you want to separate in the trace. To see a remote function's own
execution, enable tracing=True or a TracingPolicy on that function.

### Tracing policies

Use TracingPolicy when you only want to collect a percentage of calls:

```
@chalkcompute.function(
    tracing=chalkcompute.TracingPolicy(sample_rate=0.25),
)
def recommend(user_id: str) -> list[str]:
    ...
```

sample_rate must be between 0.0 and 1.0.

| Configuration                             | Behavior                                                                                                       |
| ----------------------------------------- | -------------------------------------------------------------------------------------------------------------- |
| `tracing=True`                            | Collect traces for every call.                                                                                 |
| `tracing=TracingPolicy(sample_rate=0.25)` | Collect traces for about 25% of calls.                                                                         |
| `tracing=TracingPolicy(sample_rate=0.0)`  | Trace this function only when its caller is already traced.                                                    |
| `tracing=False`                           | Turn tracing off for the function.                                                                             |
| Omitted                                   | Use the default behavior: only continue a caller's trace if tracing support is already installed in the image. |

Once a call is traced, nested function calls stay in the same trace.

tracing=True and TracingPolicy(...) automatically install tracing support
for functions deployed with @chalkcompute.function or RemoteFunction. If
you omit tracing, Chalk does not add those packages for you. The default can
still continue an existing trace if your image already includes tracing support,
for example because it installs chalkcompute or
chalk-remote-call-python[tracing].

### Troubleshooting

If you do not see remote function spans:

- Redeploy the function after changing tracing.
- Enable tracing on both outer and inner functions to see both functions'
execution in a nested trace.
- Wrap local tests in chalkcompute.span() if you want the call to be part of
an explicit trace.
- Make sure you are looking at the called function's Scaling Group page and a
time range that includes the call.

### Streaming results with async generators

If your function is an async generator (uses yield), the caller receives each
value as soon as the worker produces it, without waiting for the generator to
finish.

This makes functions a good fit for:

- Streaming LLM token output.
- Incremental ETL where downstream consumers start processing early.
- Long-running jobs that emit progress events.

### Example: audio transcription

Transcribe an audio file and yield each segment as it's decoded:

```
import chalkcompute

@chalkcompute.function(
    image=chalkcompute.Image.base("python:3.12-slim").run_commands(
        "pip install faster-whisper httpx",
    ),
    cpu="4",
    memory="8Gi",
)
async def transcribe(audio_uri: str):
    import httpx
    from faster_whisper import WhisperModel

    async with httpx.AsyncClient() as client:
        resp = await client.get(audio_uri)
        resp.raise_for_status()

    audio_path = "/tmp/audio"
    with open(audio_path, "wb") as f:
        f.write(resp.content)

    model = WhisperModel("base.en", compute_type="int8")
    segments, _info = model.transcribe(audio_path)

    for segment in segments:
        yield {
            "start": segment.start,
            "end": segment.end,
            "text": segment.text,
        }
```

The caller iterates over segments as they arrive:

```
import asyncio

transcribe.wait_ready()

async def main():
    async for segment in transcribe.remote("https://example.com/meeting-recording.wav"):
        print(f"[{segment['start']:.1f}s] {segment['text']}")

asyncio.run(main())
```

Output streams in real time:

```
[0.0s]  Welcome to the quarterly review.
[3.2s]  Let's start with the revenue numbers.
[6.1s]  Q3 came in at 4.2 million, up 18 percent.
...
```

The caller receives the first segment as soon as it's decoded. It does not
wait for the full file to finish processing. For a 30-minute recording, the
first segment may arrive in seconds.

### Nested function calls

A function can call other functions with .remote(...). Each inner call is
dispatched independently, so you can compose pipelines where each step runs on
its own infrastructure.

```
import chalkcompute

@chalkcompute.function
def translate(text: str, target_lang: str) -> str:
    from transformers import pipeline
    translator = pipeline("translation", model=f"Helsinki-NLP/opus-mt-en-{target_lang}")
    return translator(text)[0]["translation_text"]


@chalkcompute.function
async def transcribe_and_translate(audio_uri: str, target_lang: str):
    """Transcribe audio, then translate each segment."""
    translate.wait_ready()

    async for segment in transcribe.remote(audio_uri):
        translated = await translate.remote(segment["text"], target_lang)
        yield {
            **segment,
            "translated": translated,
        }
```

The caller sees a single stream of translated segments:

```
import asyncio

transcribe_and_translate.wait_ready()

async def main():
    async for seg in transcribe_and_translate.remote(
        "https://example.com/meeting.wav",
        target_lang="es",
    ):
        print(f"[{seg['start']:.1f}s] {seg['translated']}")

asyncio.run(main())
```

```
[0.0s]  Bienvenidos a la revisión trimestral.
[3.2s]  Comencemos con las cifras de ingresos.
[6.1s]  El tercer trimestre llegó a 4,2 millones, un 18 por ciento más.
```

The outer function composes two inner calls: transcribe (async generator)
and translate (single-value). Each runs on its own compute configuration.

### Error handling

If a function raises an exception, the caller receives it as a RuntimeError with the
original exception type and message:

```
@chalkcompute.function
def divide(a: float, b: float) -> float:
    return a / b

divide.wait_ready()

try:
    divide.remote(1.0, 0.0)
except RuntimeError as e:
    print(e)  # ZeroDivisionError: division by zero
```

For async generators, an exception mid-stream terminates the iterator. Any segments
already yielded are still available to the caller; only subsequent iteration raises:

```
@chalkcompute.function
async def fragile_gen():
    yield "ok-1"
    yield "ok-2"
    raise ValueError("something went wrong")

import asyncio

async def main():
    async for item in fragile_gen.remote():
        print(item)

asyncio.run(main())
# ok-1
# ok-2
# RuntimeError: ValueError: something went wrong
```

### How it works

@chalkcompute.function is backed by Chalk Compute's async execution layer.
At a high level, calling a function goes through three moving parts:

```
┌──────────┐         ┌──────────────────┐         ┌───────────┐
│  Caller  │──call──▸│  Function Queue  │──run───▸│  Worker   │
│          │◂results─│                  │◂results─│ (Python)  │
└──────────┘         └──────────────────┘         └───────────┘
```

- The caller (local script, notebook, another function) submits an
invocation and receives results over a single bidirectional stream.
- The Function Queue durably accepts the work, assigns each call an ID,
enforces configured policies (concurrency, rate limit, retry), and relays
results back to the caller.
- The worker is the Python process that actually runs your function code.

Callers and workers are loosely coupled: the caller never opens a connection
directly to a worker. Workers can scale up, scale down, or get rescheduled
mid-flight and in-flight calls are not lost. This is what enables streaming
async generators, nested calls, and policy-enforced dispatch.

### Policy enforcement

The queue enforces the policies declared on each function (rate limit,
concurrency, and retries) before dispatching work to the worker.

| Policy                               | What it caps                                                                                                                                                                |
| ------------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| [`rate_limit`](#rate-limiting)       | Outbound requests per second / minute / hour, optionally shared across functions via a `key`.                                                                               |
| [`concurrency`](#concurrency-limits) | In-flight invocations at once, optionally shared across functions via a `key`. Defaults to `1` per function.                                                                |
| [`retries`](#retry-policies)         | Automatic retry of transient failures (`UNAVAILABLE`, `DEADLINE_EXCEEDED`, `RESOURCE_EXHAUSTED`, `ABORTED`) with exponential backoff and jitter. Other errors are terminal. |

Concurrency is acquired before dispatching, so a burst of queued calls
backpressures on the semaphore rather than spawning unbounded tasks.
Rate-limit tokens are consumed per attempt: each retry takes a fresh token.

### Retry safety

Retries are safe for failures that happen before the function has started
streaming results: connection errors, DEADLINE_EXCEEDED on the initial
call, etc. If a function has already yielded chunks when it fails, those
chunks are visible to the caller and a retry would produce an ambiguous
stream. Retries therefore apply cleanly only to pre-stream failures today;
making mid-stream retries safe is a known follow-up.