WeaviateEx

A modern, idiomatic Elixir client for Weaviate vector database (v1.28+) with full Python client feature parity.

Features

Core Capabilities

• Complete API Coverage - Collections, objects, batch operations, queries, aggregations, cross-references, tenants
• RBAC & User Management - Full role-based access control, user lifecycle management, OIDC groups
• Hybrid Protocol Architecture - gRPC for high-performance data operations, HTTP for schema management
• Type-Safe - Protocol-based architecture with comprehensive typespecs
• Test-First Design - 2600+ tests with Mox-based mocking for fast, isolated testing
• Production-Ready - gRPC persistent channels, Finch HTTP pooling, proper error handling, health checks
• Easy Setup - First-class Mix tasks for managing local Weaviate stacks

Generative AI (RAG) - 20+ Providers

• OpenAI (GPT-4, GPT-3.5, O1/O3 reasoning models)
• Anthropic (Claude 3.5 Sonnet, Claude 3 Opus/Haiku)
• Cohere, Google Vertex/Gemini, AWS Bedrock/SageMaker
• Mistral, Ollama, XAI (Grok), ContextualAI
• NEW in v0.3: NVIDIA NIM, Databricks, FriendliAI
• Typed provider configurations with full parameter support
• Multimodal generation with image support

Vector Search

• Semantic Search - near_text, near_vector, near_object
• Multimodal Search - near_image (images), near_media (audio, video, thermal, depth, IMU)
• Hybrid Search - Combined keyword + vector with configurable alpha
• BM25 Keyword Search - Full-text search with AND/OR operators
• Reranking - gRPC-based result reranking with Cohere, Transformers, VoyageAI, and more
• Multi-Vector Support - ColBERT-style embeddings with Muvera encoding
• Named Vectors - Multiple vectors per object with targeting strategies

Advanced Features

• Cross-References - Full CRUD for object relationships
• Multi-Tenancy - HOT, COLD, FROZEN, OFFLOADED states
• Batch Operations - Error tracking, retry logic, rate limit handling
• Embedded Mode - Run Weaviate without Docker
• 20+ Vectorizers - OpenAI, Cohere, VoyageAI, Jina, Transformers, Ollama, and more
• gRPC Batch Streaming - High-performance bidirectional streaming (Weaviate 1.34+)

Table of Contents

• Quick Start
• Installation
• Configuration
• Usage
  • Embedded Mode
  • Health Checks
  • Server Version Detection
  • Collections (Schema Management)
  • Data Operations (CRUD)
  • Objects API
  • Batch Operations
  • Queries & Vector Search
  • Multimodal Search
  • Aggregations
  • Advanced Filtering
  • Vector Configuration
  • Backup & Restore
  • Multi-Tenancy
  • RBAC (Role-Based Access Control)
  • User Management
  • Group Management
• Examples
• Testing
• Mix Tasks
• Docker Management
• Authentication
• Connection Management
• Debug & Troubleshooting
• Documentation
• Contributing
• License

Quick Start

1. Start Weaviate locally

🧰 Prerequisite: Docker Desktop (macOS/Windows) or Docker Engine (Linux)

We ship Docker Compose profiles from the Python client under ci/. Use our Mix tasks to bring everything up:

# Start Weaviate containers (default version: 1.35.0)
mix weaviate.start

# Or specify a version
mix weaviate.start --version 1.35.0

# Inspect running services and health status
mix weaviate.status

The first run downloads the Weaviate Docker image and waits for the /v1/.well-known/ready endpoint to return 200.

When you're done:

mix weaviate.stop

Prefer direct scripts? Use ./ci/start_weaviate.sh 1.35.0 and ./ci/stop_weaviate.sh.

2. Add to Your Project

Add weaviate_ex to your mix.exs dependencies:

def deps do
  [
    {:weaviate_ex, "~> 0.7.4"}
  ]
end

Then fetch dependencies:

mix deps.get

3. Configure

The library automatically reads from environment variables (loaded from .env):

# .env file (created by install.sh)
WEAVIATE_URL=http://localhost:8080
WEAVIATE_API_KEY=  # Optional, for authenticated instances

Or configure in your Elixir config files:

# config/config.exs
config :weaviate_ex,
  url: "http://localhost:8080",
  api_key: nil,    # Optional
  strict: true     # Default: true - fails fast if Weaviate is unreachable

Strict Mode: By default, WeaviateEx validates connectivity on startup. If Weaviate is unreachable, your application won't start. Set strict: false to allow startup anyway (useful for development when Weaviate might not always be running).

4. Verify Connection

The library automatically performs a health check on startup:

[WeaviateEx] Successfully connected to Weaviate
  URL: http://localhost:8080
  Version: 1.34.0-rc.0

You can also run mix weaviate.status to see every profile that’s currently online and the ports they expose.

If configuration is missing, you'll get helpful error messages:

╔══════════════════════════════════════════════════════════════════════╗
║                  WeaviateEx Configuration Error                      ║
╠══════════════════════════════════════════════════════════════════════╣
║  Missing required configuration: WEAVIATE_URL                        ║
║                                                                      ║
║  Please set the Weaviate URL using one of these methods:             ║
║  1. Environment variable: export WEAVIATE_URL=http://localhost:8080  ║
║  2. Application configuration (config/config.exs)                    ║
║  3. Runtime configuration (config/runtime.exs)                       ║
╚══════════════════════════════════════════════════════════════════════╝

5. Shape a Tenant-Aware Collection and Load Data

alias WeaviateEx.{Collections, Objects, Batch}

# Define the collection and toggle multi-tenancy when ready
{:ok, _collection} =
  Collections.create("Article", %{
    description: "Articles by tenant",
    properties: [
      %{name: "title", dataType: ["text"]},
      %{name: "content", dataType: ["text"]}
    ]
  })

