Agenta Skill

Manage, evaluate, and deploy LLM prompts with confidence. Version control your prompts, run A/B tests, and measure quality with automated evaluation.

Quick Start

Install Agenta SDK

pip install agenta

Start Agenta locally with Docker

docker run -d -p 3000:3000 -p 8000:8000 ghcr.io/agenta-ai/agenta

Or use pip for just the SDK

pip install agenta

Initialize project

agenta init --app-name my-llm-app

When to Use This Skill

USE when:

• Managing multiple versions of prompts in production

• Need systematic A/B testing of prompt variations

• Evaluating prompt quality with automated metrics

• Collaborating on prompt development across teams

• Requiring audit trails for prompt changes

• Building LLM applications that need to iterate quickly

• Need to compare different models with same prompts

• Want a playground for rapid prompt experimentation

• Self-hosting is required for security/compliance

DON'T USE when:

• Simple single-prompt applications

• No need for prompt versioning or testing

• Already using another prompt management system

• Rapid prototyping without evaluation needs

• Cost-sensitive projects (evaluation adds API calls)

Prerequisites

SDK installation

pip install agenta>=0.10.0

For self-hosted deployment

docker pull ghcr.io/agenta-ai/agenta

Or with docker-compose

git clone https://github.com/Agenta-AI/agenta cd agenta docker-compose up -d

Environment setup

export AGENTA_HOST="http://localhost:3000" export AGENTA_API_KEY="your-api-key" # If using cloud version

For LLM providers

export OPENAI_API_KEY="sk-..." export ANTHROPIC_API_KEY="sk-ant-..."

Verify Installation

import agenta as ag from agenta import Agenta

Initialize client

client = Agenta()

Check connection

print(f"Agenta SDK version: {ag.version}") print("Connection successful!")

Core Capabilities

1. Prompt Versioning and Management

Creating Versioned Prompts:

""" Create and manage versioned prompts with Agenta. """ import agenta as ag from agenta import Agenta from typing import Optional, Dict, Any

Initialize Agenta

ag.init()

@ag.entrypoint def generate_summary( text: str, max_length: int = 100, style: str = "professional" ) -> str: """ Generate a summary with versioned prompt.

Args:
    text: Text to summarize
    max_length: Maximum summary length
    style: Writing style (professional, casual, technical)

Returns:
    Generated summary
"""
# Define prompt template (this becomes versioned)
prompt = f"""Summarize the following text in a {style} tone.

Keep the summary under {max_length} words.

Text: {text}

Summary:"""

# Call LLM (Agenta tracks this)
response = ag.llm.complete(
    prompt=prompt,
    model="gpt-4",
    temperature=0.3,
    max_tokens=max_length * 2
)

return response.text

Example usage

text = """ The company reported strong Q3 results with revenue up 25% year-over-year. Operating margins improved to 18% from 15% in the prior year. The CEO highlighted expansion into new markets and product launches. """

summary = generate_summary(text, max_length=50, style="professional") print(summary)

Managing Prompt Versions:

""" Manage multiple prompt versions programmatically. """ import agenta as ag from agenta import Agenta from dataclasses import dataclass from typing import List, Dict, Optional from datetime import datetime

@dataclass class PromptVersion: """Represents a prompt version.""" version_id: str name: str template: str parameters: Dict[str, Any] created_at: datetime is_active: bool = False

class PromptManager: """ Manage prompt versions with Agenta. """

def __init__(self, app_name: str):
    self.app_name = app_name
    self.client = Agenta()

def create_version(
    self,
    name: str,
    template: str,
    parameters: Dict[str, Any] = None
) -> PromptVersion:
    """
    Create a new prompt version.

    Args:
        name: Version name
        template: Prompt template
        parameters: Default parameters

    Returns:
        Created PromptVersion
    """
    # Create variant in Agenta
    variant = self.client.create_variant(
        app_name=self.app_name,
        variant_name=name,
        config={
            "template": template,
            "parameters": parameters or {}
        }
    )

    return PromptVersion(
        version_id=variant.id,
        name=name,
        template=template,
        parameters=parameters or {},
        created_at=datetime.now(),
        is_active=False
    )

def list_versions(self) -> List[PromptVersion]:
    """List all prompt versions."""
    variants = self.client.list_variants(app_name=self.app_name)

    versions = []
    for v in variants:
        versions.append(PromptVersion(
            version_id=v.id,
            name=v.name,
            template=v.config.get("template", ""),
            parameters=v.config.get("parameters", {}),
            created_at=v.created_at,
            is_active=v.is_default
        ))

    return versions

def set_active_version(self, version_id: str) -> None:
    """Set a version as the active/default version."""
    self.client.set_default_variant(
        app_name=self.app_name,
        variant_id=version_id
    )

def get_version(self, version_id: str) -> PromptVersion:
    """Get a specific version."""
    variant = self.client.get_variant(variant_id=version_id)

    return PromptVersion(
        version_id=variant.id,
        name=variant.name,
        template=variant.config.get("template", ""),
        parameters=variant.config.get("parameters", {}),
        created_at=variant.created_at,
        is_active=variant.is_default
    )

