adds telemetry

README.md

@@ -240,6 +240,16 @@ See the [`LangChain.ChatModels.ChatBumblebee` documentation](https://hexdocs.pm/
 
 ## Testing
 
+Before you can run the tests, make sure you have the environment variables set.
+
+You can do this by running:
+
+```
+source .envrc_template
+```
+
+Or you can copy it to `.envrc` and populate it with your private API values.
+
 To run all the tests including the ones that perform live calls against the OpenAI API, use the following command:
 
 ```

lib/chains/llm_chain.ex

@@ -427,17 +427,27 @@ defmodule LangChain.Chains.LLMChain do
             &run_until_success/1
         end
 
-      # Run the chain and return the success or error results. NOTE: We do not add
-      # the current LLM to the list and process everything through a single
-      # codepath because failing after attempted fallbacks returns a different
-      # error.
-      if Keyword.has_key?(opts, :with_fallbacks) do
-        # run function and using fallbacks as needed.
-        with_fallbacks(chain, opts, function_to_run)
-      else
-        # run it directly right now and return the success or error
-        function_to_run.(chain)
-      end
+      # Add telemetry for chain execution
+      metadata = %{
+        chain_type: "llm_chain",
+        mode: Keyword.get(opts, :mode, "default"),
+        message_count: length(chain.messages),
+        tool_count: length(chain.tools)
+      }
+
+      LangChain.Telemetry.span([:langchain, :chain, :execute], metadata, fn ->
+        # Run the chain and return the success or error results. NOTE: We do not add
+        # the current LLM to the list and process everything through a single
+        # codepath because failing after attempted fallbacks returns a different
+        # error.
+        if Keyword.has_key?(opts, :with_fallbacks) do
+          # run function and using fallbacks as needed.
+          with_fallbacks(chain, opts, function_to_run)
+        else
+          # run it directly right now and return the success or error
+          function_to_run.(chain)
+        end
+      end)
     rescue
       err in LangChainError ->
         {:error, chain, err}
@@ -1107,54 +1117,62 @@ defmodule LangChain.Chains.LLMChain do
     verbose = Keyword.get(opts, :verbose, false)
     context = Keyword.get(opts, :context, nil)
 
-    try do
-      if verbose, do: IO.inspect(function.name, label: "EXECUTING FUNCTION")
-
-      case Function.execute(function, call.arguments, context) do
-        {:ok, llm_result, processed_result} ->
-          if verbose, do: IO.inspect(processed_result, label: "FUNCTION PROCESSED RESULT")
-          # successful execution and storage of processed_content.
-          ToolResult.new!(%{
-            tool_call_id: call.call_id,
-            content: llm_result,
-            processed_content: processed_result,
-            name: function.name,
-            display_text: function.display_text
-          })
-
-        {:ok, result} ->
-          if verbose, do: IO.inspect(result, label: "FUNCTION RESULT")
-          # successful execution.
-          ToolResult.new!(%{
-            tool_call_id: call.call_id,
-            content: result,
-            name: function.name,
-            display_text: function.display_text
-          })
+    metadata = %{
+      tool_name: function.name,
+      tool_call_id: call.call_id,
+      async: function.async
+    }
 
-        {:error, reason} when is_binary(reason) ->
-          if verbose, do: IO.inspect(reason, label: "FUNCTION ERROR")
+    LangChain.Telemetry.span([:langchain, :tool, :call], metadata, fn ->
+      try do
+        if verbose, do: IO.inspect(function.name, label: "EXECUTING FUNCTION")
+
+        case Function.execute(function, call.arguments, context) do
+          {:ok, llm_result, processed_result} ->
+            if verbose, do: IO.inspect(processed_result, label: "FUNCTION PROCESSED RESULT")
+            # successful execution and storage of processed_content.
+            ToolResult.new!(%{
+              tool_call_id: call.call_id,
+              content: llm_result,
+              processed_content: processed_result,
+              name: function.name,
+              display_text: function.display_text
+            })
+
+          {:ok, result} ->
+            if verbose, do: IO.inspect(result, label: "FUNCTION RESULT")
+            # successful execution.
+            ToolResult.new!(%{
+              tool_call_id: call.call_id,
+              content: result,
+              name: function.name,
+              display_text: function.display_text
+            })
+
+          {:error, reason} when is_binary(reason) ->
+            if verbose, do: IO.inspect(reason, label: "FUNCTION ERROR")
+
+            ToolResult.new!(%{
+              tool_call_id: call.call_id,
+              content: reason,
+              name: function.name,
+              display_text: function.display_text,
+              is_error: true
+            })
+        end
+      rescue
+        err ->
+          Logger.error(
+            "Function #{function.name} failed in execution. Exception: #{LangChainError.format_exception(err, __STACKTRACE__)}"
+          )
 
           ToolResult.new!(%{
             tool_call_id: call.call_id,
-            content: reason,
-            name: function.name,
-            display_text: function.display_text,
+            content: "ERROR executing tool: #{inspect(err)}",
             is_error: true
           })
       end
-    rescue
-      err ->
-        Logger.error(
-          "Function #{function.name} failed in execution. Exception: #{LangChainError.format_exception(err, __STACKTRACE__)}"
-        )
-
-        ToolResult.new!(%{
-          tool_call_id: call.call_id,
-          content: "ERROR executing tool: #{inspect(err)}",
-          is_error: true
-        })
-    end
+    end)
   end
 
   @doc """

lib/chat_models/chat_anthropic.ex

@@ -324,19 +324,39 @@ defmodule LangChain.ChatModels.ChatAnthropic do
   end
 
   def call(%ChatAnthropic{} = anthropic, messages, functions) when is_list(messages) do
-    try do
-      # make base api request and perform high-level success/failure checks
-      case do_api_request(anthropic, messages, functions) do
-        {:error, %LangChainError{} = error} ->
-          {:error, error}
-
-        parsed_data ->
-          {:ok, parsed_data}
+    metadata = %{
+      model: anthropic.model,
+      message_count: length(messages),
+      tools_count: length(functions)
+    }
+
+    LangChain.Telemetry.span([:langchain, :llm, :call], metadata, fn ->
+      try do
+        # Track the prompt being sent
+        LangChain.Telemetry.llm_prompt(
+          %{system_time: System.system_time()},
+          %{model: anthropic.model, messages: messages}
+        )
+
+        # make base api request and perform high-level success/failure checks
+        case do_api_request(anthropic, messages, functions) do
+          {:error, %LangChainError{} = error} ->
+            {:error, error}
+
+          parsed_data ->
+            # Track the response being received
+            LangChain.Telemetry.llm_response(
+              %{system_time: System.system_time()},
+              %{model: anthropic.model, response: parsed_data}
+            )
+
+            {:ok, parsed_data}
+        end
+      rescue
+        err in LangChainError ->
+          {:error, err}
       end
-    rescue
-      err in LangChainError ->
-        {:error, err}
-    end
+    end)
   end
 
   # Call Anthropic's API.
@@ -429,6 +449,12 @@ defmodule LangChain.ChatModels.ChatAnthropic do
         tools,
         retry_count
       ) do
+    # Track the prompt being sent for streaming
+    LangChain.Telemetry.llm_prompt(
+      %{system_time: System.system_time(), streaming: true},
+      %{model: anthropic.model, messages: messages}
+    )
+
     Req.new(
       url: url(anthropic),
       json: for_api(anthropic, messages, tools),
@@ -450,6 +476,13 @@ defmodule LangChain.ChatModels.ChatAnthropic do
           get_ratelimit_info(response.headers)
         ])
 
+        # Track the stream completion
+        LangChain.Telemetry.emit_event(
+          [:langchain, :llm, :response, streaming: true],
+          %{system_time: System.system_time()},
+          %{model: anthropic.model}
+        )
+
         data
 
       # The error tuple was successfully received from the API. Unwrap it and

lib/chat_models/chat_bumblebee.ex

@@ -231,19 +231,40 @@ defmodule LangChain.ChatModels.ChatBumblebee do
   end
 
   def call(%ChatBumblebee{} = model, messages, functions) when is_list(messages) do
-    try do
-      # make base api request and perform high-level success/failure checks
-      case do_serving_request(model, messages, functions) do
-        {:error, reason} ->
-          {:error, reason}
-
-        parsed_data ->
-          {:ok, parsed_data}
+    metadata = %{
+      model: inspect(model.serving),
+      template_format: model.template_format,
+      message_count: length(messages),
+      tools_count: length(functions)
+    }
+
+    LangChain.Telemetry.span([:langchain, :llm, :call], metadata, fn ->
+      try do
+        # Track the prompt being sent
+        LangChain.Telemetry.llm_prompt(
+          %{system_time: System.system_time()},
+          %{model: inspect(model.serving), messages: messages}
+        )
+
+        # make base api request and perform high-level success/failure checks
+        case do_serving_request(model, messages, functions) do
+          {:error, reason} ->
+            {:error, reason}
+
+          parsed_data ->
+            # Track the response being received
+            LangChain.Telemetry.llm_response(
+              %{system_time: System.system_time()},
+              %{model: inspect(model.serving), response: parsed_data}
+            )
+
+            {:ok, parsed_data}
+        end
+      rescue
+        err in LangChainError ->
+          {:error, err}
       end
-    rescue
-      err in LangChainError ->
-        {:error, err}
-    end
+    end)
   end
 
   @doc false
@@ -461,6 +482,16 @@ defmodule LangChain.ChatModels.ChatBumblebee do
       when is_binary(content) do
     fire_token_usage_callback(model, token_summary)
 
+    # Track non-streaming response completion
+    LangChain.Telemetry.emit_event(
+      [:langchain, :llm, :response, :non_streaming],
+      %{system_time: System.system_time()},
+      %{
+        model: inspect(model.serving),
+        response_size: byte_size(inspect(content))
+      }
+    )
+
     case Message.new(%{role: :assistant, status: :complete, content: content}) do
       {:ok, message} ->
         # execute the callback with the final message
@@ -495,6 +526,13 @@ defmodule LangChain.ChatModels.ChatBumblebee do
       {:done, %{token_summary: token_summary}} ->
         fire_token_usage_callback(model, token_summary)
 
+        # Track stream completion
+        LangChain.Telemetry.emit_event(
+          [:langchain, :llm, :response, streaming: true],
+          %{system_time: System.system_time()},
+          %{model: inspect(model.serving)}
+        )
+
         final_delta = MessageDelta.new!(%{role: :assistant, status: :complete})
         Callbacks.fire(model.callbacks, :on_llm_new_delta, [final_delta])
         final_delta

lib/chat_models/chat_google_ai.ex

@@ -431,18 +431,38 @@ defmodule LangChain.ChatModels.ChatGoogleAI do
 
   def call(%ChatGoogleAI{} = google_ai, messages, tools)
       when is_list(messages) do
-    try do
-      case do_api_request(google_ai, messages, tools) do
-        {:error, reason} ->
-          {:error, reason}
+    metadata = %{
+      model: google_ai.model,
+      message_count: length(messages),
+      tools_count: length(tools)
+    }
+
+    LangChain.Telemetry.span([:langchain, :llm, :call], metadata, fn ->
+      try do
+        # Track the prompt being sent
+        LangChain.Telemetry.llm_prompt(
+          %{system_time: System.system_time()},
+          %{model: google_ai.model, messages: messages}
+        )
+
+        case do_api_request(google_ai, messages, tools) do
+          {:error, reason} ->
+            {:error, reason}
 
-        parsed_data ->
-          {:ok, parsed_data}
+          parsed_data ->
+            # Track the response being received
+            LangChain.Telemetry.llm_response(
+              %{system_time: System.system_time()},
+              %{model: google_ai.model, response: parsed_data}
+            )
+
+            {:ok, parsed_data}
+        end
+      rescue
+        err in LangChainError ->
+          {:error, err.message}
       end
-    rescue
-      err in LangChainError ->
-        {:error, err.message}
-    end
+    end)
   end
 
   @doc false
@@ -468,6 +488,16 @@ defmodule LangChain.ChatModels.ChatGoogleAI do
             {:error, reason}
 
           result ->
+            # Track non-streaming response completion
+            LangChain.Telemetry.emit_event(
+              [:langchain, :llm, :response, streaming: false],
+              %{system_time: System.system_time()},
+              %{
+                model: google_ai.model,
+                response_size: byte_size(inspect(result))
+              }
+            )
+
             Callbacks.fire(google_ai.callbacks, :on_llm_new_message, [result])
             result
         end