{:ok, %{"enabled" => true}} = Collections.set_multi_tenancy("Article", true)
{:ok, true} = Collections.exists?("Article")

# Create & read tenant-scoped objects with _additional metadata
{:ok, created} =
  Objects.create("Article", %{properties: %{title: "Tenant scoped", content: "Hello!"}},
    tenant: "tenant-a"
  )

{:ok, fetched} =
  Objects.get("Article", created["id"],
    tenant: "tenant-a",
    include: ["_additional", "vector"]
  )

# Batch ingest with a summary that separates successes from errors
objects =
  Enum.map(1..3, fn idx ->
    %{class: "Article", properties: %{title: "Story #{idx}"}, tenant: "tenant-a"}
  end)

{:ok, summary} = Batch.create_objects(objects, return_summary: true, tenant: "tenant-a")
summary.statistics
#=> %{processed: 3, successful: 3, failed: 0}

Installation

See INSTALL.md for detailed installation instructions covering:

• Docker installation on various platforms
• Manual Weaviate setup
• Configuration options
• Troubleshooting

Configuration

Environment Variables

Application Configuration

# config/config.exs
config :weaviate_ex,
  url: System.get_env("WEAVIATE_URL", "http://localhost:8080"),
  api_key: System.get_env("WEAVIATE_API_KEY"),
  strict: true,      # Fail on startup if unreachable
  timeout: 30_000    # Request timeout in milliseconds

gRPC Configuration

WeaviateEx v0.4.0+ uses a hybrid protocol architecture: gRPC for data operations (queries, batch, aggregations) and HTTP for schema management. gRPC provides significantly better performance for high-throughput operations.

# config/config.exs
config :weaviate_ex,
  url: "http://localhost:8080",       # HTTP endpoint for schema operations
  grpc_host: "localhost",             # gRPC host (default: derived from url)
  grpc_port: 50051,                   # gRPC port (default: 50051)
  grpc_max_message_size: 104_857_600, # Max message size in bytes (default: 100MB)
  api_key: nil                        # Used for both HTTP and gRPC auth

The gRPC connection is automatically established when you create a client:

# Connect with gRPC (automatic)
{:ok, client} = WeaviateEx.Client.connect(
  url: "http://localhost:8080",
  grpc_port: 50051
)

# Client now has both HTTP and gRPC channels
client.grpc_channel  # => gRPC channel for data operations
client.config        # => Configuration for HTTP operations

Custom Headers (v0.7.1+)

Add custom headers to all HTTP and gRPC requests for authentication, tracing, or other purposes:

# Configure additional headers in client config
{:ok, client} = WeaviateEx.Client.connect(
  url: "http://localhost:8080",
  additional_headers: %{
    "X-Custom-Header" => "custom-value",
    "X-Request-ID" => "trace-123",
    "Authorization" => "Bearer custom-token"
  }
)

# Headers are automatically included in:
# - All HTTP requests (schema operations, health checks)
# - All gRPC requests as metadata (lowercased keys)

Headers are validated on client creation - nil values will raise an ArgumentError.

gRPC Retry with Exponential Backoff (v0.7.1+)

All gRPC operations automatically retry on transient errors with exponential backoff:

# Retryable gRPC status codes:
# - UNAVAILABLE (14)     - Service temporarily unavailable
# - RESOURCE_EXHAUSTED (8) - Rate limiting
# - ABORTED (10)         - Transaction aborted
# - DEADLINE_EXCEEDED (4) - Timeout

# Default: 4 retries with exponential backoff
# Attempt 0: 1 second delay
# Attempt 1: 2 seconds
# Attempt 2: 4 seconds
# Attempt 3: 8 seconds
# Maximum delay capped at 32 seconds

# Configure retry behavior (optional)
alias WeaviateEx.GRPC.Retry

# Custom retry with options
result = Retry.with_retry(
  fn -> some_grpc_operation() end,
  max_retries: 3,
  base_delay_ms: 500
)

# Check if error is retryable
Retry.retryable?(%GRPC.RPCError{status: 14})  # => true (UNAVAILABLE)
Retry.retryable?(%GRPC.RPCError{status: 3})   # => false (INVALID_ARGUMENT)

# Calculate backoff delay
Retry.calculate_backoff(0)  # => 1000ms
Retry.calculate_backoff(2)  # => 4000ms
Retry.calculate_backoff(5)  # => 32000ms (capped)

All gRPC services (Search, Batch, Aggregate, Tenants, Health) automatically use retry logic.

Proxy Configuration (v0.5.0+)

WeaviateEx supports HTTP, HTTPS, and gRPC proxy configuration:

alias WeaviateEx.Config.Proxy

# Read from environment variables (HTTP_PROXY, HTTPS_PROXY, GRPC_PROXY)
proxy = Proxy.from_env()

# Or configure explicitly
proxy = Proxy.new(
  http: "http://proxy.example.com:8080",
  https: "https://proxy.example.com:8443",
  grpc: "http://grpc-proxy.example.com:8080"
)

# Check if proxy is configured
Proxy.configured?(proxy)  # => true

# Get Finch HTTP client options
Proxy.to_finch_opts(proxy)  # => [proxy: {:https, "proxy.example.com", 8443, []}]

# Get gRPC channel options
Proxy.to_grpc_opts(proxy)   # => [http_proxy: "http://grpc-proxy.example.com:8080"]

Environment variables are read case-insensitively (uppercase takes precedence):

• HTTP_PROXY / http_proxy - HTTP proxy URL
• HTTPS_PROXY / https_proxy - HTTPS proxy URL
• GRPC_PROXY / grpc_proxy - gRPC proxy URL