def compare_versions(
    self,
    version_ids: List[str],
    test_input: str
) -> Dict[str, str]:
    """
    Compare outputs from multiple versions.

    Args:
        version_ids: List of version IDs to compare
        test_input: Input to test with

    Returns:
        Dictionary mapping version_id to output
    """
    results = {}

    for vid in version_ids:
        version = self.get_version(vid)

        # Format prompt with test input
        prompt = version.template.format(input=test_input)

        # Generate output
        response = ag.llm.complete(prompt=prompt)
        results[vid] = response.text

    return results

Usage

manager = PromptManager("summarizer-app")

Create versions

v1 = manager.create_version( name="concise-v1", template="Summarize briefly: {input}", parameters={"max_tokens": 100} )

v2 = manager.create_version( name="detailed-v2", template="Provide a comprehensive summary with key points: {input}", parameters={"max_tokens": 300} )

List all versions

versions = manager.list_versions() for v in versions: print(f"{v.name}: {v.version_id} (active: {v.is_active})")

Set active version

manager.set_active_version(v1.version_id)

2. A/B Testing Prompts

Setting Up A/B Tests:

""" Configure and run A/B tests on prompt variations. """ import agenta as ag from agenta import Agenta from typing import Dict, List, Optional from dataclasses import dataclass import random

@dataclass class ABTestConfig: """Configuration for A/B test.""" name: str variants: Dict[str, float] # variant_id: traffic_percentage metrics: List[str] min_samples: int = 100

class ABTestRunner: """ Run A/B tests on prompt variants. """

def __init__(self, app_name: str):
    self.app_name = app_name
    self.client = Agenta()
    self.results: Dict[str, List[Dict]] = {}

def create_test(
    self,
    name: str,
    control_variant: str,
    treatment_variant: str,
    traffic_split: float = 0.5
) -> ABTestConfig:
    """
    Create an A/B test.

    Args:
        name: Test name
        control_variant: Control variant ID
        treatment_variant: Treatment variant ID
        traffic_split: Percentage for treatment (0-1)

    Returns:
        ABTestConfig
    """
    config = ABTestConfig(
        name=name,
        variants={
            control_variant: 1 - traffic_split,
            treatment_variant: traffic_split
        },
        metrics=["response_quality", "latency", "cost"]
    )

    # Initialize results tracking
    for variant in config.variants.keys():
        self.results[variant] = []

    return config

def route_request(self, config: ABTestConfig) -> str:
    """
    Route a request to a variant based on traffic split.

    Args:
        config: A/B test configuration

    Returns:
        Selected variant ID
    """
    rand = random.random()
    cumulative = 0

    for variant_id, percentage in config.variants.items():
        cumulative += percentage
        if rand <= cumulative:
            return variant_id

    # Fallback to first variant
    return list(config.variants.keys())[0]

def run_request(
    self,
    config: ABTestConfig,
    input_data: str
) -> Dict:
    """
    Run a single request in the A/B test.

    Args:
        config: A/B test configuration
        input_data: Input for the prompt

    Returns:
        Result dictionary with variant and output
    """
    import time

    # Route to variant
    variant_id = self.route_request(config)
    variant = self.client.get_variant(variant_id)

    # Prepare prompt
    prompt = variant.config.get("template", "").format(input=input_data)

    # Run with timing
    start_time = time.time()
    response = ag.llm.complete(prompt=prompt)
    latency = time.time() - start_time

    result = {
        "variant_id": variant_id,
        "input": input_data,
        "output": response.text,
        "latency": latency,
        "tokens_used": response.usage.total_tokens if hasattr(response, 'usage') else 0
    }

    # Store result
    self.results[variant_id].append(result)

    return result

def get_test_results(self, config: ABTestConfig) -> Dict:
    """
    Get aggregated results for an A/B test.

    Args:
        config: A/B test configuration

    Returns:
        Aggregated results by variant
    """
    summary = {}

    for variant_id, results in self.results.items():
        if not results:
            continue

        latencies = [r["latency"] for r in results]
        tokens = [r["tokens_used"] for r in results]

        summary[variant_id] = {
            "sample_count": len(results),
            "avg_latency": sum(latencies) / len(latencies),
            "avg_tokens": sum(tokens) / len(tokens) if tokens else 0,
            "min_latency": min(latencies),
            "max_latency": max(latencies)
        }

    return summary

def declare_winner(self, config: ABTestConfig) -> Optional[str]:
    """
    Analyze results and declare a winner.

    Args:
        config: A/B test configuration

    Returns:
        Winner variant ID or None if inconclusive
    """
    summary = self.get_test_results(config)

    # Check minimum samples
    for variant_id, stats in summary.items():
        if stats["sample_count"] < config.min_samples:
            print(f"Insufficient samples for {variant_id}")
            return None

    # Simple winner selection based on latency
    # In production, use statistical significance tests
    best_variant = min(
        summary.keys(),
        key=lambda v: summary[v]["avg_latency"]
    )

    return best_variant

Usage Example

ag.init()

runner = ABTestRunner("chatbot-app")

Create A/B test

test_config = runner.create_test( name="prompt-optimization-test", control_variant="variant-a-id", treatment_variant="variant-b-id", traffic_split=0.5 )

Run test requests

test_inputs = [ "What is machine learning?", "Explain neural networks", "How does backpropagation work?" ]

for input_text in test_inputs: result = runner.run_request(test_config, input_text) print(f"Variant: {result['variant_id']}, Latency: {result['latency']:.3f}s")

