Prompt Engineering实验：同一文本分类任务下6种提示词的Token消耗与输出质量对比

判断下面这条评论是好评、中评还是差评：
{comment}

import os, json, time
from typing import Callable
from openai import OpenAI
from dashscope import Generation

openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def call_gpt(prompt: str) -> tuple[str, dict]:
    resp = openai_client.chat.completions.create(
        model="gpt-4o-mini-2024-07-18",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.0,
        top_p=1.0,
        max_tokens=64,
    )
    text = resp.choices[0].message.content.strip()
    usage = {
        "in": resp.usage.prompt_tokens,
        "out": resp.usage.completion_tokens,
    }
    return text, usage

def call_qwen(prompt: str) -> tuple[str, dict]:
    resp = Generation.call(
        model="qwen-plus-2024-09-19",
        prompt=prompt,
        temperature=0.0,
        top_p=1.0,
        max_tokens=64,
        result_format="message",
    )
    text = resp.output.choices[0].message.content.strip()
    usage = {
        "in": resp.usage["input_tokens"],
        "out": resp.usage["output_tokens"],
    }
    return text, usage

LABELS = {"好评", "中评", "差评"}

def evaluate(caller: Callable, prompt_builder: Callable, dataset: list[dict]):
    correct = 0
    fmt_ok = 0
    total_in = total_out = 0
    errors = []
    for row in dataset:
        prompt = prompt_builder(row["comment"])
        try:
            raw, usage = caller(prompt)
        except Exception as e:
            errors.append((row["comment"], str(e)))
            continue
        total_in += usage["in"]
        total_out += usage["out"]
        pred = normalize(raw)          # 见下方 normalize
        if pred in LABELS:
            fmt_ok += 1
        if pred == row["label"]:
            correct += 1
        else:
            errors.append((row["comment"], pred, row["label"]))
    n = len(dataset)
    return {
        "acc": correct / n,
        "fmt": fmt_ok / n,
        "avg_in": total_in / n,
        "avg_out": total_out / n,
        "errors": errors[:20],
    }

def normalize(raw: str) -> str:
    for lab in ["好评", "中评", "差评"]:
        if lab in raw:
            return lab
    return "UNKNOWN"

def p1_naive(c):
    return f"判断下面这条评论是好评、中评还是差评：\n{c}"

def p2_role_format(c):
    return (
        "你是一名电商评论情感分析专家。\n"
        "请判断以下评论的情感极性，只能从『好评』『中评』『差评』中选一个词输出，"
        "不要输出任何其他内容。\n"
        f"评论：{c}\n"
        "答案："
    )

def p3_structured(c):
    return (
        "你是电商评论情感分析专家。请严格输出JSON，不要包含```json```围栏。\n"
        "JSON schema: {\"label\": \"好评|中评|差评\"}\n"
        "判定规则：\n"
        "- 好评：整体满意，无明显负面\n"
        "- 差评：核心体验差（口味/卫生/严重迟到>30min）\n"
        "- 中评：有明确正面也有明确负面，或情绪平淡\n"
        f"评论：{c}"
    )

FEWSHOT_3 = [
    ("味道很好，分量也足，下次还来", "好评"),
    ("味道一般般，不难吃也不惊艳", "中评"),
    ("等了两个小时，饭都凉了，差评", "差评"),
]

def p4_fewshot3(c):
    shots = "\n".join(f"评论：{x}\n答案：{y}" for x, y in FEWSHOT_3)
    return (
        "你是电商评论情感分析专家，只能输出『好评』『中评』『差评』之一。\n"
        f"{shots}\n"
        f"评论：{c}\n答案："
    )

FEWSHOT_6 = FEWSHOT_3 + [
    ("五星好评，就是配送慢了点", "好评"),
    ("味道还行，就是等太久了", "中评"),
    ("包装破了，汤洒了一半，再也不点了", "差评"),
]

def p5_fewshot6_cot(c):
    shots = "\n".join(
        f"评论：{x}\n分析：先看口味/服务/配送，再综合判断。\n答案：{y}"
        for x, y in FEWSHOT_6
    )
    return (
        "你是电商评论情感分析专家。请先简要分析，再给出标签。\n"
        f"{shots}\n"
        f"评论：{c}\n分析："
    )

def p6_cot_lite(c):
    return (
        "你是电商评论情感分析专家。\n"
        "步骤：\n"
        "1) 用不超过8个字写出关键证据（口味/服务/配送/包装之一）；\n"
        "2) 输出标签。\n"
        "严格按以下格式，两行，不要多余内容：\n"
        "证据：\n"
        "标签：\n"
        f"评论：{c}"
    )

import pandas as pd
dataset = pd.read_json("reviews_labeled.jsonl", lines=True).to_dict("records")

PROMPTS = {
    "P1_naive": p1_naive,
    "P2_role_format": p2_role_format,
    "P3_structured": p3_structured,
    "P4_fewshot3": p4_fewshot3,
    "P5_fewshot6_cot": p5_fewshot6_cot,
    "P6_cot_lite": p6_cot_lite,
}

results = {}
for name, builder in PROMPTS.items():
    results[name] = evaluate(call_gpt, builder, dataset)
    print(name, results[name]["acc"], results[name]["fmt"],
          results[name]["avg_in"], results[name]["avg_out"])