Runtime Configuration (Recommended for Production)

# config/runtime.exs
config :weaviate_ex,
  url: System.fetch_env!("WEAVIATE_URL"),
  api_key: System.get_env("WEAVIATE_API_KEY")

Usage

Embedded Mode

Need an ephemeral instance without Docker? WeaviateEx can download and manage the official embedded binary:

# Downloads (once) into ~/.cache/weaviate-embedded and starts the process
{:ok, embedded} =
  WeaviateEx.start_embedded(
    version: "1.34.0",
    port: 8099,
    grpc_port: 50155,
    persistence_data_path: Path.expand("tmp/weaviate-data"),
    environment_variables: %{"DISABLE_TELEMETRY" => "true"}
  )

# Talk to it just like any other instance
System.put_env("WEAVIATE_URL", "http://localhost:8099")
{:ok, meta} = WeaviateEx.health_check()

# Always stop the handle when finished
:ok = WeaviateEx.stop_embedded(embedded)

Passing version: "latest" fetches the most recent GitHub release. Binaries are cached, so subsequent calls reuse the download. You can override binary_path/persistence_data_path to control where the executable and data live.

Health Checks

Check if Weaviate is accessible and get version information:

# Get metadata (version, modules)
{:ok, meta} = WeaviateEx.health_check()
# => %{"version" => "1.34.0-rc.0", "modules" => %{}}

# Check readiness (can handle requests) - K8s readiness probe
{:ok, true} = WeaviateEx.ready?()

# Check liveness (service is up) - K8s liveness probe
{:ok, true} = WeaviateEx.alive?()

# With explicit client
{:ok, client} = WeaviateEx.Client.connect(base_url: "http://localhost:8080")
{:ok, true} = WeaviateEx.Health.alive?(client)
{:ok, true} = WeaviateEx.Health.ready?(client)

# Wait for Weaviate to become ready (useful for startup scripts)
:ok = WeaviateEx.Health.wait_until_ready(timeout: 30_000, check_interval: 1000)

# gRPC health ping (v0.7.0+)
alias WeaviateEx.GRPC.Services.Health, as: GRPCHealth
:ok = GRPCHealth.ping(client.grpc_channel)

Kubernetes Integration

The alive? and ready? functions use the standard Kubernetes probe endpoints:

• Liveness: /.well-known/live - Is the process running?
• Readiness: /.well-known/ready - Can the service handle traffic?

# Example K8s deployment liveness/readiness probes
livenessProbe:
  httpGet:
    path: /.well-known/live
    port: 8080
readinessProbe:
  httpGet:
    path: /.well-known/ready
    port: 8080

Server Version Detection

Parse and validate Weaviate server versions (v0.7.0+):

alias WeaviateEx.Version

# Parse version strings
{:ok, {1, 28, 0}} = Version.parse("1.28.0")
{:ok, {1, 28, 0}} = Version.parse("v1.28.0-rc1")  # Handles v prefix and prerelease

# Check if version meets minimum requirement
true = Version.meets_minimum?({1, 28, 0}, {1, 27, 0})
false = Version.meets_minimum?({1, 26, 0}, {1, 27, 0})

# Validate server version (minimum: 1.27.0)
:ok = Version.validate_server({1, 28, 0})
{:error, {:unsupported_version, {1, 20, 0}, {1, 27, 0}}} = Version.validate_server({1, 20, 0})

# Extract version from meta endpoint response
{:ok, meta} = WeaviateEx.health_check()
{:ok, {1, 28, 0}} = Version.get_server_version(meta)

# Get minimum supported version
Version.minimum_version()        # => {1, 27, 0}
Version.minimum_version_string() # => "1.27.0"

# Format version tuple to string
"1.28.0" = Version.format_version({1, 28, 0})

Collections (Schema Management)

Collections define the structure of your data:

# Create a collection with properties
{:ok, collection} = WeaviateEx.Collections.create("Article", %{
  description: "News articles",
  properties: [
    %{name: "title", dataType: ["text"]},
    %{name: "content", dataType: ["text"]},
    %{name: "publishedAt", dataType: ["date"]},
    %{name: "views", dataType: ["int"]}
  ],
  vectorizer: "none"  # Use "text2vec-openai" for auto-vectorization
})

# List all collections
{:ok, schema} = WeaviateEx.Collections.list()

# Get a specific collection
{:ok, collection} = WeaviateEx.Collections.get("Article")

# Add a property to existing collection
{:ok, property} = WeaviateEx.Collections.add_property("Article", %{
  name: "author",
  dataType: ["text"]
})

# Check if collection exists
{:ok, true} = WeaviateEx.Collections.exists?("Article")

# Delete a collection
{:ok, _} = WeaviateEx.Collections.delete("Article")

Object TTL (Time-To-Live)

Automatically expire and delete objects after a specified duration:

alias WeaviateEx.Config.ObjectTTL

# Create collection with 24-hour TTL using human-readable duration
{:ok, _} = WeaviateEx.Collections.create("Events", %{
  properties: [%{name: "title", dataType: ["text"]}],
  object_ttl: ObjectTTL.from_duration(hours: 24)
})

# Or specify exact seconds with creation time deletion
{:ok, _} = WeaviateEx.Collections.create("Sessions", %{
  properties: [%{name: "user_id", dataType: ["text"]}],
  object_ttl: ObjectTTL.delete_by_creation_time(3600)  # 1 hour
})

# Delete objects based on last update time
{:ok, _} = WeaviateEx.Collections.create("Cache", %{
  properties: [%{name: "data", dataType: ["text"]}],
  object_ttl: ObjectTTL.delete_by_update_time(86_400, true)  # 24h, filter expired
})

