def automatic_prompt_engineering(task_description, examples, num_candidates=10):
    # 第一步：生成候选提示词
    generation_prompt = f"""
    你是一位提示词工程专家。请为以下任务生成 {num_candidates} 个不同的提示词：

    任务描述：{task_description}

    示例输入输出：
    {format_examples(examples)}

    要求：
    1. 每个提示词采用不同的策略（角色设定、格式、推理方式等）
    2. 确保提示词清晰、完整
    3. 以 JSON 数组格式输出所有候选
    """

    candidates = llm.generate(generation_prompt)

    # 第二步：评估候选
    scores = []
    for candidate in candidates:
        score = evaluate_prompt(candidate, examples)
        scores.append((candidate, score))

    # 第三步：选择并优化
    best_candidate = max(scores, key=lambda x: x[1])

    # 第四步：微调最优候选
    feedback = analyze_failures(best_candidate[0], examples)
    refined = refine_prompt(best_candidate[0], feedback)

    return refined

def opro_optimization(initial_prompt, task_examples, iterations=5):
    history = []
    current_prompt = initial_prompt

    for i in range(iterations):
        # 评估当前提示词
        score = evaluate(current_prompt, task_examples)
        history.append({
            "prompt": current_prompt,
            "score": score,
            "iteration": i
        })

        # 生成改进建议
        improvement_prompt = f"""
        以下是提示词优化的历史记录：

        {format_history(history)}

        当前最佳提示词得分：{max(h['score'] for h in history)}

        请分析历史数据，生成一个可能获得更高分数的新提示词。
        重点改进：
        1. 分析表现好的提示词的共同特征
        2. 识别表现差的提示词的问题
        3. 结合以上分析生成改进版本
        """

        current_prompt = llm.generate(improvement_prompt)

    # 最后一轮生成的提示词也要评估，否则它永远不会进入 history
    history.append({
        "prompt": current_prompt,
        "score": evaluate(current_prompt, task_examples),
        "iteration": iterations
    })

    return max(history, key=lambda x: x['score'])['prompt']

from llama_index.core import PromptTemplate

# 查询改写模板

query_rewrite_template = PromptTemplate(
    template="""
    原始查询：{query}

    请将这个查询改写为 3 个更具体的子查询，以便更好地检索相关文档。
    输出格式：每行一个子查询。
    """
)

# RAG 回答模板

qa_template = PromptTemplate(
    template="""
    基于以下参考资料回答问题。

    参考资料：
    {context}

    问题：{query}

    要求：
    1. 仅基于提供的资料回答
    2. 如资料不足，明确说明
    3. 引用资料来源
    """
)

print(query_rewrite_template.format(query="量子计算是什么？").strip())

import os
import promptlayer

# 初始化（包装 OpenAI 客户端）

promptlayer.api_key = os.environ["PROMPTLAYER_API_KEY"]
openai = promptlayer.openai

# 调用时自动记录

response = openai.chat.completions.create(
    model="gpt-5.4",
    messages=[{"role": "user", "content": "Hello!"}],
    pl_tags=["test", "greeting"],  # 自定义标签
)

print(response)

# 功能：

# - 所有调用自动记录和可视化

# - 提示词版本管理

# - A/B 测试支持

# - 成本追踪

import wandb
from wandb.sdk.data_types.trace_tree import Trace

# 初始化

wandb.init(project="prompt_engineering")

# 记录提示词实验

trace = Trace(
    name="customer_service_v2",
    kind="llm",
    inputs={"prompt": "..."},
    outputs={"response": "..."},
    metadata={
        "model": "gpt-5.6-sol",
        "temperature": 0.7,
        "version": "2.1.0"
    }
)

# 记录指标

wandb.log({
    "accuracy": 0.85,
    "latency_ms": 1200,
    "cost_usd": 0.02
})

print(trace)

from typing import List, Dict

def evaluate_with_ground_truth(
    prompt: str,
    test_cases: List[Dict],
    llm_client
) -> Dict:
    """基于标准答案评估提示词"""

    results = {
        "correct": 0,
        "total": len(test_cases),
        "details": []
    }

    for case in test_cases:
        # 调用模型
        output = llm_client.generate(
            prompt.format(input=case["input"])
        )

        # 比对答案
        is_correct = normalize(output) == normalize(case["expected"])

        results["details"].append({
            "input": case["input"],
            "expected": case["expected"],
            "actual": output,
            "correct": is_correct
        })

        if is_correct:
            results["correct"] += 1

    results["accuracy"] = results["correct"] / results["total"]
    return results

