Single-Agent Evolution¶

This guide covers basic agent evolution patterns for optimizing a single LlmAgent.

When to Use This Pattern¶

Use single-agent evolution when:

• You have one agent that needs instruction optimization
• You can define clear scoring criteria via a critic agent
• You want straightforward instruction improvement

Prerequisites¶

• Python 3.12+
• gepa-adk installed (uv add gepa-adk)
• Ollama running locally with a model (e.g., llama3.2:latest)
• OLLAMA_API_BASE environment variable set

export OLLAMA_API_BASE=http://localhost:11434

Basic Evolution with Critic¶

The standard pattern uses a critic agent to score the evolved agent's outputs.

Step 1: Create the Agent to Evolve¶

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

agent = LlmAgent(
    name="greeter",
    model=LiteLlm(model="ollama_chat/llama3.2:latest"),
    instruction="Greet the user appropriately based on their introduction.",
)

Step 2: Create a Critic Agent¶

The critic evaluates outputs and provides scores. Use SimpleCriticOutput for basic scoring:

from gepa_adk import SimpleCriticOutput

critic = LlmAgent(
    name="critic",
    model=LiteLlm(model="ollama_chat/llama3.2:latest"),
    instruction="Score for formal, Dickens-style greetings. 0.0-1.0.",
    output_schema=SimpleCriticOutput,
)

Or define a custom critic schema for richer feedback:

from pydantic import BaseModel, Field

class CriticOutput(BaseModel):
    score: float = Field(ge=0.0, le=1.0)
    feedback: str

critic = LlmAgent(
    name="critic",
    model=LiteLlm(model="ollama_chat/llama3.2:latest"),
    instruction="""Evaluate greeting quality. Look for formal, elaborate,
Dickens-style greetings appropriate for the social context.
Score 0.0-1.0 where 1.0 is a perfect formal greeting.""",
    output_schema=CriticOutput,
)

Step 3: Prepare Training Data¶

trainset = [
    {"input": "I am His Majesty, the King."},
    {"input": "I am your mother."},
    {"input": "I am a close friend."},
]

Step 4: Run Evolution¶

from gepa_adk import evolve, run_sync, EvolutionConfig

config = EvolutionConfig(
    max_iterations=5,
    patience=2,
    reflection_model="ollama_chat/llama3.2:latest",
)

result = run_sync(evolve(agent, trainset, critic=critic, config=config))

print(f"Original score: {result.original_score:.3f}")
print(f"Final score: {result.final_score:.3f}")
print(f"Improvement: {result.improvement:.2%}")
print(f"Evolved instruction:\n{result.evolved_components['instruction']}")

Read the Iteration History¶

result.iteration_history holds one IterationRecord per iteration. Each record's failed_evaluations counts the rows whose agent run or scorer raised, or whose run returned a failed execution, in that iteration, across every evaluation the iteration made. result.baseline_failed_evaluations counts them for the initial candidate, and result.total_failed_evaluations is the baseline plus the sum over the history.

A row that fails is scored 0.0 and counts toward the aggregate, the same as a wrong answer. The count tells the two apart: a score of 0 with failed_evaluations == 0 means the agent answered and the scorer rejected the answer, while a nonzero count means that many rows got their 0.0 because the run or the scorer raised, not because of what the agent said. A run whose low scores come from failures points at the model, tools, network or scorer, not the instruction.

for record in result.iteration_history:
    print(
        record.iteration_number,
        f"score={record.score:.3f}",
        f"failed={record.failed_evaluations}",
        record.skip_reason or "",
    )
print(f"Failed rows in total: {result.total_failed_evaluations}")

Results saved with to_dict() before these counts existed (schema version 1) still load with EvolutionResult.from_dict(); their counts read as 0.