# Delete objects based on a custom date property
{:ok, _} = WeaviateEx.Collections.create("Subscriptions", %{
  properties: [
    %{name: "plan", dataType: ["text"]},
    %{name: "expires_at", dataType: ["date"]}
  ],
  object_ttl: ObjectTTL.delete_by_date_property("expires_at")
})

# Update TTL on existing collection
{:ok, _} = WeaviateEx.Collections.update_ttl("Events",
  ObjectTTL.from_duration(days: 7)
)

# Disable TTL
{:ok, _} = WeaviateEx.Collections.update_ttl("Events",
  ObjectTTL.disable()
)

Note: Objects are deleted asynchronously in the background. The filter_expired_objects option (second parameter in delete_by_* functions) controls whether expired but not yet deleted objects are excluded from search results.

Schema helpers for range filters and auto-tenant configuration:

alias WeaviateEx.Config.{AutoTenant, ObjectTTL}
alias WeaviateEx.Schema.MultiTenancyConfig
alias WeaviateEx.Property

ttl = ObjectTTL.delete_by_update_time(86_400, true)

{:ok, _} = WeaviateEx.Collections.create("Session", %{
  properties: [
    Property.number("expires_in", index_range_filters: true)
  ],
  object_ttl: ttl,
  multi_tenancy_config: MultiTenancyConfig.new(enabled: true, auto_tenant_creation: true),
  auto_tenant: AutoTenant.enable(auto_delete_timeout: 3_600)
})

Nested Properties

Define complex object structures with nested properties:

alias WeaviateEx.Property
alias WeaviateEx.Property.Nested

# Create a collection with nested object properties
{:ok, _} = WeaviateEx.Collections.create("Product", %{
  description: "Products with specifications",
  properties: [
    %{name: "name", dataType: ["text"]},
    %{name: "price", dataType: ["number"]},
    # Nested object property
    Property.object("specs", [
      Nested.new(name: "weight", data_type: :number),
      Nested.new(name: "dimensions", data_type: :text),
      Nested.new(name: "material", data_type: :text)
    ]),
    # Array of nested objects
    Property.object_array("variants", [
      Nested.new(name: "color", data_type: :text),
      Nested.new(name: "size", data_type: :text),
      Nested.new(name: "sku", data_type: :text),
      Nested.new(name: "stock", data_type: :int)
    ])
  ]
})

# Insert object with nested data
{:ok, product} = WeaviateEx.Objects.create("Product", %{
  properties: %{
    name: "Laptop Stand",
    price: 79.99,
    specs: %{
      weight: 2.5,
      dimensions: "30x25x15cm",
      material: "aluminum"
    },
    variants: [
      %{color: "silver", size: "standard", sku: "LS-001", stock: 50},
      %{color: "black", size: "large", sku: "LS-002", stock: 30}
    ]
  }
})

# Deeply nested properties (object within object)
{:ok, _} = WeaviateEx.Collections.create("Company", %{
  properties: [
    %{name: "name", dataType: ["text"]},
    Property.object("headquarters", [
      Nested.new(name: "city", data_type: :text),
      Nested.new(name: "country", data_type: :text),
      Nested.new(
        name: "address",
        data_type: :object,
        nested_properties: [
          Nested.new(name: "street", data_type: :text),
          Nested.new(name: "zip", data_type: :text)
        ]
      )
    ])
  ]
})

# Parse nested properties from API response
api_data = %{
  "name" => "specs",
  "dataType" => ["object"],
  "nestedProperties" => [
    %{"name" => "weight", "dataType" => ["number"]}
  ]
}
nested = Nested.from_api(api_data)

Data Operations (CRUD)

Simple CRUD operations with automatic UUID generation:

alias WeaviateEx.API.Data

# Create (insert) a new object
data = %{
  properties: %{
    "title" => "Hello Weaviate",
    "content" => "This is a test article",
    "views" => 0
  },
  vector: [0.1, 0.2, 0.3, 0.4, 0.5]  # Optional if using auto-vectorization
}

{:ok, object} = Data.insert(client, "Article", data)

# Named vectors (v0.7.1+) - for collections with multiple vector spaces
data_with_named_vectors = %{
  properties: %{"title" => "Multi-vector article"},
  vectors: %{
    "title_vector" => [0.1, 0.2, 0.3],
    "content_vector" => [0.4, 0.5, 0.6, 0.7]
  }
}

{:ok, object} = Data.insert(client, "MultiVectorCollection", data_with_named_vectors)
uuid = object["id"]

# Read - get object by ID
{:ok, retrieved} = Data.get_by_id(client, "Article", uuid)

# Update - partial update (PATCH)
{:ok, updated} = Data.patch(client, "Article", uuid, %{
  properties: %{"views" => 42},
  vector: [0.1, 0.2, 0.3, 0.4, 0.5]
})

# Check if object exists
{:ok, true} = Data.exists?(client, "Article", uuid)

# Delete
{:ok, _} = Data.delete_by_id(client, "Article", uuid)

Collection handles with default tenant/consistency:

collection =
  WeaviateEx.Collection.new(client, "Article",
    tenant: "tenant-a",
    consistency_level: "QUORUM"
  )

{:ok, _} = WeaviateEx.Collection.insert(collection, %{properties: %{title: "Tenant scoped"}})

Inline References During Insert (v0.7.1+)

Create objects with references in a single operation:

# Insert object with inline references
{:ok, article} = WeaviateEx.Objects.create("Article", %{
  properties: %{
    title: "My Article",
    content: "Article content..."
  },
  # Single reference
  references: %{
    "hasAuthor" => "author-uuid-here"
  }
})

