Gemini (GeminiEx v0.8.2)

View Source

Gemini Elixir Client

A comprehensive Elixir client for Google's Gemini AI API with dual authentication support, advanced streaming capabilities, type safety, and built-in telemetry.

Features

  • 🔐 Dual Authentication: Seamless support for both Gemini API keys and Vertex AI OAuth/Service Accounts
  • ⚡ Advanced Streaming: Production-grade Server-Sent Events streaming with real-time processing
  • 🛡️ Type Safety: Complete type definitions with runtime validation
  • 📊 Built-in Telemetry: Comprehensive observability and metrics out of the box
  • 💬 Chat Sessions: Multi-turn conversation management with state persistence
  • 🎭 Multimodal: Full support for text, image, audio, and video content
  • 🚀 Production Ready: Robust error handling, retry logic, and performance optimizations

Quick Start

Installation

Add to your mix.exs:

def deps do
  [
    {:gemini, "~> 0.0.1"}
  ]
end

Basic Configuration

Configure your API key in config/runtime.exs:

import Config

config :gemini,
  api_key: System.get_env("GEMINI_API_KEY")

Or set the environment variable:

export GEMINI_API_KEY="your_api_key_here"

Simple Usage

# Basic text generation
{:ok, response} = Gemini.generate("Tell me about Elixir programming")
{:ok, text} = Gemini.extract_text(response)
IO.puts(text)

# With options
{:ok, response} = Gemini.generate("Explain quantum computing", [
  model: Gemini.Config.get_model(:flash_lite_latest),
  temperature: 0.7,
  max_output_tokens: 1000
])

Streaming

# Start a streaming session
{:ok, stream_id} = Gemini.stream_generate("Write a long story", [
  on_chunk: fn chunk -> IO.write(chunk) end,
  on_complete: fn -> IO.puts("\n✅ Complete!") end
])

Authentication

This client supports two authentication methods:

1. Gemini API Key (Simple)

Best for development and simple applications:

# Environment variable (recommended)
export GEMINI_API_KEY="your_api_key"

# Application config
config :gemini, api_key: "your_api_key"

# Per-request override
Gemini.generate("Hello", api_key: "specific_key")

2. Vertex AI (Production)

Best for production Google Cloud applications:

# Service Account JSON file
export VERTEX_SERVICE_ACCOUNT="/path/to/service-account.json"
export VERTEX_PROJECT_ID="your-gcp-project"
export VERTEX_LOCATION="us-central1"

# Application config
config :gemini, :auth,
  type: :vertex_ai,
  credentials: %{
    service_account_key: System.get_env("VERTEX_SERVICE_ACCOUNT"),
    project_id: System.get_env("VERTEX_PROJECT_ID"),
    location: "us-central1"
  }

Error Handling

The client provides detailed error information with recovery suggestions:

case Gemini.generate("Hello world") do
  {:ok, response} ->
    {:ok, text} = Gemini.extract_text(response)

  {:error, %Gemini.Error{type: :rate_limit} = error} ->
    IO.puts("Rate limited. Retry after: #{error.retry_after}")

  {:error, %Gemini.Error{type: :authentication} = error} ->
    IO.puts("Auth error: #{error.message}")

  {:error, error} ->
    IO.puts("Unexpected error: #{inspect(error)}")
end

Advanced Features

Multimodal Content

content = [
  %{type: "text", text: "What's in this image?"},
  %{type: "image", source: %{type: "base64", data: base64_image}}
]

{:ok, response} = Gemini.generate(content)

Model Management

# List available models
{:ok, models} = Gemini.list_models()

# Get model details
{:ok, model_info} = Gemini.get_model(Gemini.Config.get_model(:flash_lite_latest))

# Count tokens
{:ok, token_count} = Gemini.count_tokens("Your text", model: Gemini.Config.get_model(:flash_lite_latest))

This module provides backward-compatible access to the Gemini API while routing requests through the unified coordinator for maximum flexibility and performance.

Summary

Types

options()

Options for content generation and related API calls.

Functions

async_batch_embed_contents(texts, opts \\ [])

Submit an asynchronous batch embedding job for production-scale generation.

await_batch_completion(batch_name, opts \\ [])

Poll and wait for batch completion with configurable intervals.

batch_embed_contents(texts, opts \\ [])

Generate embeddings for multiple texts in a single batch request.

chat(opts \\ [])

Start a new chat session.

configure(auth_type, credentials)

