from dataclasses import dataclass
from typing import Optional, Dict, List
import time
@dataclass
class ModelSample:
    """统一样本结构，打通评测、线上日志、人工标注"""
    sample_id: str
    query: str                  # 用户输入
    prompt: str                 # 完整提示词
    response: str               # 模型输出
    reference: Optional[str]    # 标准答案（评测使用）
    metadata: Dict              # 业务标签、场景、版本信息
    latency: Optional[float]    # 响应时延ms
    token_input: Optional[int]
    token_output: Optional[int]
    create_time: float = time.time()
    human_label: Optional[int] = None # 人工评分：1坏例，5优秀
class SampleStorage:
    def __init__(self):
        self.sample_list: List[ModelSample] = []
    def add_sample(self, sample: ModelSample):
        self.sample_list.append(sample)
    def export_for_eval(self, path: str):
        """导出数据集用于离线评测"""
        import json
        output = [s.__dict__ for s in self.sample_list]
        with open(path, "w", encoding="utf-8") as f:
            json.dump(output, f, ensure_ascii=False, indent=2)
# 使用示例
if __name__ == "__main__":
    storage = SampleStorage()
    print("=" * 55)
    print("统一样本结构演示：打通评测、日志、标注全流程")
    print("=" * 55)
    # 1. 构建一条完整样本
    print("\n[1] 构建样本 => 包含输入、输出、标准答案、性能指标")
    sample = ModelSample(
        sample_id="s001",
        query="解释什么是大模型回归测试",
        prompt="你是技术专家，请简洁回答用户问题：{query}",
        response="大模型回归测试用于验证模型更新后原有能力不下降",
        reference="回归测试验证模型迭代后，历史场景效果不退化",
        metadata={"scene": "tech_qa", "model_version": "v1.2"},
        latency=420.5,
        token_input=28,
        token_output=36
    )
    print(f"   样本ID  : {sample.sample_id}")
    print(f"   场景    : {sample.metadata['scene']}")
    print(f"   模型版本: {sample.metadata['model_version']}")
    print(f"   耗时    : {sample.latency}ms")
    print(f"   Token   : 入={sample.token_input}, 出={sample.token_output}")
    print("-" * 40)
    # 2. 加入存储，模拟线上日志持续收集
    print("\n[2] 入库 => 写入样本存储，累积评测数据集")
    storage.add_sample(sample)
    print(f"   当前样本数: {len(storage.sample_list)}")
    # 3. 再追加一条示例：模拟多场景多版本样本收集
    sample2 = ModelSample(
        sample_id="s002",
        query="用三句话概括人工智能的发展历程",
        prompt="你是一个AI历史专家，请用三句话回答：{query}",
        response="AI从符号推理起步，经历神经网络复兴，如今进入大模型时代。",
        reference="人工智能经历了符号主义、连接主义到深度学习三个主要阶段。",
        metadata={"scene": "history_qa", "model_version": "v1.3"},
        latency=315.2,
        token_input=22,
        token_output=30
    )
    storage.add_sample(sample2)
    print(f"   追加后样本数: {len(storage.sample_list)}")
    print("-" * 40)
    # 4. 模拟人工标注
    print("\n[3] 人工标注 => 对样本评分（1坏例 ~ 5优秀）")
    sample.human_label = 4
    sample2.human_label = 3
    print(f"   s001 人工评分: {sample.human_label}（良好）")
    print(f"   s002 人工评分: {sample2.human_label}（一般）")
    print("-" * 40)
    # 5. 导出为离线评测数据集
    print("\n[4] 导出 => 输出统一格式 JSON 评测数据集")
    storage.export_for_eval("eval_dataset.json")
    print(f"   已导出 2 条样本到 eval_dataset.json")
    print("-" * 40)
    # 6. 打印导出内容预览
    import json
    print("\n[5] 导出文件预览:")
    with open("eval_dataset.json", "r", encoding="utf-8") as f:
        data = json.load(f)
    # 仅打印关键字段
    for item in data:
        print(f"   ┌─ {item['sample_id']} | 场景:{item['metadata']['scene']} | "
              f"版本:{item['metadata']['model_version']} | 评分:{item['human_label']}")
        print(f"   ├─ Query : {item['query']}")
        print(f"   ├─ 模型输出: {item['response']}")
        print(f"   └─ 标准答案: {item['reference']}")
    print("\n" + "=" * 55)
    print("样本结构打通 评测日志标注 → 支撑持续迭代优化")

