123 lines
4.8 KiB
Python
123 lines
4.8 KiB
Python
import base64
|
|
import json
|
|
|
|
from openai import AsyncOpenAI
|
|
|
|
from app.core.config import settings
|
|
|
|
DISCLAIMER = "本报告仅用于娱乐占卜与自我反思，不构成医学、心理、职业、财务、投资或任何人生决策建议。"
|
|
|
|
SYSTEM_PROMPT = (
|
|
"你是“赛博先生”，一个面向普通人的娱乐型 AI 命理解读助手。"
|
|
"你的风格要像一个会聊天、懂生活的朋友：温和、具体、接地气，有一点玄学仪式感，但不要装神秘。"
|
|
"所有表达都必须使用“可能、倾向、适合、提醒你”这类非确定性措辞。"
|
|
"禁止给出医疗、心理诊断、投资、职业成败、婚恋结果、寿命、灾祸等确定性判断。"
|
|
"不要堆砌专业术语，不要写得像教材；每个结论都要落到生活、学习、事业、关系或近期行动里的具体场景。"
|
|
)
|
|
|
|
REPORT_SCHEMA = {
|
|
"name": "reading_report",
|
|
"schema": {
|
|
"type": "object",
|
|
"additionalProperties": False,
|
|
"properties": {
|
|
"quality_check": {
|
|
"type": "object",
|
|
"additionalProperties": False,
|
|
"properties": {
|
|
"can_analyze": {"type": "boolean"},
|
|
"reason": {"type": "string"},
|
|
"confidence": {"type": "number"},
|
|
},
|
|
"required": ["can_analyze", "reason", "confidence"],
|
|
},
|
|
"overall_summary": {"type": "string"},
|
|
"dimensions": {
|
|
"type": "array",
|
|
"items": {
|
|
"type": "object",
|
|
"additionalProperties": False,
|
|
"properties": {
|
|
"name": {"type": "string"},
|
|
"observations": {"type": "array", "items": {"type": "string"}},
|
|
"interpretation": {"type": "string"},
|
|
"confidence": {"type": "number"},
|
|
"advice": {"type": "string"},
|
|
},
|
|
"required": ["name", "observations", "interpretation", "confidence", "advice"],
|
|
},
|
|
},
|
|
"strengths": {"type": "array", "items": {"type": "string"}},
|
|
"challenges": {"type": "array", "items": {"type": "string"}},
|
|
"suggestions": {"type": "array", "items": {"type": "string"}},
|
|
"lucky_keywords": {"type": "array", "items": {"type": "string"}},
|
|
"disclaimer": {"type": "string"},
|
|
},
|
|
"required": [
|
|
"quality_check",
|
|
"overall_summary",
|
|
"dimensions",
|
|
"strengths",
|
|
"challenges",
|
|
"suggestions",
|
|
"lucky_keywords",
|
|
"disclaimer",
|
|
],
|
|
},
|
|
"strict": True,
|
|
}
|
|
|
|
|
|
class BaseAnalyzer:
|
|
def __init__(self) -> None:
|
|
self.client = (
|
|
AsyncOpenAI(
|
|
api_key=settings.openai_api_key,
|
|
base_url=settings.openai_base_url,
|
|
timeout=settings.openai_timeout_seconds,
|
|
)
|
|
if settings.openai_api_key
|
|
else None
|
|
)
|
|
|
|
async def _create_text_report(self, user_prompt: str) -> dict:
|
|
if not self.client:
|
|
raise RuntimeError("OpenAI client is not configured")
|
|
response = await self.client.responses.create(
|
|
model=settings.openai_model,
|
|
input=[
|
|
{"role": "system", "content": [{"type": "input_text", "text": SYSTEM_PROMPT}]},
|
|
{"role": "user", "content": [{"type": "input_text", "text": user_prompt}]},
|
|
],
|
|
text={"format": {"type": "json_schema", **REPORT_SCHEMA}},
|
|
)
|
|
data = json.loads(response.output_text)
|
|
data["disclaimer"] = DISCLAIMER
|
|
return data
|
|
|
|
async def _create_image_report(self, user_prompt: str, image_bytes: bytes, content_type: str) -> dict:
|
|
if not self.client:
|
|
raise RuntimeError("OpenAI client is not configured")
|
|
image_data = base64.b64encode(image_bytes).decode("ascii")
|
|
response = await self.client.responses.create(
|
|
model=settings.openai_model,
|
|
input=[
|
|
{"role": "system", "content": [{"type": "input_text", "text": SYSTEM_PROMPT}]},
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "input_text", "text": user_prompt},
|
|
{
|
|
"type": "input_image",
|
|
"image_url": f"data:{content_type};base64,{image_data}",
|
|
"detail": "high",
|
|
},
|
|
],
|
|
},
|
|
],
|
|
text={"format": {"type": "json_schema", **REPORT_SCHEMA}},
|
|
)
|
|
data = json.loads(response.output_text)
|
|
data["disclaimer"] = DISCLAIMER
|
|
return data
|