Skip to content

Latest commit

 

History

History
736 lines (520 loc) · 18.5 KB

File metadata and controls

736 lines (520 loc) · 18.5 KB

API Reference

This document covers every public function and class in diffprompt. All types are Pydantic models unless noted otherwise.


Data models

All models live in diffprompt.models.


TestCategory

class TestCategory(str, Enum):
    TYPICAL     = "typical"
    BOUNDARY    = "boundary"
    ADVERSARIAL = "adversarial"
    FORMAT      = "format"

The four taxonomy buckets used for test generation.

  • TYPICAL - realistic everyday inputs representing actual usage
  • BOUNDARY - inputs at the edge of what the prompt handles; too long, too short, tangentially related
  • ADVERSARIAL - inputs designed to expose failures; ambiguous, contradictory, trick questions
  • FORMAT - unusual formatting; ALL CAPS, no punctuation, emojis, very short inputs

Verdict

class Verdict(str, Enum):
    IMPROVEMENT = "improvement"
    REGRESSION  = "regression"
    NEUTRAL     = "neutral"

The judgment produced by the judge LLM for each output pair.


TestCase

class TestCase(BaseModel):
    id: str
    input: str
    category: TestCategory
    tags: dict[str, str]

A single test input. Created by generate_test_cases(), tagged by Ontology.tag().

  • id - short unique identifier (8 characters)
  • input - the raw input string sent to the LLM
  • category - which taxonomy bucket this came from
  • tags - dimension-to-value mapping; e.g. {"tone": "emotional", "complexity": "simple"}

RunResult

class RunResult(BaseModel):
    test_id: str
    prompt_version: str
    output: str
    model_used: str
    latency_ms: float | None

The output from running one test case through one prompt version.

  • test_id - matches TestCase.id; used for joining v1 and v2 results
  • prompt_version - "v1" or "v2"
  • model_used - the model that produced this output; e.g. "groq/llama-3.3-70b-versatile"
  • latency_ms - time in milliseconds for the LLM call

DiffResult

class DiffResult(BaseModel):
    test_case: TestCase
    v1_output: str
    v2_output: str
    similarity: float
    divergence: float
    verdict: Verdict
    reason: str
    judge_confidence: float
    importance_score: float
    cluster_label: int
    cluster_centrality: float

The core analysis unit. One per test case. Assembled in cli.py from embedder and judge outputs.

  • similarity - cosine similarity between v1 and v2 outputs; 0.0 to 1.0
  • divergence - 1 - similarity; how much the outputs differ
  • verdict - improvement, regression, or neutral
  • reason - one-sentence explanation from the judge LLM
  • judge_confidence - how certain the judge was; 0.0 to 1.0
  • importance_score - computed by scorer.importance_score(); used for ranking key examples
  • cluster_label - assigned by cluster_diffs(); -1 means unclustered noise
  • cluster_centrality - how central this diff is within its cluster; populated by clusterer

SliceResult

class SliceResult(BaseModel):
    dimension: str
    value: str
    label: str
    n: int
    mean_similarity: float
    variance: float
    typical_ratio: float
    confidence: float
    verdict: Verdict
    depth: int

Performance summary for one behavioral slice.

  • dimension - the tag dimension; e.g. "tone"
  • value - the tag value; e.g. "emotional"
  • label - "{dimension}:{value}"; e.g. "tone:emotional"
  • n - number of test cases in this slice
  • mean_similarity - average similarity across all diffs in this slice
  • variance - variance of similarity scores; high variance means inconsistent behavior
  • typical_ratio - fraction of diffs from TYPICAL bucket; affects confidence
  • confidence - reliability of this slice's verdict; computed from variance, typical_ratio, and n
  • depth - 1 for top-level slices; 2 or 3 for recursively split sub-slices

Cluster

class Cluster(BaseModel):
    label: int
    name: str
    description: str
    n: int
    mean_similarity: float
    test_ids: list[str]

A named failure mode; a group of diffs with similar judge reasons.

  • label - HDBSCAN cluster label (integer)
  • name - auto-generated name; e.g. "CONTEXT_LOSS", "TONE_SHIFT", "REFUSAL_SHIFT"
  • description - first 120 characters of the combined reasons from the top 3 diffs
  • test_ids - list of TestCase.id values in this cluster

KeyExample

class KeyExample(BaseModel):
    slot: str
    diff: DiffResult
    why_it_matters: str

