[NA] [SDK] feat: add greenfield optimization framework package

Implements a new optimization framework (`apps/opik-optimizer`) that
decouples optimizer algorithms from experiment execution, persistence,
and UI concerns. Integrates via the existing optimization studio pipeline
(Redis queue → Python backend → subprocess).

Key components:
- Orchestrator: central lifecycle controller with sampler, validator,
  materializer, result aggregator, and event emitter
- StupidOptimizer: 2-step test optimizer (3 candidates → best → 2 more)
- EvaluationAdapter: wraps SDK evaluate_optimization_suite_trial()
- Backend integration: new Redis queue, framework_optimizer job processor,
  framework_runner subprocess entry point

Also adds evaluate_optimization_suite_trial() to the Python SDK, combining
optimization trial linkage with evaluation suite behavior (evaluators and
execution policy from the dataset).

53 unit + integration tests passing. Verified end-to-end against Comet cloud
with real LLM calls, UI progress chart, prompt display, and score tracking.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: itamargolan

.github/workflows/build_and_push_docker.yaml

@@ -147,6 +147,7 @@ jobs:
         uses: docker/build-push-action@v6
         with:
           context: apps/${{ inputs.image }}/
+          build-contexts: ${{ inputs.image == 'opik-python-backend' && 'opik-optimizer=apps/opik-optimizer/' || '' }}
           platforms: linux/${{ matrix.platform }}
           cache-from: type=registry,ref=${{ env.DOCKER_REGISTRY }}/${{ steps.set_vars.outputs.image_name }}:main
           provenance: false

apps/opik-backend/src/main/java/com/comet/opik/domain/OptimizationService.java