# Multiple references to same property
{:ok, article} = WeaviateEx.Objects.create("Article", %{
  properties: %{title: "Collaborative Article"},
  references: %{
    "hasAuthors" => ["author-uuid-1", "author-uuid-2", "author-uuid-3"]
  }
})

# Multi-target references (pointing to specific collection)
{:ok, article} = WeaviateEx.Objects.create("Article", %{
  properties: %{title: "Related Content"},
  references: %{
    "relatedTo" => %{
      target_collection: "Category",
      uuids: "category-uuid"
    }
  }
})

# Multiple multi-target references
{:ok, article} = WeaviateEx.Objects.create("Article", %{
  properties: %{title: "Multi-related"},
  references: %{
    "mentions" => %{
      target_collection: "Person",
      uuids: ["person-1", "person-2"]
    }
  }
})

References are automatically converted to Weaviate beacon format.

Reference Operations API (v0.7.3+)

For managing references after object creation, use the References API with full multi-target support:

alias WeaviateEx.API.References
alias WeaviateEx.Data.ReferenceToMulti
alias WeaviateEx.Types.Beacon

# Add a single reference
{:ok, _} = References.add(client, "Article", article_uuid, "hasAuthor", author_uuid)

# Add a multi-target reference using ReferenceToMulti
ref = ReferenceToMulti.new("Person", person_uuid)
{:ok, _} = References.add(client, "Article", article_uuid, "hasAuthor", ref)

# Add multiple references at once
ref = ReferenceToMulti.new("Person", [person1_uuid, person2_uuid])
{:ok, _} = References.add(client, "Article", article_uuid, "hasAuthors", ref)

# Replace all references on a property
{:ok, _} = References.replace(client, "Article", article_uuid, "hasAuthors",
  [author1_uuid, author2_uuid, author3_uuid]
)

# Replace with multi-target references pointing to different collections
{:ok, _} = References.replace(client, "Article", article_uuid, "relatedTo", [
  ReferenceToMulti.new("Person", person_uuid),
  ReferenceToMulti.new("Organization", org_uuid)
])

# Delete a reference
{:ok, _} = References.delete(client, "Article", article_uuid, "hasAuthor", author_uuid)

# Batch add references
refs = [
  %{from_uuid: "article-1", from_property: "hasAuthor", to_uuid: "author-1"},
  %{from_uuid: "article-2", from_property: "hasAuthor", to_uuid: "author-2",
    target_collection: "Person"}  # For multi-target properties
]
{:ok, _} = References.add_many(client, "Article", refs)

# Parse beacon URLs
parsed = Beacon.parse("weaviate://localhost/Person/uuid-123")
# => %{collection: "Person", uuid: "uuid-123"}

# Build beacon URLs
beacon = Beacon.build("uuid-123", "Person")
# => "weaviate://localhost/Person/uuid-123"

Objects API

Full CRUD operations with explicit UUID control:

# Create with custom UUID
{:ok, object} = WeaviateEx.Objects.create("Article", %{
  id: "custom-uuid-here",  # Optional
  properties: %{
    title: "Hello Weaviate",
    content: "This is a test article",
    publishedAt: "2025-01-15T10:00:00Z"
  },
  vector: [0.1, 0.2, 0.3]  # Optional
})

# Get an object with additional fields
{:ok, object} = WeaviateEx.Objects.get("Article", uuid,
  include: "vector,classification"
)

# List objects with pagination
{:ok, result} = WeaviateEx.Objects.list("Article",
  limit: 10,
  offset: 0,
  include: "vector"
)

# Update (full replacement)
{:ok, updated} = WeaviateEx.Objects.update("Article", uuid, %{
  properties: %{
    title: "Updated Title",
    content: "Updated content"
  }
})

# Patch (partial update)
{:ok, patched} = WeaviateEx.Objects.patch("Article", uuid, %{
  properties: %{title: "New Title"}
})

# Delete
{:ok, _} = WeaviateEx.Objects.delete("Article", uuid)

# Check existence
{:ok, true} = WeaviateEx.Objects.exists?("Article", uuid)

Payload validation happens client-side: properties is required for inserts/updates, and property names id and vector are reserved (raises ArgumentError).

Complex Data Types

WeaviateEx automatically serializes complex Elixir types when creating or updating objects:

alias WeaviateEx.Types.{GeoCoordinate, PhoneNumber, Blob}

# DateTime - serialized to RFC3339/ISO8601
%{created_at: ~U[2024-01-01 00:00:00Z]}
# -> {"created_at": "2024-01-01T00:00:00Z"}

# Date - serialized as midnight UTC
%{published_date: ~D[2024-06-15]}
# -> {"published_date": "2024-06-15T00:00:00Z"}

# GeoCoordinate - serialized to lat/lon map
{:ok, geo} = GeoCoordinate.new(40.71, -74.00)
%{location: geo}
# -> {"location": {"latitude": 40.71, "longitude": -74.00}}

# PhoneNumber - serialized with input and country
phone = PhoneNumber.new("555-1234", default_country: "US")
%{contact: phone}
# -> {"contact": {"input": "555-1234", "defaultCountry": "US"}}

# Blob (binary data) - base64 encoded
blob = Blob.new(<<binary_image_data>>)
%{image: blob}
# -> {"image": "<base64 encoded string>"}

# Nested objects with complex types
{:ok, geo} = GeoCoordinate.new(40.7128, -74.0060)
{:ok, article} = WeaviateEx.Objects.create("Place", %{
  properties: %{
    name: "Central Park",
    location: geo,
    created_at: ~U[2024-01-01 00:00:00Z],
    metadata: %{
      last_visited: ~D[2024-12-25]
    }
  }
})

Deserializing Responses

