Observe

Wrap functions with Laminar to capture inputs, outputs, and errors automatically.

TypeScript

observe(options, fn, …args)

Wrap a function to create a span. Automatically captures inputs, outputs, and errors.

import { observe } from '@lmnr-ai/lmnr';

const result = await observe(
  { name: 'my_operation' },
  async (param1, param2) => {
    return param1 + param2;
  },
  1, 2
);
// Span input: [1, 2], output: 3

Parameters:

Name	Type	Default	Description
name	string	fn.name	Span name
sessionId	string	—	Associate trace with session
userId	string	—	Associate trace with user
metadata	Record<string, any>	—	Trace metadata (JSON-serializable)
tags	string[]	—	Span tags (deduplicated)
traceType	'DEFAULT' | 'EVALUATION'	'DEFAULT'	Trace type
spanType	'DEFAULT' | 'LLM' | 'TOOL' | 'EXECUTOR' | 'EVALUATOR' | 'HUMAN_EVALUATOR' | 'EVALUATION' | 'CACHED'	'DEFAULT'	Span type. See Span Types.
input	any	—	Explicit input (overrides function args)
ignoreInput	boolean	false	Don’t record input
ignoreOutput	boolean	false	Don’t record output
parentSpanContext	string | LaminarSpanContext	—	Parent context for distributed tracing
rolloutEntrypoint	boolean	false	Return a wrapped function for debugger instead of executing immediately

Returns: Promise<ReturnType<F>> (default) or wrapped function when rolloutEntrypoint: true

Behavior:

• If one non-Map object argument, records that object as input; otherwise records array
• Maps converted to entries, circular references become "[Circular reference]"
• Exceptions recorded via recordException and re-thrown
• Anonymous function name becomes "" if name not provided
• Works with both sync and async functions
• When rolloutEntrypoint is true, returns a wrapped function and does not execute immediately
• In debugger, Laminar attaches rollout.session_id from LMNR_ROLLOUT_SESSION_ID

Examples: Rollout entrypoint

import { observe } from '@lmnr-ai/lmnr';

export const generateStream = observe(
  { name: 'generateStream', rolloutEntrypoint: true },
  async (messages: string[]) => {
    // Your agent code here.
  },
);
// Start a session with: npx lmnr-cli@latest dev path/to/entrypoint.js

Examples: Group multiple LLM calls under one parent span

import { observe } from '@lmnr-ai/lmnr';

export const handle = async (userMessage: string) =>
  observe({ name: 'requestHandler' }, async () => {
    const routerResponse = await openai.chat.completions.create({
      model: 'gpt-4o-mini',
      messages: [{ role: 'user', content: `Route this: ${userMessage}` }],
    });

    const modelResponse = await openai.chat.completions.create({
      model: 'gpt-4o',
      messages: [{ role: 'user', content: userMessage }],
    });

    return {
      route: routerResponse.choices[0].message.content,
      answer: modelResponse.choices[0].message.content,
    };
  });

Examples: Nested spans for multi-step workflows

import { observe } from '@lmnr-ai/lmnr';

await observe({ name: 'checkout', sessionId: `checkout-${userId}` }, async () => {
  await observe({ name: 'validateCart' }, async () => {
    // ...
  });

  await observe({ name: 'processPayment' }, async () => {
    // ...
  });

  await observe({ name: 'createOrder' }, async () => {
    // ...
  });
});

---

observeDecorator(config)

Class method decorator version of observe.

import { observeDecorator } from '@lmnr-ai/lmnr';

class MyService {
  @observeDecorator({ name: 'process_request' })
  async processRequest(input: string) {
    return await this.doWork(input);
  }
}

Parameters: Same as observe options. Config can be an object or a function (thisArg, ...args) => config for per-call configuration.

Returns: Method decorator

Python

@observe()

Decorate a function to create a span. Automatically captures inputs, outputs, and errors.

from lmnr import observe

@observe()
def my_function(param1, param2):
    return param1 + param2

my_function(1, 2)
# Span input: {"param1": 1, "param2": 2}
# Span output: 3

Parameters:

Name	Type	Default	Description
name	str	Function name	Span name
session_id	str	None	Associate trace with session
user_id	str	None	Associate trace with user
metadata	dict	None	Trace metadata (JSON-serializable)
tags	list[str]	None	Span tags
span_type	'DEFAULT' | 'LLM' | 'TOOL' | 'EXECUTOR' | 'EVALUATOR' | 'HUMAN_EVALUATOR' | 'EVALUATION'	'DEFAULT'	Span type (PIPELINE is reserved internally). See Span Types.
ignore_input	bool	False	Don’t record input
ignore_output	bool	False	Don’t record output
ignore_inputs	list[str]	None	Specific parameters to skip
input_formatter	Callable	None	Custom input formatting function
output_formatter	Callable	None	Custom output formatting function
preserve_global_context	bool	False	Use global OTEL context instead of isolated
rollout_entrypoint	bool	False	Register function as rollout entrypoint when running in rollout mode

Returns: Decorator

Behavior:

• Function arguments serialized via orjson to lmnr.span.input
• Return value serialized to lmnr.span.output
• Inputs/outputs >10MB are truncated with warning message
• Exceptions recorded as span events with ERROR status
• Generators/async generators: yielded parts aggregated for output
• If Laminar not initialized, returns original function unchanged
• Async functions supported automatically

Examples: Group multiple LLM calls under one parent span

from lmnr import observe

@observe()
def handle(user_message: str) -> dict:
    router = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"Route this: {user_message}"}],
    )

    model = openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": user_message}],
    )

    return {
        "route": router.choices[0].message.content,
        "answer": model.choices[0].message.content,
    }

Examples: Nested spans for multi-step workflows

from lmnr import Laminar, observe

@observe()
def checkout(user_id: str):
    Laminar.set_trace_session_id(f"checkout-{user_id}")
    validate_cart()
    process_payment()
    create_order()

@observe()
def validate_cart():
    # ...
    pass

@observe()
def process_payment():
    # ...
    pass

@observe()
def create_order():
    # ...
    pass

Examples: Rollout entrypoint

from lmnr import observe

@observe(rollout_entrypoint=True)
def generate_stream(messages):
    # Your agent code here.
    pass
# Start a session with: lmnr dev path/to/entrypoint.py