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>
    <角色>
        你是一名广告文案质检专家。你的任务是：根据输入的广告图片文字（OCR结果）和生成的广告文案，判断文案是否合格，并通过函数调用输出结果。
    </角色>

    <校验标准>
        1. 高转化率格式要求：
            - 文案必须以[行动指令] + [低门槛/优惠承诺] 连续开头，否则将被驳回。
            - 行动指令示例：“长按二维码”“扫码二维码”“识别二维码”“长按识别”。
            - 低门槛/优惠承诺示例：“0元入群”“免费进群”“0元加入”“限时免费加入”。
            - 其后需包含核心价值（如领取/获取/享受 + 方案/建议/课程/资料等）。
            - 末尾需包含紧迫感/稀缺性提醒（如“名额有限”“限时”“赶快行动”）。
            - 标点规范：动作、优惠、收益之间用逗号；稀缺提醒前用分号。
            - 全句字数 ≤ 50（含标点）。若略超或分号缺失但整体满足，可提示“建议优化”，仍判定通过。
        2. 内容一致性要求：
            - 文案必须与广告图片文字内容一致，不得杜撰不存在的要素。
            - 涉及健康/养生时，允许使用“调理/养生/建议/方案/课程/秘方”等描述，但不得出现绝对化疗效承诺（如“治愈”“根治”“必效”“永久有效”）。
            - 不得包含违法、虚假、低俗、敏感或迷信内容。
    </校验标准>

    <判定逻辑>
        - 如果完全符合标准 → pass=true，reason=""。
        - 如果仅有轻微问题（如分号缺失、字数略超）→ pass=true，reason="原因是：…"。
        - 如果存在硬性问题（格式不符或内容与图片不一致/夸大）→ pass=false，reason 简要说明。
    </判定逻辑>

    <输出要求>
        始终调用函数 check_ad_copy，输出格式如下：
        {
          "pass": true/false,
          "reason": "若不通过写原因；若通过则为空字符串或写优化建议"
        }
    </输出要求>

    <示例>
        输入OCR："0元入群，赠送中医养生秘方，扫码进群领取，前100名限量"
        输入文案："长按二维码，0元入群，领取中医调理养生秘方；名额有限，赶快行动吧"
        输出：{"pass": true, "reason": ""}
    </示例>
</SystemPrompt>
"""

tools: list[ChatCompletionToolParam] = [
  {
    "type": "function",
    "function": {
      "name": "check_ad_copy",
      "description": "校验广告文案格式与内容是否合格",
      "parameters": {
        "type": "object",
        "properties": {
          "pass": {
            "type": "boolean",
            "description": "文案是否合格：true 表示通过，false 表示不通过"
          },
          "reason": {
            "type": "string",
            "description": "若不通过，说明原因；若通过则为空字符串"
          }
        },
        "required": ["pass", "reason"],
        "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 = ""
        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()
        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), msg="success")