Each record's token_usage is a TokenRollup of the tokens the iteration used, and result.token_usage sums the run, baseline included. The rollup has two splits: token_usage.evaluation covers the rows the agent evaluated and token_usage.reflection covers the reflection calls that proposed new text. The top-level counters cover both. In the evaluation split, rows_counted is the number of rows whose trace reported usage; in the reflection split, rows_counted is the number of reflection calls that reported usage. rows_unknown counts the rows or calls that reported none. Anything without usage is counted as unknown, never as zero, so a counter covers the counted rows only; it is None ("unknown" in to_dict()) when nothing reported usage. An iteration that evaluated nothing, such as a duplicate proposal, reports zero evaluation usage but still carries the cost of its reflection call.

def describe(usage):
    if usage is None or usage.total_tokens is None:
        return "tokens=unknown"
    return f"tokens={usage.total_tokens} unknown_rows={usage.rows_unknown}"


for record in result.iteration_history:
    usage = record.token_usage
    print(
        record.iteration_number,
        describe(usage),
        "evaluation:", describe(usage and usage.evaluation),
        "reflection:", describe(usage and usage.reflection),
    )
print("Run", describe(result.token_usage))

token_usage is None on a result saved before schema version 3, and a counter is None when nothing reported usage, so read it as above.

Only evaluations that capture traces are observed. A separate valset is evaluated without traces, so its rows count as unknown. A reflection call whose executor captured no events is counted as unknown in the reflection split. reflection is None when no reflection call was observed: on the baseline, or when the adapter proposes without the reflective proposer. Results saved before schema version 3 load with token_usage set to None, and results saved before schema version 6 load with their usage as the evaluation split and reflection set to None.

Rows behind each score¶

A record's score is the acceptance aggregate (a sum or a mean, per EvolutionConfig.acceptance_metric) over the rows it was scored on, and rows_evaluated says how many rows that was. An evaluated record counts the rows of its scoring batch, a "minibatch_rejected" record counts the minibatch rows it was scored on, and a skip that evaluated nothing (an empty or incomplete proposal, a reflection timeout or error, a rejected schema or a proposal_validator rejection) records 0. A "duplicate" record carries None, because its score comes from an earlier evaluation, and so does every record of a result saved before schema version 7.

With the default acceptance_metric="sum", score / rows_evaluated is the per-row mean, which compares minibatch and full-set records on one scale. Guard the division, since the count can be 0 or None:

for record in result.iteration_history:
    if record.rows_evaluated:
        print(record.iteration_number, record.score / record.rows_evaluated)

Candidate genealogy¶

Each record also names the candidate it concerns. candidate_id is the Candidate.id of the proposal the iteration produced, and parent_ids is the list of Candidate.id values it was made from: one entry, the candidate the reflector rewrote, for a mutation. With a candidate selector that is the selector's draw; without one it is the current best candidate. Evaluated, duplicate, minibatch-rejected and schema-rejected iterations carry both fields. Iterations that produced no proposal (an empty proposal, a reflection timeout or a reflection error) carry None.

for record in result.iteration_history:
    print(record.iteration_number, record.parent_ids, "->", record.candidate_id)

The genealogy fields arrived in schema version 4. Results and checkpoints saved before version 4 load with candidate_id and parent_ids set to None.

Complete Working Example¶

"""Single-agent evolution with critic scoring."""

import asyncio
import os

from google.adk.agents import LlmAgent
from google.adk.models.lite_llm import LiteLlm
from pydantic import BaseModel, Field

from gepa_adk import evolve, EvolutionConfig


class CriticOutput(BaseModel):
    score: float = Field(ge=0.0, le=1.0)
    feedback: str


async def main() -> None:
    if not os.getenv("OLLAMA_API_BASE"):
        raise ValueError("Set OLLAMA_API_BASE environment variable")

    agent = LlmAgent(
        name="greeter",
        model=LiteLlm(model="ollama_chat/llama3.2:latest"),
        instruction="Greet the user appropriately.",
    )

    critic = LlmAgent(
        name="critic",
        model=LiteLlm(model="ollama_chat/llama3.2:latest"),
        instruction="""Evaluate greeting quality. Look for formal, Dickens-style
greetings appropriate for the social context. Score 0.0-1.0.""",
        output_schema=CriticOutput,
    )

    trainset = [
        {"input": "I am His Majesty, the King."},
        {"input": "I am your mother."},
        {"input": "I am a close friend."},
    ]

    config = EvolutionConfig(
        max_iterations=5,
        patience=2,
        reflection_model="ollama_chat/llama3.2:latest",
    )

    result = await evolve(agent, trainset, critic=critic, config=config)

    print(f"Original: {result.original_score:.3f}")
    print(f"Final: {result.final_score:.3f}")
    print(f"Improvement: {result.improvement:.2%}")
    print(f"\nEvolved instruction:\n{result.evolved_components['instruction']}")