One of the three highlighted examples in the output.

  • slot - "most_important", "best_improvement", or "most_surprising"
  • diff - the full DiffResult for this example
  • why_it_matters - one-sentence explanation generated by the scorer LLM

DiffReport

class DiffReport(BaseModel):
    prompt_v1: str
    prompt_v2: str
    model: str
    judge: str
    test_cases: list[TestCase]
    diversity_score: float
    diffs: list[DiffResult]
    slices: list[SliceResult]
    clusters: list[Cluster]
    unclustered: list[DiffResult]
    key_examples: list[KeyExample]
    regression_score: float
    n_improved: int
    n_regressed: int
    n_neutral: int
    verdict: Verdict
    recommendation: str

The final report. Contains everything produced by the pipeline. Passed to the output layer for rendering.

  • diversity_score - how diverse the test suite is; 0.0 to 1.0. Below 0.4 triggers a warning.
  • regression_score - overall score 0 to 100. 100 means v2 improves everywhere. 0 means v2 regresses everywhere.
  • recommendation - plain English summary of what to do

Core functions


diffprompt.core.ontology

class Ontology:
    dimensions: dict[str, list[str]]
    anchors: dict[str, dict[str, str]]

Manages prompt-specific dimensions and input tagging.


Ontology.infer

async def infer(self, prompt: str, local_only: bool = False) -> None

Calls the LLM once to infer relevant dimensions for this prompt. Populates self.dimensions.

  • prompt - the prompt to analyze
  • local_only - if True, never call external APIs

Ontology.build_anchors

async def build_anchors(self, prompt: str, local_only: bool = False) -> None

Builds anchor phrases for each tag using zero-shot label embedding. No LLM calls. Must be called after infer().


Ontology.tag

def tag(self, input_text: str) -> dict[str, str]

Tags a single input by comparing it to anchor phrase embeddings. Returns a dimension-to-value mapping. No LLM call; pure embedding comparison.

ontology.tag("I've been feeling anxious lately")
# {"tone": "emotional", "complexity": "simple", "intent": "seeking-support"}

Ontology.to_dict / from_dict

def to_dict(self) -> dict
@classmethod
def from_dict(cls, data: dict) -> Ontology

Serialize and deserialize the ontology for caching to disk.


diffprompt.core.generator

generate_test_cases

async def generate_test_cases(
    prompt: str,
    n: int = 40,
    ontology: Ontology | None = None,
    local_only: bool = False,
) -> list[TestCase]

Generates n test cases distributed across the four taxonomy buckets (45% typical, 35% adversarial, 10% boundary, 10% format). Each test case is tagged using the ontology if provided.

  • n - total number of test cases to generate. Actual count may differ slightly due to rounding.
  • ontology - if provided, tags each generated input. If None, tags are empty.

diversity_score

def diversity_score(test_cases: list[TestCase]) -> float

Computes how diverse the test suite is. Returns 1 - mean_pairwise_similarity. Higher is more diverse. A score below 0.4 means many inputs are semantically redundant.


diffprompt.core.embedder

get_embedder

@lru_cache(maxsize=1)
def get_embedder() -> SentenceTransformer

Returns the cached all-MiniLM-L6-v2 model. Loads on first call, returns the same instance on all subsequent calls.


embed

def embed(texts: list[str]) -> np.ndarray

Embeds a list of texts. Returns a matrix of shape (len(texts), 384).


similarity

def similarity(text_a: str, text_b: str) -> float

Cosine similarity between two texts. Returns a float between 0.0 and 1.0.


batch_similarity

def batch_similarity(pairs: list[tuple[str, str]]) -> list[float]

Efficient similarity for many pairs. Embeds all texts in one pass. Returns one float per pair, in the same order as input.

scores = batch_similarity([
    ("Paris is the capital of France", "France's capital is Paris"),
    ("Hello world", "Goodbye world"),
])
# [0.94, 0.41]

diffprompt.core.runner

run_single

async def run_single(
    test_case: TestCase,
    prompt: str,
    version: str,
    model: str,
    local_only: bool = False,
) -> RunResult

Runs one test case through one prompt version. Returns a RunResult with the output and latency.

  • prompt - used as the system message
  • test_case.input - used as the user message
  • version - label for the result; "v1" or "v2"

run_both