import time
from typing import List
import openai
# 配置模型客户端
client = openai.OpenAI(
    base_url="http://localhost:8000/v1",
    api_key="dummy"
)
def llm_judge_score(query: str, pred: str, reference: str) -> float:
    """LLM-as-judge 自动打分，分数区间0~5"""
    judge_prompt = f"""
请对比模型回答和标准答案，评判回答质量，仅输出数字分数。
打分规则：
0：完全错误，严重幻觉
1：大部分错误
3：基本正确，但存在瑕疵
5：准确完整，贴合标准答案
用户问题：{query}
标准答案：{reference}
模型回答：{pred}
只输出0-5之间数字
"""
    resp = client.chat.completions.create(
        model="judge-model",
        messages=[{"role":"user","content":judge_prompt}],
        temperature=0
    )
    try:
        score = float(resp.choices[0].message.content.strip())
        return min(max(score,0),5)
    except:
        return 2.0
def run_offline_eval(dataset: List[ModelSample], model_version: str) -> dict:
    result_list = []
    for item in dataset:
        start = time.time()
        resp = client.chat.completions.create(
            model=model_version,
            messages=[{"role":"user","content":item.query}],
            temperature=0.1
        )
        latency = (time.time() - start)*1000
        pred = resp.choices[0].message.content
        score = llm_judge_score(item.query, pred, item.reference)
        result_list.append({
            "sample_id": item.sample_id,
            "score": score,
            "latency_ms": latency,
            "prediction": pred
        })
    # 汇总指标
    avg_score = sum(r["score"] for r in result_list)/len(result_list)
    avg_latency = sum(r["latency_ms"] for r in result_list)/len(result_list)
    return {
        "model_version": model_version,
        "sample_count": len(result_list),
        "avg_score": round(avg_score,2),
        "avg_latency_ms": round(avg_latency,2),
        "details": result_list
    }

def regression_check(baseline_report: dict, new_report: dict, threshold=0.3):
    """
    对比基线与新版本评测结果，判断是否出现退化
    return True 允许上线；False 存在能力退化
    """
    baseline_score = baseline_report["avg_score"]
    new_score = new_report["avg_score"]
    drop = baseline_score - new_score
    if drop > threshold:
        print(f"警告！效果下降{drop:.2f}，超过阈值{threshold}，阻断发布")
        return False
    print(f"指标下降{drop:.2f}，通过回归校验，可以灰度上线")
    return True

import logging
from logging.handlers import RotatingFileHandler
import json
# 初始化异步日志
model_logger = logging.getLogger("model_runtime")
model_logger.setLevel(logging.INFO)
handler = RotatingFileHandler("model_request.log", maxBytes=1024*1024*10, backupCount=10, encoding="utf-8")
model_logger.addHandler(handler)
def log_model_request(req_info: dict):
    """模型调用埋点日志"""
    model_logger.info(json.dumps(req_info, ensure_ascii=False))
# 接口调用处使用
def model_infer(query: str):
    start = time.time()
    resp = client.chat.completions.create(model="online-v1.2", messages=[{"role":"user", "content":query}])
    latency = (time.time()-start)*1000
    log_model_request({
        "query": query,
        "response": resp.choices[0].message.content,
        "latency_ms": latency,
        "model_version": "online-v1.2",
        "input_tokens": resp.usage.prompt_tokens,
        "output_tokens": resp.usage.completion_tokens
    })
    return resp

class FeedbackPipeline:
    def __init__(self):
        self.candidate_bad_cases = []
        self.labeled_samples = []
    def auto_filter_bad_case(self, log_item: dict) -> bool:
        """自动判断是否送入人工标注"""
        # 用户负反馈
        if log_item.get("user_dislike"):
            return True
        # 输出长度异常
        if len(log_item.get("response","")) < 10:
            return True
        return False
    def add_labeled_sample(self, sample: ModelSample):
        self.labeled_samples.append(sample)
        self.export_to_eval_set()
    def export_to_eval_set(self):
        storage = SampleStorage()
        for s in self.labeled_samples:
            storage.add_sample(s)
        storage.export_for_eval("business_eval_set.json")