if __name__ == "__main__":
    asyncio.run(main())

Configuration Options¶

EvolutionConfig Parameters¶

from gepa_adk import EvolutionConfig

config = EvolutionConfig(
    max_iterations=20,          # Maximum evolution iterations
    patience=5,                 # Stop after N iterations without improvement
    reflection_model="ollama_chat/llama3.2:latest",  # Model for generating improvements
    min_improvement_threshold=0.01,  # Minimum score gain to accept
    reflection_timeout_seconds=600,  # Seconds per reflection call (None = executor default)
    reflection_minibatch_size=None,  # Trainset rows a proposal must win on first
    proposal_validator=None,    # Callable(component, text) -> reason or None
    checkpoint_path=None,       # JSON file to checkpoint engine state to (None = off)
    resume=False,               # Continue from checkpoint_path instead of starting fresh
)

The merge and frontier fields (use_merge, max_merge_invocations, frontier_type) are covered in Merge and Pareto Selection. EvolutionConfig also has max_concurrent_evals (default 5, at least 1). evolve() passes this value to ADKAdapter to limit concurrently evaluated rows within each batch. Set it to 1 to evaluate rows serially. This setting does not apply to evolve_group() or evolve_workflow(), which use MultiAgentAdapter.

Reflection Timeout¶

reflection_timeout_seconds bounds how long the reflection agent may run for one proposal. The default, None, keeps the executor's own default of 300 seconds. Raise it when a slow local model needs more time to answer; it must be a whole number of seconds, at least 1.

A reflection that times out does not end the run. The iteration is recorded as skipped with skip_reason="reflection_timeout" and a score of 0.0, it is not retried, and it counts toward patience. Accepted candidates are kept, and the loop checks its stop conditions and continues.

Each timeout logs a reflection.timeout line with the session id, the component, timeout_seconds and timeout_source ("config" when you set the field, "executor_default" otherwise). A reflection that finishes with no text logs reflection.empty_response instead.

Reflection Errors¶

A reflection call that raises a transient provider error (HTTP 429, 500, 502, 503 or 504, RESOURCE_EXHAUSTED, UNAVAILABLE, a rate limit, a dropped connection, a connection error or a refused connection) is retried once after a two-second backoff. When the retry fails too, the iteration is recorded as skipped with skip_reason="reflection_error" and a score of 0.0, counts toward patience, and the loop continues.

A provider that fails the same way every time would otherwise use up all of patience without saying why. After EvolutionConfig.max_consecutive_reflection_errors (default 3) consecutive reflection_error skips whose causes have the same exception type and message, the engine raises a non-retryable ReflectionError with the partial result described below. A different error, or any other iteration in between, resets the count. Set a higher value to tolerate longer outages.

Any other exception from the reflection agent, such as a bad request or a programming error, is not retried and raises ReflectionError. When that happens after the baseline was scored, the error's partial_result holds an EvolutionResult with stop_reason=StopReason.ERROR, the recorded iterations and the best candidate found so far:

from gepa_adk.domain.exceptions import EvolutionError

try:
    result = await evolve(agent, trainset, critic=critic, config=config)
except EvolutionError as error:
    result = error.partial_result  # None if the baseline never finished

Incomplete Proposals¶

