Gemini.Tools.Executor (GeminiEx v0.8.4)

View Source

Executes function calls from Gemini API responses against a registry of implementations.

This module provides the core function execution infrastructure for tool calling. It handles:

  • Executing single function calls against a function registry
  • Batch execution of multiple calls (sequential or parallel)
  • Building function responses for multi-turn conversations
  • Error handling and recovery

Function Registry

A function registry is a map from function names to implementations:

registry = %{
  "get_weather" => fn args -> WeatherService.get(args["location"]) end,
  "search_database" => fn args -> Database.search(args["query"]) end
}

You can also use create_registry/1 for convenience:

registry = Executor.create_registry(
  get_weather: &WeatherService.get(&1["location"]),
  search_database: &Database.search(&1["query"])
)

Examples

# Single execution
{:ok, call} = FunctionCall.new(call_id: "1", name: "add", args: %{"a" => 1, "b" => 2})
registry = %{"add" => fn args -> args["a"] + args["b"] end}

{:ok, result} = Executor.execute(call, registry)
#=> {:ok, 3}

# Batch execution
calls = [call1, call2, call3]
results = Executor.execute_all(calls, registry)

# Parallel execution for I/O-bound functions
results = Executor.execute_all_parallel(calls, registry)

# Build responses for Gemini API
responses = Executor.build_responses(calls, results)

Summary

Types

execution_result()
function_impl()
function_registry()

Functions

build_responses(calls, results)

Build FunctionResponse structs from execution results.

create_registry(functions)

Create a function registry from a keyword list or map.

execute(function_call, registry)

Execute a single function call against the registry.

execute_all(calls, registry)

Execute multiple function calls sequentially.

execute_all_parallel(calls, registry, opts \\ [])

Execute multiple function calls in parallel.

Types

execution_result()

@type execution_result() :: {:ok, term()} | {:error, term()}

function_impl()

@type function_impl() :: (map() -> term()) | {module(), atom(), list()}

function_registry()

@type function_registry() :: %{required(String.t()) => function_impl()}

Functions

build_responses(calls, results)

@spec build_responses([Altar.ADM.FunctionCall.t()], [execution_result()]) :: [
  Gemini.Types.FunctionResponse.t()
]

Build FunctionResponse structs from execution results.

Creates responses suitable for sending back to the Gemini API in multi-turn function calling conversations.

Parameters

Returns

List of FunctionResponse structs.

Examples

calls = [call1, call2]
results = Executor.execute_all(calls, registry)
responses = Executor.build_responses(calls, results)

# Use responses in next Gemini API call
contents = [previous_response, %{role: "function", parts: responses}]

create_registry(functions)

@spec create_registry(keyword() | map()) :: function_registry()

Create a function registry from a keyword list or map.

Converts atom keys to strings for consistent lookup.

Examples

# From keyword list
registry = Executor.create_registry(
  add: fn args -> args["a"] + args["b"] end,
  multiply: fn args -> args["a"] * args["b"] end
)

# From map with string keys
registry = Executor.create_registry(%{
  "add" => fn args -> args["a"] + args["b"] end
})

execute(function_call, registry)

@spec execute(Altar.ADM.FunctionCall.t(), function_registry()) :: execution_result()

Execute a single function call against the registry.

Parameters

  • call: A FunctionCall struct with name and args
  • registry: Map from function names to implementations

Returns

  • {:ok, result} - Function executed successfully
  • {:error, {:unknown_function, name}} - Function not found in registry
  • {:error, {:execution_error, exception}} - Function raised an exception

Examples

{:ok, call} = FunctionCall.new(call_id: "1", name: "double", args: %{"n" => 5})
registry = %{"double" => fn args -> args["n"] * 2 end}

{:ok, 10} = Executor.execute(call, registry)

execute_all(calls, registry)

@spec execute_all([Altar.ADM.FunctionCall.t()], function_registry()) :: [
  execution_result()
]

Execute multiple function calls sequentially.

Returns results in the same order as the input calls.

Parameters

  • calls: List of FunctionCall structs
  • registry: Function registry

Returns

List of execution results, one for each call.

Examples

results = Executor.execute_all([call1, call2], registry)
[{:ok, result1}, {:ok, result2}] = results

execute_all_parallel(calls, registry, opts \\ [])

@spec execute_all_parallel(
  [Altar.ADM.FunctionCall.t()],
  function_registry(),
  keyword()
) :: [
  execution_result()
]

Execute multiple function calls in parallel.

Uses Task.async_stream for concurrent execution. Best for I/O-bound functions like HTTP requests or database queries.

Parameters

  • calls: List of FunctionCall structs
  • registry: Function registry
  • opts: Options passed to Task.async_stream (default: max_concurrency: 10)

Returns

List of execution results, in the same order as input calls.

Examples

# Execute 3 slow operations in parallel
results = Executor.execute_all_parallel([call1, call2, call3], registry)