Fix unit test

* feat: refactor to dataclasses

* feat: spans

* test: fix part 1

* test: fix part 2

* test: fix part 3

* test: fix part n

* test: add langgraph agent test

* chore: bump and linters

* chore: bump

* fix: correctly capture a parent id when a custom trace_id is set

* test: multiple spans parent_ids

* fix: exception serialization

* refactor: ai_trace_name -> ai_span_name and ai_generation_id -> ai_span_id

* fix: logs typos

* Add minor breaking change note to changelog

* fix: naming

---------

Co-authored-by: Michael Matloka <michael@matloka.com>

CHANGELOG.md

@@ -1,3 +1,18 @@
+## 3.11.0 - 2025-01-28
+
+1. Add the `$ai_span` event to the LangChain callback handler to capture the input and output of intermediary chains.
+   > LLM observability naming change: event property `$ai_trace_name` is now `$ai_span_name`.
+
+2. Fix serialiazation of Pydantic models in methods.
+
+## 3.10.0 - 2025-01-24
+
+1. Add `$ai_error` and `$ai_is_error` properties to LangChain callback handler, OpenAI, and Anthropic.
+
+## 3.9.3 - 2025-01-23
+
+1. Fix capturing of multiple traces in the LangChain callback handler.
+
 ## 3.9.2 - 2025-01-22
 
 1. Fix importing of LangChain callback handler under certain circumstances.

posthog/__init__.py

