Skip to content

Extending Evolvable Surfaces

This guide explains how to add new evolvable surfaces to gepa-adk by implementing the ComponentHandler protocol.

When to Use

Create a custom ComponentHandler when you need to evolve an agent attribute beyond the built-in surfaces (instruction, output_schema, generate_content_config). For example, evolving temperature, tool configurations, or custom metadata.

Available Built-in Handlers

Handler Component Purpose
InstructionHandler instruction Agent system prompt
OutputSchemaHandler output_schema Pydantic output schema
GenerateContentConfigHandler generate_content_config LLM generation parameters
MappingComponentHandler each registered mapping key Prompt strings held in a caller-owned mapping

Protocol Definition

The ComponentHandler protocol defines three methods for the serialize/apply/restore cycle:

# Protocol definition (from gepa_adk.ports.component_handler — shown for reference)
class ComponentHandler(Protocol):
    def serialize(self, agent: LlmAgent) -> str:
        """Extract current component value as a string."""
        ...

    def apply(self, agent: LlmAgent, value: str) -> Any:
        """Apply a new value to the agent, return the original for restoration."""
        ...

    def restore(self, agent: LlmAgent, original: Any) -> None:
        """Reinstate the original value after evaluation."""
        ...

Contract

  • serialize() must never raise exceptions — return an empty string or a sensible default for missing values.
  • apply() must never raise exceptions — log a warning and keep the original on failure.
  • restore() must always succeed — None values reset to the component default.
  • All methods are synchronous — no I/O operations.

Step-by-Step Implementation

Here is a TemperatureHandler that evolves the generate_content_config.temperature parameter:

from typing import Any

from gepa_adk import ComponentHandler  # verify structural subtyping


class TemperatureHandler:
    """Handler for evolving the temperature parameter."""

    def serialize(self, agent) -> str:
        config = getattr(agent, "generate_content_config", None)
        if config and config.temperature is not None:
            return str(config.temperature)
        return "1.0"

    def apply(self, agent, value: str) -> Any:
        config = getattr(agent, "generate_content_config", None)
        original = config.temperature if config else 1.0
        try:
            new_temp = float(value)
            if config:
                config.temperature = new_temp
        except (ValueError, TypeError):
            pass  # keep original on invalid input
        return original

    def restore(self, agent, original) -> None:
        config = getattr(agent, "generate_content_config", None)
        if config:
            config.temperature = original

Note that TemperatureHandler is a plain class — it does not inherit from ComponentHandler. gepa-adk uses structural subtyping (ADR-002): any class with matching method signatures satisfies the protocol automatically.

Registration

Register your handler with the ComponentHandlerRegistry so the evolution engine discovers it:

from gepa_adk.adapters import register_handler

register_handler("temperature", TemperatureHandler())

The engine uses get_handler() to look up handlers by component name:

from gepa_adk.adapters import get_handler

handler = get_handler("temperature")
original = handler.apply(agent, "0.5")
# ... evaluate agent ...
handler.restore(agent, original)

You can also create a dedicated registry if you need isolation from the global default:

from gepa_adk.adapters import ComponentHandlerRegistry

my_registry = ComponentHandlerRegistry()
my_registry.register("temperature", TemperatureHandler())

Runnable Example

This example demonstrates the full serialize/apply/restore cycle without requiring an LLM API key:

from google.adk.agents import LlmAgent
from google.adk.models.lite_llm import LiteLlm
from google.genai.types import GenerateContentConfig

from typing import Any

from gepa_adk import ComponentHandler
from gepa_adk.adapters import register_handler


class TemperatureHandler:
    def serialize(self, agent) -> str:
        config = getattr(agent, "generate_content_config", None)
        if config and config.temperature is not None:
            return str(config.temperature)
        return "1.0"

    def apply(self, agent, value: str) -> Any:
        config = getattr(agent, "generate_content_config", None)
        original = config.temperature if config else 1.0
        try:
            new_temp = float(value)
            if config:
                config.temperature = new_temp
        except (ValueError, TypeError):
            pass
        return original

    def restore(self, agent, original) -> None:
        config = getattr(agent, "generate_content_config", None)
        if config:
            config.temperature = original


# Verify protocol compliance
handler = TemperatureHandler()
assert isinstance(handler, ComponentHandler)

# Create a test agent
agent = LlmAgent(
    name="demo",
    model=LiteLlm(model="ollama_chat/llama3.2:latest"),
    instruction="Be helpful",
    generate_content_config=GenerateContentConfig(temperature=0.7),
)

# Serialize → Apply → Restore cycle
print(f"Original: {handler.serialize(agent)}")  # "0.7"

original = handler.apply(agent, "0.3")
print(f"After apply: {handler.serialize(agent)}")  # "0.3"

handler.restore(agent, original)
print(f"After restore: {handler.serialize(agent)}")  # "0.7"

# Register for use in evolution
register_handler("temperature", handler)

To integrate with evolve(), include the component name in your candidate:

from gepa_adk import EvolutionConfig, evolve, run_sync

