Simplify observation API

README.md

@@ -7,8 +7,7 @@
 SessionBat is a Python SDK for recording AI session activity and sending it to SessionBat.
 
 It is designed for teams that want to debug and understand what your AI app actually did, including:
-- user messages
-- assistant responses
+- model calls and responses
 - tool calls
 - document retrievals
 
@@ -70,8 +69,6 @@ session = client.session(
 
 interaction = session.interaction(interaction_id="interaction_123")
 
-interaction.user_message("I am locked out of my account")
-
 interaction.retrieval(
     query="reset password locked out",
     documents=[
@@ -93,10 +90,10 @@ interaction.tool_call(
     metrics={"latency_ms": 117, "http_status": 200},
 )
 
-interaction.assistant_response(
+interaction.llm(
     model="gpt-5.4-mini",
-    request={"messages": [{"role": "user", "content": "I am locked out of my account"}]},
-    response={"text": "I found your account. Use the reset link and follow the email prompt."},
+    input={"messages": [{"role": "user", "content": "I am locked out of my account"}]},
+    output={"text": "I found your account. Use the reset link and follow the email prompt."},
     metrics={"latency_ms": 820, "input_tokens": 142, "output_tokens": 36},
 )
 ```
@@ -112,7 +109,7 @@ That emits structured events like:
   "context": {"environment": "prod", "user_id": "user_123", "workspace_id": "ws_456"},
   "observation": {
     "kind": "llm",
-    "name": "assistant_response"
+    "name": null
   }
 }
 ```
@@ -154,9 +151,7 @@ delivery before exiting a short-lived process.
 
 A `Session` groups interactions. A `Interaction` records completed observations:
 
-- `interaction.user_message(content)`
-- `interaction.message(role=..., content=...)`
-- `interaction.assistant_response(...)`
+- `interaction.llm(...)`
 - `interaction.tool_call(...)`
 - `interaction.retrieval(...)`
 

examples/failure_model_error.py

@@ -21,11 +21,9 @@ def main() -> None:
     )
     interaction = session.interaction(interaction_id="interaction_model_error")
 
-    interaction.user_message("What invoices are overdue for Acme Corp?")
-
-    interaction.assistant_response(
+    interaction.llm(
         model="gpt-5.4-mini",
-        request={
+        input={
             "messages": [
                 {"role": "user", "content": "What invoices are overdue for Acme Corp?"},
             ]

examples/failure_retrieval_miss.py

@@ -21,8 +21,6 @@ def main() -> None:
     )
     interaction = session.interaction(interaction_id="interaction_retrieval_miss")
 
-    interaction.user_message("How do I export audit logs for SSO users?")
-
     interaction.retrieval(
         query="export audit logs for sso users",
         documents=[],
@@ -34,23 +32,19 @@ def main() -> None:
         metrics={"latency_ms": 63, "documents_found": 0},
     )
 
-    interaction.assistant_response(
+    interaction.llm(
         model="gpt-5.4-mini",
-        request={
+        input={
             "messages": [
                 {"role": "user", "content": "How do I export audit logs for SSO users?"},
             ]
         },
-        response={
+        output={
             "text": "I could not find documentation for exporting audit logs for SSO users.",
         },
         metadata={"provider": "openai"},
         metrics={"latency_ms": 704, "input_tokens": 118, "output_tokens": 24},
     )
-    interaction.user_message(
-        "Can an admin download SAML access logs anywhere?",
-        metadata={"sequence": "follow_up"},
-    )
 
 
 if __name__ == "__main__":

examples/failure_tool_loop.py

@@ -21,8 +21,6 @@ def main() -> None:
     )
     interaction = session.interaction(interaction_id="interaction_tool_loop")
 
-    interaction.user_message("Why is my workspace usage total wrong?")
-
     interaction.tool_call(
         tool_name="get_workspace_usage",
         input={"workspace_id": "ws_999"},
@@ -47,14 +45,14 @@ def main() -> None:
         metrics={"latency_ms": 139, "http_status": 200, "attempt": 3},
     )
 
-    interaction.assistant_response(
+    interaction.llm(
         model="gpt-5.4-mini",
-        request={
+        input={
             "messages": [
                 {"role": "user", "content": "Why is my workspace usage total wrong?"},
             ]
         },
-        response={
+        output={
             "text": (
                 "I checked the usage totals again, but I still cannot determine the discrepancy."
             ),

examples/manual_trace.py

@@ -21,8 +21,6 @@ def main() -> None:
     )
     interaction = session.interaction(interaction_id="interaction_123")
 
-    interaction.user_message("I am locked out of my account")
-
     interaction.retrieval(
         query="reset password locked out",
         documents=[
@@ -48,10 +46,10 @@ def main() -> None:
         metrics={"latency_ms": 117, "http_status": 200},
     )
 
-    interaction.assistant_response(
+    interaction.llm(
         model="gpt-5.4-mini",
-        request={"messages": [{"role": "user", "content": "I am locked out of my account"}]},
-        response={"text": "I found your account. Use the reset link and follow the email prompt."},
+        input={"messages": [{"role": "user", "content": "I am locked out of my account"}]},
+        output={"text": "I found your account. Use the reset link and follow the email prompt."},
         metadata={"provider": "openai"},
         metrics={"latency_ms": 820, "input_tokens": 142, "output_tokens": 36},
     )

src/sessionbat/client.py

@@ -7,8 +7,7 @@
 from .models import Envelope, isoformat, new_id, utc_now
 from .transports import DEFAULT_INGESTION_ENDPOINT, IngestionTransport, Transport
 
-ObservationKind = Literal["message", "llm", "tool", "retrieval"]
-MessageRole = Literal["user", "assistant", "system", "tool"]
+ObservationKind = Literal["llm", "tool", "retrieval"]
 
 
 def _merge_tags(*tag_sets: list[str] | None) -> list[str]:
@@ -145,47 +144,12 @@ class Interaction:
     tags: list[str] = field(default_factory=list)
     context: dict[str, Any] = field(default_factory=dict)
 
-    def message(
-        self,
-        *,
-        role: MessageRole,
-        content: Any,
-        name: str | None = None,
-        metadata: dict[str, Any] | None = None,
-        tags: list[str] | None = None,
-        context: dict[str, Any] | None = None,
-    ) -> str:
-        return self._record(
-            kind="message",
-            name=name or f"{role}_message",
-            input={"content": content},
-            metadata=_merge_dicts({"role": role}, metadata),
-            tags=tags,
-            context=context,
-        )
-
-    def user_message(
-        self,
-        content: Any,
-        *,
-        metadata: dict[str, Any] | None = None,
-        tags: list[str] | None = None,
-        context: dict[str, Any] | None = None,
-    ) -> str:
-        return self.message(
-            role="user",
-            content=content,
-            metadata=metadata,
-            tags=tags,
-            context=context,
-        )
-
-    def assistant_response(
+    def llm(
         self,
         *,
         model: str,
-        request: dict[str, Any] | None = None,
-        response: dict[str, Any] | None = None,
+        input: dict[str, Any] | None = None,
+        output: dict[str, Any] | None = None,
         error: dict[str, Any] | None = None,
         metadata: dict[str, Any] | None = None,
         metrics: dict[str, Any] | None = None,
@@ -194,9 +158,8 @@ def assistant_response(
     ) -> str:
         return self._record(
             kind="llm",
-            name="assistant_response",
-            input=request,
-            output=response,
+            input=input,
+            output=output,
             error=error,
             metadata=_merge_dicts({"model": model}, metadata),
             metrics=metrics,
@@ -241,7 +204,6 @@ def retrieval(
     ) -> str:
         return self._record(
             kind="retrieval",
-            name="retrieval",
             input={"query": query},
             output={"documents": documents or []},
             error=error,
@@ -255,7 +217,7 @@ def _record(
         self,
         *,
         kind: ObservationKind,
-        name: str,
+        name: str | None = None,
         input: dict[str, Any] | None = None,
         output: dict[str, Any] | None = None,
         error: dict[str, Any] | None = None,

src/sessionbat/langchain.py

@@ -137,12 +137,6 @@ def on_chat_model_start(
             extra={"langchain_callback": "on_chat_model_start", "chat_model": True},
             kwargs=kwargs,
         )
-        self._record_messages_from_chat_input(
-            messages,
-            interaction=interaction,
-            metadata=callback_metadata,
-            tags=tags,
-        )
         self._runs[_run_id(run_id)] = _RunState(
             kind="llm",
             name=_serialized_name(serialized, "chat_model"),
@@ -170,10 +164,10 @@ def on_llm_end(
         state = self._pop_run(run_id, kind="llm", name="llm", parent_run_id=parent_run_id)
         response_payload = _llm_response_payload(response)
         model_metadata = _merge_dicts(state.metadata, metadata)
-        state.interaction.assistant_response(
+        state.interaction.llm(
             model=_extract_model(response=response, metadata=model_metadata),
-            request=state.input,
-            response=response_payload,
+            input=state.input,
+            output=response_payload,
             metadata=_merge_dicts(
                 state.metadata,
                 self._callback_metadata(
@@ -200,10 +194,10 @@ def on_llm_error(
         **kwargs: Any,
     ) -> None:
         state = self._pop_run(run_id, kind="llm", name="llm", parent_run_id=parent_run_id)
-        state.interaction.assistant_response(
+        state.interaction.llm(
             model=_extract_model(metadata=_merge_dicts(state.metadata, metadata)),
-            request=state.input,
-            response=_jsonable(kwargs.get("response")),
+            input=state.input,
+            output=_jsonable(kwargs.get("response")),
             error=_error_payload(error),
             metadata=_merge_dicts(
                 state.metadata,
@@ -493,32 +487,6 @@ def _callback_metadata(
     def _tags(self, *tag_sets: list[str] | None) -> list[str]:
         return _merge_tags(self.tags, *tag_sets)
 
-    def _record_messages_from_chat_input(
-        self,
-        message_batches: list[list[Any]],
-        *,
-        interaction: SessionBatInteraction,
-        metadata: dict[str, Any],
-        tags: list[str] | None,
-    ) -> None:
-        for message in _system_and_user_messages(message_batches):
-            role = _message_role(message)
-            if role is None:
-                continue
-            interaction.message(
-                role=role,
-                content=_message_content(message),
-                metadata=_merge_dicts(
-                    metadata,
-                    {
-                        "langchain_message_type": _message_type(message),
-                        "langchain_message_id": getattr(message, "id", None),
-                    },
-                ),
-                tags=self._tags(tags),
-                context=self.context,
-            )
-
     def _interaction_for(
         self,
         run_id: UUID,
@@ -816,45 +784,6 @@ def _document_payload(document: Any) -> dict[str, Any]:
     return {"content": _jsonable(document)}
 
 
-def _system_and_user_messages(message_batches: list[list[Any]]) -> list[Any]:
-    if not message_batches:
-        return []
-    messages = message_batches[-1]
-    return [message for message in messages if _message_role(message) in {"system", "user"}]
-
-
-def _message_role(message: Any) -> str | None:
-    message_type = _message_type(message)
-    if message_type in {"human", "user"}:
-        return "user"
-    if message_type == "system":
-        return "system"
-    return None
-
-
-def _message_type(message: Any) -> str | None:
-    message_type = getattr(message, "type", None)
-    if isinstance(message_type, str):
-        return message_type
-    role = getattr(message, "role", None)
-    if isinstance(role, str):
-        return role
-    if isinstance(message, Mapping):
-        value = message.get("type") or message.get("role")
-        if isinstance(value, str):
-            return value
-    return None
-
-
-def _message_content(message: Any) -> Any:
-    content = getattr(message, "content", None)
-    if content is not None:
-        return _jsonable(content)
-    if isinstance(message, Mapping) and "content" in message:
-        return _jsonable(message["content"])
-    return _jsonable(message)
-
-
 def _jsonable(value: Any, *, _depth: int = 0) -> Any:
     if _depth > 8:
         return repr(value)