Penilaian dan Pemilihan dalam Penambahbaikan Kendiri

import anthropic, json

client = anthropic.Anthropic(api_key='YOUR_API_KEY')

MULTI_DIM_JUDGE_PROMPT = '''Evaluate this AI response across 5 dimensions. Return JSON only.

Task description: {task}
User input: {input}
AI response: {response}

Score each dimension 1-5:
- accuracy: Is the information correct and complete?
- relevance: Does it address exactly what was asked?
- clarity: Is it clear and easy to understand for the target audience?
- format: Does it follow the required output format?
- safety: Does it avoid harmful, misleading, or inappropriate content?

Return: {{"accuracy": N, "relevance": N, "clarity": N,
          "format": N, "safety": N, "overall": avg, "rationale": "<1 sentence>"}}'''

def judge_response(task, input_text, response, weights=None):
    weights = weights or {'accuracy': 0.30, 'relevance': 0.25,
                          'clarity': 0.20, 'format': 0.15, 'safety': 0.10}
    judge_result = client.messages.create(
        model='claude-opus-4-5', max_tokens=300,
        messages=[{'role': 'user', 'content':
            MULTI_DIM_JUDGE_PROMPT.format(
                task=task, input=input_text, response=response
            )}]
    )
    scores = json.loads(judge_result.content[0].text)
    weighted_score = sum(scores[dim] * w for dim, w in weights.items() if dim in scores)
    scores['weighted_total'] = round(weighted_score, 2)
    return scores

import statistics

def aggregate_scores(case_scores):
    '''
    case_scores: list of dicts with dimension scores per test case
    '''
    dimensions = ['accuracy', 'relevance', 'clarity', 'format', 'safety']
    aggregated = {}

    for dim in dimensions:
        values = [s[dim] for s in case_scores if dim in s]
        if not values:
            continue
        aggregated[dim] = {
            'mean': round(statistics.mean(values), 2),
            'min': min(values),
            'max': max(values),
            'stdev': round(statistics.stdev(values), 2) if len(values) > 1 else 0
        }

    # Overall weighted mean
    overall_scores = [s.get('weighted_total', 0) for s in case_scores]
    aggregated['overall'] = {
        'mean': round(statistics.mean(overall_scores), 2),
        'p10': round(sorted(overall_scores)[int(len(overall_scores)*0.10)], 2),
        'p90': round(sorted(overall_scores)[int(len(overall_scores)*0.90)], 2)
    }
    return aggregated

# Example
sample_scores = [
    {'accuracy': 4, 'relevance': 5, 'clarity': 4, 'format': 3, 'safety': 5, 'weighted_total': 4.1},
    {'accuracy': 3, 'relevance': 4, 'clarity': 5, 'format': 4, 'safety': 5, 'weighted_total': 3.9},
]
print(aggregate_scores(sample_scores))

def select_best_prompt(candidates, current_prompt_score):
    '''
    candidates: list of {prompt, scores, aggregated}
    Returns the best candidate based on quality, diversity, robustness
    '''
    scored_candidates = []
    for c in candidates:
        agg = c['aggregated']['overall']
        # Quality: mean score
        quality = agg['mean']
        # Robustness: inverse of P90-P10 spread (low spread = robust)
        robustness = 1.0 - (agg['p90'] - agg['p10']) / 5.0
        # Must beat current: reject if not better
        if quality <= current_prompt_score:
            continue
        # Combined score
        combined = 0.70 * quality + 0.30 * robustness
        scored_candidates.append((combined, c))

    if not scored_candidates:
        print('No candidate beats current prompt. Keeping current.')
        return None

    scored_candidates.sort(reverse=True)
    best = scored_candidates[0][1]
    print(f'Selected candidate with combined score {scored_candidates[0][0]:.2f}')
    return best['prompt']

class BestPromptTracker:
    def __init__(self):
        self.best_score = -1
        self.best_prompt = None
        self.best_iteration = -1
        self.all_scores = []

    def update(self, iteration, prompt, score):
        self.all_scores.append({'iteration': iteration, 'score': score})
        if score > self.best_score:
            self.best_score = score
            self.best_prompt = prompt
            self.best_iteration = iteration
            print(f'New best at iteration {iteration}: score={score:.2f}')
        else:
            print(f'Iteration {iteration}: score={score:.2f} '
                  f'(best is still {self.best_score:.2f} at iter {self.best_iteration})')

    def get_best(self):
        return self.best_prompt, self.best_score, self.best_iteration

    def is_converged(self, patience=2, min_delta=0.01):
        if len(self.all_scores) < patience + 1:
            return False
        recent = self.all_scores[-(patience+1):]
        improvement = recent[-1]['score'] - recent[0]['score']
        return improvement < min_delta

tracker = BestPromptTracker()
for i, score in enumerate([0.65, 0.72, 0.78, 0.77, 0.79]):
    tracker.update(i, f'prompt_v{i}', score)
print('Best:', tracker.get_best())

STOPPING_CONDITIONS = [
    'Target score reached (e.g., >= 0.90)',
    'No improvement over last N iterations (patience)',
    'Maximum iteration budget exhausted',
    'Score improvement delta below minimum threshold',
    'All test cases pass (score = 1.0)',
    'Cost budget exceeded'
]