# Register handler before evolution
register_handler("temperature", TemperatureHandler())

config = EvolutionConfig(max_iterations=5)
result = run_sync(evolve(agent, trainset, config=config, components=["temperature"]))

Evolving a caller-owned mapping

Some prompts do not live on the agent. A tool may keep a dict of prompt strings that it sends to another service. MappingComponentHandler evolves one key of such a mapping, and register_mapping_components() registers one handler per key in the default registry, so each key becomes an ordinary component name. The handler ignores its agent argument.

from google.adk.agents import LlmAgent
from google.adk.models.lite_llm import LiteLlm
from google.adk.tools import FunctionTool

from gepa_adk import (
    EvolutionConfig,
    LabelAgreementScorer,
    evolve,
    register_mapping_components,
    run_sync,
)

# The caller owns this dict; the tool reads it on every call.
prompts = {
    "summarize": "Summarize the ticket in one sentence.",
    "classify": "Label the ticket as bug, feature or question.",
}


def triage(ticket: str) -> str:
    """Send the ticket to the triage service with the current prompts."""
    return call_triage_service(ticket, prompts["summarize"], prompts["classify"])


agent = LlmAgent(
    name="triager",
    model=LiteLlm(model="ollama_chat/llama3.2:latest"),
    instruction="Pass the user's ticket to the triage tool and return its answer.",
    tools=[FunctionTool(triage)],
)

names = register_mapping_components(prompts)  # ["summarize", "classify"]

result = run_sync(
    evolve(
        agent,
        trainset,
        scorer=LabelAgreementScorer(),
        components=names,
        component_selector="round_robin",
        config=EvolutionConfig(reflection_model="ollama_chat/llama3.2:latest"),
    )
)

# Evaluation writes candidate text into `prompts` and restores it afterwards,
# so the mapping holds the original text here. Apply the result yourself.
prompts.update({name: result.evolved_components[name] for name in names})

register_mapping_components() rejects an empty mapping, a key that is not a Python identifier, a non-string value, and the keys instruction and output_schema, which are built-in component names. evolve() consults only the default registry, so leave registry= unset for a real run. Pass registry= to build an isolated ComponentHandlerRegistry for a test or a manual serialize/apply/restore cycle.

Common Pitfalls

Avoid These Mistakes

Raising exceptions instead of returning defaults. The serialize() and apply() methods must never raise. On failure, serialize() should return an empty string or a sensible default, and apply() should log a warning and keep the original value.

Forgetting to restore. Always implement restore(). The evolution engine calls it after every evaluation to reset the agent to its original state. A missing or broken restore() causes state corruption across iterations.

Not handling None values. Agent attributes may be None (e.g., no generate_content_config set). Guard with getattr() and None checks.

Inheriting from the Protocol. Do NOT write class MyHandler(ComponentHandler):. gepa-adk uses structural subtyping (ADR-002) — just implement the methods with matching signatures. Inheriting from a Protocol is misleading and unnecessary.

Contract Test Skeleton

When adding a new handler, write contract tests to verify protocol compliance. This skeleton follows the three-class template established in the project.

Exemplar Reference

This skeleton follows the three-class template used in tests/contracts/test_stopper_protocol.py and tests/contracts/test_candidate_selector_protocol.py — always check the latest exemplar before starting.

import pytest

from gepa_adk import ComponentHandler

pytestmark = pytest.mark.contract


class TestMyHandlerRuntimeCheckable:
    """Positive compliance: isinstance checks."""

    def test_satisfies_component_handler_protocol(self):
        handler = MyHandler()
        assert isinstance(handler, ComponentHandler)

    def test_protocol_has_required_methods(self):
        handler = MyHandler()
        assert hasattr(handler, "serialize")
        assert hasattr(handler, "apply")
        assert hasattr(handler, "restore")


class TestMyHandlerBehavior:
    """Behavioral expectations: return types, state transitions."""

    def test_serialize_returns_string(self):
        handler = MyHandler()
        result = handler.serialize(agent)
        assert isinstance(result, str)

    def test_apply_returns_original(self):
        handler = MyHandler()
        original = handler.apply(agent, "new_value")
        assert original is not None  # must return previous value

    def test_apply_restore_idempotent(self):
        handler = MyHandler()
        original_value = handler.serialize(agent)
        returned = handler.apply(agent, "new_value")
        handler.restore(agent, returned)
        assert handler.serialize(agent) == original_value

    def test_serialize_returns_empty_for_missing(self):
        handler = MyHandler()
        result = handler.serialize(agent_without_component)
        assert result == ""


class TestMyHandlerNonCompliance:
    """Negative cases: missing methods fail isinstance."""

    def test_missing_apply_fails(self):
        class Incomplete:
            def serialize(self, agent):
                return ""

            def restore(self, agent, original):
                pass

        assert not isinstance(Incomplete(), ComponentHandler)

    def test_missing_restore_fails(self):
        class Incomplete:
            def serialize(self, agent):
                return ""

            def apply(self, agent, value):
                return None

        assert not isinstance(Incomplete(), ComponentHandler)

API Reference