Method set_input

agentlightning/emitter/annotation.py:143–171

Record input arguments on the current span. Positional arguments are stored under the `input.args. ` attributes, and keyword arguments are stored under `input. ` attributes. This is intended for use inside a `with operation(...) as op` block. Args:

(self, *args: Any, **kwargs: Any)

        return self._span

    def set_input(self, *args: Any, **kwargs: Any) -> None:
        """Record input arguments on the current span.

        Positional arguments are stored under the `input.args.<index>` attributes,
        and keyword arguments are stored under `input.<name>` attributes.

        This is intended for use inside a `with operation(...) as op` block.

        Args:
            *args: Positional arguments to record.
            **kwargs: Keyword arguments to record.
        """
        if not self._recording_context:
            raise RuntimeError("No recording context found. Cannot set input.")

        prefix = LightningSpanAttributes.OPERATION_INPUT.value
        attributes: Dict[str, Any] = {}
        if args:
            for idx, value in enumerate(args):
                flattened = flatten_attributes({str(idx): value})
                for nested_key, nested_value in flattened.items():
                    attributes[f"{prefix}.args.{nested_key}"] = nested_value
        if kwargs:
            for key, value in kwargs.items():
                flattened = flatten_attributes({key: value})
                for nested_key, nested_value in flattened.items():
                    attributes[f"{prefix}.{nested_key}"] = nested_value
        if attributes:
            self._recording_context.record_attributes(sanitize_attributes(attributes))

    def set_output(self, output: Any) -> None:
        """Record the output value on the current span.