Get results

results = runner.get_test_results(test_config) print("\nTest Results:") for variant, stats in results.items(): print(f" {variant}: {stats}")

3. Evaluation Metrics and Testing

Automated Evaluation Pipeline:

""" Evaluate prompts with automated metrics. """ import agenta as ag from agenta import Agenta from typing import List, Dict, Callable, Any from dataclasses import dataclass import json

@dataclass class EvaluationResult: """Result of an evaluation.""" metric_name: str score: float details: Dict[str, Any]

class MetricEvaluator: """Base class for evaluation metrics."""

def __init__(self, name: str):
    self.name = name

def evaluate(
    self,
    output: str,
    expected: str = None,
    context: Dict = None
) -> EvaluationResult:
    raise NotImplementedError

class ExactMatchMetric(MetricEvaluator): """Exact match evaluation."""

def __init__(self):
    super().__init__("exact_match")

def evaluate(self, output: str, expected: str = None, context: Dict = None) -> EvaluationResult:
    if expected is None:
        return EvaluationResult(self.name, 0.0, {"error": "No expected value"})

    match = output.strip().lower() == expected.strip().lower()

    return EvaluationResult(
        metric_name=self.name,
        score=1.0 if match else 0.0,
        details={"match": match}
    )

class ContainsMetric(MetricEvaluator): """Check if output contains expected keywords."""

def __init__(self, keywords: List[str]):
    super().__init__("contains_keywords")
    self.keywords = keywords

def evaluate(self, output: str, expected: str = None, context: Dict = None) -> EvaluationResult:
    output_lower = output.lower()
    found = [kw for kw in self.keywords if kw.lower() in output_lower]
    score = len(found) / len(self.keywords)

    return EvaluationResult(
        metric_name=self.name,
        score=score,
        details={
            "found_keywords": found,
            "missing_keywords": [kw for kw in self.keywords if kw.lower() not in output_lower]
        }
    )

class LengthMetric(MetricEvaluator): """Evaluate output length."""

def __init__(self, min_length: int = 10, max_length: int = 500):
    super().__init__("length")
    self.min_length = min_length
    self.max_length = max_length

def evaluate(self, output: str, expected: str = None, context: Dict = None) -> EvaluationResult:
    length = len(output.split())

    if self.min_length <= length <= self.max_length:
        score = 1.0
    elif length < self.min_length:
        score = length / self.min_length
    else:
        score = max(0, 1 - (length - self.max_length) / self.max_length)

    return EvaluationResult(
        metric_name=self.name,
        score=score,
        details={
            "word_count": length,
            "min_length": self.min_length,
            "max_length": self.max_length
        }
    )

class LLMJudgeMetric(MetricEvaluator): """Use an LLM to judge output quality."""

def __init__(self, criteria: str = "helpfulness"):
    super().__init__(f"llm_judge_{criteria}")
    self.criteria = criteria

def evaluate(self, output: str, expected: str = None, context: Dict = None) -> EvaluationResult:
    judge_prompt = f"""Evaluate the following response on {self.criteria}.

Score from 0.0 to 1.0.

Response: {output}

{f'Expected: {expected}' if expected else ''}

Provide your evaluation as JSON: {{"score": 0.0-1.0, "reasoning": "..."}} """

    response = ag.llm.complete(
        prompt=judge_prompt,
        model="gpt-4",
        temperature=0
    )

    try:
        result = json.loads(response.text)
        score = float(result.get("score", 0.5))
        reasoning = result.get("reasoning", "")
    except (json.JSONDecodeError, ValueError):
        score = 0.5
        reasoning = "Failed to parse judge response"

    return EvaluationResult(
        metric_name=self.name,
        score=score,
        details={"reasoning": reasoning, "criteria": self.criteria}
    )

class EvaluationPipeline: """ Pipeline for running multiple evaluations. """

def __init__(self, app_name: str):
    self.app_name = app_name
    self.client = Agenta()
    self.metrics: List[MetricEvaluator] = []

def add_metric(self, metric: MetricEvaluator) -> 'EvaluationPipeline':
    """Add a metric to the pipeline."""
    self.metrics.append(metric)
    return self

def evaluate_single(
    self,
    output: str,
    expected: str = None,
    context: Dict = None
) -> Dict[str, EvaluationResult]:
    """
    Evaluate a single output with all metrics.

    Args:
        output: Generated output
        expected: Expected output (optional)
        context: Additional context

    Returns:
        Dictionary of metric results
    """
    results = {}

    for metric in self.metrics:
        result = metric.evaluate(output, expected, context)
        results[metric.name] = result

    return results

def evaluate_batch(
    self,
    test_cases: List[Dict]
) -> Dict[str, List[EvaluationResult]]:
    """
    Evaluate a batch of test cases.

    Args:
        test_cases: List of {input, output, expected} dicts

    Returns:
        Aggregated results by metric
    """
    all_results = {metric.name: [] for metric in self.metrics}

    for case in test_cases:
        results = self.evaluate_single(
            output=case.get("output", ""),
            expected=case.get("expected"),
            context=case.get("context")
        )

        for metric_name, result in results.items():
            all_results[metric_name].append(result)

    return all_results