A reflection cut off at the model's output-token limit (a MAX_TOKENS finish reason on its last response, or litellm's length, recorded as LENGTH), or one whose text opens a reasoning tag such as <think> and never closes it, is not evaluated. The iteration is recorded as skipped with skip_reason="incomplete_proposal" and a score of 0.0, the truncated text is kept in the record's component_text, it is not retried, and it counts toward patience. Each one logs a reflection.incomplete line with the component, finish_reason (the finish reason name or unterminated_<tag>) and response_length. The reasoning_tags parameter of create_adk_reflection_fn sets which tags are checked; the default is ("think", "thinking", "reasoning").

Reflection Minibatch¶

By default every proposal is evaluated on the full trainset. Set reflection_minibatch_size to gate each proposal on a sample first, as the GEPA reference does:

config = EvolutionConfig(
    max_iterations=30,
    reflection_minibatch_size=4,  # Sample 4 trainset rows per iteration
    seed=42,                      # Makes the sampled rows reproducible
)

Each iteration draws a fresh seeded sample of k rows from the trainset rows the parent has scores for, where k is the smaller of the setting and the trainset size. A parent with at most k scored rows is compared on all of them. The proposal runs on those rows, and its mean score is compared with its parent's cached scores on the same rows. Only a strictly higher mean lets the proposal go on. A proposal that loses or ties is recorded with skip_reason="minibatch_rejected" and counts toward patience. Its record's score covers the sampled rows only, and its rows_evaluated is their count.

Cost per iteration, with a trainset of n rows:

• A rejected proposal costs k rows.
• A proposal that passes costs k + n rows when the valset is the trainset. The full trainset batch is reused for scoring, as without the gate.
• A proposal that passes costs k rows plus the valset pass when you pass a separate valset. No full trainset pass runs: the proposal's minibatch rows, evaluated with traces, are its reflection batch.

With a separate valset, an accepted candidate therefore has scores for its k sampled rows only. The next gate samples among those rows, the reflective dataset for its next proposal covers them, and trainset_score on the result is the mean over them.

A setting of None, or one at least the trainset size, keeps the full evaluation for every proposal.

The minibatch works alongside two other settings:

• reflection_max_trials caps how many trials the reflection prompt reads from the parent's cached batch: the full trainset, or the sampled rows of a candidate accepted with a separate valset. It does not change what is evaluated.
• SubsetEvaluationPolicy trims the valset scoring pass when a candidate_selector is set (see Evaluation Policies). It is independent of the minibatch, which gates the trainset pass.

Proposal Validator¶

Set proposal_validator to reject a proposal the reflector produced before it costs any evaluation. The engine calls it with each evolved component's name and proposed text. Return None to accept, or a short reason to reject:

def validate(component: str, text: str) -> str | None:
    if "{question}" not in text:
        return "dropped the {question} placeholder"
    if text.startswith("I need the missing inputs"):
        return "asked for its inputs instead of answering"
    return None

config = EvolutionConfig(max_iterations=30, proposal_validator=validate)

A rejected proposal is recorded with skip_reason="proposal_rejected", a score of 0.0, the rejected text as component_text and the reason in rejection_reason. It is not evaluated, it counts toward patience, and it reaches on_iteration like any other skipped iteration. The check runs before output_schema validation. An exception raised by the validator ends the run.

Checkpoint and Resume¶

Set checkpoint_path to have the engine save its own state to one JSON file, so a run that crashes at iteration 40 of 50 can continue instead of starting over. Set resume=True on the next run to continue from that file:

config = EvolutionConfig(
    max_iterations=50,
    checkpoint_path="runs/checkpoint.json",  # str or Path
    resume=True,  # Omit on the first run
)

The engine writes the file after the baseline evaluation and again after every recorded iteration, skipped iterations included. Each write goes to a temporary file beside the checkpoint and then replaces it with os.replace, so the file always holds the last completed write, never a partial one. The parent directory is created when needed. A write failure raises; nothing is swallowed. A reflection batch whose outputs or metadata cannot be written as JSON therefore makes the write raise too.

The file records the iteration number, the stagnation counter, the best candidate and its scores, the iteration history, the score of every candidate scored so far, the evaluation counter, the best candidate's reflection batch (with its trajectories) and the random state that draws minibatch rows.

