types

Author: nimarb

langfuse/_client/observe.py

@@ -377,7 +377,18 @@ def _sync_observe(
         func: F,
         *,
         name: Optional[str],
-        as_type: Optional[str],
+        as_type: Optional[
+            Literal[
+                "span",
+                "generation",
+                "event",
+                "agent",
+                "tool",
+                "chain",
+                "retriever",
+                "embedding",
+            ]
+        ],
         capture_input: bool,
         capture_output: bool,
         transform_to_string: Optional[Callable[[Iterable], str]] = None,

langfuse/_client/span.py

@@ -1333,7 +1333,7 @@ def update(
 class LangfuseAgent(LangfuseGraphObservation):
     """Specialized span for agent observations in agentic workflows."""
 
-    def __init__(self, **kwargs) -> None:
+    def __init__(self, **kwargs: Any) -> None:
         """Initialize a new LangfuseAgent span."""
         kwargs["observation_type"] = "agent"
         super().__init__(**kwargs)
@@ -1342,7 +1342,7 @@ def __init__(self, **kwargs) -> None:
 class LangfuseTool(LangfuseGraphObservation):
     """Specialized span for tool observations in agentic workflows."""
 
-    def __init__(self, **kwargs) -> None:
+    def __init__(self, **kwargs: Any) -> None:
         """Initialize a new LangfuseTool span."""
         kwargs["observation_type"] = "tool"
         super().__init__(**kwargs)
@@ -1351,7 +1351,7 @@ def __init__(self, **kwargs) -> None:
 class LangfuseChain(LangfuseGraphObservation):
     """Specialized span for chain observations in agentic workflows."""
 
-    def __init__(self, **kwargs) -> None:
+    def __init__(self, **kwargs: Any) -> None:
         """Initialize a new LangfuseChain span."""
         kwargs["observation_type"] = "chain"
         super().__init__(**kwargs)
@@ -1360,7 +1360,7 @@ def __init__(self, **kwargs) -> None:
 class LangfuseRetriever(LangfuseGraphObservation):
     """Specialized span for retriever observations in agentic workflows."""
 
-    def __init__(self, **kwargs) -> None:
+    def __init__(self, **kwargs: Any) -> None:
         """Initialize a new LangfuseRetriever span."""
         kwargs["observation_type"] = "retriever"
         super().__init__(**kwargs)
@@ -1369,7 +1369,7 @@ def __init__(self, **kwargs) -> None:
 class LangfuseEmbedding(LangfuseGraphObservation):
     """Specialized span for embedding observations in agentic workflows."""
 
-    def __init__(self, **kwargs) -> None:
+    def __init__(self, **kwargs: Any) -> None:
         """Initialize a new LangfuseEmbedding span."""
         kwargs["observation_type"] = "embedding"
         super().__init__(**kwargs)