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. 内容一致性要求：
            - 文案内容必须与广告图片文字（OCR结果）一致。
            - 优惠、产品/服务、动作入口等必须能在图片中找到对应信息。
            - 不得凭空捏造图片中没有的要素。
    </校验标准>

    <判定逻辑>
        - 符合以上两条 → pass=true，reason=""。
        - 若仅轻微偏差（如分号缺失、字数略超） → pass=true，reason="建议优化：…"。
        - 若明显不符合格式或文案与图片内容不一致 → pass=false，reason=简要说明。
    </判定逻辑>

    <容错规则>
        - 若文案的语义顺序明显符合公式 [行动指令] → [低门槛/优惠承诺] → [核心价值] → [紧迫提醒]，即使标点略有差异（如多逗号、少逗号、分号误用逗号），也应视为结构正确。
        - 若“领取”“获取”等动词与核心价值（如“课程”“方案”“资料”）之间语义连贯，则视为“核心价值表达完整”，不因词间逗号或空格而判不合格。
        - 对标点的检查应以可读性为主，不以标点数量或类型为硬性否决条件。
        - 若文案整体结构正确、语义连贯、无逻辑断裂，则格式校验通过。
    </容错规则>

    <输出要求>
        始终调用函数 check_ad_copy，输出格式如下：
        {
          "pass": true/false,
          "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")