Configure authentication for the client.

count_tokens(contents, opts \\ [])

Count tokens in the given content.

create_cache(contents, opts \\ [])

Create a cached content resource for reuse across requests.

delete_cache(name, opts \\ [])

Delete cached content.

embed_content(text, opts \\ [])

Generate an embedding for the given text content.

extract_text(arg1)

Extract text from a GenerateContentResponse or raw streaming data.

extract_thought_signatures(arg1)

Extract thought signatures from a GenerateContentResponse.

generate(contents, opts \\ [])

Generate content using the configured authentication.

generate_content_with_auto_tools(contents, opts \\ [])

Generate content with automatic tool execution.

get_batch_embeddings(batch)

Retrieve embeddings from a completed batch job.

get_batch_status(batch_name, opts \\ [])

Get the current status of an async batch embedding job.

get_cache(name, opts \\ [])

Get a cached content by name.

get_model(model_name)

Get information about a specific model.

get_stream_status(stream_id)

Get stream status.

list_caches(opts \\ [])

List cached contents.

list_models(opts \\ [])

List available models.

model_exists?(model_name)

Check if a model exists.

send_message(chat, message)

Send a message in a chat session.

start_link()

Start the streaming manager (for compatibility).

start_stream(contents, opts \\ [])

Start a managed streaming session.

stream_generate(contents, opts \\ [])

Generate content with streaming response (synchronous collection).

stream_generate_with_auto_tools(contents, opts \\ [])

Start a streaming session with automatic tool execution.

subscribe_stream(stream_id)

Subscribe to streaming events.

text(contents, opts \\ [])

Generate text content and return only the text.

update_cache(name, opts)

Update cached content TTL/expiry.

Types

options()

@type options() :: [
  model: String.t(),
  generation_config: Gemini.Types.GenerationConfig.t() | nil,
  safety_settings: [Gemini.Types.SafetySetting.t()],
  system_instruction: Gemini.Types.Content.t() | String.t() | nil,
  tools: [map()],
  tool_config: map() | nil,
  api_key: String.t(),
  auth: :gemini | :vertex_ai,
  temperature: float(),
  max_output_tokens: non_neg_integer(),
  top_p: float(),
  top_k: non_neg_integer()
]

Options for content generation and related API calls.

  • :model - Model name (string, defaults to configured default model)
  • :generation_config - GenerationConfig struct (Gemini.Types.GenerationConfig.t())
  • :safety_settings - List of SafetySetting structs ([Gemini.Types.SafetySetting.t()])

  • :system_instruction - System instruction as Content struct or string (Gemini.Types.Content.t() | String.t() | nil)

  • :tools - List of tool definitions ([map()])

  • :tool_config - Tool configuration (map() | nil)

  • :api_key - Override API key (string)

  • :auth - Authentication strategy (:gemini | :vertex_ai)

  • :temperature - Generation temperature (float, 0.0-1.0)
  • :max_output_tokens - Maximum tokens to generate (non_neg_integer)
  • :top_p - Top-p sampling parameter (float)
  • :top_k - Top-k sampling parameter (non_neg_integer)

Functions

async_batch_embed_contents(texts, opts \\ [])

@spec async_batch_embed_contents([String.t()], options()) ::
  {:ok, map()} | {:error, Gemini.Error.t()}

Submit an asynchronous batch embedding job for production-scale generation.

Processes large batches with 50% cost savings compared to interactive API.

See Gemini.options/0 for available options.

Examples

{:ok, batch} = Gemini.async_batch_embed_contents(
  ["Text 1", "Text 2", "Text 3"],
  display_name: "My Batch",
  task_type: :retrieval_document
)

await_batch_completion(batch_name, opts \\ [])

@spec await_batch_completion(String.t(), options()) :: {:ok, map()} | {:error, term()}

Poll and wait for batch completion with configurable intervals.

Examples

{:ok, completed} = Gemini.await_batch_completion(
  batch.name,
  poll_interval: 10_000,
  timeout: 600_000
)

batch_embed_contents(texts, opts \\ [])

@spec batch_embed_contents([String.t()], options()) ::
  {:ok, map()} | {:error, Gemini.Error.t()}

Generate embeddings for multiple texts in a single batch request.

See Gemini.options/0 for available options.

Examples

{:ok, response} = Gemini.batch_embed_contents([
  "What is AI?",
  "How does ML work?"
])

chat(opts \\ [])

