pq-web-ai/app/providers/evaluation_provider.py

from openai import OpenAI
from ..schemas.base import DataResponse, CopywritingEvaluationPayload
from ..core.config import get_settings
from ..core.logger import get_logger
from openai.types.chat import ChatCompletionToolParam
import json

settings = get_settings()
logger = get_logger("evaluation_provider")

SYSTEM_PROMPT = """
<SystemPrompt>
    <角色>
        你是一名广告文案质检与优化专家。你的任务是：
        1. 根据输入的广告图片文字（OCR结果）和广告文案，对文案进行校验；
        2. 当文案不符合规则时，给出具体原因（reason），并根据这些问题自动修正文案；
        3. 当文案符合规则时，直接通过；
        4. 最终输出校验状态与合格文案。
    </角色>

    <校验标准>
        1. 结构要求（只看语义组成，不看标点）：
            - 文案语义上应包含以下要素：
              a. 行动指令（可选）：长按二维码 / 扫码二维码 / 识别二维码 / 点击领取 / 立即添加；
              b. 低门槛或优惠承诺（可选）：0元入群 / 免费进群 / 0元加入 / 限时免费加入；
              c. 核心价值/具体收益（必有）：例如“领取中医调理养发方案”“获取控糖建议”“享受养生课程”等；
              d. 紧迫感/稀缺性提醒（必有）：例如“名额有限”“限时”“马上领取”“赶快行动”等；
            - 只要语义上具备这些组成部分即可，不因标点、顺序或语气词差异判不合格。
        2. 内容一致性要求：
            - 文案的核心价值必须与广告图片（OCR文字）内容一致；
            - 优惠、产品/服务、动作入口等信息必须在图片中能找到对应要素；
            - 不得凭空编造页面中不存在的内容。
        3. 逻辑合理性：
            - 文案应语义连贯，表达完整，无逻辑冲突。
    </校验标准>

    <修正规则>
        - 当文案未通过校验时，应严格根据 reason 中列出的问题逐条修复：
            1. 阅读 reason，识别缺失或错误的要素；
            2. 在修正文案中补充或修改这些要素；
            3. 确保修正后的文案：
                - 符合公式 [行动指令]，[低门槛/优惠承诺]，[核心价值/具体收益]；[紧迫感/稀缺性提醒]；
                - 保持与图片（OCR结果）内容一致；
                - 语义自然、逻辑通顺；
            4. 不可仅复述原文；修正文案必须显式解决 reason 指出的所有问题。
        - 修正完成后，应重新验证：
            - 若问题已解决，返回合格结果；
            - 若问题仍存在，则继续修正，直至完全合格。
    </修正规则>

    <判定逻辑>
        - 若文案语义完整、内容一致 → pass=true，reason=""，corrected_copy=""；
        - 若存在轻微偏差（结构略乱或词序不同但语义正确） → pass=true，reason=问题说明，corrected_copy=优化后版本；
        - 若存在严重问题（缺少核心价值或与图片内容不符） → pass=false，reason=问题说明（简单说明），corrected_copy=修正后合格文案；
        - 修正文案应完整保留图片中的核心卖点信息。
    </判定逻辑>

    <输出要求>
        始终调用函数 check_ad_copy，输出格式如下：
        {
          "pass": true/false,
          "reason": "若不通过写原因；若通过则为空字符串或给出优化建议",
          "corrected_copy": "最终合格的一句广告文案（若原文合格则为原文）"
        }
    </输出要求>

    <示例>
        输入OCR："0元入群领取改善发质方案"
        输入文案："长按二维码，0元入群，领取中医调理养发方案；名额有限，马上行动！"
        reason："核心价值与图片不一致，未体现‘改善发质’"
        修正输出：
        {
          "pass": true,
          "reason": "",
          "corrected_copy": "长按二维码，0元入群，领取中医调理改善发质方案；名额有限，马上行动！"
        }
    </示例>
</SystemPrompt>
"""

tools: list[ChatCompletionToolParam] = [
  {
    "type": "function",
    "function": {
      "name": "check_ad_copy",
      "description": "校验并在必要时修正广告文案，确保结构与内容合规",
      "parameters": {
        "type": "object",
        "properties": {
          "pass": { "type": "boolean" },
          "reason": { "type": "string" },
          "corrected_copy": { "type": "string" }
        },
        "required": ["pass", "reason", "corrected_copy"],
        "additionalProperties": False
      }
    }
  }
]

class EvaluationProvider:
  print("EvaluationProvider called")

  def copywriting_evaluation(self, image_url: str, text: str, model: str) -> DataResponse:
        client = OpenAI(
            api_key = settings.dashscope_api_key or "",
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
        )

        if not client:
            logger.error("OpenAI client is not initialized.")
            return DataResponse(code=1, data=None, msg=f"OpenAI client is not initialized")
        
        completion = client.chat.completions.create(
            model=model,
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {
                    "role": "user",
                    "content": [
                        {"type": "image_url", "image_url": {"url": image_url}},
                        {"type": "text", "text": text}
                    ],
                },
            ],
            tools=tools,
            tool_choice={
                "type": "function",
                "function": {"name": "check_ad_copy"}
            },
            temperature=0.3
        )

        msg = completion.choices[0].message
        print(msg)
        # Safely parse tool call arguments (if any)
        content = True
        reason = ""
        corrected_msg = ""
        try:
            tool_calls = getattr(msg, "tool_calls", None) or []
            if tool_calls:
                call = tool_calls[0]
                arg_str = getattr(getattr(call, "function", None), "arguments", None)
                if isinstance(arg_str, str) and arg_str.strip():
                    args = json.loads(arg_str)
                    if isinstance(args, dict):
                        content = bool(args.get("pass", True))
                        reason = str(args.get("reason", "")).strip()
                        corrected_msg = str(args.get("corrected_copy", "")).strip()
        except Exception as e:
            logger.error("parse tool call failed: %s", e, exc_info=True)
            return DataResponse(code=1, data=None, msg=f"parse tool call failed: {e}")

        print("✅ PASS:\n", content)
        print("✅ REASON:\n", reason)

        return DataResponse(code=0, data=CopywritingEvaluationPayload(content=content, reason=reason, corrected_msg=corrected_msg), msg="success")