Convert Weaviate response data back to rich Elixir types:

alias WeaviateEx.Types.Deserialize

# Parse individual values
{:ok, dt} = Deserialize.deserialize("2024-01-01T00:00:00Z", :date)
# => {:ok, ~U[2024-01-01 00:00:00Z]}

{:ok, geo} = Deserialize.deserialize(
  %{"latitude" => 52.37, "longitude" => 4.90},
  :geo_coordinates
)
# => {:ok, %GeoCoordinate{latitude: 52.37, longitude: 4.90}}

# Deserialize properties with schema hints
schema = %{"created_at" => :date, "location" => :geo_coordinates}
{:ok, props} = Deserialize.deserialize_properties(raw_props, schema)

# Auto-detect types based on value structure
{:ok, props} = Deserialize.auto_deserialize(response["properties"])

Batch Operations

Efficient bulk operations for importing large datasets:

# Batch create multiple objects
objects = [
  %{class: "Article", properties: %{title: "Article 1", content: "Content 1"}},
  %{class: "Article", properties: %{title: "Article 2", content: "Content 2"}},
  %{class: "Article", properties: %{title: "Article 3", content: "Content 3"}}
]

{:ok, summary} = WeaviateEx.Batch.create_objects(objects, return_summary: true)

# Check rolled-up stats and per-object errors
summary.statistics
#=> %{processed: 3, successful: 3, failed: 0}

Enum.each(summary.errors, fn error ->
  Logger.warn("[Batch error] #{error.id} => #{Enum.join(error.messages, "; ")}")
end)

If every object in the batch fails, `Batch.create_objects/2` returns
`{:error, %WeaviateEx.Error{type: :batch_all_failed}}`.

# Batch delete with criteria (WHERE filter)
{:ok, result} = WeaviateEx.Batch.delete_objects(%{
  class: "Article",
  where: %{
    path: ["status"],
    operator: "Equal",
    valueText: "draft"
  }
})

Concurrent Batch Operations

High-throughput parallel batch processing with failure tracking:

alias WeaviateEx.Batch.Concurrent
alias WeaviateEx.Batch.Queue

# Concurrent batch insertion with parallel processing
objects = Enum.map(1..10_000, fn i ->
  %{class: "Article", properties: %{title: "Article #{i}", content: "Content #{i}"}}
end)

{:ok, result} = Concurrent.insert_many(client, "Article", objects,
  max_concurrency: 8,    # Parallel batch requests
  batch_size: 200,       # Objects per request
  ordered: false,        # Don't maintain order (faster)
  timeout: 60_000        # Timeout per batch
)

# Check results
IO.puts(Concurrent.Result.summary(result))
# => "Inserted 10000/10000 objects in 50 batches (1234ms). Failures: 0, Batch errors: 0"

if Concurrent.Result.all_successful?(result) do
  IO.puts("All objects inserted successfully!")
else
  IO.puts("Some failures occurred")
  for failed <- result.failed do
    IO.puts("Failed: #{failed.id} - #{failed.error}")
  end
end

# Batch Queue for failure tracking and re-queuing
queue = Queue.new()

# Add objects to queue
queue = Enum.reduce(objects, queue, fn obj, q ->
  Queue.enqueue(q, obj)
end)

# Dequeue a batch for processing
{batch, queue} = Queue.dequeue_batch(queue, 100)

# Process batch and mark failures
queue = Enum.reduce(failed_objects, queue, fn {obj, reason}, q ->
  Queue.mark_failed(q, obj, reason)
end)

# Re-queue failed objects for retry (with max retry limit)
queue = Queue.requeue_failed(queue, max_retries: 3)

# Get queue statistics
IO.puts("Pending: #{Queue.pending_count(queue)}")
IO.puts("Failed: #{Queue.failed_count(queue)}")
IO.puts("Empty: #{Queue.empty?(queue)}")

# Rate limit detection
alias WeaviateEx.Batch.RateLimit

response = %{status: 429, headers: [{"retry-after", "5"}]}
case RateLimit.detect(response) do
  :ok -> IO.puts("No rate limit")
  {:rate_limited, wait_ms} ->
    IO.puts("Rate limited, wait #{wait_ms}ms")
    Process.sleep(wait_ms)
end

# Server queue monitoring for dynamic batch sizing
alias WeaviateEx.API.Cluster

{:ok, stats} = Cluster.batch_stats(client)
IO.puts("Queue length: #{stats.queue_length}")
IO.puts("Rate: #{stats.rate_per_second}/s")
IO.puts("Failed: #{stats.failed_count}")

gRPC Batch Streaming (v0.6.0+)

Bidirectional gRPC streaming for high-throughput batch operations (requires Weaviate 1.34+):

alias WeaviateEx.Batch.Stream

# Create a streaming batch session
{:ok, stream} = Stream.new(client, "Article",
  buffer_size: 200,           # Objects per batch
  flush_interval_ms: 1000,    # Auto-flush interval
  auto_flush: true            # Enable automatic flushing
)

# Add objects to the stream buffer
{:ok, stream} = Stream.add(stream, %{
  properties: %{title: "Article 1", content: "Content 1"}
})

{:ok, stream} = Stream.add(stream, %{
  properties: %{title: "Article 2", content: "Content 2"}
})

# Manually flush when buffer reaches threshold
{:ok, stream} = Stream.flush(stream)

# Add many objects efficiently
objects = Enum.map(1..1000, fn i ->
  %{properties: %{title: "Article #{i}", content: "Content #{i}"}}
end)

{:ok, stream} = Enum.reduce(objects, {:ok, stream}, fn obj, {:ok, s} ->
  Stream.add(s, obj)
end)

# Close stream and get final results
{:ok, results} = Stream.close(stream)

