defmodule LangChain.Telemetry do @moduledoc """ Telemetry events for LangChain. This module defines telemetry events that other applications can attach to. It provides a standardized way to emit events for various operations in the LangChain library without implementing tracing functionality. ## Event Naming Events follow the convention: `[:langchain, component, operation, stage]` ## Core Events * `[:langchain, :llm, :call, :start]` - Emitted when an LLM call starts * `[:langchain, :llm, :call, :stop]` - Emitted when an LLM call completes * `[:langchain, :llm, :call, :exception]` - Emitted when an LLM call raises an exception * `[:langchain, :llm, :prompt]` - Emitted when a prompt is sent to an LLM * `[:langchain, :llm, :response]` - Emitted when a response is received from an LLM * `[:langchain, :llm, :response, :non_streaming]` - Emitted when a non-streaming response is received from an LLM * `[:langchain, :llm, :response, :streaming]` - Emitted when a streaming response is received from an LLM * `[:langchain, :chain, :execute, :start]` - Emitted when a chain execution starts * `[:langchain, :chain, :execute, :stop]` - Emitted when a chain execution completes * `[:langchain, :chain, :execute, :exception]` - Emitted when a chain execution raises an exception * `[:langchain, :tool, :call, :start]` - Emitted when a tool call starts * `[:langchain, :tool, :call, :stop]` - Emitted when a tool call completes * `[:langchain, :tool, :call, :exception]` - Emitted when a tool call raises an exception * `[:langchain, :message, :process, :start]` - Emitted when a message processor (e.g. `LangChain.MessageProcessors.JsonProcessor`) starts processing a received message * `[:langchain, :message, :process, :stop]` - Emitted when message processing completes * `[:langchain, :message, :process, :exception]` - Emitted when message processing raises an exception * `[:langchain, :llm, :stream, :first_token]` - Emitted once per streaming LLM call when the first delta is received. Carries a `duration` measurement (time from the call's start to the first streamed chunk, in native units) — the basis for a time-to-first-token metric. ## Reserved events (not currently emitted) The following event names — and the `*_start` helper functions that would emit them (`memory_read_start/1`, `memory_write_start/1`, `retriever_get_relevant_documents_start/1`) — are **reserved for future use and are not emitted by LangChain today.** They are kept so the naming convention is stable if/when those subsystems are instrumented. Do not attach handlers expecting them to fire yet: * `[:langchain, :memory, :read, :start | :stop | :exception]` * `[:langchain, :memory, :write, :start | :stop | :exception]` * `[:langchain, :retriever, :get_relevant_documents, :start | :stop | :exception]` ## Metadata Fields The following metadata fields are automatically injected or available in events: * `:call_id` - A UUID (via `Ecto.UUID.generate/0`) that correlates start, stop, and exception events within a single `span/3` or `start_event/2` call. Automatically injected via `Map.put_new/3`, so callers can supply their own ID to override. * `:provider` - The LLM provider name (e.g. `"openai"`, `"anthropic"`, `"google"`). Included in LLM call metadata by each chat model implementation via the `ChatModel.provider/0` callback. * `:custom_context` - User-supplied context data from `LLMChain.custom_context`. Included in chain execution and tool call metadata. Not included in LLM-level telemetry (correlate via `call_id` instead). * `:token_usage` - A `%TokenUsage{}` struct with input/output token counts. Included in LLM call `:stop` events and chain execution `:stop` events when available (via the `:enrich_stop` callback). `nil` when the model does not report usage. * `:request_options` - A map of the standard request parameters the chat model set (`:temperature`, `:max_tokens`, `:top_p`, `:seed`, ...), extracted from the model struct by `LangChain.ChatModels.ChatModel.request_options/1` and injected on LLM call events. Absent parameters are omitted; an empty map means none were captured. The OpenTelemetry layer maps these to `gen_ai.request.*` span attributes. * `:output_type` - `"text"` or `"json"`, from `ChatModel.output_type/1`, injected on LLM call events. Maps to `gen_ai.output.type`. * `:endpoint` - The request URL, injected on LLM call events when the chat model exposes an `:endpoint`. The OpenTelemetry layer derives `server.address` / `server.port` from it. * `:last_message` - The final assembled `%Message{}` from the LLM response. Included in chain execution `:stop` events. For streaming responses this is the fully assembled message (not individual deltas). ## Privacy Note Message content is intentionally excluded from the lifecycle events (`:start` / `:stop` / `:exception`) to avoid unconditional exposure of user/PII data. Message content is only available through the purpose-specific `[:langchain, :llm, :prompt]` and `[:langchain, :llm, :response]` events — subscribing to these is an explicit opt-in. ## Expected Metadata Shape by Event * **LLM call `:start`**: `%{model: String.t(), provider: String.t(), message_count: integer(), tools_count: integer(), request_options: map(), output_type: String.t(), endpoint: String.t() | nil, call_id: String.t()}` * **LLM call `:stop`** (includes enriched fields): `%{model: String.t(), provider: String.t(), message_count: integer(), tools_count: integer(), request_options: map(), output_type: String.t(), endpoint: String.t() | nil, call_id: String.t(), token_usage: TokenUsage.t() | nil, result: term()}` * **Chain execution `:start`**: `%{chain_type: String.t(), mode: term(), message_count: integer(), tools_count: integer(), custom_context: term(), call_id: String.t()}` * **Chain execution `:stop`** (includes enriched fields): `%{chain_type: String.t(), mode: term(), message_count: integer(), tools_count: integer(), custom_context: term(), call_id: String.t(), last_message: Message.t() | nil, token_usage: TokenUsage.t() | nil, result: term()}` * **Tool call** (`:start` / `:stop` / `:exception`): `%{tool_name: String.t(), tool_call_id: String.t(), tool_description: String.t() | nil, async: boolean(), custom_context: term(), call_id: String.t()}` ## Usage To attach to these events in your application: ```elixir :telemetry.attach( "my-handler-id", [:langchain, :llm, :call, :stop], &MyApp.handle_llm_call/4, nil ) def handle_llm_call(_event_name, measurements, metadata, _config) do # Process the event IO.inspect(measurements) IO.inspect(metadata) end ``` """ @doc """ Emits a telemetry event with the given name, measurements, and metadata. ## Parameters * `event_name` - The name of the event as a list of atoms * `measurements` - A map of measurements for the event * `metadata` - A map of metadata for the event ## Examples iex> LangChain.Telemetry.emit_event([:langchain, :llm, :call, :start], %{system_time: System.system_time()}, %{model: "gpt-4"}) """ @spec emit_event(list(atom()), map(), map()) :: :ok def emit_event(event_name, measurements, metadata) do :telemetry.execute(event_name, measurements, metadata) end @doc """ Emits a start event and returns a function to emit the corresponding stop event. This is useful for span-like events where you want to measure the duration of an operation. ## Parameters * `event_prefix` - The prefix for the event name as a list of atoms * `metadata` - A map of metadata for the event ## Returns A function that accepts additional metadata to be merged with the original metadata and emits the stop event with the duration measurement. ## Examples iex> stop_fun = LangChain.Telemetry.start_event([:langchain, :llm, :call], %{model: "gpt-4"}) iex> # Do some work iex> stop_fun.(%{result: "success"}) """ @spec start_event(list(atom()), map()) :: (map() -> :ok) def start_event(event_prefix, metadata) do start_time = System.monotonic_time() start_system_time = System.system_time() # Inject a call_id if not already present, so start and stop events share the same ID metadata = Map.put_new(metadata, :call_id, Ecto.UUID.generate()) emit_event(event_prefix ++ [:start], %{system_time: start_system_time}, metadata) fn additional_metadata -> end_time = System.monotonic_time() duration = end_time - start_time emit_event( event_prefix ++ [:stop], %{duration: duration, system_time: System.system_time()}, Map.merge(metadata, additional_metadata) ) end end @doc """ Wraps a function call with start and stop telemetry events. ## Parameters * `event_prefix` - The prefix for the event name as a list of atoms * `metadata` - A map of metadata for the event * `fun` - The function to execute * `opts` - Optional keyword list: * `:enrich_stop` - A 1-arity function that receives the result and returns a map of additional metadata to merge into the stop event. Useful for extracting data (e.g. token usage) from the result into top-level metadata. ## Returns The result of the function call. ## Examples iex> LangChain.Telemetry.span([:langchain, :llm, :call], %{model: "gpt-4"}, fn -> ...> # Call the LLM ...> {:ok, "response"} ...> end) # With enrich_stop to surface token usage: iex> LangChain.Telemetry.span([:langchain, :llm, :call], %{model: "gpt-4"}, fn -> ...> {:ok, response} ...> end, enrich_stop: fn {:ok, msg} -> %{token_usage: msg.metadata[:usage]} end) """ @spec span(list(atom()), map(), (-> result), keyword()) :: result when result: any() def span(event_prefix, metadata, fun, opts \\ []) do # Inject call_id once here so it's shared across start, stop, and exception events. # start_event/2 also calls put_new, but since we set it first, the same ID is reused. metadata = Map.put_new(metadata, :call_id, Ecto.UUID.generate()) # Capture the start time here too so the `:exception` event can report a # duration. Failed operations otherwise carry no duration, which would make # them invisible to duration-based metrics (and hide error latency). exception_start_time = System.monotonic_time() stop = start_event(event_prefix, metadata) try do result = fun.() additional_metadata = case Keyword.get(opts, :enrich_stop) do nil -> %{result: result} enrich_fn when is_function(enrich_fn, 1) -> try do Map.merge(%{result: result}, enrich_fn.(result)) rescue _ -> %{result: result} end end stop.(additional_metadata) result rescue exception -> stacktrace = __STACKTRACE__ emit_event( event_prefix ++ [:exception], %{ duration: System.monotonic_time() - exception_start_time, system_time: System.system_time() }, Map.merge(metadata, %{ kind: :error, error: exception, stacktrace: stacktrace }) ) reraise exception, stacktrace catch # `rescue` above only traps raised exceptions. A function that `exit`s (a # linked crash, a `Task.await` timeout) or `throw`s would otherwise emit # neither `:stop` nor `:exception` — and a span-based consumer that opened a # span on `:start` would never end it, leaking the span and its attached # context for the rest of a long-lived process. Emit `:exception` here too so # the span is closed, then re-propagate the exit/throw unchanged. kind, reason -> stacktrace = __STACKTRACE__ emit_event( event_prefix ++ [:exception], %{ duration: System.monotonic_time() - exception_start_time, system_time: System.system_time() }, Map.merge(metadata, %{ kind: kind, # No exception struct exists for a throw/exit; consumers key off # `:kind`/`:reason`. A `nil` `:error` tells them to skip # exception-only rendering while still ending their span. error: nil, reason: reason, stacktrace: stacktrace }) ) :erlang.raise(kind, reason, stacktrace) end end # LLM Events @doc """ Emits an LLM call start event. """ @spec llm_call_start(map()) :: (map() -> :ok) def llm_call_start(metadata) do start_event([:langchain, :llm, :call], metadata) end @doc """ Emits an LLM prompt event. """ @spec llm_prompt(map(), map()) :: :ok def llm_prompt(measurements, metadata) do emit_event([:langchain, :llm, :prompt], measurements, metadata) end @doc """ Emits an LLM response event. """ @spec llm_response(map(), map()) :: :ok def llm_response(measurements, metadata) do emit_event([:langchain, :llm, :response], measurements, metadata) end # Chain Events @doc """ Emits a chain execution start event. """ @spec chain_execute_start(map()) :: (map() -> :ok) def chain_execute_start(metadata) do start_event([:langchain, :chain, :execute], metadata) end # Message Events @doc """ Emits a message processing start event. > #### Unused convenience helper {: .info} > > This helper is not called internally. The `[:langchain, :message, :process, …]` > events themselves **are** emitted — `LangChain.MessageProcessors.JsonProcessor` > emits the full span directly via `span/4` when it runs in a chain's > `message_processors`. The helper is kept for callers who want to emit the same > event from their own message processors. """ @spec message_process_start(map()) :: (map() -> :ok) def message_process_start(metadata) do start_event([:langchain, :message, :process], metadata) end # Tool Events @doc """ Emits a tool call start event. """ @spec tool_call_start(map()) :: (map() -> :ok) def tool_call_start(metadata) do start_event([:langchain, :tool, :call], metadata) end @doc """ Emits a tool call event. """ @spec tool_call(map(), map()) :: :ok def tool_call(measurements, metadata) do emit_event([:langchain, :tool, :call], measurements, metadata) end # Memory Events (reserved — not currently emitted by LangChain) @doc """ Emits a memory read start event. > #### Reserved {: .info} > > LangChain does not call this today. See "Reserved events" in the module doc. """ @spec memory_read_start(map()) :: (map() -> :ok) def memory_read_start(metadata) do start_event([:langchain, :memory, :read], metadata) end @doc """ Emits a memory write start event. > #### Reserved {: .info} > > LangChain does not call this today. See "Reserved events" in the module doc. """ @spec memory_write_start(map()) :: (map() -> :ok) def memory_write_start(metadata) do start_event([:langchain, :memory, :write], metadata) end # Retriever Events (reserved — not currently emitted by LangChain) @doc """ Emits a retriever get relevant documents start event. > #### Reserved {: .info} > > LangChain does not call this today. See "Reserved events" in the module doc. """ @spec retriever_get_relevant_documents_start(map()) :: (map() -> :ok) def retriever_get_relevant_documents_start(metadata) do start_event([:langchain, :retriever, :get_relevant_documents], metadata) end end