def reduce_judge_bias(question, response, judge_model):
    """使用位置交换和多次采样减少偏差"""

    scores = []

    # 多次采样
    for _ in range(3):
        score = judge_model.evaluate(question, response)
        scores.append(score)

    # 返回中位数
    return {
        "median_score": np.median(scores),
        "variance": np.var(scores)
    }

from __future__ import annotations

from typing import Dict, List

def ab_comparison(prompt_a: str, prompt_b: str, test_cases: List) -> Dict:
    """A/B 对比评估"""

    results = {"a_wins": 0, "b_wins": 0, "ties": 0}

    for case in test_cases:
        output_a = llm.generate(prompt_a.format(**case))
        output_b = llm.generate(prompt_b.format(**case))

        # 使用 LLM-as-Judge 进行对比
        comparison = judge_llm.compare(
            question=case["question"],
            response_a=output_a,
            response_b=output_b
        )

        if comparison["winner"] == "A":
            results["a_wins"] += 1
        elif comparison["winner"] == "B":
            results["b_wins"] += 1
        else:
            results["ties"] += 1

    return results

from __future__ import annotations

from typing import Dict

def consistency_evaluation(
    prompt: str,
    test_input: str,
    n_runs: int = 10,
    temperature: float = 0.7
) -> Dict:
    """评估输出一致性"""

    outputs = []
    for _ in range(n_runs):
        output = llm.generate(prompt.format(input=test_input))
        outputs.append(output)

    # 计算语义相似度矩阵
    embeddings = [get_embedding(o) for o in outputs]
    similarities = cosine_similarity(embeddings)
    # 只取上三角：矩阵对角线是每个输出与自身的相似度（恒为 1.0），
    # 直接对整个矩阵求均值会被这 n 个 1.0 拉高
    pairwise = similarities[np.triu_indices(len(outputs), k=1)]

    return {
        "mean_similarity": np.mean(pairwise),
        "min_similarity": np.min(pairwise),
        "outputs": outputs
    }

test_suite = {
    "metadata": {
        "name": "客服问答评估集",
        "version": "1.2",
        "created_at": "2026-01-14"
    },
    "test_cases": [
        # 基础案例
        {
            "id": "basic_001",
            "category": "product_inquiry",
            "difficulty": "easy",
            "input": "这款手机支持 5G 吗？",
            "expected": {"answer_contains": ["支持", "5G"]},
            "tags": ["product", "specification"]
        },
        # 边界案例
        {
            "id": "edge_001",
            "category": "out_of_scope",
            "difficulty": "hard",
            "input": "你觉得人生的意义是什么？",
            "expected": {"should_redirect": True},
            "tags": ["off-topic", "boundary"]
        }
    ]
}

from __future__ import annotations

from typing import Dict

class PromptEvaluationPipeline:
    def __init__(self, llm_client, judge_client):
        self.llm = llm_client
        self.judge = judge_client

    def run_evaluation(
        self,
        prompt: str,
        test_suite: Dict,
        config: EvalConfig
    ) -> EvaluationReport:
        """运行完整的评估流程"""

        results = []

        for case in test_suite["test_cases"]:
            # 1. 生成输出
            output = self.llm.generate(prompt.format(**case))

            # 2. 多维度评估
            scores = self.judge.evaluate(
                question=case["input"],
                response=output,
                criteria=config.criteria
            )

            # 3. 记录结果
            results.append({
                "case_id": case["id"],
                "output": output,
                "scores": scores
            })

        # 4. 汇总统计
        return self.aggregate_results(results)

# 生产环境输出采样评估

from __future__ import annotations

import random
from typing import Dict, List

def production_sampling_eval(
    output_log: List[Dict],
    sample_rate: float = 0.01
) -> Dict:
    """对生产输出进行抽样评估"""

    sample = random.sample(
        output_log,
        int(len(output_log) * sample_rate)
    )

    scores = []
    for record in sample:
        score = judge_llm.evaluate(
            record["input"],
            record["output"]
        )
        scores.append(score)

    return {
        "sample_size": len(sample),
        "avg_score": np.mean(scores),
        "score_distribution": get_distribution(scores)
    }

from deepeval import evaluate
from deepeval.metrics import AnswerRelevancyMetric
from deepeval.test_case import LLMTestCase

# 定义测试用例

