W&B Prompts：LLM应用可观测性调试工具

import wandb # 初始化W&B Run，Prompts数据自动关联到此Runrun = wandb.init(project='llm-prompt-experiments', name='gpt4-v3-eval') # 使用wandb.Prompt追踪Prompt调用prompt = wandb.Prompt(    template='请分析以下文本的情感: {text}',    model='gpt-4',    temperature=0.7)response = prompt.call(text='这个产品非常好用')print(f'Response: {response.text}')print(f'Tokens: {response.usage}')run.finish()

import wandbimport time run = wandb.init(project='chatbot-tracing') # 嵌套调用追踪示例with wandb.PromptTracer(name='intent_recognition') as tracer:    intent_result = tracer.call(        template='识别用户意图: {input}',        input='帮我订一张明天去北京的机票'    )        with wandb.PromptTracer(name='entity_extraction', parent=tracer) as sub_tracer:        entity_result = sub_tracer.call(            template='提取实体: {input}',            input=intent_result.text        )        with wandb.PromptTracer(name='response_generation', parent=tracer) as sub_tracer:        final_result = sub_tracer.call(            template='生成回复: {intent} {entities}',            intent=intent_result.text,            entities=entity_result.text        ) # 自动记录调用树结构和每级的Token消耗print(f'Total tokens: {tracer.total_tokens}')print(f'Total latency: {tracer.total_latency_ms}ms')run.finish()

import wandb # 创建带版本管理的Promptprompt_v1 = wandb.Prompt(    name='sentiment_analysis',    template='分析以下评论的情感(正面/负面/中性): {review}',    model='gpt-4',    config={'temperature': 0.3, 'max_tokens': 50}) # 修改Prompt创建新版本prompt_v2 = wandb.Prompt(    name='sentiment_analysis',    template='请判断以下产品评论的情感倾向，只能回答正面、负面或中性: {review}',    model='gpt-4',    config={'temperature': 0.1, 'max_tokens': 30}) # 在W&B Run中对比两个版本run = wandb.init(project='prompt-versioning')test_reviews = ['这个产品太好了！', '质量很差，失望', '一般般吧'] for review in test_reviews:    r1 = prompt_v1.call(review=review)    r2 = prompt_v2.call(review=review)    wandb.log({        'review': review,        'v1_response': r1.text,        'v2_response': r2.text,        'v1_tokens': r1.usage['total_tokens'],        'v2_tokens': r2.usage['total_tokens']    })run.finish()

import wandb # 配置质量监控run = wandb.init(project='quality-monitoring') # 定义评估指标def evaluate_accuracy(response, reference):    return 1.0 if response.strip().lower() == reference.strip().lower() else 0.0 def evaluate_toxicity(response):    toxic_keywords = ['恨', '杀', '死', '蠢']    return 0.0 if any(kw in response for kw in toxic_keywords) else 1.0 # 模拟生产环境监控test_cases = [    {'input': '1+1=?', 'expected': '2'},    {'input': '中国首都是？', 'expected': '北京'},] for case in test_cases:    prompt = wandb.Prompt(template='回答: {input}', model='gpt-4')    result = prompt.call(input=case['input'])        accuracy = evaluate_accuracy(result.text, case['expected'])    toxicity = evaluate_toxicity(result.text)        wandb.log({        'accuracy': accuracy,        'toxicity_score': toxicity,        'latency_ms': result.latency_ms,        'tokens': result.usage['total_tokens']    }) # 设置回归检测告警wandb.alerts.create(    name='accuracy_regression',    condition='accuracy < 0.9',    channel='slack')run.finish()