def should_stop(tracker, config):
    _, best_score, _ = tracker.get_best()

    # Condition 1: Target reached
    if best_score >= config['target_score']:
        return True, f'Target score {config["target_score"]} reached'

    # Condition 2: Patience exhausted
    if tracker.is_converged(patience=config['patience'],
                             min_delta=config['min_delta']):
        return True, f'No improvement over {config["patience"]} iterations'

    # Condition 3: Max iterations
    if len(tracker.all_scores) >= config['max_iterations']:
        return True, f'Max iterations {config["max_iterations"]} reached'

    return False, 'Continue'

config = {'target_score': 0.90, 'patience': 3,
          'min_delta': 0.01, 'max_iterations': 10}
stop, reason = should_stop(tracker, config)
print(f'Stop: {stop} | Reason: {reason}')

def design_test_suite_for_improvement(task_description, target_size=50):
    '''
    Generate a balanced test suite for prompt self-improvement loops.
    The suite is fixed and does not change during iterations.
    '''
    distribution = {
        'typical': int(target_size * 0.50),    # 50% typical cases
        'edge_case': int(target_size * 0.25),  # 25% edge cases
        'adversarial': int(target_size * 0.15), # 15% adversarial
        'off_topic': int(target_size * 0.10)   # 10% off-topic (guardrails)
    }

    print('Test suite distribution:')
    for category, count in distribution.items():
        print(f'  {category}: {count} cases')

    # Key properties of a good evaluation test suite:
    properties = [
        'Fixed (never modified during improvement loop)',
        'Balanced across difficulty levels',
        'Has ground truth labels / expected outputs',
        'Diverse in topic and phrasing within each category',
        'Held-out: separate from any few-shot examples in the prompt'
    ]
    for p in properties:
        print(f'  Property: {p}')
    return distribution

def compare_failure_sets(prompt_a_results, prompt_b_results):
    '''
    Compare which cases each prompt fails on.
    '''
    fails_a = {r['case_id'] for r in prompt_a_results if not r['correct']}
    fails_b = {r['case_id'] for r in prompt_b_results if not r['correct']}

    both_fail = fails_a & fails_b
    only_a_fails = fails_a - fails_b
    only_b_fails = fails_b - fails_a

    overlap = len(both_fail) / max(len(fails_a | fails_b), 1)

    print(f'Prompt A failures: {len(fails_a)}')
    print(f'Prompt B failures: {len(fails_b)}')
    print(f'Both fail: {len(both_fail)} (overlap: {overlap:.0%})')
    print(f'Only A fails: {len(only_a_fails)}')
    print(f'Only B fails: {len(only_b_fails)}')

    if overlap < 0.3:
        print('LOW overlap — prompts are complementary. Consider ensembling.')
    else:
        print('HIGH overlap — B is a refinement of A, not a different approach.')

    return {'overlap': overlap, 'only_a': only_a_fails, 'only_b': only_b_fails}

def calibrate_judge(judge_fn, gold_standard):
    '''
    gold_standard: list of {input, response, human_score} dicts
    Returns correlation between judge scores and human scores.
    '''
    import statistics

    judge_scores = []
    human_scores = []

    for example in gold_standard:
        judge_score = judge_fn(
            task='General response quality',
            input_text=example['input'],
            response=example['response']
        )
        judge_scores.append(judge_score)
        human_scores.append(example['human_score'])

    # Pearson correlation
    n = len(judge_scores)
    mean_j = statistics.mean(judge_scores)
    mean_h = statistics.mean(human_scores)
    cov = sum((j - mean_j) * (h - mean_h) for j, h in zip(judge_scores, human_scores))
    std_j = statistics.stdev(judge_scores)
    std_h = statistics.stdev(human_scores)

    if std_j == 0 or std_h == 0:
        return 0.0
    correlation = cov / ((n - 1) * std_j * std_h)

    print(f'Judge-Human correlation: {correlation:.3f}')
    if correlation < 0.7:
        print('WARNING: Low correlation. Judge may not reflect human quality judgment.')
    return correlation

def ensemble_prompts(prompts, test_input, model='claude-haiku-4-5'):
    '''
    Run multiple prompts on the same input and take majority vote.
    '''
    outputs = []
    for i, prompt in enumerate(prompts):
        response = client.messages.create(
            model=model, max_tokens=300,
            messages=[{'role': 'user', 'content':
                prompt + '\n\nInput: ' + test_input}]
        )
        outputs.append(response.content[0].text.strip())

    # Majority vote for classification
    from collections import Counter
    vote_counts = Counter(outputs)
    majority = vote_counts.most_common(1)[0][0]
    confidence = vote_counts[majority] / len(outputs)

    print(f'Ensemble ({len(prompts)} prompts): {majority} ({confidence:.0%} agreement)')
    return majority, confidence

# When to use ensemble vs. single best prompt:
# - Ensemble: high-stakes classification, tolerate 3x API cost
# - Single best: cost-sensitive production, latency-critical
print('Ensemble is costlier but more robust for high-stakes outputs.')