agenta

LLM prompt management and evaluation platform. Version prompts, run A/B tests, evaluate with metrics, and deploy with confidence using Agenta's self-hosted solution.

Jump to install

Repository: vamseeachanta/workspace-hub
Last source activity: June 14, 2026 at 09:00
Detected SKILL.md language: English
Stars: 16
Forks: 7

Install options

The review-first prompt is selected by default. You can switch to a direct command or download a local copy.

Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.

A direct command skips the review prompt. Inspect the source before running it.

npx skills add https://github.com/vamseeachanta/workspace-hub --skill agenta

The command stays on one line. Scroll horizontally to inspect it before copying.

Prefer a local copy? Download the files currently available to SkillsMP.

Review the source files

Read SKILL.md and any companion files shown by SkillsMP before deciding whether to install.

File Explorer

2 files

SKILL.md

Source instructions · Read-only preview

Back to search Back to installation

communication-voice

Write outbound email and external messages in Vamsee Achanta's voice — a subtle offer to help, never bold or rash claims. Load before drafting ANY email, LinkedIn/Collide reply, proposal note, or outreach sent under his name.

2026-07-30

hf-dataset-publishing

Save/publish analysis or computation results from ANY ecosystem repo to Hugging Face as a queryable, viewer-renderable dataset. Use when the user wants to "save results to hugging face", "publish dataset to HF", "hugging face data saving", "save analysis results", "hf dataset", "make results queryable", or "render via datasets-server API". Reshapes nested results into flat parquet tables, writes a dataset card with a viewer `configs:` block and provenance, applies license/public-vs-private routing, enforces a domain data-quality gate (faithful-to-source != correct), publishes to `aceengineer/<repo>-<projection>`, and verifies via the datasets-server API.

2026-07-28

github-repo-management

Clone, create, fork, configure, and manage GitHub repositories. Manage remotes, secrets, releases, and workflows. Works with gh CLI or falls back to git + GitHub REST API via curl.

2026-07-23