@spec chat(options()) :: {:ok, Gemini.Chat.t()}

Start a new chat session.

See Gemini.options/0 for available options.

configure(auth_type, credentials)

@spec configure(atom(), map()) :: :ok

Configure authentication for the client.

Examples

# Gemini API
Gemini.configure(:gemini, %{api_key: "your_api_key"})

# Vertex AI
Gemini.configure(:vertex_ai, %{
  service_account_key: "/path/to/key.json",
  project_id: "your-project",
  location: "us-central1"
})

count_tokens(contents, opts \\ [])

@spec count_tokens(String.t() | [Gemini.Types.Content.t()], options()) ::
  {:ok, map()} | {:error, Gemini.Error.t()}

Count tokens in the given content.

See Gemini.options/0 for available options.

create_cache(contents, opts \\ [])

@spec create_cache(
  [Gemini.Types.Content.t()] | [map()] | String.t(),
  keyword()
) :: {:ok, map()} | {:error, term()}

Create a cached content resource for reuse across requests.

delete_cache(name, opts \\ [])

@spec delete_cache(
  String.t(),
  keyword()
) :: :ok | {:error, term()}

Delete cached content.

embed_content(text, opts \\ [])

@spec embed_content(String.t(), options()) ::
  {:ok, map()} | {:error, Gemini.Error.t()}

Generate an embedding for the given text content.

See Gemini.options/0 for available options.

Examples

{:ok, response} = Gemini.embed_content("What is AI?")
{:ok, values} = EmbedContentResponse.get_values(response)

extract_text(arg1)

@spec extract_text(Gemini.Types.Response.GenerateContentResponse.t() | map()) ::
  {:ok, String.t()} | {:error, String.t()}

Extract text from a GenerateContentResponse or raw streaming data.

extract_thought_signatures(arg1)

@spec extract_thought_signatures(
  Gemini.Types.Response.GenerateContentResponse.t()
  | nil
) :: [
  String.t()
]

Extract thought signatures from a GenerateContentResponse.

Gemini 3 models return thought_signature fields on parts that must be echoed back in subsequent turns to maintain reasoning context.

Parameters

  • response: GenerateContentResponse struct

Returns

  • List of thought signature strings found in the response

Examples

{:ok, response} = Gemini.generate("Complex question", model: "gemini-3-pro-preview")
signatures = Gemini.extract_thought_signatures(response)
# => ["sig_abc123", "sig_def456"]

generate(contents, opts \\ [])

@spec generate(String.t() | [Gemini.Types.Content.t()], options()) ::
  {:ok, Gemini.Types.Response.GenerateContentResponse.t()}
  | {:error, Gemini.Error.t()}

Generate content using the configured authentication.

See Gemini.options/0 for available options.

generate_content_with_auto_tools(contents, opts \\ [])

@spec generate_content_with_auto_tools(
  String.t() | [Gemini.Types.Content.t()],
  options()
) ::
  {:ok, Gemini.Types.Response.GenerateContentResponse.t()}
  | {:error, Gemini.Error.t()}

Generate content with automatic tool execution.

This function provides a seamless, Python-SDK-like experience by automatically handling the tool-calling loop. When the model returns function calls, they are executed automatically and the conversation continues until a final text response is received.

Parameters

  • contents: String prompt or list of Content structs
  • opts: Standard generation options plus:
    • :turn_limit - Maximum number of tool-calling turns (default: 10)
    • :tools - List of tool declarations (required for tool calling)
    • :tool_config - Tool configuration (optional)

Examples

# Register a tool first
{:ok, declaration} = Altar.ADM.new_function_declaration(%{
  name: "get_weather",
  description: "Gets weather for a location",
  parameters: %{
    type: "object",
    properties: %{location: %{type: "string"}},
    required: ["location"]
  }
})
:ok = Gemini.Tools.register(declaration, &MyApp.get_weather/1)

# Use automatic tool execution
{:ok, response} = Gemini.generate_content_with_auto_tools(
  "What's the weather in San Francisco?",
  tools: [declaration],
  model: "gemini-flash-lite-latest"
)

Returns

  • {:ok, GenerateContentResponse.t()}: Final text response after all tool calls
  • {:error, term()}: Error during generation or tool execution

get_batch_embeddings(batch)

@spec get_batch_embeddings(map()) :: {:ok, [map()]} | {:error, String.t()}

Retrieve embeddings from a completed batch job.

Examples