def get_summary(self, batch_results: Dict[str, List[EvaluationResult]]) -> Dict:
    """
    Get summary statistics from batch evaluation.

    Args:
        batch_results: Results from evaluate_batch

    Returns:
        Summary statistics
    """
    summary = {}

    for metric_name, results in batch_results.items():
        scores = [r.score for r in results]
        summary[metric_name] = {
            "mean": sum(scores) / len(scores) if scores else 0,
            "min": min(scores) if scores else 0,
            "max": max(scores) if scores else 0,
            "count": len(scores)
        }

    return summary

Usage

ag.init()

Create evaluation pipeline

pipeline = EvaluationPipeline("qa-bot") pipeline.add_metric(ContainsMetric(["answer", "explanation"])) pipeline.add_metric(LengthMetric(min_length=20, max_length=200)) pipeline.add_metric(LLMJudgeMetric(criteria="helpfulness"))

Test cases

test_cases = [ { "input": "What is Python?", "output": "Python is a programming language known for its simplicity. The answer is that it's versatile. Here's an explanation: it's widely used in data science and web development.", "expected": "Python is a high-level programming language" }, { "input": "Explain recursion", "output": "Recursion is a function calling itself. The answer involves base cases and recursive calls. Explanation: it's useful for tree structures.", "expected": "A function that calls itself" } ]

Run evaluation

results = pipeline.evaluate_batch(test_cases) summary = pipeline.get_summary(results)

print("Evaluation Summary:") for metric, stats in summary.items(): print(f" {metric}: mean={stats['mean']:.2f}, min={stats['min']:.2f}, max={stats['max']:.2f}")

4. Playground and Experimentation

Creating Interactive Playground:

""" Build an interactive playground for prompt experimentation. """ import agenta as ag from agenta import Agenta from typing import Dict, List, Any, Optional from dataclasses import dataclass, field from datetime import datetime import json

@dataclass class ExperimentRun: """Single experiment run.""" run_id: str prompt: str parameters: Dict[str, Any] output: str metrics: Dict[str, float] timestamp: datetime = field(default_factory=datetime.now)

class Playground: """ Interactive playground for prompt experimentation. """

def __init__(self, app_name: str):
    self.app_name = app_name
    self.client = Agenta()
    self.experiments: List[ExperimentRun] = []
    self.current_prompt = ""
    self.current_params = {}

def set_prompt(self, prompt: str) -> 'Playground':
    """Set the current prompt template."""
    self.current_prompt = prompt
    return self

def set_parameters(self, **params) -> 'Playground':
    """Set LLM parameters."""
    self.current_params.update(params)
    return self

def run(self, input_data: str) -> ExperimentRun:
    """
    Run the current prompt with input.

    Args:
        input_data: Input to format into prompt

    Returns:
        ExperimentRun with results
    """
    import time
    import uuid

    # Format prompt
    formatted_prompt = self.current_prompt.format(input=input_data)

    # Run with timing
    start_time = time.time()
    response = ag.llm.complete(
        prompt=formatted_prompt,
        **self.current_params
    )
    latency = time.time() - start_time

    # Create run record
    run = ExperimentRun(
        run_id=str(uuid.uuid4())[:8],
        prompt=formatted_prompt,
        parameters=self.current_params.copy(),
        output=response.text,
        metrics={
            "latency": latency,
            "output_length": len(response.text),
            "tokens": response.usage.total_tokens if hasattr(response, 'usage') else 0
        }
    )

    self.experiments.append(run)

    return run

def compare(
    self,
    prompts: List[str],
    test_input: str,
    parameters: Dict = None
) -> List[ExperimentRun]:
    """
    Compare multiple prompts with same input.

    Args:
        prompts: List of prompt templates
        test_input: Input to test
        parameters: Shared parameters

    Returns:
        List of ExperimentRuns
    """
    runs = []
    original_prompt = self.current_prompt
    original_params = self.current_params.copy()

    if parameters:
        self.set_parameters(**parameters)

    for prompt in prompts:
        self.set_prompt(prompt)
        run = self.run(test_input)
        runs.append(run)

    # Restore original state
    self.current_prompt = original_prompt
    self.current_params = original_params

    return runs

def parameter_sweep(
    self,
    param_name: str,
    values: List[Any],
    test_input: str
) -> List[ExperimentRun]:
    """
    Sweep over parameter values.

    Args:
        param_name: Parameter to sweep
        values: List of values to try
        test_input: Input for testing

    Returns:
        List of ExperimentRuns
    """
    runs = []
    original_value = self.current_params.get(param_name)

    for value in values:
        self.current_params[param_name] = value
        run = self.run(test_input)
        runs.append(run)

    # Restore original value
    if original_value is not None:
        self.current_params[param_name] = original_value
    else:
        self.current_params.pop(param_name, None)

    return runs

def get_history(self, limit: int = 10) -> List[ExperimentRun]:
    """Get recent experiment history."""
    return self.experiments[-limit:]

def export_experiments(self, filepath: str) -> None:
    """Export experiments to JSON file."""
    data = []
    for exp in self.experiments:
        data.append({
            "run_id": exp.run_id,
            "prompt": exp.prompt,
            "parameters": exp.parameters,
            "output": exp.output,
            "metrics": exp.metrics,
            "timestamp": exp.timestamp.isoformat()
        })

    with open(filepath, 'w') as f:
        json.dump(data, f, indent=2)