With resume=True the engine restores that state instead of evaluating the baseline. Iteration numbers, the stagnation counter and the evaluation counter continue where they stopped, and a candidate that was already scored is not evaluated again, including the best candidate whose reflection batch is restored from the file. on_iteration is not called for restored history. With resume=False, an existing file is overwritten on the first write.

Resume refuses with a ConfigurationError, rather than starting over, when:

• resume=True is set without checkpoint_path;
• the file does not exist;
• the file's checkpoint_version is not one this gepa-adk reads;
• the checkpoint came from a different initial candidate;
• the trainset or valset size differs from the checkpointed run.

Pareto state is not checkpointed yet, so the engine refuses a checkpoint_path together with a candidate_selector.

The checkpoint also carries the run's token rollup, so result.token_usage on a resumed run covers the whole run. A checkpoint written before 2.5.0 has no rollup, and the pre-resume usage then reads as unknown.

Inspecting a checkpoint¶

inspect_checkpoint reads a checkpoint file without building an engine or calling a model, so you can see how far a run got before you resume it:

from gepa_adk import inspect_checkpoint

summary = inspect_checkpoint("runs/checkpoint.json")
print(summary.iteration, summary.best_score)
print(summary.best_candidate.components["instruction"])
print(summary.total_evaluations, summary.token_usage.total_tokens)

It returns a frozen CheckpointSummary and raises the same ConfigurationError as resume when the file is missing or its checkpoint_version is not one this gepa-adk reads.

Using Validation Sets¶

Split data for more robust optimization:

# Given a larger dataset of examples
examples = [
    {"input": "I am the Mayor."},
    {"input": "I am your neighbor."},
    {"input": "I am a stranger."},
    {"input": "I am the postman."},
    {"input": "I am a visiting dignitary."},
    {"input": "I am your teacher."},
    {"input": "I am the shopkeeper."},
    {"input": "I am a lost traveler."},
    {"input": "I am your cousin."},
    {"input": "I am the village elder."},
]

trainset = examples[:8]   # 80% for training
valset = examples[8:]     # 20% for validation

result = run_sync(evolve(agent, trainset, valset=valset, critic=critic, config=config))
print(result.valset_score)

result.valset_score is the best candidate's mean per-row score on the scoring set: the valset, or the trainset when you pass no valset. result.final_score is the same candidate's acceptance aggregate, a sum under the default acceptance_metric="sum".

Stop Callbacks¶

Add custom stopping conditions:

from gepa_adk.adapters.stoppers import ScoreThresholdStopper

config = EvolutionConfig(
    max_iterations=50,
    patience=10,
    reflection_model="ollama_chat/llama3.2:latest",
    stop_callbacks=[ScoreThresholdStopper(0.95)],  # Stop at 95% score
)

See the Stop Callbacks Guide for more options.

Evaluation Policies¶

AsyncGEPAEngine accepts an evaluation_policy that picks which validation examples to score each iteration, such as SubsetEvaluationPolicy for large validation sets. A policy takes effect only through the Pareto state that a candidate_selector creates, so a policy must be paired with a selector. evolve() does not expose this parameter; build the engine directly.

from gepa_adk.adapters.selection import (
    ParetoCandidateSelector,
    SubsetEvaluationPolicy,
)
from gepa_adk.engine import AsyncGEPAEngine

engine = AsyncGEPAEngine(
    adapter=my_adapter,
    config=config,
    initial_candidate=candidate,
    batch=trainset,
    valset=valset,
    candidate_selector=ParetoCandidateSelector(),
    evaluation_policy=SubsetEvaluationPolicy(subset_size=0.2),
)

Valid combinations¶

The last row applies to every explicit policy, including FullEvaluationPolicy().

Async vs Sync¶

Use evolve() for async contexts, run_sync(evolve(...)) for scripts:

# Async
result = await evolve(agent, trainset, critic=critic, config=config)

# Sync (wraps async internally)
result = run_sync(evolve(agent, trainset, critic=critic, config=config))

Other evolve() Options¶

evolve() takes these keyword arguments besides the ones explained elsewhere on this page (valset, critic, config, candidate_selector, components, registry):

