# Generated by SnakeBridge v0.13.0 - DO NOT EDIT MANUALLY # Regenerate with: mix compile # Library: dspy 3.1.2 # Python module: dspy.adapters.utils defmodule Dspy.Adapters.Utils do @moduledoc """ Submodule bindings for `dspy.adapters.utils`. ## Version - Requested: 3.1.2 - Observed at generation: 3.1.2 ## Runtime Options All functions accept a `__runtime__` option for controlling execution behavior: Dspy.Adapters.Utils.some_function(args, __runtime__: [timeout: 120_000]) ### Supported runtime options - `:timeout` - Call timeout in milliseconds (default: 120,000ms / 2 minutes) - `:timeout_profile` - Use a named profile (`:default`, `:ml_inference`, `:batch_job`, `:streaming`) - `:stream_timeout` - Timeout for streaming operations (default: 1,800,000ms / 30 minutes) - `:session_id` - Override the session ID for this call - `:pool_name` - Target a specific Snakepit pool (multi-pool setups) - `:affinity` - Override session affinity (`:hint`, `:strict_queue`, `:strict_fail_fast`) ### Timeout Profiles - `:default` - 2 minute timeout for regular calls - `:ml_inference` - 10 minute timeout for ML/LLM workloads - `:batch_job` - Unlimited timeout for long-running jobs - `:streaming` - 2 minute timeout, 30 minute stream_timeout ### Example with timeout override # For a long-running ML inference call Dspy.Adapters.Utils.predict(data, __runtime__: [timeout_profile: :ml_inference]) # Or explicit timeout Dspy.Adapters.Utils.predict(data, __runtime__: [timeout: 600_000]) # Route to a pool and enforce strict affinity Dspy.Adapters.Utils.predict(data, __runtime__: [pool_name: :strict_pool, affinity: :strict_queue]) See `SnakeBridge.Defaults` for global timeout configuration. """ @doc false def __snakebridge_python_name__, do: "dspy.adapters.utils" @doc false def __snakebridge_library__, do: "dspy" @doc """ Finds the enum member corresponding to the specified identifier, which may be the enum member's name or value. ## Parameters - `enum` - The enum to search for the member. - `identifier` - If the enum is explicitly-valued, this is the value of the enum member to find. If the enum is auto-valued, this is the name of the enum member to find. ## Returns - `term()` """ @spec find_enum_member(term(), term(), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()} def find_enum_member(enum, identifier, opts \\ []) do SnakeBridge.Runtime.call(__MODULE__, :find_enum_member, [enum, identifier], opts) end @doc """ Formats the value of the specified field according to the field's DSPy type (input or output), annotation (e.g. str, int, etc.), and the type of the value itself. ## Parameters - `field_info` (term()) - `value` (term()) - `assume_text` (term() default: True) ## Returns - `term()` """ @spec format_field_value(term(), term()) :: {:ok, term()} | {:error, Snakepit.Error.t()} @spec format_field_value(term(), term(), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()} @spec format_field_value(term(), term(), term()) :: {:ok, term()} | {:error, Snakepit.Error.t()} @spec format_field_value(term(), term(), term(), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()} def format_field_value(field_info, value) do SnakeBridge.Runtime.call(__MODULE__, :format_field_value, [field_info, value], []) end def format_field_value(field_info, value, opts) when is_list(opts) and (opts == [] or (is_tuple(hd(opts)) and tuple_size(hd(opts)) == 2 and is_atom(elem(hd(opts), 0)))) do SnakeBridge.Runtime.call(__MODULE__, :format_field_value, [field_info, value], opts) end def format_field_value(field_info, value, assume_text) do SnakeBridge.Runtime.call( __MODULE__, :format_field_value, [field_info, value, assume_text], [] ) end def format_field_value(field_info, value, assume_text, opts) when is_list(opts) and (opts == [] or (is_tuple(hd(opts)) and tuple_size(hd(opts)) == 2 and is_atom(elem(hd(opts), 0)))) do SnakeBridge.Runtime.call( __MODULE__, :format_field_value, [field_info, value, assume_text], opts ) end @doc """ Python binding for `dspy.adapters.utils.get_annotation_name`. ## Parameters - `annotation` (term()) ## Returns - `term()` """ @spec get_annotation_name(term(), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()} def get_annotation_name(annotation, opts \\ []) do SnakeBridge.Runtime.call(__MODULE__, :get_annotation_name, [annotation], opts) end @doc """ Python binding for `dspy.adapters.utils.get_field_description_string`. ## Parameters - `fields` (%{optional(term()) => term()}) ## Returns - `String.t()` """ @spec get_field_description_string(%{optional(term()) => term()}, keyword()) :: {:ok, String.t()} | {:error, Snakepit.Error.t()} def get_field_description_string(fields, opts \\ []) do SnakeBridge.Runtime.call(__MODULE__, :get_field_description_string, [fields], opts) end @doc """ Python binding for `dspy.adapters.utils.parse_value`. ## Parameters - `value` (term()) - `annotation` (term()) ## Returns - `term()` """ @spec parse_value(term(), term(), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()} def parse_value(value, annotation, opts \\ []) do SnakeBridge.Runtime.call(__MODULE__, :parse_value, [value, annotation], opts) end @doc """ Formats the specified value so that it can be serialized as a JSON string. ## Parameters - `value` - The value to format as a JSON string. ## Returns - `term()` """ @spec serialize_for_json(term(), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()} def serialize_for_json(value, opts \\ []) do SnakeBridge.Runtime.call(__MODULE__, :serialize_for_json, [value], opts) end @doc """ Python binding for `dspy.adapters.utils.translate_field_type`. ## Parameters - `field_name` (term()) - `field_info` (term()) ## Returns - `term()` """ @spec translate_field_type(term(), term(), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()} def translate_field_type(field_name, field_info, opts \\ []) do SnakeBridge.Runtime.call(__MODULE__, :translate_field_type, [field_name, field_info], opts) end end