def find_best_run(self, metric: str = "latency", minimize: bool = True) -> Optional[ExperimentRun]:
    """
    Find the best run based on a metric.

    Args:
        metric: Metric to optimize
        minimize: Whether to minimize (True) or maximize (False)

    Returns:
        Best ExperimentRun or None
    """
    if not self.experiments:
        return None

    valid_runs = [e for e in self.experiments if metric in e.metrics]

    if not valid_runs:
        return None

    if minimize:
        return min(valid_runs, key=lambda e: e.metrics[metric])
    else:
        return max(valid_runs, key=lambda e: e.metrics[metric])

Usage

ag.init()

playground = Playground("experiment-app")

Set up experiment

playground.set_prompt("Answer this question concisely: {input}") playground.set_parameters(model="gpt-4", temperature=0.3, max_tokens=100)

Run single experiment

run = playground.run("What is machine learning?") print(f"Output: {run.output}") print(f"Latency: {run.metrics['latency']:.3f}s")

Compare prompts

comparison_runs = playground.compare( prompts=[ "Answer briefly: {input}", "Explain in detail: {input}", "Give a one-sentence answer: {input}" ], test_input="What is deep learning?" )

print("\nPrompt Comparison:") for i, run in enumerate(comparison_runs): print(f" Prompt {i+1}: {run.metrics['latency']:.3f}s, {run.metrics['output_length']} chars")

Parameter sweep

temperature_runs = playground.parameter_sweep( param_name="temperature", values=[0.0, 0.3, 0.7, 1.0], test_input="Write a creative story opening" )

print("\nTemperature Sweep:") for run in temperature_runs: print(f" temp={run.parameters['temperature']}: {run.output[:50]}...")

Find best run

best = playground.find_best_run(metric="latency", minimize=True) if best: print(f"\nBest run: {best.run_id} with latency {best.metrics['latency']:.3f}s")

Export experiments

playground.export_experiments("experiments.json")

5. Model Comparison

Comparing Different LLM Models:

""" Compare performance across different LLM models. """ import agenta as ag from agenta import Agenta from typing import Dict, List, Any from dataclasses import dataclass import time

@dataclass class ModelResult: """Result from a single model run.""" model: str output: str latency: float tokens: int cost: float

class ModelComparator: """ Compare prompts across different models. """

# Cost per 1K tokens (approximate)
MODEL_COSTS = {
    "gpt-4": {"input": 0.03, "output": 0.06},
    "gpt-4-turbo": {"input": 0.01, "output": 0.03},
    "gpt-3.5-turbo": {"input": 0.0005, "output": 0.0015},
    "claude-3-opus": {"input": 0.015, "output": 0.075},
    "claude-3-sonnet": {"input": 0.003, "output": 0.015},
    "claude-3-haiku": {"input": 0.00025, "output": 0.00125}
}

def __init__(self, models: List[str] = None):
    self.models = models or ["gpt-4", "gpt-3.5-turbo"]
    self.results: Dict[str, List[ModelResult]] = {m: [] for m in self.models}

def _estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
    """Estimate cost for a model run."""
    costs = self.MODEL_COSTS.get(model, {"input": 0.01, "output": 0.03})
    return (input_tokens / 1000 * costs["input"] +
            output_tokens / 1000 * costs["output"])

def run_comparison(
    self,
    prompt: str,
    temperature: float = 0.3,
    max_tokens: int = 200
) -> Dict[str, ModelResult]:
    """
    Run the same prompt across all models.

    Args:
        prompt: Prompt to test
        temperature: Temperature setting
        max_tokens: Maximum output tokens

    Returns:
        Results for each model
    """
    results = {}

    for model in self.models:
        start_time = time.time()

        try:
            response = ag.llm.complete(
                prompt=prompt,
                model=model,
                temperature=temperature,
                max_tokens=max_tokens
            )

            latency = time.time() - start_time

            # Get token counts
            input_tokens = len(prompt.split()) * 1.3  # Rough estimate
            output_tokens = len(response.text.split()) * 1.3

            if hasattr(response, 'usage'):
                input_tokens = response.usage.prompt_tokens
                output_tokens = response.usage.completion_tokens

            result = ModelResult(
                model=model,
                output=response.text,
                latency=latency,
                tokens=int(input_tokens + output_tokens),
                cost=self._estimate_cost(model, input_tokens, output_tokens)
            )

        except Exception as e:
            result = ModelResult(
                model=model,
                output=f"Error: {str(e)}",
                latency=0,
                tokens=0,
                cost=0
            )

        results[model] = result
        self.results[model].append(result)

    return results

def run_benchmark(
    self,
    prompts: List[str],
    temperature: float = 0.3
) -> Dict[str, Dict]:
    """
    Run benchmark across multiple prompts.

    Args:
        prompts: List of prompts to test
        temperature: Temperature setting

    Returns:
        Aggregated benchmark results
    """
    for prompt in prompts:
        self.run_comparison(prompt, temperature)

    return self.get_summary()

