diff --git a/langfuse/_client/observe.py b/langfuse/_client/observe.py index 5aa88ce2e..a609d2d9b 100644 --- a/langfuse/_client/observe.py +++ b/langfuse/_client/observe.py @@ -93,6 +93,7 @@ def observe( capture_input: Optional[bool] = None, capture_output: Optional[bool] = None, transform_to_string: Optional[Callable[[Iterable], str]] = None, + named_inputs: bool = False, ) -> Callable[[F], F]: ... def observe( @@ -104,6 +105,7 @@ def observe( capture_input: Optional[bool] = None, capture_output: Optional[bool] = None, transform_to_string: Optional[Callable[[Iterable], str]] = None, + named_inputs: bool = False, ) -> Union[F, Callable[[F], F]]: """Wrap a function to create and manage Langfuse tracing around its execution, supporting both synchronous and asynchronous functions. @@ -132,6 +134,8 @@ def observe( transform_to_string (Optional[Callable[[Iterable], str]]): For functions returning generators, joins the yielded chunks into the string stored as output. Without it, chunks are concatenated if all are strings, otherwise stored as a list. + named_inputs (bool): Record input as `{"param": value}` instead of `{"args": [...], "kwargs": {...}}`. + Falls back to the latter if the arguments can't be bound to the signature. Defaults to False. Returns: Callable: A wrapped version of the original function that automatically creates and manages Langfuse spans. @@ -224,6 +228,7 @@ def decorator(func: F) -> F: capture_input=should_capture_input, capture_output=should_capture_output, transform_to_string=transform_to_string, + named_inputs=named_inputs, ) if _is_coroutine_function(func) else self._sync_observe( @@ -233,6 +238,7 @@ def decorator(func: F) -> F: capture_input=should_capture_input, capture_output=should_capture_output, transform_to_string=transform_to_string, + named_inputs=named_inputs, ) ) @@ -260,6 +266,7 @@ def _async_observe( capture_input: bool, capture_output: bool, transform_to_string: Optional[Callable[[Iterable], str]] = None, + named_inputs: bool = False, ) -> F: @wraps(func) async def async_wrapper(*args: Tuple[Any], **kwargs: Dict[str, Any]) -> Any: @@ -281,6 +288,8 @@ async def async_wrapper(*args: Tuple[Any], **kwargs: Dict[str, Any]) -> Any: is_method=self._is_method(func), func_args=args, func_kwargs=kwargs, + func=func, + named_inputs=named_inputs, ) if capture_input else None @@ -353,6 +362,7 @@ def _sync_observe( capture_input: bool, capture_output: bool, transform_to_string: Optional[Callable[[Iterable], str]] = None, + named_inputs: bool = False, ) -> F: @wraps(func) def sync_wrapper(*args: Any, **kwargs: Any) -> Any: @@ -372,6 +382,8 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any: is_method=self._is_method(func), func_args=args, func_kwargs=kwargs, + func=func, + named_inputs=named_inputs, ) if capture_input else None @@ -448,7 +460,30 @@ def _get_input_from_func_args( is_method: bool = False, func_args: Tuple = (), func_kwargs: Dict = {}, + func: Optional[Callable] = None, + named_inputs: bool = False, ) -> Dict: + if named_inputs and func is not None: + try: + signature = inspect.signature(func) + bound = signature.bind( + *func_args, + **{ + k: v + for k, v in func_kwargs.items() + if k != "langfuse_public_key" + }, + ) + except Exception: + pass # fall back to the args/kwargs format below + else: + arguments = dict(bound.arguments) + first_param = next(iter(signature.parameters), None) + if is_method and first_param in ("self", "cls"): + arguments.pop(first_param, None) + + return arguments + # Remove implicitly passed "self" or "cls" argument for instance or class methods logged_args = func_args[1:] if is_method else func_args diff --git a/tests/unit/test_observe.py b/tests/unit/test_observe.py index 21ff1418d..e86ef6f4d 100644 --- a/tests/unit/test_observe.py +++ b/tests/unit/test_observe.py @@ -523,3 +523,128 @@ def marked() -> Any: span = _finished_spans_by_name(memory_exporter, "marked")[0] assert span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT] == "done" + + +def _observation_input(memory_exporter: Any, name: str) -> Any: + span = _finished_spans_by_name(memory_exporter, name)[0] + return json.loads(span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_INPUT]) + + +def test_named_inputs_records_passed_arguments_by_name( + langfuse_memory_client: Any, memory_exporter: Any +) -> None: + @observe(name="positional", named_inputs=True) + def positional(query: str, lang: str, limit: int = 10) -> str: + return query + + @observe(name="keyword", named_inputs=True) + def keyword(query: str, lang: str, limit: int = 10) -> str: + return query + + positional("how do I return an item?", "en") + keyword(lang="en", query="how do I return an item?") + + langfuse_memory_client.flush() + + expected = {"query": "how do I return an item?", "lang": "en"} + assert _observation_input(memory_exporter, "positional") == expected + assert _observation_input(memory_exporter, "keyword") == expected + + +def test_named_inputs_excludes_self( + langfuse_memory_client: Any, memory_exporter: Any +) -> None: + class Assistant: + @observe(name="answer", named_inputs=True) + def answer(self, query: str) -> str: + return query + + Assistant().answer("q") + + langfuse_memory_client.flush() + + assert _observation_input(memory_exporter, "answer") == {"query": "q"} + + +@pytest.mark.asyncio +async def test_named_inputs_for_async_function( + langfuse_memory_client: Any, memory_exporter: Any +) -> None: + @observe(name="async_answer", named_inputs=True) + async def async_answer(query: str, lang: str) -> str: + return query + + await async_answer("q", lang="en") + + langfuse_memory_client.flush() + + assert _observation_input(memory_exporter, "async_answer") == { + "query": "q", + "lang": "en", + } + + +def test_named_inputs_falls_back_when_binding_fails( + langfuse_memory_client: Any, memory_exporter: Any +) -> None: + @observe(name="strict", named_inputs=True) + def strict(query: str) -> str: + return query + + with pytest.raises(TypeError): + cast(Any, strict)("q", extra=1) + + langfuse_memory_client.flush() + + assert _observation_input(memory_exporter, "strict") == { + "args": ["q"], + "kwargs": {"extra": 1}, + } + + +def test_inputs_are_not_named_by_default( + langfuse_memory_client: Any, memory_exporter: Any +) -> None: + @observe(name="default_input") + def default_input(query: str, lang: str) -> str: + return query + + default_input("q", lang="en") + + langfuse_memory_client.flush() + + assert _observation_input(memory_exporter, "default_input") == { + "args": ["q"], + "kwargs": {"lang": "en"}, + } + + +def test_named_inputs_excludes_langfuse_public_key( + langfuse_memory_client: Any, memory_exporter: Any +) -> None: + @observe(name="with_public_key", named_inputs=True) + def with_public_key(query: str) -> str: + return query + + cast(Any, with_public_key)("q", langfuse_public_key="test-public-key") + + langfuse_memory_client.flush() + + assert _observation_input(memory_exporter, "with_public_key") == {"query": "q"} + + +def test_named_inputs_keeps_first_argument_when_cls_is_not_first( + langfuse_memory_client: Any, memory_exporter: Any +) -> None: + @observe(name="parse", named_inputs=True) + def parse(data: str, *, cls: str) -> str: + return data + + parse(data="payload", cls="Model") + + langfuse_memory_client.flush() + + assert _observation_input(memory_exporter, "parse") == { + "data": "payload", + "cls": "Model", + }