test_cases = [
    LLMTestCase(
        input="什么是机器学习？",
        actual_output="机器学习是人工智能的一个分支，允许系统从数据中学习。",
        expected_output="机器学习是人工智能的一个分支，允许系统从数据中学习。"
    ),
    LLMTestCase(
        input="解释神经网络如何工作",
        actual_output="神经网络由相互连接的节点组成，模仿生物神经处理信息。",
        expected_output="神经网络由相互连接的节点组成，模仿生物神经..."
    )
]

# 定义评估指标

answer_relevancy_metric = AnswerRelevancyMetric(
    threshold=0.8,
    model="gpt-5.4-mini",
    include_reason=True
)

# 评估

results = evaluate(
    test_cases=test_cases,
    metrics=[answer_relevancy_metric]
)

# 分析结果

print(results)

from typing import List, Dict
from dataclasses import dataclass
from enum import Enum
import json

class EvaluationStatus(Enum):
    """评估状态"""
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class EvaluationConfig:
    """评估配置"""
    prompt_version: str
    test_set_path: str
    metrics: List[str]
    min_pass_rate: float = 0.8
    min_quality_score: float = 7.0
    model: str = "gpt-5.4-mini"
    temperature: float = 0.7
    max_tokens: int = 1024
    timeout_seconds: int = 300

class PromptEvaluationPipeline:
    """提示词评估流水线"""

    def __init__(self, llm_client, storage_client):
        self.llm = llm_client
        self.storage = storage_client
        self.history = []

    def run_evaluation(self, config: EvaluationConfig) -> Dict:
        """运行完整评估"""

        # 1. 加载测试集
        test_cases = self._load_test_set(config.test_set_path)

        # 2. 读取提示词
        prompt_template = self._load_prompt(config.prompt_version)

        # 3. 批量运行测试
        outputs = self._batch_generate(prompt_template, test_cases, config)

        # 4. 评估输出
        metric_results = self._evaluate_outputs(
            outputs,
            test_cases,
            config.metrics
        )

        # 5. 聚合结果
        report = self._aggregate_results(metric_results, config)

        # 6. 持久化结果
        self._save_results(config.prompt_version, report)

        return report

    def _batch_generate(self, prompt_template: str, test_cases: List,
                       config: EvaluationConfig) -> List[str]:
        """批量生成模型输出"""
        outputs = []

        for case in test_cases:
            try:
                # 格式化提示词
                formatted_prompt = prompt_template.format(**case)

                # 调用模型
                response = self.llm.generate(
                    prompt=formatted_prompt,
                    temperature=config.temperature,
                    max_tokens=config.max_tokens
                )

                outputs.append(response)

            except Exception as e:
                print(f"Error processing case {case['id']}: {e}")
                outputs.append(None)

        return outputs

    def _evaluate_outputs(self, outputs: List[str], test_cases: List,
                         metrics: List[str]) -> Dict:
        """评估输出"""
        results = {metric: [] for metric in metrics}

        for output, case in zip(outputs, test_cases):
            if output is None:
                continue

            for metric in metrics:
                if metric == "accuracy":
                    score = self._metric_accuracy(output, case["expected"])
                elif metric == "relevance":
                    score = self._metric_relevance(output, case["input"])
                elif metric == "consistency":
                    score = self._metric_consistency(output)
                elif metric == "safety":
                    score = self._metric_safety(output)
                else:
                    score = 0.0

                results[metric].append(score)

        return results

    def _aggregate_results(self, metric_results: Dict,
                          config: EvaluationConfig) -> Dict:
        """聚合评估结果"""

        import numpy as np

        aggregated = {}
        for metric, scores in metric_results.items():
            aggregated[metric] = {
                "mean": np.mean(scores),
                "std": np.std(scores),
                "min": np.min(scores),
                "max": np.max(scores)
            }

        # 计算是否通过
        pass_rate = np.mean([
            1 if s >= config.min_quality_score else 0
            for scores in metric_results.values()
            for s in scores
        ])

        return {
            "config": config.__dict__,
            "metrics": aggregated,
            "pass_rate": pass_rate,
            "passed": pass_rate >= config.min_pass_rate,
            "timestamp": self._get_timestamp()
        }

    def _save_results(self, version: str, report: Dict):
        """保存评估结果"""
        self.storage.save(
            f"evaluations/{version}/{self._get_timestamp()}.json",
            json.dumps(report, indent=2)
        )
        self.history.append(report)

    # 评估指标实现方法
    def _metric_accuracy(self, output: str, expected: str) -> float:
        """准确性评估"""
        # 实现准确性计算逻辑
        pass

    def _metric_relevance(self, output: str, input_text: str) -> float:
        """相关性评估"""
        # 实现相关性计算逻辑
        pass

    def _metric_consistency(self, output: str) -> float:
        """一致性评估"""
        # 实现一致性计算逻辑
        pass

    def _metric_safety(self, output: str) -> float:
        """安全性评估"""
        # 实现安全性计算逻辑
        pass

    def _get_timestamp(self) -> str:
        from datetime import datetime
        return datetime.now().isoformat()