async def run_both(
    test_cases: list[TestCase],
    prompt_v1: str,
    prompt_v2: str,
    model: str = "groq/llama-3.3-70b-versatile",
    local_only: bool = False,
    concurrency: int = 5,
) -> tuple[dict[str, RunResult], dict[str, RunResult]]

Runs all test cases through both prompts concurrently. Returns two dicts keyed by test_id.

  • concurrency - max concurrent LLM calls. Increase for faster runs, decrease to avoid rate limits.
v1_results, v2_results = await run_both(test_cases, prompt_v1, prompt_v2)
v1_output = v1_results["a3f8b2"].output
v2_output = v2_results["a3f8b2"].output

diffprompt.core.judge

judge_single

async def judge_single(
    test_case: TestCase,
    v1_output: str,
    v2_output: str,
    similarity: float,
    local_only: bool = False,
) -> tuple[Verdict, str, float]

Judges one output pair. Returns (verdict, reason, confidence).

Short-circuits to (NEUTRAL, "outputs are semantically identical", 1.0) if similarity > 0.95. Automatically escalates to the Groq 70B model if confidence is below 0.65.


diffprompt.core.clusterer

cluster_diffs

def cluster_diffs(diffs: list[DiffResult]) -> tuple[list[Cluster], list[DiffResult]]

Clusters diffs by their judge reasons using HDBSCAN + UMAP. Returns (clusters, unclustered) where unclustered contains noise points (HDBSCAN label -1).

Requires hdbscan and umap-learn to be installed. Returns ([], diffs) if fewer than 4 diffs are provided.


diffprompt.core.slicer

compute_slices

def compute_slices(diffs: list[DiffResult]) -> list[SliceResult]

Groups diffs by their tag dimensions and computes performance per slice. Returns slices sorted by mean_similarity ascending (worst first). Recursively splits high-variance slices up to depth 3.

Returns an empty list if diffs have no tags.


diffprompt.core.scorer

regression_score

def regression_score(diffs: list[DiffResult]) -> float

Computes overall regression score from 0 to 100. Weighted by divergence so large behavioral changes matter more than small ones. Returns 50.0 for empty input and 100.0 if all outputs are identical.

Formula: ((weighted_sum / total_weight) + 1) / 2 * 100 where improvements contribute +divergence and regressions contribute -divergence.


importance_score

def importance_score(diff: DiffResult) -> float

Ranks a single diff by how informative it is. Returns a float between 0 and 1.

Formula: 0.4 * divergence + 0.3 * cluster_centrality + 0.3 * surprise where surprise = divergence * (1 - input_length / 50). Short inputs that changed a lot are surprising.


select_key_examples

async def select_key_examples(
    diffs: list[DiffResult],
    top_n: int = 3,
    local_only: bool = False,
) -> list[KeyExample]

Selects up to three key examples and generates a "why it matters" sentence for each via LLM call.

  • Slot 1: Most Important (highest importance_score)
  • Slot 2: Best Improvement (highest divergence among improvements; omitted if none)
  • Slot 3: Most Surprising (high divergence on short input, excluding slot 1)

diffprompt.models.cascade

call_cascade

async def call_cascade(
    prompt: str,
    system: str | None = None,
    local_model: str = "qwen2.5:7b",
    groq_model: str = "llama-3.3-70b-versatile",
    local_only: bool = False,
) -> tuple[str, str]

The main LLM entry point. Tries Ollama first, falls back to Groq. Returns (output, model_used).

  • prompt - the user message
  • system - optional system message
  • local_only - if True, raises RuntimeError when Ollama is unavailable instead of falling back to Groq

Raises RuntimeError if all models fail.


call_ollama

async def call_ollama(
    model: str,
    prompt: str,
    system: str | None = None,
) -> str | None

Calls a local Ollama model. Returns None (never raises) if Ollama is not running or the call fails.


call_groq

async def call_groq(
    model: str,
    prompt: str,
    system: str | None = None,
) -> str | None

Calls the Groq API. Returns None if GROQ_API_KEY is not set or the call fails. Reads the key from os.getenv("GROQ_API_KEY").


evolve

diffprompt evolve never uses an LLM to judge or score outputs. Fitness comes only from deterministic checks and embedding similarity - see diffprompt.core.fitness below. The only LLM call in this path is generation (running a prompt variant to get its output), same as diff's runner.

CheckType

class CheckType(str, Enum):
    REGEX       = "regex"
    JSON_SCHEMA = "json_schema"
    KEYWORD     = "keyword"
    NUMERIC     = "numeric"

