vamseeachanta 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.1": {"input": 0.01, "output": 0.03},
"gpt-4.1-mini": {"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-4.1-mini"]
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.1", "claude-3-sonnet", "gpt-4.1-mini"]
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-4.1-mini"])
# 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:
- post
…(truncated)