@@ -36,7 +36,6 @@ def capture(
     distinct_id,  # type: str
     event,  # type: str
     properties=None,  # type: Optional[Dict]
-    context=None,  # type: Optional[Dict]
     timestamp=None,  # type: Optional[datetime.datetime]
     uuid=None,  # type: Optional[str]
     groups=None,  # type: Optional[Dict]
@@ -69,7 +68,6 @@ def capture(
         distinct_id=distinct_id,
         event=event,
         properties=properties,
-        context=context,
         timestamp=timestamp,
         uuid=uuid,
         groups=groups,
@@ -81,7 +79,6 @@ def capture(
 def identify(
     distinct_id,  # type: str
     properties=None,  # type: Optional[Dict]
-    context=None,  # type: Optional[Dict]
     timestamp=None,  # type: Optional[datetime.datetime]
     uuid=None,  # type: Optional[str]
     disable_geoip=None,  # type: Optional[bool]
@@ -106,7 +103,6 @@ def identify(
         "identify",
         distinct_id=distinct_id,
         properties=properties,
-        context=context,
         timestamp=timestamp,
         uuid=uuid,
         disable_geoip=disable_geoip,
@@ -116,7 +112,6 @@ def identify(
 def set(
     distinct_id,  # type: str
     properties=None,  # type: Optional[Dict]
-    context=None,  # type: Optional[Dict]
     timestamp=None,  # type: Optional[datetime.datetime]
     uuid=None,  # type: Optional[str]
     disable_geoip=None,  # type: Optional[bool]
@@ -141,7 +136,6 @@ def set(
         "set",
         distinct_id=distinct_id,
         properties=properties,
-        context=context,
         timestamp=timestamp,
         uuid=uuid,
         disable_geoip=disable_geoip,
@@ -151,7 +145,6 @@ def set(
 def set_once(
     distinct_id,  # type: str
     properties=None,  # type: Optional[Dict]
-    context=None,  # type: Optional[Dict]
     timestamp=None,  # type: Optional[datetime.datetime]
     uuid=None,  # type: Optional[str]
     disable_geoip=None,  # type: Optional[bool]
@@ -176,7 +169,6 @@ def set_once(
         "set_once",
         distinct_id=distinct_id,
         properties=properties,
-        context=context,
         timestamp=timestamp,
         uuid=uuid,
         disable_geoip=disable_geoip,
@@ -187,7 +179,6 @@ def group_identify(
     group_type,  # type: str
     group_key,  # type: str
     properties=None,  # type: Optional[Dict]
-    context=None,  # type: Optional[Dict]
     timestamp=None,  # type: Optional[datetime.datetime]
     uuid=None,  # type: Optional[str]
     disable_geoip=None,  # type: Optional[bool]
@@ -213,7 +204,6 @@ def group_identify(
         group_type=group_type,
         group_key=group_key,
         properties=properties,
-        context=context,
         timestamp=timestamp,
         uuid=uuid,
         disable_geoip=disable_geoip,
@@ -223,7 +213,6 @@ def group_identify(
 def alias(
     previous_id,  # type: str
     distinct_id,  # type: str
-    context=None,  # type: Optional[Dict]
     timestamp=None,  # type: Optional[datetime.datetime]
     uuid=None,  # type: Optional[str]
     disable_geoip=None,  # type: Optional[bool]
@@ -249,7 +238,6 @@ def alias(
         "alias",
         previous_id=previous_id,
         distinct_id=distinct_id,
-        context=context,
         timestamp=timestamp,
         uuid=uuid,
         disable_geoip=disable_geoip,

posthog/ai/langchain/callbacks.py

@@ -5,14 +5,14 @@ except ImportError:
 
 import logging
 import time
-import uuid
+from dataclasses import dataclass
 from typing import (
     Any,
     Dict,
     List,
     Optional,
+    Sequence,
     Tuple,
-    TypedDict,
     Union,
     cast,
 )
@@ -20,6 +20,7 @@ from uuid import UUID
 
 from langchain.callbacks.base import BaseCallbackHandler
 from langchain.schema.agent import AgentAction, AgentFinish
+from langchain_core.documents import Document
 from langchain_core.messages import AIMessage, BaseMessage, FunctionMessage, HumanMessage, SystemMessage, ToolMessage
 from langchain_core.outputs import ChatGeneration, LLMResult
 from pydantic import BaseModel
@@ -31,17 +32,38 @@ from posthog.client import Client
 log = logging.getLogger("posthog")
 
 
-class RunMetadata(TypedDict, total=False):
-    messages: Union[List[Dict[str, Any]], List[str]]
-    provider: str
-    model: str
-    model_params: Dict[str, Any]
-    base_url: str
+@dataclass
+class SpanMetadata:
+    name: str
+    """Name of the run: chain name, model name, etc."""
     start_time: float
-    end_time: float
+    """Start time of the run."""
+    end_time: Optional[float]
+    """End time of the run."""
+    input: Optional[Any]
+    """Input of the run: messages, prompt variables, etc."""
+
+    @property
+    def latency(self) -> float:
+        if not self.end_time:
+            return 0
+        return self.end_time - self.start_time
 
 
-RunStorage = Dict[UUID, RunMetadata]
+@dataclass
+class GenerationMetadata(SpanMetadata):
+    provider: Optional[str] = None
+    """Provider of the run: OpenAI, Anthropic"""
+    model: Optional[str] = None
+    """Model used in the run"""
+    model_params: Optional[Dict[str, Any]] = None
+    """Model parameters of the run: temperature, max_tokens, etc."""
+    base_url: Optional[str] = None
+    """Base URL of the provider's API used in the run."""
+
+
+RunMetadata = Union[SpanMetadata, GenerationMetadata]
+RunMetadataStorage = Dict[UUID, RunMetadata]
 
 
 class CallbackHandler(BaseCallbackHandler):
@@ -67,7 +89,7 @@ class CallbackHandler(BaseCallbackHandler):
     _properties: Optional[Dict[str, Any]]
     """Global properties to be sent with every event."""
 
-    _runs: RunStorage
+    _runs: RunMetadataStorage
     """Mapping of run IDs to run metadata as run metadata is only available on the start of generation."""
 
     _parent_tree: Dict[UUID, UUID]
@@ -95,11 +117,12 @@ class CallbackHandler(BaseCallbackHandler):
             privacy_mode: Whether to redact the input and output of the trace.
             groups: Optional additional PostHog groups to use for the trace.
         """
-        self._client = client or default_client
+        posthog_client = client or default_client
+        if posthog_client is None:
+            raise ValueError("PostHog client is required")
+        self._client = posthog_client
         self._distinct_id = distinct_id
         self._trace_id = trace_id
-        self._trace_name = None
-        self._trace_input = None
         self._properties = properties or {}
         self._privacy_mode = privacy_mode
         self._groups = groups or {}
@@ -118,9 +141,29 @@ class CallbackHandler(BaseCallbackHandler):
     ):
         self._log_debug_event("on_chain_start", run_id, parent_run_id, inputs=inputs)
         self._set_parent_of_run(run_id, parent_run_id)
-        if parent_run_id is None and self._trace_name is None:
-            self._trace_name = self._get_langchain_run_name(serialized, **kwargs)
-            self._trace_input = inputs
+        self._set_trace_or_span_metadata(serialized, inputs, run_id, parent_run_id, **kwargs)
+
+    def on_chain_end(
+        self,
+        outputs: Dict[str, Any],
+        *,
+        run_id: UUID,
+        parent_run_id: Optional[UUID] = None,
+        **kwargs: Any,
+    ):
+        self._log_debug_event("on_chain_end", run_id, parent_run_id, outputs=outputs)
+        self._pop_run_and_capture_trace_or_span(run_id, parent_run_id, outputs)
+
+    def on_chain_error(
+        self,
+        error: BaseException,
+        *,
+        run_id: UUID,
+        parent_run_id: Optional[UUID] = None,
+        **kwargs: Any,
+    ):
+        self._log_debug_event("on_chain_error", run_id, parent_run_id, error=error)
+        self._pop_run_and_capture_trace_or_span(run_id, parent_run_id, error)
 
     def on_chat_model_start(
         self,
@@ -134,7 +177,7 @@ class CallbackHandler(BaseCallbackHandler):
         self._log_debug_event("on_chat_model_start", run_id, parent_run_id, messages=messages)
         self._set_parent_of_run(run_id, parent_run_id)
         input = [_convert_message_to_dict(message) for row in messages for message in row]
-        self._set_run_metadata(serialized, run_id, input, **kwargs)
+        self._set_llm_metadata(serialized, run_id, input, **kwargs)
 
     def on_llm_start(
         self,
@@ -147,7 +190,7 @@ class CallbackHandler(BaseCallbackHandler):
     ):
         self._log_debug_event("on_llm_start", run_id, parent_run_id, prompts=prompts)
         self._set_parent_of_run(run_id, parent_run_id)
-        self._set_run_metadata(serialized, run_id, prompts, **kwargs)
+        self._set_llm_metadata(serialized, run_id, prompts, **kwargs)
 
     def on_llm_new_token(
         self,
@@ -160,66 +203,6 @@ class CallbackHandler(BaseCallbackHandler):
         """Run on new LLM token. Only available when streaming is enabled."""
         self._log_debug_event("on_llm_new_token", run_id, parent_run_id, token=token)
 
-    def on_tool_start(
-        self,
-        serialized: Optional[Dict[str, Any]],
-        input_str: str,
-        *,
-        run_id: UUID,
-        parent_run_id: Optional[UUID] = None,
-        metadata: Optional[Dict[str, Any]] = None,
-        **kwargs: Any,
-    ) -> Any:
-        self._log_debug_event("on_tool_start", run_id, parent_run_id, input_str=input_str)
-
-    def on_tool_end(
-        self,
-        output: str,
-        *,
-        run_id: UUID,
-        parent_run_id: Optional[UUID] = None,
-        **kwargs: Any,
-    ) -> Any:
-        self._log_debug_event("on_tool_end", run_id, parent_run_id, output=output)
-
-    def on_tool_error(
-        self,
-        error: Union[Exception, KeyboardInterrupt],
-        *,
-        run_id: UUID,
-        parent_run_id: Optional[UUID] = None,
-        **kwargs: Any,
-    ) -> Any:
-        self._log_debug_event("on_tool_error", run_id, parent_run_id, error=error)
-
-    def on_chain_end(
-        self,
-        outputs: Dict[str, Any],
-        *,
-        run_id: UUID,
-        parent_run_id: Optional[UUID] = None,
-        **kwargs: Any,
-    ):
-        self._log_debug_event("on_chain_end", run_id, parent_run_id, outputs=outputs)
-        self._pop_parent_of_run(run_id)
-
-        if parent_run_id is None:
-            self._capture_trace(run_id, outputs=outputs)
-
-    def on_chain_error(
-        self,
-        error: BaseException,
-        *,
-        run_id: UUID,
-        parent_run_id: Optional[UUID] = None,
-        **kwargs: Any,
-    ):
-        self._log_debug_event("on_chain_error", run_id, parent_run_id, error=error)
-        self._pop_parent_of_run(run_id)
-
-        if parent_run_id is None:
-            self._capture_trace(run_id, outputs=None)
-
     def on_llm_end(
         self,
         response: LLMResult,
@@ -232,45 +215,7 @@ class CallbackHandler(BaseCallbackHandler):
         The callback works for both streaming and non-streaming runs. For streaming runs, the chain must set `stream_usage=True` in the LLM.
         """
         self._log_debug_event("on_llm_end", run_id, parent_run_id, response=response, kwargs=kwargs)
-        trace_id = self._get_trace_id(run_id)
-        self._pop_parent_of_run(run_id)
-        run = self._pop_run_metadata(run_id)
-        if not run:
-            return
-
-        latency = run.get("end_time", 0) - run.get("start_time", 0)
-        input_tokens, output_tokens = _parse_usage(response)
-
-        generation_result = response.generations[-1]
-        if isinstance(generation_result[-1], ChatGeneration):
-            output = [
-                _convert_message_to_dict(cast(ChatGeneration, generation).message) for generation in generation_result
-            ]
-        else:
-            output = [_extract_raw_esponse(generation) for generation in generation_result]
-
-        event_properties = {
-            "$ai_provider": run.get("provider"),
-            "$ai_model": run.get("model"),
-            "$ai_model_parameters": run.get("model_params"),
-            "$ai_input": with_privacy_mode(self._client, self._privacy_mode, run.get("messages")),
-            "$ai_output_choices": with_privacy_mode(self._client, self._privacy_mode, output),
-            "$ai_http_status": 200,
-            "$ai_input_tokens": input_tokens,
-            "$ai_output_tokens": output_tokens,
-            "$ai_latency": latency,
-            "$ai_trace_id": trace_id,
-            "$ai_base_url": run.get("base_url"),
-            **self._properties,
-        }
-        if self._distinct_id is None:
-            event_properties["$process_person_profile"] = False
-        self._client.capture(
-            distinct_id=self._distinct_id or trace_id,
-            event="$ai_generation",
-            properties=event_properties,
-            groups=self._groups,
-        )
+        self._pop_run_and_capture_generation(run_id, parent_run_id, response)
 
     def on_llm_error(
         self,
@@ -281,32 +226,43 @@ class CallbackHandler(BaseCallbackHandler):
         **kwargs: Any,
     ):
         self._log_debug_event("on_llm_error", run_id, parent_run_id, error=error)
-        trace_id = self._get_trace_id(run_id)
-        self._pop_parent_of_run(run_id)
-        run = self._pop_run_metadata(run_id)
-        if not run:
-            return
+        self._pop_run_and_capture_generation(run_id, parent_run_id, error)
 
-        latency = run.get("end_time", 0) - run.get("start_time", 0)
-        event_properties = {
-            "$ai_provider": run.get("provider"),
-            "$ai_model": run.get("model"),
-            "$ai_model_parameters": run.get("model_params"),
-            "$ai_input": with_privacy_mode(self._client, self._privacy_mode, run.get("messages")),
-            "$ai_http_status": _get_http_status(error),
-            "$ai_latency": latency,
-            "$ai_trace_id": trace_id,
-            "$ai_base_url": run.get("base_url"),
-            **self._properties,
-        }
-        if self._distinct_id is None:
-            event_properties["$process_person_profile"] = False
-        self._client.capture(
-            distinct_id=self._distinct_id or trace_id,
-            event="$ai_generation",
-            properties=event_properties,
-            groups=self._groups,
-        )
+    def on_tool_start(
+        self,
+        serialized: Optional[Dict[str, Any]],
+        input_str: str,
+        *,
+        run_id: UUID,
+        parent_run_id: Optional[UUID] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+        **kwargs: Any,
+    ) -> Any:
+        self._log_debug_event("on_tool_start", run_id, parent_run_id, input_str=input_str)
+        self._set_trace_or_span_metadata(serialized, input_str, run_id, parent_run_id, **kwargs)
+
+    def on_tool_end(
+        self,
+        output: str,
+        *,
+        run_id: UUID,
+        parent_run_id: Optional[UUID] = None,
+        **kwargs: Any,
+    ) -> Any:
+        self._log_debug_event("on_tool_end", run_id, parent_run_id, output=output)
+        self._pop_run_and_capture_trace_or_span(run_id, parent_run_id, output)
+
+    def on_tool_error(
+        self,
+        error: BaseException,
+        *,
+        run_id: UUID,
+        parent_run_id: Optional[UUID] = None,
+        tags: Optional[list[str]] = None,
+        **kwargs: Any,
+    ) -> Any:
+        self._log_debug_event("on_tool_error", run_id, parent_run_id, error=error)
+        self._pop_run_and_capture_trace_or_span(run_id, parent_run_id, error)
 
     def on_retriever_start(
         self,
@@ -319,17 +275,31 @@ class CallbackHandler(BaseCallbackHandler):
         **kwargs: Any,
     ) -> Any:
         self._log_debug_event("on_retriever_start", run_id, parent_run_id, query=query)
+        self._set_trace_or_span_metadata(serialized, query, run_id, parent_run_id, **kwargs)
 
-    def on_retriever_error(
+    def on_retriever_end(
         self,
-        error: Union[Exception, KeyboardInterrupt],
+        documents: Sequence[Document],
         *,
         run_id: UUID,
         parent_run_id: Optional[UUID] = None,
         **kwargs: Any,
+    ):
+        self._log_debug_event("on_retriever_end", run_id, parent_run_id, documents=documents)
+        self._pop_run_and_capture_trace_or_span(run_id, parent_run_id, documents)
+
+    def on_retriever_error(
+        self,
+        error: BaseException,
+        *,
+        run_id: UUID,
+        parent_run_id: Optional[UUID] = None,
+        tags: Optional[list[str]] = None,
+        **kwargs: Any,
     ) -> Any:
         """Run when Retriever errors."""
         self._log_debug_event("on_retriever_error", run_id, parent_run_id, error=error)
+        self._pop_run_and_capture_trace_or_span(run_id, parent_run_id, error)
 
     def on_agent_action(
         self,
@@ -341,6 +311,8 @@ class CallbackHandler(BaseCallbackHandler):
     ) -> Any:
         """Run on agent action."""
         self._log_debug_event("on_agent_action", run_id, parent_run_id, action=action)
+        self._set_parent_of_run(run_id, parent_run_id)
+        self._set_trace_or_span_metadata(None, action, run_id, parent_run_id, **kwargs)
 
     def on_agent_finish(
         self,
@@ -351,6 +323,7 @@ class CallbackHandler(BaseCallbackHandler):
         **kwargs: Any,
     ) -> Any:
         self._log_debug_event("on_agent_finish", run_id, parent_run_id, finish=finish)
+        self._pop_run_and_capture_trace_or_span(run_id, parent_run_id, finish)
 
     def _set_parent_of_run(self, run_id: UUID, parent_run_id: Optional[UUID] = None):
         """
@@ -377,7 +350,19 @@ class CallbackHandler(BaseCallbackHandler):
             id = self._parent_tree[id]
         return id
 
-    def _set_run_metadata(
+    def _set_trace_or_span_metadata(
+        self,
+        serialized: Optional[Dict[str, Any]],
+        input: Any,
+        run_id: UUID,
+        parent_run_id: Optional[UUID] = None,
+        **kwargs,
+    ):
+        default_name = "trace" if parent_run_id is None else "span"
+        run_name = _get_langchain_run_name(serialized, **kwargs) or default_name
+        self._runs[run_id] = SpanMetadata(name=run_name, input=input, start_time=time.time(), end_time=None)
+
+    def _set_llm_metadata(
         self,
         serialized: Dict[str, Any],
         run_id: UUID,
@@ -386,24 +371,22 @@ class CallbackHandler(BaseCallbackHandler):
         invocation_params: Optional[Dict[str, Any]] = None,
         **kwargs,
     ):
-        run: RunMetadata = {
-            "messages": messages,
-            "start_time": time.time(),
-        }
+        run_name = _get_langchain_run_name(serialized, **kwargs) or "generation"
+        generation = GenerationMetadata(name=run_name, input=messages, start_time=time.time(), end_time=None)
         if isinstance(invocation_params, dict):
-            run["model_params"] = get_model_params(invocation_params)
+            generation.model_params = get_model_params(invocation_params)
         if isinstance(metadata, dict):
             if model := metadata.get("ls_model_name"):
-                run["model"] = model
+                generation.model = model
             if provider := metadata.get("ls_provider"):
-                run["provider"] = provider
+                generation.provider = provider
         try:
             base_url = serialized["kwargs"]["openai_api_base"]
             if base_url is not None:
-                run["base_url"] = base_url
+                generation.base_url = base_url
         except KeyError:
             pass
-        self._runs[run_id] = run
+        self._runs[run_id] = generation
 
     def _pop_run_metadata(self, run_id: UUID) -> Optional[RunMetadata]:
         end_time = time.time()
@@ -412,59 +395,140 @@ class CallbackHandler(BaseCallbackHandler):
         except KeyError:
             log.warning(f"No run metadata found for run {run_id}")
             return None
-        run["end_time"] = end_time
+        run.end_time = end_time
         return run
 
     def _get_trace_id(self, run_id: UUID):
         trace_id = self._trace_id or self._find_root_run(run_id)
         if not trace_id:
-            trace_id = uuid.uuid4()
+            return run_id
         return trace_id
 
-    def _get_langchain_run_name(self, serialized: Optional[Dict[str, Any]], **kwargs: Any) -> str:
-        """Retrieve the name of a serialized LangChain runnable.
-
-        The prioritization for the determination of the run name is as follows:
-        - The value assigned to the "name" key in `kwargs`.
-        - The value assigned to the "name" key in `serialized`.
-        - The last entry of the value assigned to the "id" key in `serialized`.
-        - "<unknown>".
-
-        Args:
-            serialized (Optional[Dict[str, Any]]): A dictionary containing the runnable's serialized data.
-            **kwargs (Any): Additional keyword arguments, potentially including the 'name' override.
-
-        Returns:
-            str: The determined name of the Langchain runnable.
+    def _get_parent_run_id(self, trace_id: Any, run_id: UUID, parent_run_id: Optional[UUID]):
         """
-        if "name" in kwargs and kwargs["name"] is not None:
-            return kwargs["name"]
+        Replace the parent run ID with the trace ID for second level runs when a custom trace ID is set.
+        """
+        if parent_run_id is not None and parent_run_id not in self._parent_tree:
+            return trace_id
+        return parent_run_id
 
-        try:
-            return serialized["name"]
-        except (KeyError, TypeError):
-            pass
-
-        try:
-            return serialized["id"][-1]
-        except (KeyError, TypeError):
-            pass
-
-    def _capture_trace(self, run_id: UUID, *, outputs: Optional[Dict[str, Any]]):
+    def _pop_run_and_capture_trace_or_span(self, run_id: UUID, parent_run_id: Optional[UUID], outputs: Any):
         trace_id = self._get_trace_id(run_id)
+        self._pop_parent_of_run(run_id)
+        run = self._pop_run_metadata(run_id)
+        if not run:
+            return
+        if isinstance(run, GenerationMetadata):
+            log.warning(f"Run {run_id} is a generation, but attempted to be captured as a trace or span.")
+            return
+        self._capture_trace_or_span(
+            trace_id, run_id, run, outputs, self._get_parent_run_id(trace_id, run_id, parent_run_id)
+        )
+
+    def _capture_trace_or_span(
+        self,
+        trace_id: Any,
+        run_id: UUID,
+        run: SpanMetadata,
+        outputs: Any,
+        parent_run_id: Optional[UUID],
+    ):
+        event_name = "$ai_trace" if parent_run_id is None else "$ai_span"
         event_properties = {
-            "$ai_trace_name": self._trace_name,
             "$ai_trace_id": trace_id,
-            "$ai_input_state": with_privacy_mode(self._client, self._privacy_mode, self._trace_input),
-            **self._properties,
+            "$ai_input_state": with_privacy_mode(self._client, self._privacy_mode, run.input),
+            "$ai_latency": run.latency,
+            "$ai_span_name": run.name,
+            "$ai_span_id": run_id,
         }
-        if outputs is not None:
+        if parent_run_id is not None:
+            event_properties["$ai_parent_id"] = parent_run_id
+        if self._properties:
+            event_properties.update(self._properties)
+
+        if isinstance(outputs, BaseException):
+            event_properties["$ai_error"] = _stringify_exception(outputs)
+            event_properties["$ai_is_error"] = True
+        elif outputs is not None:
             event_properties["$ai_output_state"] = with_privacy_mode(self._client, self._privacy_mode, outputs)
+
         if self._distinct_id is None:
             event_properties["$process_person_profile"] = False
+
+        self._client.capture(
+            distinct_id=self._distinct_id or run_id,
+            event=event_name,
+            properties=event_properties,
+            groups=self._groups,
+        )
+
+    def _pop_run_and_capture_generation(
+        self, run_id: UUID, parent_run_id: Optional[UUID], response: Union[LLMResult, BaseException]
+    ):
+        trace_id = self._get_trace_id(run_id)
+        self._pop_parent_of_run(run_id)
+        run = self._pop_run_metadata(run_id)
+        if not run:
+            return
+        if not isinstance(run, GenerationMetadata):
+            log.warning(f"Run {run_id} is not a generation, but attempted to be captured as a generation.")
+            return
+        self._capture_generation(
+            trace_id, run_id, run, response, self._get_parent_run_id(trace_id, run_id, parent_run_id)
+        )
+
+    def _capture_generation(
+        self,
+        trace_id: Any,
+        run_id: UUID,
+        run: GenerationMetadata,
+        output: Union[LLMResult, BaseException],
+        parent_run_id: Optional[UUID] = None,
+    ):
+        event_properties = {
+            "$ai_trace_id": trace_id,
+            "$ai_span_id": run_id,
+            "$ai_span_name": run.name,
+            "$ai_parent_id": parent_run_id,
+            "$ai_provider": run.provider,
+            "$ai_model": run.model,
+            "$ai_model_parameters": run.model_params,
+            "$ai_input": with_privacy_mode(self._client, self._privacy_mode, run.input),
+            "$ai_http_status": 200,
+            "$ai_latency": run.latency,
+            "$ai_base_url": run.base_url,
+        }
+
+        if isinstance(output, BaseException):
+            event_properties["$ai_http_status"] = _get_http_status(output)
+            event_properties["$ai_error"] = _stringify_exception(output)
+            event_properties["$ai_is_error"] = True
+        else:
+            # Add usage
+            input_tokens, output_tokens = _parse_usage(output)
+            event_properties["$ai_input_tokens"] = input_tokens
+            event_properties["$ai_output_tokens"] = output_tokens
+
+            # Generation results
+            generation_result = output.generations[-1]
+            if isinstance(generation_result[-1], ChatGeneration):
+                completions = [
+                    _convert_message_to_dict(cast(ChatGeneration, generation).message)
+                    for generation in generation_result
+                ]
+            else:
+                completions = [_extract_raw_esponse(generation) for generation in generation_result]
+            event_properties["$ai_output_choices"] = with_privacy_mode(self._client, self._privacy_mode, completions)
+
+        if self._properties:
+            event_properties.update(self._properties)
+
+        if self._distinct_id is None:
+            event_properties["$process_person_profile"] = False
+
         self._client.capture(
             distinct_id=self._distinct_id or trace_id,
-            event="$ai_trace",
+            event="$ai_generation",
             properties=event_properties,
             groups=self._groups,
         )
@@ -595,3 +659,41 @@ def _get_http_status(error: BaseException) -> int:
     # Google: https://github.com/googleapis/python-api-core/blob/main/google/api_core/exceptions.py
     status_code = getattr(error, "status_code", getattr(error, "code", 0))
     return status_code
+
+
+def _get_langchain_run_name(serialized: Optional[Dict[str, Any]], **kwargs: Any) -> Optional[str]:
+    """Retrieve the name of a serialized LangChain runnable.
+
+    The prioritization for the determination of the run name is as follows:
+    - The value assigned to the "name" key in `kwargs`.
+    - The value assigned to the "name" key in `serialized`.
+    - The last entry of the value assigned to the "id" key in `serialized`.
+    - "<unknown>".
+
+    Args:
+        serialized (Optional[Dict[str, Any]]): A dictionary containing the runnable's serialized data.
+        **kwargs (Any): Additional keyword arguments, potentially including the 'name' override.
+
+    Returns:
+        str: The determined name of the Langchain runnable.
+    """
+    if "name" in kwargs and kwargs["name"] is not None:
+        return kwargs["name"]
+    if serialized is None:
+        return None
+    try:
+        return serialized["name"]
+    except (KeyError, TypeError):
+        pass
+    try:
+        return serialized["id"][-1]
+    except (KeyError, TypeError):
+        pass
+    return None
+
+
+def _stringify_exception(exception: BaseException) -> str:
+    description = str(exception)
+    if description:
+        return f"{exception.__class__.__name__}: {description}"
+    return exception.__class__.__name__

posthog/ai/utils.py

@@ -116,12 +116,17 @@ def call_llm_and_track_usage(
     error = None
     http_status = 200
     usage: Dict[str, Any] = {}
+    error_params: Dict[str, any] = {}
 
     try:
         response = call_method(**kwargs)
     except Exception as exc:
         error = exc
         http_status = getattr(exc, "status_code", 0)  # default to 0 becuase its likely an SDK error
+        error_params = {
+            "$ai_is_error": True,
+            "$ai_error": exc.__str__(),
+        }
     finally:
         end_time = time.time()
         latency = end_time - start_time
@@ -149,6 +154,7 @@ def call_llm_and_track_usage(
             "$ai_trace_id": posthog_trace_id,
             "$ai_base_url": str(base_url),
             **(posthog_properties or {}),
+            **(error_params or {}),
         }
 
         if posthog_distinct_id is None:
@@ -186,12 +192,17 @@ async def call_llm_and_track_usage_async(
     error = None
     http_status = 200
     usage: Dict[str, Any] = {}
+    error_params: Dict[str, any] = {}
 
     try:
         response = await call_async_method(**kwargs)
     except Exception as exc:
         error = exc
         http_status = getattr(exc, "status_code", 0)  # default to 0 because its likely an SDK error
+        error_params = {
+            "$ai_is_error": True,
+            "$ai_error": exc.__str__(),
+        }
     finally:
         end_time = time.time()
         latency = end_time - start_time
@@ -219,6 +230,7 @@ async def call_llm_and_track_usage_async(
             "$ai_trace_id": posthog_trace_id,
             "$ai_base_url": str(base_url),
             **(posthog_properties or {}),
+            **(error_params or {}),
         }
 
         if posthog_distinct_id is None:

posthog/client.py

@@ -146,15 +146,13 @@ class Client(object):
                 if send:
                     consumer.start()
 
-    def identify(self, distinct_id=None, properties=None, context=None, timestamp=None, uuid=None, disable_geoip=None):
+    def identify(self, distinct_id=None, properties=None, timestamp=None, uuid=None, disable_geoip=None):
         properties = properties or {}
-        context = context or {}
         require("distinct_id", distinct_id, ID_TYPES)
         require("properties", properties, dict)
 
         msg = {
             "timestamp": timestamp,
-            "context": context,
             "distinct_id": distinct_id,
             "$set": properties,
             "event": "$identify",
@@ -211,7 +209,6 @@ class Client(object):
         distinct_id=None,
         event=None,
         properties=None,
-        context=None,
         timestamp=None,
         uuid=None,
         groups=None,
@@ -219,7 +216,6 @@ class Client(object):
         disable_geoip=None,
     ):
         properties = properties or {}
-        context = context or {}
         require("distinct_id", distinct_id, ID_TYPES)
         require("properties", properties, dict)
         require("event", event, string_types)
@@ -227,7 +223,6 @@ class Client(object):
         msg = {
             "properties": properties,
             "timestamp": timestamp,
-            "context": context,
             "distinct_id": distinct_id,
             "event": event,
             "uuid": uuid,
@@ -263,15 +258,13 @@ class Client(object):
 
         return self._enqueue(msg, disable_geoip)
 
-    def set(self, distinct_id=None, properties=None, context=None, timestamp=None, uuid=None, disable_geoip=None):
+    def set(self, distinct_id=None, properties=None, timestamp=None, uuid=None, disable_geoip=None):
         properties = properties or {}
-        context = context or {}
         require("distinct_id", distinct_id, ID_TYPES)
         require("properties", properties, dict)
 
         msg = {
             "timestamp": timestamp,
-            "context": context,
             "distinct_id": distinct_id,
             "$set": properties,
             "event": "$set",
@@ -280,15 +273,13 @@ class Client(object):
 
         return self._enqueue(msg, disable_geoip)
 
-    def set_once(self, distinct_id=None, properties=None, context=None, timestamp=None, uuid=None, disable_geoip=None):
+    def set_once(self, distinct_id=None, properties=None, timestamp=None, uuid=None, disable_geoip=None):
         properties = properties or {}
-        context = context or {}
         require("distinct_id", distinct_id, ID_TYPES)
         require("properties", properties, dict)
 
         msg = {
             "timestamp": timestamp,
-            "context": context,
             "distinct_id": distinct_id,
             "$set_once": properties,
             "event": "$set_once",
@@ -302,14 +293,12 @@ class Client(object):
         group_type=None,
         group_key=None,
         properties=None,
-        context=None,
         timestamp=None,
         uuid=None,
         disable_geoip=None,
         distinct_id=None,
     ):
         properties = properties or {}
-        context = context or {}
         require("group_type", group_type, ID_TYPES)
         require("group_key", group_key, ID_TYPES)
         require("properties", properties, dict)
@@ -328,15 +317,12 @@ class Client(object):
             },
             "distinct_id": distinct_id,
             "timestamp": timestamp,
-            "context": context,
             "uuid": uuid,
         }
 
         return self._enqueue(msg, disable_geoip)
 
-    def alias(self, previous_id=None, distinct_id=None, context=None, timestamp=None, uuid=None, disable_geoip=None):
-        context = context or {}
-
+    def alias(self, previous_id=None, distinct_id=None, timestamp=None, uuid=None, disable_geoip=None):
         require("previous_id", previous_id, ID_TYPES)
         require("distinct_id", distinct_id, ID_TYPES)
 
@@ -346,18 +332,14 @@ class Client(object):
                 "alias": distinct_id,
             },
             "timestamp": timestamp,
-            "context": context,
             "event": "$create_alias",
             "distinct_id": previous_id,
         }
 
         return self._enqueue(msg, disable_geoip)
 
-    def page(
-        self, distinct_id=None, url=None, properties=None, context=None, timestamp=None, uuid=None, disable_geoip=None
-    ):
+    def page(self, distinct_id=None, url=None, properties=None, timestamp=None, uuid=None, disable_geoip=None):
         properties = properties or {}
-        context = context or {}
 
         require("distinct_id", distinct_id, ID_TYPES)
         require("properties", properties, dict)
@@ -369,7 +351,6 @@ class Client(object):
             "event": "$pageview",
             "properties": properties,
             "timestamp": timestamp,
-            "context": context,
             "distinct_id": distinct_id,
             "uuid": uuid,
         }
@@ -446,7 +427,6 @@ class Client(object):
             timestamp = datetime.now(tz=tzutc())
 
         require("timestamp", timestamp, datetime)
-        require("context", msg["context"], dict)
 
         # add common
         timestamp = guess_timezone(timestamp)

posthog/test/ai/anthropic/test_anthropic.py

@@ -325,3 +325,17 @@ async def test_async_streaming_system_prompt(mock_client, mock_anthropic_stream)
         {"role": "system", "content": "You must always answer with 'Bar'."},
         {"role": "user", "content": "Foo"},
     ]
+
+
+def test_error(mock_client, mock_anthropic_response):
+    with patch("anthropic.resources.Messages.create", side_effect=Exception("Test error")):
+        client = Anthropic(api_key="test-key", posthog_client=mock_client)
+        with pytest.raises(Exception):
+            client.messages.create(model="claude-3-opus-20240229", messages=[{"role": "user", "content": "Hello"}])
+
+        assert mock_client.capture.call_count == 1
+
+        call_args = mock_client.capture.call_args[1]
+        props = call_args["properties"]
+        assert props["$ai_is_error"] is True
+        assert props["$ai_error"] == "Test error"

posthog/test/ai/openai/test_openai.py

@@ -173,3 +173,17 @@ def test_privacy_mode_global(mock_client, mock_openai_response):
         props = call_args["properties"]
         assert props["$ai_input"] is None
         assert props["$ai_output_choices"] is None
+
+
+def test_error(mock_client, mock_openai_response):
+    with patch("openai.resources.chat.completions.Completions.create", side_effect=Exception("Test error")):
+        client = OpenAI(api_key="test-key", posthog_client=mock_client)
+        with pytest.raises(Exception):
+            client.chat.completions.create(model="gpt-4", messages=[{"role": "user", "content": "Hello"}])
+
+        assert mock_client.capture.call_count == 1
+
+        call_args = mock_client.capture.call_args[1]
+        props = call_args["properties"]
+        assert props["$ai_is_error"] is True
+        assert props["$ai_error"] == "Test error"

posthog/test/test_client.py

@@ -581,7 +581,6 @@ class TestClient(unittest.TestCase):
             "distinct_id",
             "python test event",
             {"property": "value"},
-            {"ip": "192.168.0.1"},
             datetime(2014, 9, 3),
             "new-uuid",
         )
@@ -590,7 +589,6 @@ class TestClient(unittest.TestCase):
 
         self.assertEqual(msg["timestamp"], "2014-09-03T00:00:00+00:00")
         self.assertEqual(msg["properties"]["property"], "value")
-        self.assertEqual(msg["context"]["ip"], "192.168.0.1")
         self.assertEqual(msg["event"], "python test event")
         self.assertEqual(msg["properties"]["$lib"], "posthog-python")
         self.assertEqual(msg["properties"]["$lib_version"], VERSION)
@@ -622,14 +620,11 @@ class TestClient(unittest.TestCase):
 
     def test_advanced_identify(self):
         client = self.client
-        success, msg = client.identify(
-            "distinct_id", {"trait": "value"}, {"ip": "192.168.0.1"}, datetime(2014, 9, 3), "new-uuid"
-        )
+        success, msg = client.identify("distinct_id", {"trait": "value"}, datetime(2014, 9, 3), "new-uuid")
 
         self.assertTrue(success)
 
         self.assertEqual(msg["timestamp"], "2014-09-03T00:00:00+00:00")
-        self.assertEqual(msg["context"]["ip"], "192.168.0.1")
         self.assertEqual(msg["$set"]["trait"], "value")
         self.assertEqual(msg["properties"]["$lib"], "posthog-python")
         self.assertEqual(msg["properties"]["$lib_version"], VERSION)
@@ -651,14 +646,11 @@ class TestClient(unittest.TestCase):
 
     def test_advanced_set(self):
         client = self.client
-        success, msg = client.set(
-            "distinct_id", {"trait": "value"}, {"ip": "192.168.0.1"}, datetime(2014, 9, 3), "new-uuid"
-        )
+        success, msg = client.set("distinct_id", {"trait": "value"}, datetime(2014, 9, 3), "new-uuid")
 
         self.assertTrue(success)
 
         self.assertEqual(msg["timestamp"], "2014-09-03T00:00:00+00:00")
-        self.assertEqual(msg["context"]["ip"], "192.168.0.1")
         self.assertEqual(msg["$set"]["trait"], "value")
         self.assertEqual(msg["properties"]["$lib"], "posthog-python")
         self.assertEqual(msg["properties"]["$lib_version"], VERSION)
@@ -680,14 +672,11 @@ class TestClient(unittest.TestCase):
 
     def test_advanced_set_once(self):
         client = self.client
-        success, msg = client.set_once(
-            "distinct_id", {"trait": "value"}, {"ip": "192.168.0.1"}, datetime(2014, 9, 3), "new-uuid"
-        )
+        success, msg = client.set_once("distinct_id", {"trait": "value"}, datetime(2014, 9, 3), "new-uuid")
 
         self.assertTrue(success)
 
         self.assertEqual(msg["timestamp"], "2014-09-03T00:00:00+00:00")
-        self.assertEqual(msg["context"]["ip"], "192.168.0.1")
         self.assertEqual(msg["$set_once"]["trait"], "value")
         self.assertEqual(msg["properties"]["$lib"], "posthog-python")
         self.assertEqual(msg["properties"]["$lib_version"], VERSION)
@@ -736,7 +725,7 @@ class TestClient(unittest.TestCase):
 
     def test_advanced_group_identify(self):
         success, msg = self.client.group_identify(
-            "organization", "id:5", {"trait": "value"}, {"ip": "192.168.0.1"}, datetime(2014, 9, 3), "new-uuid"
+            "organization", "id:5", {"trait": "value"}, datetime(2014, 9, 3), "new-uuid"
         )
 
         self.assertTrue(success)
@@ -754,14 +743,12 @@ class TestClient(unittest.TestCase):
             },
         )
         self.assertEqual(msg["timestamp"], "2014-09-03T00:00:00+00:00")
-        self.assertEqual(msg["context"]["ip"], "192.168.0.1")
 
     def test_advanced_group_identify_with_distinct_id(self):
         success, msg = self.client.group_identify(
             "organization",
             "id:5",
             {"trait": "value"},
-            {"ip": "192.168.0.1"},
             datetime(2014, 9, 3),
             "new-uuid",
             distinct_id="distinct_id",
@@ -783,7 +770,6 @@ class TestClient(unittest.TestCase):
             },
         )
         self.assertEqual(msg["timestamp"], "2014-09-03T00:00:00+00:00")
-        self.assertEqual(msg["context"]["ip"], "192.168.0.1")
 
     def test_basic_alias(self):
         client = self.client
@@ -819,7 +805,6 @@ class TestClient(unittest.TestCase):
             "distinct_id",
             "https://posthog.com/contact",
             {"property": "value"},
-            {"ip": "192.168.0.1"},
             datetime(2014, 9, 3),
             "new-uuid",
         )
@@ -827,7 +812,6 @@ class TestClient(unittest.TestCase):
         self.assertTrue(success)
 
         self.assertEqual(msg["timestamp"], "2014-09-03T00:00:00+00:00")
-        self.assertEqual(msg["context"]["ip"], "192.168.0.1")
         self.assertEqual(msg["properties"]["$current_url"], "https://posthog.com/contact")
         self.assertEqual(msg["properties"]["property"], "value")
         self.assertEqual(msg["properties"]["$lib"], "posthog-python")

posthog/test/test_utils.py

@@ -1,10 +1,13 @@
 import unittest
 from datetime import date, datetime, timedelta
 from decimal import Decimal
+from typing import Optional
 from uuid import UUID
 
 import six
 from dateutil.tz import tzutc
+from pydantic import BaseModel
+from pydantic.v1 import BaseModel as BaseModelV1
 
 from posthog import utils
 
@@ -81,6 +84,32 @@ class TestUtils(unittest.TestCase):
         self.assertEqual("http://posthog.io", utils.remove_trailing_slash("http://posthog.io/"))
         self.assertEqual("http://posthog.io", utils.remove_trailing_slash("http://posthog.io"))
 
+    def test_clean_pydantic(self):
+        class ModelV2(BaseModel):
+            foo: str
+            bar: int
+            baz: Optional[str] = None
+
+        class ModelV1(BaseModelV1):
+            foo: int
+            bar: str
+
+        class NestedModel(BaseModel):
+            foo: ModelV2
+
+        self.assertEqual(utils.clean(ModelV2(foo="1", bar=2)), {"foo": "1", "bar": 2, "baz": None})
+        self.assertEqual(utils.clean(ModelV1(foo=1, bar="2")), {"foo": 1, "bar": "2"})
+        self.assertEqual(
+            utils.clean(NestedModel(foo=ModelV2(foo="1", bar=2, baz="3"))), {"foo": {"foo": "1", "bar": 2, "baz": "3"}}
+        )
+
+        class Dummy:
+            def model_dump(self, required_param):
+                pass
+
+        # Skips a class with a defined non-Pydantic `model_dump` method.
+        self.assertEqual(utils.clean({"test": Dummy()}), {})
+
 
 class TestSizeLimitedDict(unittest.TestCase):
     def test_size_limited_dict(self):

posthog/utils.py

@@ -51,14 +51,24 @@ def clean(item):
         return float(item)
     if isinstance(item, UUID):
         return str(item)
-    elif isinstance(item, (six.string_types, bool, numbers.Number, datetime, date, type(None))):
+    if isinstance(item, (six.string_types, bool, numbers.Number, datetime, date, type(None))):
         return item
-    elif isinstance(item, (set, list, tuple)):
+    if isinstance(item, (set, list, tuple)):
         return _clean_list(item)
-    elif isinstance(item, dict):
+    # Pydantic model
+    try:
+        # v2+
+        if hasattr(item, "model_dump") and callable(item.model_dump):
+            item = item.model_dump()
+        # v1
+        elif hasattr(item, "dict") and callable(item.dict):
+            item = item.dict()
+    except TypeError as e:
+        log.debug(f"Could not serialize Pydantic-like model: {e}")
+        pass
+    if isinstance(item, dict):
         return _clean_dict(item)
-    else:
-        return _coerce_unicode(item)
+    return _coerce_unicode(item)
 
 
 def _clean_list(list_):

posthog/version.py

@@ -1,4 +1,4 @@
-VERSION = "3.9.2"
+VERSION = "3.12.0"
 
 if __name__ == "__main__":
     print(VERSION, end="")  # noqa: T201

setup.py

@@ -46,6 +46,7 @@ extras_require = {
         "langchain-community>=0.2.0",
         "langchain-openai>=0.2.0",
         "langchain-anthropic>=0.2.0",
+        "pydantic",
     ],
     "sentry": ["sentry-sdk", "django"],
     "langchain": ["langchain>=0.2.0"],