# 使用示例

config = EvaluationConfig(
    prompt_version="v2.3",
    test_set_path="test_cases.json",
    metrics=["accuracy", "relevance", "consistency", "safety"],
    min_pass_rate=0.85,
    model="gpt-5.4-mini"
)

# 实例化评估管道

pipeline = PromptEvaluationPipeline(llm_client, storage_client)

# 运行评估

report = pipeline.run_evaluation(config)

# 输出结果

if report["passed"]:
    print("✓ 评估通过！可以发布")
else:
    print("✗ 评估失败，需要改进")
    for metric, stats in report["metrics"].items():
        print(f"{metric}: {stats['mean']:.2f} ± {stats['std']:.2f}")

class RegressionTestSuite:
    """回归测试套件"""

    def __init__(self, baseline_version: str):
        self.baseline = self._load_baseline(baseline_version)

    def run_regression_test(self, new_version: str,
                           test_cases: List) -> Dict:
        """对比新版本与基准版本"""

        baseline_results = self.baseline["metrics"]
        new_results = self._evaluate_version(new_version, test_cases)

        regression_report = {
            "improved": [],
            "degraded": [],
            "unchanged": [],
            "overall_change": 0.0
        }

        for metric in baseline_results:
            baseline_score = baseline_results[metric]["mean"]
            new_score = new_results[metric]["mean"]

            change = new_score - baseline_score

            if change > 0.05:  # 改进超过 5%
                regression_report["improved"].append({
                    "metric": metric,
                    "baseline": baseline_score,
                    "new": new_score,
                    "improvement": change
                })
            elif change < -0.05:  # 退化超过 5%
                regression_report["degraded"].append({
                    "metric": metric,
                    "baseline": baseline_score,
                    "new": new_score,
                    "degradation": -change
                })
            else:
                regression_report["unchanged"].append(metric)

        # 检查是否出现回归
        has_regression = len(regression_report["degraded"]) > 0

        regression_report["passed"] = not has_regression

        return regression_report

    def _load_baseline(self, version: str) -> Dict:
        """加载基准版本的评估结果"""
        # 从存储中加载
        pass

    def _evaluate_version(self, version: str, test_cases: List) -> Dict:
        """评估版本"""
        # 使用前面的评估流水线
        pass

# 在 CI/CD 中使用

def ci_check_regression(new_version: str):
    """在部署前检查回归"""

    baseline = "v2.0"  # 当前生产版本
    suite = RegressionTestSuite(baseline)

    # 加载完整测试集
    test_cases = load_full_test_set()

    report = suite.run_regression_test(new_version, test_cases)

    if not report["passed"]:
        print("❌ 检测到性能退化，阻止部署：")
        for item in report["degraded"]:
            print(f"  - {item['metric']}: "
                  f"{item['baseline']:.3f} → {item['new']:.3f} "
                  f"(-{item['degradation']:.1%})")
        return False

    print("✅ 回归测试通过，可以部署")
    return True

class PromptManager:
    def __init__(self, environment="development"):
        self.env = environment
        self.config = load_env_config(environment)

    def get_prompt(self, prompt_name):
        if self.env == "production":
            return load_prompt(f"production/{prompt_name}.yaml")
        elif self.env == "staging":
            return load_prompt(f"staging/{prompt_name}.yaml")
        else:
            return load_prompt(f"development/{prompt_name}.yaml")

    def deploy_to_staging(self, prompt_name, version):
        """将指定版本部署到预发布环境"""
        source = f"versions/{prompt_name}/{version}.yaml"
        target = f"staging/{prompt_name}.yaml"
        symlink(source, target)

    def deploy_to_production(self, prompt_name):
        """将预发布版本提升到生产环境"""
        source = f"staging/{prompt_name}.yaml"
        target = f"production/{prompt_name}.yaml"
        symlink(source, target)
        log_deployment(prompt_name, get_version(source))