from openai import OpenAI from ..schemas.base import DataResponse 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("understand_image_provider") SYSTEM_PROMPT = """ <角色> 你是一名资深广告文案专家。你的任务是根据输入的一张广告图片中的文字内容,生成一句简洁有力的广告文案。 <受众> 目标用户:50岁以上中老年人。语言需亲切、直白、易理解,避免专业术语与复杂长句。 <结构公式> [行动指令] + [低门槛/优惠承诺] + [核心价值/具体收益] + [紧迫感/稀缺性提醒] <约束> 1. 文案必须准确传达广告图片中的产品/服务信息,不得杜撰不存在的内容。 2. 加入紧迫感或稀缺性(如“限时”“名额有限”“马上领取”等),但不得虚构或夸大事实。 3. 避免医疗或功效的绝对化/保证性用语(如“治愈”“根治”“无副作用”“永久有效”)。 4. 不得包含违法、虚假、低俗、敏感、歧视性内容,不引导危险行为,不传播迷信。 5. 涉及健康/养生场景时,表述应为辅助/改善/建议性质,不承诺疗效;避免“祖传秘方”等违规表述。 6. 仅输出一句中文广告文案,简短醒目,适合作为宣传主标题。 7. 文案必须注意标点与短句分隔:动作、优惠承诺、核心收益之间用逗号分隔;紧迫感/稀缺性提醒用分号与前半部分隔开,避免把多个短语连写在一起,字数50字以内。 <示例 few-shot="true"> 长按二维码,0元入群,领取中医调理养生秘方;名额有限,赶快行动吧 <输出要求> 始终通过工具调用(function calling)输出,参数仅包含生成的一句文案。 """ tools: list[ChatCompletionToolParam] = [ { "type": "function", "function": { "name": "generate_ocr_text", "description": "生成一句适合中老年用户的广告文案(遵循结构公式与约束)", "parameters": { "type": "object", "properties": { "ocr_text": { "type": "string", "description": "最终的一句广告文案(中文,简短醒目,合规)" } }, "required": ["ocr_text"], "additionalProperties": False } } } ] class UnderstandImageProvider: print("UnderstandImageProvider called") def understand_image(self, image_url: 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.") completion = client.chat.completions.create( model=model, messages=[ {"role": "system", "content": SYSTEM_PROMPT}, { "role": "user", "content": [{ "type": "image_url", "image_url": { "url": image_url } }], }, ], tools=tools, tool_choice={ "type": "function", "function": {"name": "generate_ocr_text"} }, temperature=1.3 ) msg = completion.choices[0].message # Safely parse tool call arguments (if any) ocr_text = "" 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): ocr_text = str(args.get("ocr_text", "")).strip() except Exception as e: logger.error("parse tool call failed: %s", e, exc_info=True) # Fallback: if no tool-calls returned, try to read text content content = getattr(msg, "content", None) if not ocr_text and isinstance(content, str): ocr_text = content.strip() print("✅ OCR_TEXT:\n", ocr_text) return DataResponse(code=200, data=ocr_text, msg="Image understood successfully")