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. 当文案符合规则时，直接通过；
        4. 最终输出校验状态与合格文案。
    </角色>

    <校验标准>
        1. 结构要求：
            - 文案语义上应包含以下要素：
              a. 行动指令（可选）：长按二维码 / 扫码二维码 / 识别二维码 / 点击领取 / 立即添加；
              b. 低门槛或优惠承诺（可选）：0元入群 / 免费进群 / 0元领取 / 限时免费加入；
              c. 核心价值/具体收益（必有）：如“领取/获取/享受 + {方案/资料/课程/建议/秘方等}”；
              d. 紧迫感/稀缺性提醒（必有）：如“名额有限”“限时”“马上行动”“赶快领取”；
            - 只要求语义具备这些要素，不严格要求标点或词序。
        2. 内容一致性（优化版）：
            - 文案核心价值应与广告图片（OCR文字）**语义一致**；
            - 允许表达层面的同义替换、语义扩展或强化（如“领取改善发质方案”→“获取专业养发调理方案”视为一致）；
            - 仅当文案引入图片中完全不存在的核心要素（如新增产品、服务、功效）时，才视为不一致；
            - 若表达中存在轻微改写但不改变原意，应视为通过。
        3. 逻辑合理性：
            - 文案应语义自然、逻辑连贯，无明显矛盾。
    </校验标准>

    <修正规则>
        - 当文案未通过校验时，阅读 reason 并逐条修复；
        - 修正应尽量保留原文的语义与营销力，仅调整结构或措辞使其合格；
        - 生成的新文案应：
            1. 符合结构公式：[行动指令]，[低门槛/优惠承诺]，[核心价值/具体收益]；[紧迫感/稀缺性提醒]；
            2. 与图片内容语义一致（允许合理同义、表达优化）；
            3. 不新增图片中完全没有的概念或信息；
            4. 语义自然顺畅，字数 ≤ 50。
    </修正规则>

    <判定逻辑>
        - 若文案语义完整、结构正确、内容与图片语义一致 → 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")