Current section
Files
Jump to
Current section
Files
lib/snakebridge_generated/dspy/teleprompt/utils/__init__.ex
# Generated by SnakeBridge v0.13.0 - DO NOT EDIT MANUALLY
# Regenerate with: mix compile
# Library: dspy 3.1.2
# Python module: dspy.teleprompt.utils
defmodule Dspy.Teleprompt.Utils do
@moduledoc """
Submodule bindings for `dspy.teleprompt.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.Teleprompt.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.Teleprompt.Utils.predict(data, __runtime__: [timeout_profile: :ml_inference])
# Or explicit timeout
Dspy.Teleprompt.Utils.predict(data, __runtime__: [timeout: 600_000])
# Route to a pool and enforce strict affinity
Dspy.Teleprompt.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.teleprompt.utils"
@doc false
def __snakebridge_library__, do: "dspy"
@doc """
Calculate the average and best quality of the last n programs proposed. This is useful for seeing if our proposals
are actually 'improving' overtime or not.
## Parameters
- `base_program` (term())
- `trial_logs` (term())
- `evaluate` (term())
- `trainset` (term())
- `devset` (term())
- `n` (term())
## Returns
- `term()`
"""
@spec calculate_last_n_proposed_quality(
term(),
term(),
term(),
term(),
term(),
term(),
keyword()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
def calculate_last_n_proposed_quality(
base_program,
trial_logs,
evaluate,
trainset,
devset,
n,
opts \\ []
) do
SnakeBridge.Runtime.call(
__MODULE__,
:calculate_last_n_proposed_quality,
[base_program, trial_logs, evaluate, trainset, devset, n],
opts
)
end
@doc """
Create a minibatch from the trainset.
## Parameters
- `trainset` (term())
- `batch_size` (term() default: 50)
- `rng` (term() default: None)
## Returns
- `term()`
"""
@spec create_minibatch(term()) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_minibatch(term(), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_minibatch(term(), term()) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_minibatch(term(), term(), keyword()) ::
{:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_minibatch(term(), term(), term()) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_minibatch(term(), term(), term(), keyword()) ::
{:ok, term()} | {:error, Snakepit.Error.t()}
def create_minibatch(trainset) do
SnakeBridge.Runtime.call(__MODULE__, :create_minibatch, [trainset], [])
end
def create_minibatch(trainset, 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__, :create_minibatch, [trainset], opts)
end
def create_minibatch(trainset, batch_size) do
SnakeBridge.Runtime.call(__MODULE__, :create_minibatch, [trainset, batch_size], [])
end
def create_minibatch(trainset, batch_size, 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__, :create_minibatch, [trainset, batch_size], opts)
end
def create_minibatch(trainset, batch_size, rng) do
SnakeBridge.Runtime.call(__MODULE__, :create_minibatch, [trainset, batch_size, rng], [])
end
def create_minibatch(trainset, batch_size, rng, 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__, :create_minibatch, [trainset, batch_size, rng], opts)
end
@doc """
This function is copied from random_search.py, and creates fewshot examples in the same way that random search does.
This allows us to take advantage of using the same fewshot examples when we use the same random seed in our optimizers.
## Parameters
- `student` (term())
- `num_candidate_sets` (term())
- `trainset` (term())
- `max_labeled_demos` (term())
- `max_bootstrapped_demos` (term())
- `metric` (term())
- `teacher_settings` (term())
- `max_errors` (term() default: None)
- `max_rounds` (term() default: 1)
- `labeled_sample` (term() default: True)
- `min_num_samples` (term() default: 1)
- `metric_threshold` (term() default: None)
- `teacher` (term() default: None)
- `include_non_bootstrapped` (term() default: True)
- `seed` (term() default: 0)
- `rng` (term() default: None)
## Returns
- `term()`
"""
@spec create_n_fewshot_demo_sets(term(), term(), term(), term(), term(), term(), term()) ::
{:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
keyword()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(term(), term(), term(), term(), term(), term(), term(), term()) ::
{:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
keyword()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
keyword()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
keyword()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
keyword()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
keyword()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
keyword()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
keyword()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
keyword()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec create_n_fewshot_demo_sets(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
term(),
keyword()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings
) do
SnakeBridge.Runtime.call(
__MODULE__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings
],
[]
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
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__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings
],
opts
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors
) do
SnakeBridge.Runtime.call(
__MODULE__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors
],
[]
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
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__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors
],
opts
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds
) do
SnakeBridge.Runtime.call(
__MODULE__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds
],
[]
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
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__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds
],
opts
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample
) do
SnakeBridge.Runtime.call(
__MODULE__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample
],
[]
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
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__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample
],
opts
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples
) do
SnakeBridge.Runtime.call(
__MODULE__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples
],
[]
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
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__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples
],
opts
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold
) do
SnakeBridge.Runtime.call(
__MODULE__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold
],
[]
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold,
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__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold
],
opts
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold,
teacher
) do
SnakeBridge.Runtime.call(
__MODULE__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold,
teacher
],
[]
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold,
teacher,
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__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold,
teacher
],
opts
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold,
teacher,
include_non_bootstrapped
) do
SnakeBridge.Runtime.call(
__MODULE__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold,
teacher,
include_non_bootstrapped
],
[]
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold,
teacher,
include_non_bootstrapped,
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__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold,
teacher,
include_non_bootstrapped
],
opts
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold,
teacher,
include_non_bootstrapped,
seed
) do
SnakeBridge.Runtime.call(
__MODULE__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold,
teacher,
include_non_bootstrapped,
seed
],
[]
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold,
teacher,
include_non_bootstrapped,
seed,
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__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold,
teacher,
include_non_bootstrapped,
seed
],
opts
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold,
teacher,
include_non_bootstrapped,
seed,
rng
) do
SnakeBridge.Runtime.call(
__MODULE__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold,
teacher,
include_non_bootstrapped,
seed,
rng
],
[]
)
end
def create_n_fewshot_demo_sets(
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold,
teacher,
include_non_bootstrapped,
seed,
rng,
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__,
:create_n_fewshot_demo_sets,
[
student,
num_candidate_sets,
trainset,
max_labeled_demos,
max_bootstrapped_demos,
metric,
teacher_settings,
max_errors,
max_rounds,
labeled_sample,
min_num_samples,
metric_threshold,
teacher,
include_non_bootstrapped,
seed,
rng
],
opts
)
end
@doc """
Evaluate a candidate program on the trainset, using the specified batch size.
## Parameters
- `batch_size` (term())
- `trainset` (term())
- `candidate_program` (term())
- `evaluate` (term())
- `rng` (term() default: None)
## Returns
- `term()`
"""
@spec eval_candidate_program(term(), term(), term(), term()) ::
{:ok, term()} | {:error, Snakepit.Error.t()}
@spec eval_candidate_program(term(), term(), term(), term(), keyword()) ::
{:ok, term()} | {:error, Snakepit.Error.t()}
@spec eval_candidate_program(term(), term(), term(), term(), term()) ::
{:ok, term()} | {:error, Snakepit.Error.t()}
@spec eval_candidate_program(term(), term(), term(), term(), term(), keyword()) ::
{:ok, term()} | {:error, Snakepit.Error.t()}
def eval_candidate_program(batch_size, trainset, candidate_program, evaluate) do
SnakeBridge.Runtime.call(
__MODULE__,
:eval_candidate_program,
[batch_size, trainset, candidate_program, evaluate],
[]
)
end
def eval_candidate_program(batch_size, trainset, candidate_program, evaluate, 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__,
:eval_candidate_program,
[batch_size, trainset, candidate_program, evaluate],
opts
)
end
def eval_candidate_program(batch_size, trainset, candidate_program, evaluate, rng) do
SnakeBridge.Runtime.call(
__MODULE__,
:eval_candidate_program,
[batch_size, trainset, candidate_program, evaluate, rng],
[]
)
end
def eval_candidate_program(batch_size, trainset, candidate_program, evaluate, rng, 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__,
:eval_candidate_program,
[batch_size, trainset, candidate_program, evaluate, rng],
opts
)
end
@doc """
Evaluation of candidate_program with pruning implemented
## Parameters
- `trial` (term())
- `trial_logs` (term())
- `trainset` (term())
- `candidate_program` (term())
- `evaluate` (term())
- `trial_num` (term())
- `batch_size` (term() default: 100)
## Returns
- `term()`
"""
@spec eval_candidate_program_with_pruning(term(), term(), term(), term(), term(), term()) ::
{:ok, term()} | {:error, Snakepit.Error.t()}
@spec eval_candidate_program_with_pruning(
term(),
term(),
term(),
term(),
term(),
term(),
keyword()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec eval_candidate_program_with_pruning(
term(),
term(),
term(),
term(),
term(),
term(),
term()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
@spec eval_candidate_program_with_pruning(
term(),
term(),
term(),
term(),
term(),
term(),
term(),
keyword()
) :: {:ok, term()} | {:error, Snakepit.Error.t()}
def eval_candidate_program_with_pruning(
trial,
trial_logs,
trainset,
candidate_program,
evaluate,
trial_num
) do
SnakeBridge.Runtime.call(
__MODULE__,
:eval_candidate_program_with_pruning,
[trial, trial_logs, trainset, candidate_program, evaluate, trial_num],
[]
)
end
def eval_candidate_program_with_pruning(
trial,
trial_logs,
trainset,
candidate_program,
evaluate,
trial_num,
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__,
:eval_candidate_program_with_pruning,
[trial, trial_logs, trainset, candidate_program, evaluate, trial_num],
opts
)
end
def eval_candidate_program_with_pruning(
trial,
trial_logs,
trainset,
candidate_program,
evaluate,
trial_num,
batch_size
) do
SnakeBridge.Runtime.call(
__MODULE__,
:eval_candidate_program_with_pruning,
[trial, trial_logs, trainset, candidate_program, evaluate, trial_num, batch_size],
[]
)
end
def eval_candidate_program_with_pruning(
trial,
trial_logs,
trainset,
candidate_program,
evaluate,
trial_num,
batch_size,
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__,
:eval_candidate_program_with_pruning,
[trial, trial_logs, trainset, candidate_program, evaluate, trial_num, batch_size],
opts
)
end
@doc """
Used as a helper function for bayesian + minibatching optimizers. Returns the program with the highest average score from the batches evaluated so far.
## Parameters
- `param_score_dict` (term())
- `fully_evaled_param_combos` (term())
## Returns
- `term()`
"""
@spec get_program_with_highest_avg_score(term(), term(), keyword()) ::
{:ok, term()} | {:error, Snakepit.Error.t()}
def get_program_with_highest_avg_score(param_score_dict, fully_evaled_param_combos, opts \\ []) do
SnakeBridge.Runtime.call(
__MODULE__,
:get_program_with_highest_avg_score,
[param_score_dict, fully_evaled_param_combos],
opts
)
end
@doc """
Python binding for `dspy.teleprompt.utils.get_prompt_model`.
## Parameters
- `prompt_model` (term())
## Returns
- `term()`
"""
@spec get_prompt_model(term(), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()}
def get_prompt_model(prompt_model, opts \\ []) do
SnakeBridge.Runtime.call(__MODULE__, :get_prompt_model, [prompt_model], opts)
end
@doc """
Python binding for `dspy.teleprompt.utils.get_signature`.
## Parameters
- `predictor` (term())
## Returns
- `term()`
"""
@spec get_signature(term(), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()}
def get_signature(predictor, opts \\ []) do
SnakeBridge.Runtime.call(__MODULE__, :get_signature, [predictor], opts)
end
@doc """
Get a full trace of the task model's history for a given candidate program.
## Parameters
- `candidate_program` (term())
- `task_model` (term())
- `devset` (term())
- `evaluate` (term())
## Returns
- `term()`
"""
@spec get_task_model_history_for_full_example(term(), term(), term(), term(), keyword()) ::
{:ok, term()} | {:error, Snakepit.Error.t()}
def get_task_model_history_for_full_example(
candidate_program,
task_model,
devset,
evaluate,
opts \\ []
) do
SnakeBridge.Runtime.call(
__MODULE__,
:get_task_model_history_for_full_example,
[candidate_program, task_model, devset, evaluate],
opts
)
end
@doc """
Extract total input tokens and output tokens from a model's interaction history.
Returns (total_input_tokens, total_output_tokens).
## Parameters
- `model` (term())
## Returns
- `{integer(), integer()}`
"""
@spec get_token_usage(term(), keyword()) ::
{:ok, {integer(), integer()}} | {:error, Snakepit.Error.t()}
def get_token_usage(model, opts \\ []) do
SnakeBridge.Runtime.call(__MODULE__, :get_token_usage, [model], opts)
end
@doc """
Extract total input and output tokens used by each model and log to trial_logs[trial_num]["token_usage"].
## Parameters
- `trial_logs` (term())
- `trial_num` (term())
- `model_dict` (term())
## Returns
- `term()`
"""
@spec log_token_usage(term(), term(), term(), keyword()) ::
{:ok, term()} | {:error, Snakepit.Error.t()}
def log_token_usage(trial_logs, trial_num, model_dict, opts \\ []) do
SnakeBridge.Runtime.call(
__MODULE__,
:log_token_usage,
[trial_logs, trial_num, model_dict],
opts
)
end
@doc """
Python binding for `dspy.teleprompt.utils.new_getfile`.
## Parameters
- `object` (term())
## Returns
- `term()`
"""
@spec new_getfile(term(), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()}
def new_getfile(object, opts \\ []) do
SnakeBridge.Runtime.call(__MODULE__, :new_getfile, [object], opts)
end
@doc """
Work out which source or compiled file an object was defined in.
## Parameters
- `object` (term())
## Returns
- `term()`
"""
@spec old_getfile(term(), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()}
def old_getfile(object, opts \\ []) do
SnakeBridge.Runtime.call(__MODULE__, :old_getfile, [object], opts)
end
@doc """
Print out the program's instructions & prefixes for each module.
## Parameters
- `program` (term())
## Returns
- `term()`
"""
@spec print_full_program(term(), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()}
def print_full_program(program, opts \\ []) do
SnakeBridge.Runtime.call(__MODULE__, :print_full_program, [program], opts)
end
@doc """
Save the candidate program to the log directory.
## Parameters
- `program` (term())
- `log_dir` (term())
- `trial_num` (term())
- `note` (term() default: None)
## Returns
- `term()`
"""
@spec save_candidate_program(term(), term(), term()) ::
{:ok, term()} | {:error, Snakepit.Error.t()}
@spec save_candidate_program(term(), term(), term(), keyword()) ::
{:ok, term()} | {:error, Snakepit.Error.t()}
@spec save_candidate_program(term(), term(), term(), term()) ::
{:ok, term()} | {:error, Snakepit.Error.t()}
@spec save_candidate_program(term(), term(), term(), term(), keyword()) ::
{:ok, term()} | {:error, Snakepit.Error.t()}
def save_candidate_program(program, log_dir, trial_num) do
SnakeBridge.Runtime.call(
__MODULE__,
:save_candidate_program,
[program, log_dir, trial_num],
[]
)
end
def save_candidate_program(program, log_dir, trial_num, 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__,
:save_candidate_program,
[program, log_dir, trial_num],
opts
)
end
def save_candidate_program(program, log_dir, trial_num, note) do
SnakeBridge.Runtime.call(
__MODULE__,
:save_candidate_program,
[program, log_dir, trial_num, note],
[]
)
end
def save_candidate_program(program, log_dir, trial_num, note, 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__,
:save_candidate_program,
[program, log_dir, trial_num, note],
opts
)
end
@doc """
Python binding for `dspy.teleprompt.utils.save_file_to_log_dir`.
## Parameters
- `source_file_path` (term())
- `log_dir` (term())
## Returns
- `term()`
"""
@spec save_file_to_log_dir(term(), term(), keyword()) ::
{:ok, term()} | {:error, Snakepit.Error.t()}
def save_file_to_log_dir(source_file_path, log_dir, opts \\ []) do
SnakeBridge.Runtime.call(__MODULE__, :save_file_to_log_dir, [source_file_path, log_dir], opts)
end
@doc """
Python binding for `dspy.teleprompt.utils.set_signature`.
## Parameters
- `predictor` (term())
- `updated_signature` (term())
## Returns
- `term()`
"""
@spec set_signature(term(), term(), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()}
def set_signature(predictor, updated_signature, opts \\ []) do
SnakeBridge.Runtime.call(__MODULE__, :set_signature, [predictor, updated_signature], opts)
end
@doc """
Setup logger, which will log our print statements to a txt file at our log_dir for later viewing
## Parameters
- `log_dir` (term())
## Returns
- `term()`
"""
@spec setup_logging(term(), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()}
def setup_logging(log_dir, opts \\ []) do
SnakeBridge.Runtime.call(__MODULE__, :setup_logging, [log_dir], opts)
end
end