Check

class Check(BaseModel):
    type: CheckType
    weight: float               # default 1.0
    pattern: str | None         # regex
    ignore_case: bool           # regex, default False
    schema_: dict | None        # json_schema (yaml/json key: "schema")
    keywords: list[str] | None  # keyword
    match_any: bool             # keyword, default False (False = ALL must be present)
    expected: float | None      # numeric
    tolerance: float            # numeric, default 0.0
    comparator: Literal["eq", "gte", "lte"]  # numeric, default "eq"
    extract_pattern: str | None # numeric - regex w/ one capture group to pull the number from the output

One deterministic, pass/fail (0.0 or 1.0) check against a prompt's output. weight is how much this check counts within its task's score.

GoldenTask

class GoldenTask(BaseModel):
    id: str                        # auto-generated 8-char id if omitted
    input: str
    weight: float                  # default 1.0 - this task's weight in the aggregate fitness
    golden_answer: str | None      # compared via embedding similarity
    golden_answer_weight: float    # default 1.0 - weight of the embedding term within this task
    checks: list[Check]

Must have at least one check or a golden_answer - otherwise there's nothing to score it on (raises ValueError).

EvolveReport

class EvolveReport(BaseModel):
    original_prompt: str
    final_prompt: str
    final_fitness: float
    model: str
    population_size: int
    n_generations_run: int
    stopped_reason: Literal["max_generations", "patience"]
    generations: list[GenerationRecord]
    golden_tasks_path: str
    n_tasks: int

The output of evolve, passed to output.evolve_terminal.render() / output.evolve_exporter.render_html().

GenerationRecord

class GenerationRecord(BaseModel):
    generation: int
    best_fitness: float   # best-of-all-time as of this generation (monotonic, via elitism)
    mean_fitness: float   # this generation's population mean
    best_prompt: str

diffprompt.core.checks

score_check

def score_check(check: Check, output: str) -> float

Evaluates one Check against one output string. Returns 0.0 or 1.0. Pure, deterministic, no LLM.


diffprompt.core.golden_tasks

load_golden_tasks

def load_golden_tasks(path: str) -> list[GoldenTask]

Loads golden tasks from .yaml/.yml (a tasks: list) or .jsonl (one task object per line). Raises FileNotFoundError, or ValueError if the format is unsupported or the file is empty.


diffprompt.core.fitness

score_task

def score_task(task: GoldenTask, output: str) -> float

Weighted blend of the task's checks and (if present) its embedding similarity to golden_answer, normalized by total weight. Returns 0.0–1.0.

fitness

async def fitness(
    prompt: str,
    tasks: list[GoldenTask],
    model: str = "groq/llama-3.3-70b-versatile",
    local_only: bool = False,
    concurrency: int = 5,
) -> float

Runs prompt against every task's input (via runner.run_prompt_on_tasks), scores each output with score_task, and returns the task-weighted average - the number the genetic algorithm selects on. Returns 0.0 for an empty task list.


diffprompt.core.mutate

TRANSFORMS: dict[str, Callable[[list[str], random.Random], list[str]]]

The fixed, hardcoded template menu: reorder, tighten_word_limit, explicit_constraint, add_format, remove_format. Each operates on a prompt as a list of non-empty lines.

mutate

def mutate(prompt: str, rng: random.Random) -> str

Applies one randomly-chosen transform from TRANSFORMS.


diffprompt.core.population

init_population

def init_population(base_prompt: str, n: int, rng: random.Random) -> list[str]

Variant 0 is always base_prompt, unmodified. Variants 1..n-1 each apply one distinct transform from TRANSFORMS, cycling through the menu if n exceeds its size.

crossover

def crossover(parent_a: str, parent_b: str, rng: random.Random) -> str

Single-point splice: head of parent_a's instruction lines + tail of parent_b's.

select

def select(scored: list[tuple[str, float]], k: int) -> list[tuple[str, float]]

Keeps the top-k (prompt, fitness) pairs by fitness, descending.

breed_next_generation

def breed_next_generation(
    survivors: list[tuple[str, float]],
    elite_prompt: str,
    population_size: int,
    rng: random.Random,
    mutation_rate: float = 0.3,
) -> list[str]

Slot 0 is always elite_prompt, unmodified (elitism). The rest are bred from survivors via crossover, then mutated with probability mutation_rate.