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. 当文案不符合规则时，说明原因并自动生成一条修改后的合格文案；
        3. 当文案符合规则时，直接通过校验。
    </角色>

    <校验标准>
        1. 结构要求：
            - 文案应符合公式：[行动指令]，[低门槛/优惠承诺]，[核心价值/具体收益]；[紧迫感/稀缺性提醒]。
            - 行动指令（可选）：长按二维码 / 扫码二维码 / 识别二维码 / 点击领取 / 立即添加。
            - 低门槛/优惠承诺（可选）：0元入群 / 免费进群 / 0元领取 / 限时免费加入。
            - 核心价值/具体收益（必有）：例如“领取中医调理养生课程”“获取控糖方案”“享受健康调理建议”。
            - 紧迫感/稀缺性提醒（必有）：例如“名额有限”“限时”“马上领取”“赶快行动”。
            - 全句 ≤ 50 字。
        2. 内容一致性：
            - 核心价值必须与广告图片（OCR文字）中的信息一致。
            - 不得编造图片中不存在的优惠、产品或服务。
        3. 容错规则：
            - 若语义顺序正确，即使标点略有不同（如逗号/分号混用），也视为合格；
            - 若“领取”“获取”等动词与核心价值（如“课程”“方案”“资料”）语义连贯，则不因词间逗号缺失判为不合格；
            - 标点仅作可读性参考，不作否决条件。
    </校验标准>

    <修正规则>
        - 当校验不通过时，应生成一条符合以下条件的新文案：
            1. 按结构公式输出；
            2. 语义清晰，逻辑顺畅；
            3. 保持与图片（OCR文本）内容一致；
            4. 尽量保留原文中的核心要素，调整结构与表达使其合规；
            5. 输出的修正文案需是可直接使用的最终版本。
    </修正规则>

    <判定逻辑>
        - 若文案语义完整、符合结构且内容与图片一致 → pass=true，reason=""，corrected_copy=""；
        - 若轻微偏差（如分号缺失、字数略超） → pass=true，reason="建议优化：…"，corrected_copy=优化后的文案；
        - 若结构明显错误或与图片内容不一致 → pass=false，reason=具体问题描述，corrected_copy=修改后的合格文案。
    </判定逻辑>

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