Merge and Pareto Selection¶

By default every mutation starts from the current best candidate and the engine keeps no frontier. Passing candidate_selector= to evolve() makes the engine keep a Pareto state of every scored candidate and draw each mutation's parent from it; GEPA Fundamentals explains the frontier, and the glossary defines pareto_frontier, candidate and acceptance_metric. A proposal is still accepted only when it beats the best score by more than min_improvement_threshold, so result.evolved_components remains the best candidate.

candidate_selector takes an instance of a CandidateSelectorProtocol or a name that create_candidate_selector resolves with the run's seeded random state:

from gepa_adk import EvolutionConfig, evolve, run_sync

config = EvolutionConfig(
    max_iterations=30,
    reflection_model="ollama_chat/llama3.2:latest",
    use_merge=True,             # Try crossover merges after accepted proposals
    max_merge_invocations=5,    # At most 5 merge attempts per run
)

result = run_sync(evolve(
    agent,
    trainset,
    valset=valset,
    critic=critic,
    config=config,
    candidate_selector="pareto",
    components=["instruction", "output_schema"],  # Merges need two or more
))

A selector cannot be combined with checkpoint_path, because Pareto state is not checkpointed (see Checkpoint and Resume).

Merges¶

use_merge=True makes the engine create a MergeProposer, which picks two frontier candidates that share a common ancestor and combines the components each one improved into one merged candidate. Each accepted proposal schedules one merge, and the engine attempts it in the same iteration.

• Merges need a candidate_selector. The attempt reads the Pareto state, which exists only with a selector. Without one, use_merge=True schedules merges that never run.
• Merges need at least two evolved components. The proposer takes each component's whole value from one parent or the other, so with a single component every merge equals a parent and is skipped as a duplicate; pass components= with two or more names, as the example does.
• max_merge_invocations caps attempts per run (default 10). Once that many merges were attempted, no more are scheduled. use_merge=True with max_merge_invocations=0 raises ConfigurationError.
• A merge can find nothing to merge. When no pair qualifies (no common ancestor, too small a frontier) the proposer returns nothing and the scheduled merge is dropped without counting as an attempt.
• A merged candidate is scored and accepted like a proposal. It is skipped without evaluation when its id was already scored or its output_schema fails validation; that still counts as an attempt.
• A merge writes no iteration record. The iteration's record describes its own proposal; merge activity appears in the merge_scheduling.* and merge.proposal_skipped log events.

Frontier types and objective scores¶

EvolutionConfig.frontier_type takes a FrontierType (or its string value) and picks which scores the Pareto state tracks:

Objective scores come from the adapter: an EvaluationBatch may carry objective_scores, one {objective name: score} dict per example. OBJECTIVE, HYBRID and CARTESIAN consume them and require them: the Pareto state raises ConfigurationError when a candidate arrives without them. The ADKAdapter that evolve() builds automatically forwards each example's non-empty metadata["dimension_scores"] dictionary to objective_scores. A critic that returns these dimensions can use candidate_selector="pareto" with frontier_type=FrontierType.OBJECTIVE (or HYBRID or CARTESIAN) without another flag. If only some examples provide non-empty dimension dictionaries, evaluation raises ConfigurationError naming the example index that lacks them. When no example provides dimensions, objective scores remain None; use the default INSTANCE frontier for scalar-only scoring. Custom adapters can also report objective scores through AsyncGEPAEngine directly as in Evaluation Policies.

The scores are kept on the results as well. Each IterationRecord has objective_scores from its scoring evaluation, and EvolutionResult has objective_scores from the best candidate's evaluation. Both are None when the adapter reported none.

Advanced: Output Schema Evolution¶

You can evolve the agent's output schema in addition to instructions.

When to Use¶

• Optimize field definitions and descriptions
• Refine data structure for better outputs
• Co-evolve instruction and schema together

Example¶

from pydantic import BaseModel, Field

class TaskOutput(BaseModel):
    result: str
    confidence: float = Field(ge=0.0, le=1.0)

