nat.plugins.eval.profiler.decorators.latency

Latency sensitivity decorator for marking functions with latency requirements.

This module provides the @latency_sensitive decorator that allows marking functions with integer latency sensitivity levels. The sensitivity propagates through the call stack with max-based merging, where higher values take precedence.

Use cases: - LLM routing: Direct high-sensitivity requests to low-latency backends - Execution optimization: Adjust timeouts, batch sizes based on sensitivity - Observability: Track which parts of workflows have strict latency requirements

Example:

Basic usage with integers:

from nat.plugins.eval.profiler.decorators.latency import latency_sensitive

@latency_sensitive(3)
async def critical_llm_call():
    return await llm.generate()

Using integer values:

@latency_sensitive(1)
def background_task():
    pass

Reading current sensitivity:

from nat.builder.context import Context

def my_function():
    sensitivity = Context.get().latency_sensitivity
    if sensitivity >= 3:
        # Use fast path
        pass

Attributes

F

Functions

latency_sensitive(→ collections.abc.Callable[[F], F])

Decorator to mark a function with a latency sensitivity level.

Module Contents

F

latency_sensitive(

sensitivity: int,

) → collections.abc.Callable[[F], F]

Decorator to mark a function with a latency sensitivity level.

The sensitivity is pushed onto the context stack for the duration of the function execution. The effective sensitivity is the maximum value across all pushed levels.

Args:

sensitivity: Latency sensitivity level as an integer (e.g. 1=low, 2=medium, 3=high)

Returns:

Decorated function that pushes sensitivity onto context stack

Raises:

TypeError: If sensitivity is not an int

Example:

from nat.plugins.eval.profiler.decorators.latency import latency_sensitive >>> from nat.builder.context import Context >>> >>> @latency_sensitive(3) … def critical_function(): … return Context.get().latency_sensitivity >>> >>> @latency_sensitive(1) … async def background_task(): … return await do_work()