View all from this repository

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
        """
        
        variant = .client.create_variant(
            app_name=.app_name,
            variant_name=name,
            config={
                : template,
                : parameters  {}
            }
        )

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

     () -> [PromptVersion]:
        
        variants = .client.list_variants(app_name=.app_name)

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

         versions

     () -> :
        
        .client.set_default_variant(
            app_name=.app_name,
            variant_id=version_id
        )

     () -> PromptVersion:
        
        variant = .client.get_variant(variant_id=version_id)

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

     () -> [, ]:
        
        results = {}

         vid  version_ids:
            version = .get_version(vid)

            
            prompt = version.template.(=test_input)

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

         results



manager = PromptManager()


v1 = manager.create_version(
    name=,
    template=,
    parameters={: }
)

v2 = manager.create_version(
    name=,
    template=,
    parameters={: }
)


versions = manager.list_versions()
 v  versions:
    ()


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:  - traffic_split,
                treatment_variant: traffic_split
            },
            metrics=[, , ]
        )

        
         variant  config.variants.keys():
            .results[variant] = []

         config

     () -> :
        
        rand = random.random()
        cumulative = 

         variant_id, percentage  config.variants.items():
            cumulative += percentage
             rand <= cumulative:
                 variant_id

        
         (config.variants.keys())[]

     () -> :
        
         time

        
        variant_id = .route_request(config)
        variant = .client.get_variant(variant_id)

        
        prompt = variant.config.get(, ).(=input_data)

        
        start_time = time.time()
        response = ag.llm.complete(prompt=prompt)
        latency = time.time() - start_time

        result = {
            : variant_id,
            : input_data,
            : response.text,
            : latency,
            : response.usage.total_tokens  (response, )  
        }

        
        .results[variant_id].append(result)

         result

     () -> :
        
        summary = {}

         variant_id, results  .results.items():
              results:
                

            latencies = [r[]  r  results]
            tokens = [r[]  r  results]

            summary[variant_id] = {
                : (results),
                : (latencies) / (latencies),
                : (tokens) / (tokens)  tokens  ,
                : (latencies),
                : (latencies)
            }

         summary

     () -> []:
        
        summary = .get_test_results(config)

        
         variant_id, stats  summary.items():
             stats[] < config.min_samples:
                ()
                 

        
        
        best_variant = (
            summary.keys(),
            key= v: summary[v][]
        )

         best_variant



ag.init()

runner = ABTestRunner()


test_config = runner.create_test(
    name=,
    control_variant=,
    treatment_variant=,
    traffic_split=
)


test_inputs = [
    ,
    ,
    
]

 input_text  test_inputs:
    result = runner.run_request(test_config, input_text)
    ()


results = runner.get_test_results(test_config)
()
 variant, stats  results.items():
    ()

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

     () -> EvaluationResult:
         expected  :
             EvaluationResult(.name, , {: })

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

         EvaluationResult(
            metric_name=.name,
            score=    ,
            details={: }
        )


 ():
    

     ():
        ().__init__()
        .keywords = keywords

     () -> EvaluationResult:
        output_lower = output.lower()
        found = [kw  kw  .keywords  kw.lower()  output_lower]
        score = (found) / (.keywords)

         EvaluationResult(
            metric_name=.name,
            score=score,
            details={
                : found,
                : [kw  kw  .keywords  kw.lower()   output_lower]
            }
        )


 ():
    

     ():
        ().__init__()
        .min_length = min_length
        .max_length = max_length

     () -> EvaluationResult:
        length = (output.split())

         .min_length <= length <= .max_length:
            score = 
         length < .min_length:
            score = length / .min_length
        :
            score = (,  - (length - .max_length) / .max_length)

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


 ():
    

     ():
        ().__init__()
        .criteria = criteria

     () -> EvaluationResult:
        judge_prompt = 

        response = ag.llm.complete(
            prompt=judge_prompt,
            model=,
            temperature=
        )

        :
            result = json.loads(response.text)
            score = (result.get(, ))
            reasoning = result.get(, )
         (json.JSONDecodeError, ValueError):
            score = 
            reasoning = 

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


 :
    

     ():
        .app_name = app_name
        .client = Agenta()
        .metrics: [MetricEvaluator] = []

     () -> :
        
        .metrics.append(metric)
         

     () -> [, EvaluationResult]:
        
        results = {}

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

         results

     () -> [, [EvaluationResult]]:
        
        all_results = {metric.name: []  metric  .metrics}

           test_cases:
            results = .evaluate_single(
                output=.get(, ),
                expected=.get(),
                context=.get()
            )

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

         all_results

     () -> :
        
        summary = {}

         metric_name, results  batch_results.items():
            scores = [r.score  r  results]
            summary[metric_name] = {
                : (scores) / (scores)  scores  ,
                : (scores)  scores  ,
                : (scores)  scores  ,
                : (scores)
            }

         summary



ag.init()


pipeline = EvaluationPipeline()
pipeline.add_metric(ContainsMetric([, ]))
pipeline.add_metric(LengthMetric(min_length=, max_length=))
pipeline.add_metric(LLMJudgeMetric(criteria=))


test_cases = [
    {
        : ,
        : ,
        : 
    },
    {
        : ,
        : ,
        : 
    }
]


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

()
 metric, stats  summary.items():
    ()

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':
        
        .current_prompt = prompt
         

     () -> :
        
        .current_params.update(params)
         

     () -> ExperimentRun:
        
         time
         uuid

        
        formatted_prompt = .current_prompt.(=input_data)

        
        start_time = time.time()
        response = ag.llm.complete(
            prompt=formatted_prompt,
            **.current_params
        )
        latency = time.time() - start_time

        
        run = ExperimentRun(
            run_id=(uuid.uuid4())[:],
            prompt=formatted_prompt,
            parameters=.current_params.copy(),
            output=response.text,
            metrics={
                : latency,
                : (response.text),
                : response.usage.total_tokens  (response, )  
            }
        )

        .experiments.append(run)

         run

     () -> [ExperimentRun]:
        
        runs = []
        original_prompt = .current_prompt
        original_params = .current_params.copy()

         parameters:
            .set_parameters(**parameters)

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

        
        .current_prompt = original_prompt
        .current_params = original_params

         runs

     () -> [ExperimentRun]:
        
        runs = []
        original_value = .current_params.get(param_name)

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

        
         original_value   :
            .current_params[param_name] = original_value
        :
            .current_params.pop(param_name, )

         runs

     () -> [ExperimentRun]:
        
         .experiments[-limit:]

     () -> :
        
        data = []
         exp  .experiments:
            data.append({
                : exp.run_id,
                : exp.prompt,
                : exp.parameters,
                : exp.output,
                : exp.metrics,
                : exp.timestamp.isoformat()
            })

         (filepath, )  f:
            json.dump(data, f, indent=)

     () -> [ExperimentRun]:
        
          .experiments:
             

        valid_runs = [e  e  .experiments  metric  e.metrics]

          valid_runs:
             

         minimize:
             (valid_runs, key= e: e.metrics[metric])
        :
             (valid_runs, key= e: e.metrics[metric])



ag.init()

playground = Playground()


playground.set_prompt()
playground.set_parameters(model=, temperature=, max_tokens=)


run = playground.run()
()
()


comparison_runs = playground.compare(
    prompts=[
        ,
        ,
        
    ],
    test_input=
)

()
 i, run  (comparison_runs):
    ()


temperature_runs = playground.parameter_sweep(
    param_name=,
    values=[, , , ],
    test_input=
)

()
 run  temperature_runs:
    ()


best = playground.find_best_run(metric=, minimize=)
 best:
    ()


playground.export_experiments()

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},
        : {: , : }
    }

     ():
        .models = models  [, ]
        .results: [, [ModelResult]] = {m: []  m  .models}

     () -> :
        
        costs = .MODEL_COSTS.get(model, {: , : })
         (input_tokens /  * costs[] +
                output_tokens /  * costs[])

     () -> [, ModelResult]:
        
        results = {}

         model  .models:
            start_time = time.time()

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

                latency = time.time() - start_time

                
                input_tokens = (prompt.split()) *   
                output_tokens = (response.text.split()) * 

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

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

             Exception  e:
                result = ModelResult(
                    model=model,
                    output=,
                    latency=,
                    tokens=,
                    cost=
                )

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

         results

     () -> [, ]:
        
         prompt  prompts:
            .run_comparison(prompt, temperature)

         .get_summary()

     () -> [, ]:
        
        summary = {}

         model, results  .results.items():
              results:
                

            valid_results = [r  r  results  r.latency > ]

              valid_results:
                

            summary[model] = {
                : (valid_results),
                : (r.latency  r  valid_results) / (valid_results),
                : (r.tokens  r  valid_results) / (valid_results),
                : (r.cost  r  valid_results),
                : (r.latency  r  valid_results),
                : (r.latency  r  valid_results)
            }

         summary

     () -> :
        
        summary = .get_summary()

          summary:
             .models[]

         priority == :
             (summary.keys(), key= m: summary[m][])
         priority == :
             (summary.keys(), key= m: summary[m][])
         priority == :
            
            quality_order = [, , , , ]
             model  quality_order:
                 model  summary:
                     model
        :  
            
            scores = {}
            max_latency = (s[]  s  summary.values())
            max_cost = (s[]  s  summary.values())  

             model, stats  summary.items():
                norm_latency = stats[] / max_latency
                norm_cost = stats[] / max_cost
                scores[model] = norm_latency *  + norm_cost * 

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

         .models[]



ag.init()

comparator = ModelComparator(models=[, ])


results = comparator.run_comparison()

()
 model, result  results.items():
    ()
    ()
    ()
    ()
    ()


benchmark_prompts = [
    ,
    ,
    
]

comparator.run_benchmark(benchmark_prompts)

()
summary = comparator.get_summary()
 model, stats  summary.items():
    ()
    ()
    ()
    ()


recommended = comparator.recommend_model(priority=)
()

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[] = .base_url
             .config.api_key:
                os.environ[] = .config.api_key

            
            ag.init()
            .client = Agenta()

            
            response = requests.get()
             response.status_code == 

         Exception  e:
            ()
             

     () -> :
        
         .client.create_app(
            name=name,
            description=description
        )

     () -> :
        
        
        variants = .client.list_variants(app_name=app_name)
        variant = ((v  v  variants  v.name == variant_name), )

          variant:
             ValueError()

        
         .client.deploy_variant(
            variant_id=variant.,
            environment=environment
        )

     () -> :
        
        response = requests.get(
            ,
            headers={: }  .config.api_key  {}
        )

         response.json()

     () -> :
        
        config = {}

         tracing_endpoint:
            config[] = {
                : ,
                : tracing_endpoint
            }

         metrics_endpoint:
            config[] = {
                : ,
                : metrics_endpoint
            }

        response = requests.post(
            ,
            json=config,
            headers={: }  .config.api_key  {}
        )

         response.status_code != :
             Exception()


 () -> :
    
    compose = 
     compose



config = DeploymentConfig(
    host=,
    port=,
    api_key=,  
    database_url=,
    redis_url=,
    enable_tracing=
)


compose_yaml = generate_docker_compose(config)
()
(compose_yaml)

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
            )
        :
            variant = client.get_default_variant(app_name=)

        
        template = variant.config.get(, )
        prompt = template.(=request.)

        
        params = variant.config.get(, {})
         request.parameters:
            params.update(request.parameters)

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

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

     Exception  e:
         HTTPException(status_code=, detail=(e))



  ():
    
    variants = client.list_variants(app_name=)
     [{: v.name, : v., : v.is_default}  v  variants]

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]

        
         variant_name:
            variant = .client.get_variant_by_name(
                app_name=.app_name,
                variant_name=variant_name
            )
        :
            variant = .client.get_default_variant(app_name=.app_name)

        
        template = variant.config.get(, )
        prompt = PromptTemplate.from_template(template)

        
        ._cache[cache_key] = prompt

         prompt

     ():
        
        prompt = .get_prompt(variant_name)
        llm = ChatOpenAI(model=model, temperature=temperature)

         prompt | llm | StrOutputParser()



ag.init()

loader = AgentaPromptLoader()


prompt = loader.get_prompt()
()


chain = loader.create_chain(variant_name=)
result = chain.invoke({: })
()

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

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.

name	agenta
description	LLM prompt management and evaluation platform. Version prompts, run A/B tests, evaluate with metrics, and deploy with confidence using Agenta's self-hosted solution.
version	1.0.0
author	workspace-hub
category	ai-prompting
type	skill
trigger	manual
auto_execute	false
capabilities