agent = LlmAgent(
    name="task-agent",
    model=LiteLlm(model="ollama_chat/llama3.2:latest"),
    instruction="Complete the task.",
    output_schema=TaskOutput,
)

# Evolve just the output schema
result = run_sync(evolve(
    agent,
    trainset,
    critic=critic,
    components=["output_schema"],
    config=config,
))

print(result.evolved_components["output_schema"])

A proposal whose schema text fails validation is not evaluated. Its iteration is recorded in result.iteration_history with skip_reason="schema_validation_failed" and counts toward patience, so the run still ends by max_iterations or patience like any other run.

Evolving Both¶

result = run_sync(evolve(
    agent,
    trainset,
    critic=critic,
    components=["instruction", "output_schema"],
    config=config,
))

Using Evolved Schemas¶

from gepa_adk.utils.schema_utils import deserialize_schema

EvolvedSchema = deserialize_schema(result.evolved_components["output_schema"])

evolved_agent = LlmAgent(
    name="evolved-agent",
    model=LiteLlm(model="ollama_chat/llama3.2:latest"),
    instruction=result.evolved_components["instruction"],
    output_schema=EvolvedSchema,
)

Advanced: Generation Config Evolution¶

Evolve LLM parameters like temperature and top_p.

Example¶

from google.genai.types import GenerateContentConfig

agent = LlmAgent(
    name="creative-agent",
    model=LiteLlm(model="ollama_chat/llama3.2:latest"),
    instruction="Write creatively.",
    generate_content_config=GenerateContentConfig(
        temperature=0.7,
        top_p=0.9,
    ),
)

result = run_sync(evolve(
    agent,
    trainset,
    critic=critic,
    components=["generate_content_config"],
    config=config,
))

print(result.evolved_components["generate_content_config"])

Advanced: Evolving Prompts Held by a Tool¶

Evolve prompt strings that a tool owns, such as the prompts it sends to an external service, instead of the agent's own instruction.

Example¶

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,
)

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)  # each key becomes a component
config = EvolutionConfig(max_iterations=10, acceptance_metric="mean")

result = run_sync(evolve(
    agent,
    trainset,
    scorer=LabelAgreementScorer(),
    components=names,
    component_selector="round_robin",
    config=config,
))

# Evaluation restores `prompts` after each run; apply the result yourself.
prompts.update({name: result.evolved_components[name] for name in names})

The round-robin selector rotates across the keys, so each iteration reflects on one prompt. During evaluation the candidate text is written into prompts before each run and restored afterwards, so the dict holds the original text when evolve() returns.

One registry per round¶

register_mapping_components() registers into the default registry unless you pass registry=. When one process runs several rounds that register the same key names, give each round its own registry and pass it to evolve(): registering the same names into the default registry again replaces the first round's handler, which still holds the first mapping by reference.

from gepa_adk.adapters.components import ComponentHandlerRegistry

registry = ComponentHandlerRegistry()
names = register_mapping_components(prompts, registry=registry)
result = run_sync(evolve(
    agent,
    trainset,
    scorer=LabelAgreementScorer(),
    components=names,
    config=config,
    registry=registry,
))

evolve() resolves every name in components through that registry, so a new registry holds only the names you register into it; a name it does not hold raises ConfigurationError. evolve_group() and evolve_workflow() take the same registry= keyword.

How trainset input reaches the tool¶

The trainset input string is the user message the agent receives. The agent decides to call the tool and passes it whatever arguments it chooses, usually that text. When the tool needs structured state, serialize it into the input string as JSON and let the agent or the tool parse it:

import json

row = {
    "input": json.dumps({"ticket": "App crashes on login", "priority": "high"}),
    "expected": "bug",
}

gepa-adk does not pass a dict through to the tool.

Related Guides¶

• Critic Agents — Detailed critic patterns
• Multi-Agent — Evolve multiple agents together
• Workflows — Optimize agent pipelines

API Reference¶

• evolve() — Async evolution
• run_sync() — Sync wrapper for async evolution
• EvolutionConfig — Configuration
• EvolutionResult — Results
• register_mapping_components() — Register a caller-owned mapping