{:ok, batch} = Gemini.get_batch_status(batch_id)
if batch.state == :completed do
  {:ok, embeddings} = Gemini.get_batch_embeddings(batch)
end

get_batch_status(batch_name, opts \\ [])

@spec get_batch_status(String.t(), options()) ::
  {:ok, map()} | {:error, Gemini.Error.t()}

Get the current status of an async batch embedding job.

Examples

{:ok, batch} = Gemini.get_batch_status("batches/abc123")
IO.puts("State: #{batch.state}")

get_cache(name, opts \\ [])

@spec get_cache(
  String.t(),
  keyword()
) :: {:ok, map()} | {:error, term()}

Get a cached content by name.

get_model(model_name)

@spec get_model(String.t()) :: {:ok, map()} | {:error, Gemini.Error.t()}

Get information about a specific model.

get_stream_status(stream_id)

@spec get_stream_status(String.t()) :: {:ok, map()} | {:error, Gemini.Error.t()}

Get stream status.

list_caches(opts \\ [])

@spec list_caches(keyword()) :: {:ok, map()} | {:error, term()}

List cached contents.

list_models(opts \\ [])

@spec list_models(options()) :: {:ok, map()} | {:error, Gemini.Error.t()}

List available models.

See Gemini.options/0 for available options.

model_exists?(model_name)

@spec model_exists?(String.t()) :: {:ok, boolean()}

Check if a model exists.

send_message(chat, message)

@spec send_message(Gemini.Chat.t(), String.t()) ::
  {:ok, Gemini.Types.Response.GenerateContentResponse.t(), Gemini.Chat.t()}
  | {:error, Gemini.Error.t()}

Send a message in a chat session.

start_link()

@spec start_link() :: {:ok, pid()} | {:error, term()}

Start the streaming manager (for compatibility).

start_stream(contents, opts \\ [])

@spec start_stream(String.t() | [Gemini.Types.Content.t()], options()) ::
  {:ok, String.t()} | {:error, Gemini.Error.t()}

Start a managed streaming session.

See Gemini.options/0 for available options.

stream_generate(contents, opts \\ [])

@spec stream_generate(String.t() | [Gemini.Types.Content.t()], options()) ::
  {:ok, [map()]} | {:error, Gemini.Error.t()}

Generate content with streaming response (synchronous collection).

See Gemini.options/0 for available options.

stream_generate_with_auto_tools(contents, opts \\ [])

@spec stream_generate_with_auto_tools(
  String.t() | [Gemini.Types.Content.t()],
  options()
) ::
  {:ok, String.t()} | {:error, Gemini.Error.t()}

Start a streaming session with automatic tool execution.

This function provides streaming support for the automatic tool-calling loop. When the model returns function calls, they are executed automatically and the conversation continues until a final text response is streamed to the subscriber.

Parameters

  • contents: String prompt or list of Content structs
  • opts: Standard generation options plus:
    • :turn_limit - Maximum number of tool-calling turns (default: 10)
    • :tools - List of tool declarations (required for tool calling)
    • :tool_config - Tool configuration (optional)

Examples

# Register a tool first
{:ok, declaration} = Altar.ADM.new_function_declaration(%{
  name: "get_weather",
  description: "Gets weather for a location",
  parameters: %{
    type: "object",
    properties: %{location: %{type: "string"}},
    required: ["location"]
  }
})
:ok = Gemini.Tools.register(declaration, &MyApp.get_weather/1)

# Start streaming with automatic tool execution
{:ok, stream_id} = Gemini.stream_generate_with_auto_tools(
  "What's the weather in San Francisco?",
  tools: [declaration],
  model: "gemini-flash-lite-latest"
)

# Subscribe to receive only the final text response
:ok = Gemini.subscribe_stream(stream_id)

Returns

  • {:ok, stream_id}: Stream started successfully
  • {:error, term()}: Error during stream setup

subscribe_stream(stream_id)

@spec subscribe_stream(String.t()) :: :ok | {:error, Gemini.Error.t()}

Subscribe to streaming events.

text(contents, opts \\ [])

@spec text(String.t() | [Gemini.Types.Content.t()], options()) ::
  {:ok, String.t()} | {:error, Gemini.Error.t()}

Generate text content and return only the text.

See Gemini.options/0 for available options.

update_cache(name, opts)

@spec update_cache(
  String.t(),
  keyword()
) :: {:ok, map()} | {:error, term()}

Update cached content TTL/expiry.