# Results include success/failure for each object
Enum.each(results, fn result ->
  case result do
    %{status: :success, uuid: uuid} ->
      IO.puts("Created: #{uuid}")
    %{status: :failed, error: error} ->
      IO.puts("Failed: #{error}")
  end
end)

When the server sends backoff messages, the stream automatically updates its buffer size to the server-provided batch size for subsequent flushes.

Low-Level gRPC Streaming

For advanced use cases, access the underlying gRPC stream directly:

alias WeaviateEx.GRPC.Services.BatchStream

# Open a bidirectional stream
{:ok, stream_handle} = BatchStream.open(client.grpc_channel)

# Send objects
:ok = BatchStream.send_objects(stream_handle, [
  %{collection: "Article", properties: %{title: "Test"}, uuid: nil, vector: nil}
])

# Send cross-references
:ok = BatchStream.send_references(stream_handle, [
  %{from_collection: "Article", from_uuid: "...", to_collection: "Author", to_uuid: "..."}
])

# Receive results
{:ok, results} = BatchStream.receive_results(stream_handle, timeout: 5000)

# Close the stream
:ok = BatchStream.close(stream_handle)

Background Batch Processing (v0.7.0+)

For high-throughput scenarios, use the background batcher for continuous async processing:

alias WeaviateEx.Batch.Background

# Start a background batch processor
{:ok, batcher} = WeaviateEx.Batch.background(client, "Article",
  batch_size: 100,
  concurrent_requests: 2,
  flush_interval: 1000
)

# Add objects asynchronously (non-blocking)
for article <- articles do
  :ok = Background.add_object(batcher, %{
    title: article.title,
    content: article.content
  })
end

# Add objects with explicit UUID and vector
:ok = Background.add_object(batcher, %{title: "Test"},
  uuid: "550e8400-e29b-41d4-a716-446655440000",
  vector: [0.1, 0.2, 0.3]
)

# Add references (automatically ordered after related objects)
:ok = Background.add_reference(batcher, article_uuid, "hasAuthor", author_uuid)

# Force immediate flush
:ok = Background.flush(batcher)

# Get current results
results = Background.get_results(batcher)
IO.puts("Imported #{map_size(results.successful_uuids)} objects")

# Stop and get final results (with flush)
results = Background.stop(batcher, flush: true)

Batch Safety Features (v0.7.4+)

WeaviateEx implements production-grade batch safety for reliable large-scale operations:

Memory Management

# MAX_STORED_RESULTS limit (100,000) prevents memory exhaustion
# Automatic eviction of oldest entries when limit exceeded
alias WeaviateEx.Batch.ErrorTracking.Results

# Check the limit
Results.max_stored_results()
#=> 100_000

# Results automatically evict oldest entries when limit is exceeded
# This prevents unbounded memory growth during large batch operations

Auto-Retry for Failed Objects

alias WeaviateEx.Batch.Dynamic

# Dynamic batcher with auto-retry enabled (default)
{:ok, batcher} = Dynamic.start(
  client: client,
  auto_retry: true,           # Enable automatic retry (default: true)
  max_retries: 5,             # Maximum retry attempts (default: 3)
  retry_delay_ms: 2000,       # Base delay for backoff (default: 1000ms)
  on_permanent_failure: fn objects ->
    Logger.error("Permanent failures: #{length(objects)}")
    # Handle objects that exceeded max_retries
  end
)

# Add objects - failed objects are automatically re-queued
Dynamic.add_object(batcher, "Article", %{title: "Test"})

# Retryable errors include:
# - Rate limit errors (429, "rate limit exceeded", etc.)
# - Transient gRPC errors (UNAVAILABLE, RESOURCE_EXHAUSTED, ABORTED, DEADLINE_EXCEEDED)

RetryQueue for Manual Control

alias WeaviateEx.Batch.RetryQueue

# Start a retry queue for manual control
{:ok, retry_queue} = RetryQueue.start_link(
  client: client,
  max_retries: 3,
  base_delay_ms: 1000,
  on_permanent_failure: fn objects ->
    Logger.error("Failed after max retries: #{length(objects)}")
  end
)

# Enqueue failed objects for retry
:ok = RetryQueue.enqueue_failed(retry_queue, failed_objects)

# Check retry count for a specific object
count = RetryQueue.get_retry_count(retry_queue, "uuid-123")

# Drain all queued objects for manual processing
{:ok, objects} = RetryQueue.drain(retry_queue)

# Clear the queue
:ok = RetryQueue.clear(retry_queue)

Configurable Batch Options

alias WeaviateEx.Batch.Config

# Create a batch configuration
config = Config.new(
  max_stored_results: 50_000,    # Custom limit
  auto_retry: true,
  max_retries: 5,
  retry_delay_ms: 2000,
  on_permanent_failure: fn objects ->
    Logger.error("Failed: #{length(objects)}")
  end
)

# Access configuration values
Config.auto_retry_enabled?(config)  #=> true
Config.default_max_retries()        #=> 3

Queries & Vector Search

Powerful query capabilities with semantic search:

alias WeaviateEx.Query

# Simple query with field selection
query = Query.get("Article")
  |> Query.fields(["title", "content", "publishedAt"])
  |> Query.limit(10)

{:ok, results} = Query.execute(query)

# Semantic search with near_text (requires vectorizer)
query = Query.get("Article")
  |> Query.near_text("artificial intelligence", certainty: 0.7)
  |> Query.fields(["title", "content"])
  |> Query.additional(["certainty", "distance"])
  |> Query.limit(5)

{:ok, results} = Query.execute(query)

# Vector search with custom vectors
query = Query.get("Article")
  |> Query.near_vector([0.1, 0.2, 0.3], certainty: 0.8)
  |> Query.fields(["title"])