def get_summary(self) -> Dict[str, Dict]:
    """Get summary statistics for all models."""
    summary = {}

    for model, results in self.results.items():
        if not results:
            continue

        valid_results = [r for r in results if r.latency > 0]

        if not valid_results:
            continue

        summary[model] = {
            "runs": len(valid_results),
            "avg_latency": sum(r.latency for r in valid_results) / len(valid_results),
            "avg_tokens": sum(r.tokens for r in valid_results) / len(valid_results),
            "total_cost": sum(r.cost for r in valid_results),
            "min_latency": min(r.latency for r in valid_results),
            "max_latency": max(r.latency for r in valid_results)
        }

    return summary

def recommend_model(
    self,
    priority: str = "balanced"
) -> str:
    """
    Recommend best model based on priority.

    Args:
        priority: "speed", "cost", "quality", or "balanced"

    Returns:
        Recommended model name
    """
    summary = self.get_summary()

    if not summary:
        return self.models[0]

    if priority == "speed":
        return min(summary.keys(), key=lambda m: summary[m]["avg_latency"])
    elif priority == "cost":
        return min(summary.keys(), key=lambda m: summary[m]["total_cost"])
    elif priority == "quality":
        # Assume larger models = better quality
        quality_order = ["gpt-4", "claude-3-opus", "gpt-4-turbo", "claude-3-sonnet", "gpt-3.5-turbo"]
        for model in quality_order:
            if model in summary:
                return model
    else:  # balanced
        # Score based on normalized latency and cost
        scores = {}
        max_latency = max(s["avg_latency"] for s in summary.values())
        max_cost = max(s["total_cost"] for s in summary.values()) or 1

        for model, stats in summary.items():
            norm_latency = stats["avg_latency"] / max_latency
            norm_cost = stats["total_cost"] / max_cost
            scores[model] = norm_latency * 0.5 + norm_cost * 0.5

        return min(scores.keys(), key=lambda m: scores[m])

    return self.models[0]

Usage

ag.init()

comparator = ModelComparator(models=["gpt-4", "gpt-3.5-turbo"])

Single comparison

results = comparator.run_comparison("Explain quantum computing in simple terms")

print("Single Comparison Results:") for model, result in results.items(): print(f" {model}:") print(f" Latency: {result.latency:.3f}s") print(f" Tokens: {result.tokens}") print(f" Cost: ${result.cost:.4f}") print(f" Output: {result.output[:100]}...")

Benchmark

benchmark_prompts = [ "What is machine learning?", "Explain the difference between AI and ML", "Write a haiku about technology" ]

comparator.run_benchmark(benchmark_prompts)

print("\nBenchmark Summary:") summary = comparator.get_summary() for model, stats in summary.items(): print(f" {model}:") print(f" Runs: {stats['runs']}") print(f" Avg Latency: {stats['avg_latency']:.3f}s") print(f" Total Cost: ${stats['total_cost']:.4f}")

Get recommendation

recommended = comparator.recommend_model(priority="balanced") print(f"\nRecommended model (balanced): {recommended}")

6. Self-Hosted Deployment

Setting Up Self-Hosted Agenta:

""" Configure and manage self-hosted Agenta deployment. """ import agenta as ag from agenta import Agenta from typing import Dict, Any, Optional import os import requests from dataclasses import dataclass

@dataclass class DeploymentConfig: """Configuration for self-hosted deployment.""" host: str port: int api_key: Optional[str] database_url: str redis_url: Optional[str] enable_tracing: bool = True

class SelfHostedManager: """ Manage self-hosted Agenta deployment. """

def __init__(self, config: DeploymentConfig):
    self.config = config
    self.base_url = f"http://{config.host}:{config.port}"
    self.client = None

def initialize(self) -> bool:
    """
    Initialize connection to self-hosted instance.

    Returns:
        True if successful
    """
    try:
        # Set environment for SDK
        os.environ["AGENTA_HOST"] = self.base_url
        if self.config.api_key:
            os.environ["AGENTA_API_KEY"] = self.config.api_key

        # Initialize Agenta
        ag.init()
        self.client = Agenta()

        # Test connection
        response = requests.get(f"{self.base_url}/api/health")
        return response.status_code == 200

    except Exception as e:
        print(f"Initialization failed: {e}")
        return False

def create_app(
    self,
    name: str,
    description: str = ""
) -> Dict:
    """
    Create a new application.

    Args:
        name: Application name
        description: Application description

    Returns:
        Created application details
    """
    return self.client.create_app(
        name=name,
        description=description
    )

def deploy_variant(
    self,
    app_name: str,
    variant_name: str,
    environment: str = "production"
) -> Dict:
    """
    Deploy a variant to an environment.

    Args:
        app_name: Application name
        variant_name: Variant to deploy
        environment: Target environment

    Returns:
        Deployment details
    """
    # Get variant
    variants = self.client.list_variants(app_name=app_name)
    variant = next((v for v in variants if v.name == variant_name), None)

    if not variant:
        raise ValueError(f"Variant '{variant_name}' not found")

    # Deploy
    return self.client.deploy_variant(
        variant_id=variant.id,
        environment=environment
    )

def get_deployment_status(self, app_name: str) -> Dict:
    """
    Get deployment status for an application.

    Args:
        app_name: Application name

    Returns:
        Deployment status
    """
    response = requests.get(
        f"{self.base_url}/api/apps/{app_name}/deployments",
        headers={"Authorization": f"Bearer {self.config.api_key}"} if self.config.api_key else {}
    )

    return response.json()

