feat(observe): add named_inputs option to record input by parameter name

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
 

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",
+    }