{:ok, results} = Query.execute(query)

# Hybrid search (combines keyword + vector)
query = Query.get("Article")
  |> Query.hybrid("machine learning", alpha: 0.5)  # alpha: 0=keyword, 1=vector
  |> Query.fields(["title", "content"])

{:ok, results} = Query.execute(query)

# BM25 keyword search
query = Query.get("Article")
  |> Query.bm25("elixir programming")
  |> Query.fields(["title", "content"])

{:ok, results} = Query.execute(query)

# Semantic direction with Move (v0.5.0+)
query = Query.get("Article")
  |> Query.near_text("technology",
       move_to: [concepts: ["artificial intelligence", "machine learning"], force: 0.8],
       move_away: [concepts: ["politics", "sports"], force: 0.5]
     )
  |> Query.fields(["title", "content"])

{:ok, results} = Query.execute(query)

# Queries with filters (WHERE clause)
query = Query.get("Article")
  |> Query.where(%{
    path: ["publishedAt"],
    operator: "GreaterThan",
    valueDate: "2025-01-01T00:00:00Z"
  })
  |> Query.fields(["title", "publishedAt"])
  |> Query.sort([%{path: ["publishedAt"], order: "desc"}])

{:ok, results} = Query.execute(query)

Fetch Objects by IDs

alias WeaviateEx.API.Data

ids = [
  "550e8400-e29b-41d4-a716-446655440001",
  "550e8400-e29b-41d4-a716-446655440002"
]

{:ok, objects} = Data.fetch_objects_by_ids(client, "Article", ids,
  return_properties: ["title", "content"]
)

# Results preserve the input ID order.

# Using the Objects module (no client needed)
{:ok, objects} = WeaviateEx.Objects.fetch_objects_by_ids("Article", ids,
  return_properties: ["title", "content"]
)

gRPC vs GraphQL

When you pass a WeaviateEx.Client, Query.execute/2 uses gRPC and now supports filters, group_by, target vectors, near_image/near_media, references, vector metadata, reranking, and generative search (RAG). If a query includes options not yet supported in gRPC (for example sorting or cursor pagination), it automatically falls back to GraphQL.

Reranking

Improve search result relevance using reranker models:

alias WeaviateEx.Query
alias WeaviateEx.Query.Rerank

# Basic reranking - re-scores results using the "content" property
rerank = Rerank.new("content")

{:ok, results} = Query.get("Article")
|> Query.near_text("machine learning")
|> Query.fields(["title", "content"])
|> Query.limit(10)
|> Query.rerank(rerank)
|> Query.execute(client)

# With custom rerank query (different from search query)
rerank = Rerank.new("content", query: "latest AI applications in healthcare")

{:ok, results} = Query.get("Article")
|> Query.hybrid("AI trends", alpha: 0.5)
|> Query.fields(["title", "content"])
|> Query.rerank(rerank)
|> Query.execute(client)

# Access rerank scores in results
for result <- results do
  score = result["_additional"]["rerankScore"]
  IO.puts("Rerank score: #{score}")
end

Note: Requires a reranker module configured on the collection. See WeaviateEx.API.RerankerConfig for available rerankers: cohere, transformers, voyageai, jinaai, nvidia, contextualai.

gRPC Generative Search (v0.7.4+)

Generative queries now use gRPC for improved performance (~2-3x lower latency):

alias WeaviateEx.GRPC.Services.Search
alias WeaviateEx.Query.GenerativeResult

# Build a search request with generative config
request = Search.build_near_text_request("Article", "machine learning",
  limit: 5,
  return_properties: ["title", "content"],
  generative: %{
    single_prompt: "Summarize this article: {content}",
    provider: :openai,
    model: "gpt-4",
    temperature: 0.7
  }
)

# Execute the search
{:ok, reply} = Search.execute(channel, request)

# Parse the generative results
result = GenerativeResult.from_grpc_response(reply)

# Access per-object generations
for gen <- result.generated_per_object do
  IO.puts("Generated: #{gen}")
end

# Grouped generation
request = Search.build_near_text_request("Article", "AI trends",
  generative: %{
    grouped_task: "Synthesize the key themes from these articles",
    grouped_properties: ["title", "content"],
    provider: :anthropic,
    model: "claude-3-5-sonnet-20241022"
  }
)

{:ok, reply} = Search.execute(channel, request)
result = GenerativeResult.from_grpc_response(reply)
IO.puts("Grouped summary: #{result.generated}")

Supported providers: :openai, :anthropic, :cohere, :mistral, :ollama, :google, :aws, :databricks, :friendliai, :nvidia, :xai, :contextualai, :anyscale.

Multi-Vector Collections (v0.7.0+)

Query collections with multiple named vectors:

alias WeaviateEx.Query
alias WeaviateEx.Query.TargetVectors

# Single target vector
query = Query.get("MultiVectorCollection")
  |> Query.near_text("search term", target_vectors: "content_vector")
  |> Query.fields(["title", "content"])

{:ok, results} = Query.execute(query, client)

# Combined vectors with average method
target = TargetVectors.combine(["title_vector", "content_vector"], method: :average)

query = Query.get("MultiVectorCollection")
  |> Query.near_vector(embedding, target_vectors: target)
  |> Query.fields(["title"])

{:ok, results} = Query.execute(query, client)

# Weighted combination
target = TargetVectors.weighted(%{
  "title_vector" => 0.7,
  "content_vector" => 0.3
})

query = Query.get("MultiVectorCollection")
  |> Query.near_text("search", target_vectors: target)
  |> Query.fields(["title", "content"])

{:ok, results} = Query.execute(query, client)