def configure_observability(
    self,
    tracing_endpoint: str = None,
    metrics_endpoint: str = None
) -> None:
    """
    Configure observability endpoints.

    Args:
        tracing_endpoint: Endpoint for traces (e.g., Jaeger)
        metrics_endpoint: Endpoint for metrics (e.g., Prometheus)
    """
    config = {}

    if tracing_endpoint:
        config["tracing"] = {
            "enabled": True,
            "endpoint": tracing_endpoint
        }

    if metrics_endpoint:
        config["metrics"] = {
            "enabled": True,
            "endpoint": metrics_endpoint
        }

    response = requests.post(
        f"{self.base_url}/api/config/observability",
        json=config,
        headers={"Authorization": f"Bearer {self.config.api_key}"} if self.config.api_key else {}
    )

    if response.status_code != 200:
        raise Exception(f"Failed to configure observability: {response.text}")

def generate_docker_compose(config: DeploymentConfig) -> str: """ Generate docker-compose.yml for self-hosted deployment.

Args:
    config: Deployment configuration

Returns:
    Docker compose YAML content
"""
compose = f"""version: '3.8'

services: agenta-backend: image: ghcr.io/agenta-ai/agenta-backend:latest ports: - "{config.port}:8000" environment: - DATABASE_URL={config.database_url} - REDIS_URL={config.redis_url or "redis://redis:6379"} - ENABLE_TRACING={str(config.enable_tracing).lower()} depends_on: - postgres - redis

agenta-frontend: image: ghcr.io/agenta-ai/agenta-frontend:latest ports: - "3000:3000" environment: - NEXT_PUBLIC_API_URL=http://agenta-backend:8000

postgres: image: postgres:15 environment: - POSTGRES_DB=agenta - POSTGRES_USER=agenta - POSTGRES_PASSWORD=agenta_password volumes: - postgres_data:/var/lib/postgresql/data

redis: image: redis:7 volumes: - redis_data:/data

volumes: postgres_data: redis_data: """ return compose

Usage