import wandbimport numpy as np # A/B测试实验设计run = wandb.init(project='prompt-ab-test', config={    'test_name': 'instruction_style_comparison',    'variant_a': 'concise_instruction',    'variant_b': 'detailed_instruction',    'traffic_split': 0.5,    'min_sample_size': 200}) # 模拟A/B测试数据np.random.seed(42)n_samples = 300 # Variant A: 简洁指令a_accuracies = np.random.binomial(1, 0.82, n_samples)# Variant B: 详细指令b_accuracies = np.random.binomial(1, 0.87, n_samples) # 计算统计显著性p_a = a_accuracies.mean()p_b = b_accuracies.mean()p_pooled = (a_accuracies.sum() + b_accuracies.sum()) / (2 * n_samples)se = np.sqrt(p_pooled * (1 - p_pooled) * (2 / n_samples))z_score = (p_b - p_a) / se wandb.log({    'variant_a_accuracy': p_a,    'variant_b_accuracy': p_b,    'accuracy_diff': p_b - p_a,    'z_score': z_score,    'significant': abs(z_score) > 1.96,    'n_samples': n_samples})print(f'A accuracy: {p_a:.3f}, B accuracy: {p_b:.3f}')print(f'Z-score: {z_score:.3f}, Significant: {abs(z_score) > 1.96}')run.finish()

import wandbimport numpy as npimport time class PromptExperimentManager:    """W&B Prompts实验管理器：版本管理、追踪和A/B测试"""     def __init__(self, project_name):        self.project = project_name        self.versions = {}     def create_prompt_version(self, name, template, model, config=None):        version_id = f"{name}_v{len(self.versions.get(name, [])) + 1}"        self.versions.setdefault(name, []).append({            'id': version_id,            'template': template,            'model': model,            'config': config or {}        })        return version_id     def run_ab_test(self, version_a_id, version_b_id, test_data, n_samples=200):        run = wandb.init(project=self.project, name=f'ab_test_{version_a_id}_vs_{version_b_id}')        a_data = self._find_version(version_a_id)        b_data = self._find_version(version_b_id)         results_a = []        results_b = []        for item in test_data[:n_samples]:            start = time.time()            pred_a = self._simulate_call(a_data, item['input'])            latency_a = (time.time() - start) * 1000            results_a.append({'correct': pred_a == item['expected'], 'latency': latency_a})             start = time.time()            pred_b = self._simulate_call(b_data, item['input'])            latency_b = (time.time() - start) * 1000            results_b.append({'correct': pred_b == item['expected'], 'latency': latency_b})         acc_a = np.mean([r['correct'] for r in results_a])        acc_b = np.mean([r['correct'] for r in results_b])        lat_a = np.mean([r['latency'] for r in results_a])        lat_b = np.mean([r['latency'] for r in results_b])         p_pool = (sum(r['correct'] for r in results_a) + sum(r['correct'] for r in results_b)) / (2 * n_samples)        se = np.sqrt(p_pool * (1 - p_pool) * (2 / n_samples)) if p_pool > 0 else 0.001        z = (acc_b - acc_a) / se         wandb.log({            'accuracy_a': acc_a, 'accuracy_b': acc_b,            'latency_a_ms': lat_a, 'latency_b_ms': lat_b,            'z_score': z, 'significant': abs(z) > 1.96        })        run.finish()        return {'acc_a': acc_a, 'acc_b': acc_b, 'z_score': z, 'significant': abs(z) > 1.96}     def _find_version(self, version_id):        for versions in self.versions.values():            for v in versions:                if v['id'] == version_id:                    return v        return None     def _simulate_call(self, version, input_text):        np.random.seed(hash(input_text + version['id']) % 2**32)        base_acc = version['config'].get('base_accuracy', 0.8)        return 'correct' if np.random.random() < base_acc else 'wrong'  mgr = PromptExperimentManager('prompt-experiments')va = mgr.create_prompt_version('qa', 'Answer: {q}', 'gpt-4', {'base_accuracy': 0.82})vb = mgr.create_prompt_version('qa', 'Please answer concisely: {q}', 'gpt-4', {'base_accuracy': 0.87}) test_data = [{'input': f'question_{i}', 'expected': 'correct'} for i in range(300)]result = mgr.run_ab_test(va, vb, test_data, n_samples=200)print(f"A acc: {result['acc_a']:.3f}, B acc: {result['acc_b']:.3f}")print(f"Z: {result['z_score']:.3f}, Significant: {result['significant']}")