@@ -366,8 +366,9 @@ private void enqueueStudioOptimizationJob(Optimization optimization, String work
                 .opikApiKey(opikApiKey)
                 .build();
 
-        // Enqueue to Redis RQ
-        queueProducer.enqueue(Queue.OPTIMIZER_CLOUD, jobMessage)
+        // Route to the appropriate queue based on optimizer type
+        var queue = resolveQueue(optimization);
+        queueProducer.enqueue(queue, jobMessage)
                 .doOnSuccess(
                         jobId -> log.info("Studio optimization job enqueued successfully for id: '{}', jobId: '{}'",
                                 optimization.id(), jobId))
@@ -379,6 +380,20 @@ private void enqueueStudioOptimizationJob(Optimization optimization, String work
                 .subscribe();
     }
 
+    private static final java.util.Set<String> LEGACY_OPTIMIZER_TYPES = java.util.Set.of(
+            "gepa", "evolutionary", "hierarchical_reflective");
+
+    private Queue resolveQueue(Optimization optimization) {
+        if (optimization.studioConfig() != null
+                && optimization.studioConfig().optimizer() != null) {
+            var optimizerType = optimization.studioConfig().optimizer().type();
+            if (optimizerType != null && !LEGACY_OPTIMIZER_TYPES.contains(optimizerType.toLowerCase())) {
+                return Queue.OPTIMIZER_FRAMEWORK;
+            }
+        }
+        return Queue.OPTIMIZER_CLOUD;
+    }
+
     private void cancelOptimization(UUID optimizationId, String workspaceId) {
         var optimizationUpdate = OptimizationUpdate.builder()
                 .status(OptimizationStatus.CANCELLED)

apps/opik-backend/src/main/java/com/comet/opik/infrastructure/queues/Queue.java

@@ -6,6 +6,7 @@
 public enum Queue {
 
     OPTIMIZER_CLOUD("opik:optimizer-cloud", "opik_backend.rq_worker.process_optimizer_job"),
+    OPTIMIZER_FRAMEWORK("opik:optimizer-framework", "opik_backend.rq_worker.process_framework_optimizer_job"),
     ;
 
     @JsonValue

apps/opik-optimizer/pyproject.toml

@@ -0,0 +1,11 @@
+[project]
+name = "opik-optimizer-framework"
+version = "0.1.0"
+requires-python = ">=3.11"
+dependencies = ["opik>=1.7.17", "litellm"]
+
+[project.optional-dependencies]
+dev = ["pytest>=8.0", "pytest-cov>=5.0"]
+
+[tool.setuptools.packages.find]
+where = ["src"]

apps/opik-optimizer/scripts/run_optimization_e2e.py

@@ -0,0 +1,221 @@
+#!/usr/bin/env python
+"""
+End-to-end test script for the optimization framework.
+
+Creates an evaluation suite with LLMJudge assertions (including item-level
+overrides), then runs a real optimization against a local Opik backend so
+you can see the results in the Optimization Studio UI.
+
+Prerequisites:
+  - Local Opik backend running (http://localhost:8080)
+  - OPENAI_API_KEY set in environment (or another LLM provider supported by litellm)
+  - pip install -e apps/opik-optimizer  (the framework package)
+  - pip install -e sdks/python          (the Opik SDK)
+
+Usage:
+  export OPENAI_API_KEY=sk-...
+  python apps/opik-optimizer/scripts/run_optimization_e2e.py
+"""
+
+import logging
+import os
+import sys
+import time
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+)
+logger = logging.getLogger("e2e-test")
+
+# -- Configuration ----------------------------------------------------------
+
+OPIK_URL = os.environ.get("OPIK_URL_OVERRIDE")  # None = use SDK default (cloud)
+OPIK_WORKSPACE = os.environ.get("OPIK_WORKSPACE", "default")
+OPIK_API_KEY = os.environ.get("OPIK_API_KEY")
+
+SUITE_NAME = f"customer-support-regression-tests-{int(time.time())}"
+OPTIMIZATION_NAME = "e2e-framework-test"
+OBJECTIVE_NAME = "llm_judge"
+
+# The model litellm will call for the optimization task.
+MODEL = os.environ.get("OPIK_TEST_MODEL", "gpt-4o-mini")
+
+PROMPT_MESSAGES = [
+    {
+        "role": "system",
+        "content": (
+            "You are a helpful customer support agent for an e-commerce company. "
+            "Be professional, empathetic, and provide clear, actionable responses. "
+            "If you don't know something, be honest about it."
+        ),
+    },
+    {
+        "role": "user",
+        "content": "Customer question: {question}\nAdditional context: {context}",
+    },
+]
+
+# ---------------------------------------------------------------------------
+
+
+def _update_optimization_status(client, optimization_id, status):
+    """Update optimization status via the SDK's REST client."""
+    client.rest_client.optimizations.update_optimizations_by_id(
+        optimization_id, status=status,
+    )
+
+
+def main():
+    import opik
+    from opik.evaluation.suite_evaluators import LLMJudge
+    from opik_optimizer_framework import OptimizationContext, run_optimization
+
+    # 1. Connect to Opik
+    if OPIK_URL:
+        os.environ["OPIK_URL_OVERRIDE"] = OPIK_URL
+    logger.info("Connecting to Opik (workspace: %s, url: %s)", OPIK_WORKSPACE, OPIK_URL or "cloud default")
+
+    client = opik.Opik(workspace=OPIK_WORKSPACE, api_key=OPIK_API_KEY)
+
+    # 2. Create evaluation suite — exact dataset from the example script
+    logger.info("Creating evaluation suite '%s'", SUITE_NAME)
+    suite = client.create_evaluation_suite(
+        name=SUITE_NAME,
+        description="Regression tests for customer support agent responses",
+        evaluators=[
+            LLMJudge(
+                assertions=[
+                    "Response is relevant to the user question",
+                ]
+            )
+        ],
+        execution_policy={"runs_per_item": 1, "pass_threshold": 1},
+    )
+
+    # Test case 1: Refund request
+    # No item-level evaluators — suite-level evaluators are used
+    suite.add_item(
+        data={
+            "question": "I received a damaged product. How can I get a refund?",
+            "context": "Order #12345, placed 3 days ago",
+        },
+    )
+
+    # Test case 2: Shipping inquiry
+    # No item-level evaluators — suite-level evaluators are used
+    suite.add_item(
+        data={
+            "question": "Where is my package? It was supposed to arrive yesterday.",
+            "context": "Tracking number: TRK789456",
+        },
+    )
+
+    # Test case 3: Account security (CRITICAL)
+    # Item-level evaluators OVERRIDE suite-level ones for this item
+    # Also uses a stricter execution policy (5 runs, 4 must pass)
+    suite.add_item(
+        data={
+            "question": "I think someone hacked my account. I see orders I didn't make!",
+            "context": "Customer reports unauthorized activity",
+        },
+        evaluators=[
+            LLMJudge(
+                assertions=[
+                    "The response treats the security concern with appropriate urgency",
+                    "The response advises immediate steps to secure the account",
+                    "The response mentions that unauthorized orders will be investigated",
+                ]
+            )
+        ],
+        execution_policy={"runs_per_item": 5, "pass_threshold": 4},
+    )
+
+    # Test case 4: Product question
+    # No item-level evaluators — suite-level evaluators are used
+    suite.add_item(
+        data={
+            "question": "Is the XYZ Wireless Headphones compatible with iPhone 15?",
+            "context": "Product SKU: WH-2024-BLK",
+        },
+    )
+
+    # Test case 5: Subscription cancellation
+    # No item-level evaluators — suite-level evaluators are used
+    suite.add_item(
+        data={
+            "question": "I want to cancel my premium subscription. This is too expensive.",
+            "context": "Customer has been subscribed for 6 months",
+        },
+    )
+
+    # Get dataset item IDs from the underlying dataset
+    dataset_items = suite.dataset.get_items()
+    dataset_item_ids = [str(item["id"]) for item in dataset_items]
+    logger.info("Suite has %d items", len(dataset_item_ids))
+
+    # 3. Create the optimization record (status is set to "running" automatically)
+    logger.info("Creating optimization record")
+    optimization = client.create_optimization(
+        dataset_name=SUITE_NAME,
+        objective_name=OBJECTIVE_NAME,
+        name=OPTIMIZATION_NAME,
+    )
+    optimization_id = optimization.id
+    logger.info("Optimization created: %s", optimization_id)
+
+    # 4. Run the framework — evaluators come from the suite itself
+    context = OptimizationContext(
+        optimization_id=optimization_id,
+        dataset_name=SUITE_NAME,
+        prompt_messages=PROMPT_MESSAGES,
+        model=MODEL,
+        model_parameters={"temperature": 0.7, "max_tokens": 256},
+        metric_type=OBJECTIVE_NAME,
+        metric_parameters={},
+        optimizer_type="stupid",
+        optimizer_parameters={},
+    )
+
+    logger.info("Starting optimization (optimizer_type=stupid, model=%s)", MODEL)
+    try:
+        result = run_optimization(
+            context=context,
+            client=client,
+            dataset_item_ids=dataset_item_ids,
+        )
+        _update_optimization_status(client, optimization_id, "completed")
+    except Exception:
+        logger.exception("Optimization failed")
+        _update_optimization_status(client, optimization_id, "error")
+        client.end()
+        sys.exit(1)
+
+    # 5. Print results
+    print("\n" + "=" * 60)
+    print("OPTIMIZATION COMPLETE")
+    print("=" * 60)
+    print(f"  Optimization ID : {optimization_id}")
+    print(f"  Final score     : {result.score:.4f}")
+    print(f"  Initial score   : {result.initial_score}")
+    print(f"  Total trials    : {len(result.all_trials)}")
+
+    if result.best_trial:
+        print(f"\n  Best trial:")
+        print(f"    Score         : {result.best_trial.score:.4f}")
+        print(f"    Experiment    : {result.best_trial.experiment_name}")
+        print(f"    Prompt        :")
+        for msg in result.best_trial.prompt_messages:
+            print(f"      [{msg['role']}] {msg['content'][:80]}...")
+
+    print(f"\n  View in UI: {OPIK_URL or 'https://www.comet.com/opik'}")
+    print("=" * 60)
+
+    client.end()
+
+
+if __name__ == "__main__":
+    if not os.environ.get("OPENAI_API_KEY") and "gpt" in MODEL.lower():
+        print("ERROR: OPENAI_API_KEY not set. Export it or set OPIK_TEST_MODEL to another provider.")
+        sys.exit(1)
+    main()

apps/opik-optimizer/src/opik_optimizer_framework/__init__.py

@@ -0,0 +1,15 @@
+from opik_optimizer_framework.orchestrator import run_optimization
+from opik_optimizer_framework.types import (
+    OptimizationContext,
+    OptimizationResult,
+    OptimizationState,
+    TrialResult,
+)
+
+__all__ = [
+    "run_optimization",
+    "OptimizationContext",
+    "OptimizationResult",
+    "OptimizationState",
+    "TrialResult",
+]

apps/opik-optimizer/src/opik_optimizer_framework/candidate_materializer.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+import uuid
+from dataclasses import asdict
+
+from opik_optimizer_framework.types import Candidate, CandidateConfig
+from opik_optimizer_framework.util.hashing import canonical_config_hash
+
+
+def materialize_candidate(
+    config: CandidateConfig,
+    step_index: int,
+    parent_candidate_ids: list[str] | None = None,
+) -> Candidate:
+    """Create a Candidate with a generated UUID and computed config hash.
+
+    This is a stub materializer: generates candidate_id, computes config_hash,
+    but does not create a mask_id.
+    """
+    config_dict = asdict(config)
+    config_hash = canonical_config_hash(config_dict)
+
+    return Candidate(
+        candidate_id=str(uuid.uuid4()),
+        config=config,
+        config_hash=config_hash,
+        step_index=step_index,
+        parent_candidate_ids=parent_candidate_ids or [],
+    )

apps/opik-optimizer/src/opik_optimizer_framework/candidate_validator.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import logging
+from dataclasses import asdict
+
+from opik_optimizer_framework.types import CandidateConfig, OptimizationState
+from opik_optimizer_framework.util.hashing import canonical_config_hash
+
+logger = logging.getLogger(__name__)
+
+
+def validate_candidate(
+    config: CandidateConfig,
+    state: OptimizationState,
+) -> tuple[bool, str | None]:
+    """Validate a candidate configuration.
+
+    Returns (True, None) if valid, or (False, reason) if rejected.
+    Checks:
+    1. Shape validation: non-empty messages, each has role+content
+    2. Dedup: config hash not already in state.seen_hashes
+    """
+    if not config.prompt_messages:
+        return False, "empty_messages"
+
+    for i, msg in enumerate(config.prompt_messages):
+        if "role" not in msg or "content" not in msg:
+            return False, f"message_{i}_missing_role_or_content"
+        if not msg["role"] or not msg["content"]:
+            return False, f"message_{i}_empty_role_or_content"
+
+    config_dict = asdict(config)
+    config_h = canonical_config_hash(config_dict)
+    if config_h in state.seen_hashes:
+        return False, "duplicate_config_hash"
+
+    return True, None