config = DeploymentConfig( host="localhost", port=8000, api_key=None, # Optional for local deployment database_url="postgresql://agenta:agenta_password@postgres:5432/agenta", redis_url="redis://redis:6379", enable_tracing=True )

Generate docker-compose

compose_yaml = generate_docker_compose(config) print("Docker Compose Configuration:") print(compose_yaml)

Initialize manager (after deploying with docker-compose)

manager = SelfHostedManager(config)

if manager.initialize():

print("Connected to self-hosted Agenta!")

# Create app

app = manager.create_app("my-llm-app", "Production LLM application")

# Deploy variant

deployment = manager.deploy_variant("my-llm-app", "v1", "production")

print(f"Deployed: {deployment}")

Integration Examples

FastAPI Integration

""" Integrate Agenta with FastAPI for production deployments. """ from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional import agenta as ag from agenta import Agenta

app = FastAPI(title="Agenta-Powered API")

Initialize Agenta

ag.init() client = Agenta()

class QueryRequest(BaseModel): """Request model for queries.""" input: str variant: Optional[str] = None parameters: Optional[dict] = None

class QueryResponse(BaseModel): """Response model.""" output: str variant_used: str latency: float

@app.post("/generate", response_model=QueryResponse) async def generate(request: QueryRequest): """Generate response using Agenta-managed prompts.""" import time

try:
    # Get variant (default or specified)
    if request.variant:
        variant = client.get_variant_by_name(
            app_name="production-app",
            variant_name=request.variant
        )
    else:
        variant = client.get_default_variant(app_name="production-app")

    # Get prompt template
    template = variant.config.get("template", "{input}")
    prompt = template.format(input=request.input)

    # Get parameters
    params = variant.config.get("parameters", {})
    if request.parameters:
        params.update(request.parameters)

    # Generate
    start_time = time.time()
    response = ag.llm.complete(prompt=prompt, **params)
    latency = time.time() - start_time

    return QueryResponse(
        output=response.text,
        variant_used=variant.name,
        latency=latency
    )

except Exception as e:
    raise HTTPException(status_code=500, detail=str(e))

@app.get("/variants") async def list_variants(): """List available variants.""" variants = client.list_variants(app_name="production-app") return [{"name": v.name, "id": v.id, "is_default": v.is_default} for v in variants]

Run with: uvicorn api:app --reload

Langchain Integration

""" Use Agenta for prompt management in Langchain applications. """ import agenta as ag from agenta import Agenta from langchain_core.prompts import PromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser from typing import Dict, Any

class AgentaPromptLoader: """ Load prompts from Agenta into Langchain. """

def __init__(self, app_name: str):
    self.app_name = app_name
    self.client = Agenta()
    self._cache: Dict[str, PromptTemplate] = {}

def get_prompt(
    self,
    variant_name: str = None,
    use_cache: bool = True
) -> PromptTemplate:
    """
    Get a Langchain PromptTemplate from Agenta.

    Args:
        variant_name: Variant to load (None for default)
        use_cache: Whether to use cached prompts

    Returns:
        Langchain PromptTemplate
    """
    cache_key = variant_name or "default"

    if use_cache and cache_key in self._cache:
        return self._cache[cache_key]

    # Get variant from Agenta
    if variant_name:
        variant = self.client.get_variant_by_name(
            app_name=self.app_name,
            variant_name=variant_name
        )
    else:
        variant = self.client.get_default_variant(app_name=self.app_name)

    # Create Langchain prompt
    template = variant.config.get("template", "{input}")
    prompt = PromptTemplate.from_template(template)

    # Cache
    self._cache[cache_key] = prompt

    return prompt

def create_chain(
    self,
    variant_name: str = None,
    model: str = "gpt-4",
    temperature: float = 0.3
):
    """
    Create a Langchain chain from Agenta prompt.

    Args:
        variant_name: Variant to use
        model: Model name
        temperature: Temperature setting

    Returns:
        Langchain chain
    """
    prompt = self.get_prompt(variant_name)
    llm = ChatOpenAI(model=model, temperature=temperature)

    return prompt | llm | StrOutputParser()

Usage

ag.init()

loader = AgentaPromptLoader("qa-app")

Get prompt template

prompt = loader.get_prompt("concise-v1") print(f"Template: {prompt.template}")

Create and use chain

chain = loader.create_chain(variant_name="detailed-v2") result = chain.invoke({"input": "What is machine learning?"}) print(f"Result: {result}")

Best Practices

1. Prompt Versioning Strategy

"""Best practices for prompt versioning."""

DO: Use semantic versioning for prompts

version_naming = { "v1.0.0": "Initial production version", "v1.1.0": "Added context handling", "v1.1.1": "Fixed edge case in formatting", "v2.0.0": "Major rewrite with new approach" }

DO: Include metadata with versions

def create_versioned_prompt(name: str, template: str, metadata: dict): return { "name": name, "template": template, "metadata": { "created_by": metadata.get("author"), "description": metadata.get("description"), "changelog": metadata.get("changelog"), "test_results": metadata.get("test_results") } }

DO: Test before promoting to production

def promote_to_production(variant_id: str, min_eval_score: float = 0.8): # Run evaluation score = run_evaluation(variant_id)

if score >= min_eval_score:
    client.set_default_variant(variant_id)
    return True
return False

2. Evaluation Strategy

"""Best practices for prompt evaluation."""

DO: Define clear evaluation criteria

evaluation_criteria = { "accuracy": {"weight": 0.4, "threshold": 0.8}, "relevance": {"weight": 0.3, "threshold": 0.7}, "coherence": {"weight": 0.2, "threshold": 0.7}, "safety": {"weight": 0.1, "threshold": 0.9} }

DO: Use diverse test sets

def create_evaluation_set(): return [ {"input": "...", "expected": "...", "category": "basic"}, {"input": "...", "expected": "...", "category": "edge_case"}, {"input": "...", "expected": "...", "category": "adversarial"} ]

DO: Track evaluation over time

def track_evaluation_history(app_name: str, variant_id: str, results: dict): # Store results with timestamp for trend analysis pass

3. A/B Testing Guidelines

"""Best practices for A/B testing prompts."""

DO: Calculate required sample size

def calculate_sample_size( baseline_metric: float, minimum_detectable_effect: float, alpha: float = 0.05, power: float = 0.8 ) -> int: # Statistical calculation for required samples pass

DO: Use proper statistical tests

def analyze_ab_test(control_results: list, treatment_results: list): from scipy import stats

# T-test for continuous metrics
t_stat, p_value = stats.ttest_ind(control_results, treatment_results)

return {
    "significant": p_value < 0.05,
    "p_value": p_value,
    "effect_size": (sum(treatment_results)/len(treatment_results) -
                   sum(control_results)/len(control_results))
}

Troubleshooting

Connection Issues

Problem: Cannot connect to Agenta host

Solution: Verify host and network settings

def diagnose_connection(host: str): import requests

try:
    response = requests.get(f"{host}/api/health", timeout=5)
    if response.status_code == 200:
        print("Connection successful")
    else:
        print(f"Server returned: {response.status_code}")
except requests.exceptions.ConnectionError:
    print("Cannot reach server - check host/port")
except requests.exceptions.Timeout:
    print("Connection timed out - server may be overloaded")

Evaluation Failures

Problem: Evaluations failing or inconsistent

Solution: Add retry logic and validation

def robust_evaluation(prompt: str, max_retries: int = 3): for attempt in range(max_retries): try: result = ag.llm.complete(prompt=prompt) if validate_result(result): return result except Exception as e: if attempt == max_retries - 1: raise time.sleep(2 ** attempt)

Version Conflicts

Problem: Multiple team members editing same variant

Solution: Use branching strategy

def create_branch_variant(base_variant: str, branch_name: str): # Clone variant for isolated development base = client.get_variant_by_name(app_name, base_variant) return client.create_variant( app_name=app_name, variant_name=f"{base_variant}-{branch_name}", config=base.config )

Resources

• Agenta Documentation: https://docs.agenta.ai/

• GitHub Repository: https://github.com/Agenta-AI/agenta

• Self-Hosting Guide: https://docs.agenta.ai/self-hosting

• API Reference: https://docs.agenta.ai/api-reference

Version History

• 1.0.0 (2026-01-17): Initial release with versioning, A/B testing, evaluation, playground, model comparison, self-hosting

This skill provides comprehensive patterns for LLM prompt management with Agenta, refined from production prompt engineering workflows.