284 lines
15 KiB
Python
284 lines
15 KiB
Python
import json
|
||
import base64
|
||
import logging
|
||
import re
|
||
import urllib.error
|
||
import urllib.request
|
||
|
||
from fastapi import HTTPException, status
|
||
|
||
from .config import settings
|
||
from .schemas import QAGenerateRequest
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _prompt(payload: QAGenerateRequest) -> str:
|
||
style = "带真实体验的轻度种草语气" if payload.seed_style else "客观自然，不刻意种草"
|
||
return f"""你是电商问大家内容策划。请根据商品条件生成真实、具体、互不重复的消费者提问和参考回答。
|
||
商品名称：{payload.product_name or '未提供'}
|
||
商品品类：{payload.category or '未提供'}
|
||
商品功效：{payload.benefits or '未提供'}
|
||
商品卖点：{payload.selling_points or '未提供'}
|
||
语言风格：{payload.language_style or '客观自然'}
|
||
目标人群：{payload.target_group or '普通消费者'}
|
||
使用场景：{payload.usage_scene or '日常使用'}
|
||
购买渠道：{payload.purchase_channel or '电商平台'}
|
||
回答要求：{payload.answer_requirement or '直接回应问题，避免夸大功效'}
|
||
回答语气：{style}
|
||
真实问答参考：{payload.reference_qa or '未提供'}
|
||
|
||
请生成 {payload.question_count} 个问题，每个问题生成 {payload.answers_per_question} 个不同角度的参考回答。
|
||
约束：问题尽量不超过20字，回答尽量不超过50字；不虚构成分、检测、认证和医疗功效；表达像真实买家；不要重复商品名称作为开头。
|
||
如果提供了真实问答参考，请优先学习其中的提问角度、真实语气和回答结构，但不要照抄原文，不要把参考中的其他商品信息误套到当前商品。
|
||
只返回严格 JSON，不要 Markdown 代码块或额外解释，格式必须是：
|
||
{{"items":[{{"question":"问题文本","answers":["回答1","回答2"]}}]}}"""
|
||
|
||
|
||
def _parse_content(content: str, expected_answers: int) -> list[dict]:
|
||
cleaned = content.strip()
|
||
if cleaned.startswith("```"):
|
||
cleaned = re.sub(r"^```(?:json)?\s*|\s*```$", "", cleaned, flags=re.I | re.S).strip()
|
||
try:
|
||
parsed = json.loads(cleaned)
|
||
except json.JSONDecodeError as exc:
|
||
start, end = cleaned.find("{"), cleaned.rfind("}")
|
||
if start < 0 or end <= start:
|
||
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="大模型返回格式无法解析") from exc
|
||
parsed = json.loads(cleaned[start:end + 1])
|
||
raw_items = parsed.get("items") if isinstance(parsed, dict) else parsed
|
||
if not isinstance(raw_items, list):
|
||
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="大模型未返回有效问答列表")
|
||
items = []
|
||
for index, raw in enumerate(raw_items):
|
||
if not isinstance(raw, dict) or not str(raw.get("question", "")).strip():
|
||
continue
|
||
answers = raw.get("answers", [])
|
||
if isinstance(answers, str):
|
||
answers = [answers]
|
||
answers = [str(answer).strip() for answer in answers if str(answer).strip()][:expected_answers]
|
||
if answers:
|
||
items.append({"id": f"Q{index + 1:03d}", "question": str(raw["question"]).strip(), "answers": answers, "source": "AI生成", "status": "待采纳", "accepted": False})
|
||
if not items:
|
||
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="大模型未返回有效问答内容")
|
||
return items
|
||
|
||
|
||
def _generate_qa_batch(payload: QAGenerateRequest) -> list[dict]:
|
||
if not settings.llm_api_key:
|
||
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="尚未配置大模型 API Key，请先设置 LLM_API_KEY")
|
||
url = settings.llm_base_url.rstrip("/") + "/chat/completions"
|
||
request_body = {"model": settings.llm_model, "temperature": 0.8, "messages": [{"role": "system", "content": "你只输出合法 JSON。"}, {"role": "user", "content": _prompt(payload)}]}
|
||
request = urllib.request.Request(url, data=json.dumps(request_body).encode("utf-8"), headers={"Content-Type": "application/json", "Authorization": f"Bearer {settings.llm_api_key}"}, method="POST")
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=settings.llm_timeout_seconds) as response:
|
||
result = json.loads(response.read().decode("utf-8"))
|
||
except urllib.error.HTTPError as exc:
|
||
logger.error("LLM HTTP error: status=%s url=%s", exc.code, url)
|
||
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"大模型服务返回错误（HTTP {exc.code}）") from exc
|
||
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
|
||
logger.exception("LLM request failed: url=%s reason=%s", url, exc)
|
||
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="大模型服务暂时不可用，请稍后重试") from exc
|
||
try:
|
||
content = result["choices"][0]["message"]["content"]
|
||
except (KeyError, IndexError, TypeError) as exc:
|
||
logger.error("LLM response missing choices/message/content: keys=%s", list(result) if isinstance(result, dict) else type(result).__name__)
|
||
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="大模型响应缺少内容") from exc
|
||
return _parse_content(content, payload.answers_per_question)
|
||
|
||
|
||
def generate_qa(payload: QAGenerateRequest) -> list[dict]:
|
||
"""Generate the requested amount in small batches so large requests are not truncated."""
|
||
target = payload.question_count
|
||
items: list[dict] = []
|
||
seen: set[str] = set()
|
||
attempts = 0
|
||
max_attempts = max((target + 19) // 20 + 2, 3)
|
||
while len(items) < target and attempts < max_attempts:
|
||
batch_size = min(20, target - len(items))
|
||
batch_payload = payload.model_copy(update={"question_count": batch_size})
|
||
for item in _generate_qa_batch(batch_payload):
|
||
key = item["question"].strip()
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
items.append(item)
|
||
if len(items) >= target:
|
||
break
|
||
attempts += 1
|
||
if len(items) < target:
|
||
logger.warning("LLM returned %s/%s unique QA items", len(items), target)
|
||
for index, item in enumerate(items[:target], start=1):
|
||
item["id"] = f"Q{index:03d}"
|
||
return items[:target]
|
||
|
||
|
||
def _generate_questions_batch(payload: dict, count: int, reference_qa: str = "") -> list[str]:
|
||
"""Generate only consumer questions for the ask-task workflow."""
|
||
if not settings.llm_api_key:
|
||
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="尚未配置大模型 API Key，无法生成真实提问")
|
||
prompt = f"""你是电商问大家提问策划。请根据商品信息生成 {count} 个真实、具体、互不重复的消费者提问。
|
||
商品名称：{payload.get('product_name') or '未提供'}
|
||
商品品类：{payload.get('category') or '未提供'}
|
||
商品功效：{payload.get('benefits') or '未提供'}
|
||
商品卖点：{payload.get('selling_points') or '未提供'}
|
||
语言风格：{payload.get('language_style') or '客观自然'}
|
||
目标人群：{payload.get('target_group') or '普通消费者'}
|
||
使用场景：{payload.get('usage_scene') or '日常使用'}
|
||
购买渠道：{payload.get('purchase_channel') or '电商平台'}
|
||
真实问答参考：{reference_qa or '未提供'}
|
||
只返回严格 JSON，不要 Markdown 或解释，格式：{{"questions":["问题1","问题2"]}}。问题应像真实买家提问，尽量不超过30字，不要虚构具体参数或功效。"""
|
||
url = settings.llm_base_url.rstrip("/") + "/chat/completions"
|
||
body = {"model": settings.llm_model, "temperature": 0.85, "messages": [{"role": "system", "content": "你只输出合法 JSON。"}, {"role": "user", "content": prompt}]}
|
||
request = urllib.request.Request(url, data=json.dumps(body).encode("utf-8"), headers={"Content-Type": "application/json", "Authorization": f"Bearer {settings.llm_api_key}"}, method="POST")
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=settings.llm_timeout_seconds) as response:
|
||
result = json.loads(response.read().decode("utf-8"))
|
||
content = result["choices"][0]["message"]["content"]
|
||
cleaned = re.sub(r"^```(?:json)?\s*|\s*```$", "", str(content).strip(), flags=re.I | re.S).strip()
|
||
parsed = json.loads(cleaned)
|
||
questions = parsed.get("questions") if isinstance(parsed, dict) else parsed
|
||
if not isinstance(questions, list):
|
||
raise ValueError("questions is not a list")
|
||
output = []
|
||
for value in questions:
|
||
text = str(value).strip()
|
||
if text and text not in output:
|
||
output.append(text[:200])
|
||
if not output:
|
||
raise ValueError("empty questions")
|
||
return output[:count]
|
||
except urllib.error.HTTPError as exc:
|
||
logger.error("LLM question generation HTTP error: status=%s url=%s", exc.code, url)
|
||
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"大模型服务返回错误（HTTP {exc.code}）") from exc
|
||
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, KeyError, IndexError, TypeError, ValueError) as exc:
|
||
logger.exception("LLM question generation failed")
|
||
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="大模型提问生成失败，请稍后重试") from exc
|
||
|
||
|
||
def generate_questions(payload: dict, count: int, reference_qa: str = "") -> list[str]:
|
||
"""Generate questions in batches and merge them until the requested count is reached."""
|
||
target = int(count)
|
||
output: list[str] = []
|
||
seen: set[str] = set()
|
||
attempts = 0
|
||
max_attempts = max((target + 19) // 20 + 2, 3)
|
||
while len(output) < target and attempts < max_attempts:
|
||
batch = _generate_questions_batch(payload, min(20, target - len(output)), reference_qa)
|
||
for question in batch:
|
||
if question not in seen:
|
||
seen.add(question)
|
||
output.append(question)
|
||
if len(output) >= target:
|
||
break
|
||
attempts += 1
|
||
if len(output) < target:
|
||
logger.warning("LLM returned %s/%s unique questions", len(output), target)
|
||
return output[:target]
|
||
|
||
|
||
def generate_answer_for_question(product_title: str, question: str) -> str:
|
||
"""Generate one concise answer. Falls back to a safe draft during the simulated-OCR MVP."""
|
||
if not settings.llm_api_key:
|
||
return f"结合{product_title}的使用体验来看，建议按自身肤质先少量试用，确认合适后再正常使用。"
|
||
url = settings.llm_base_url.rstrip("/") + "/chat/completions"
|
||
prompt = f"商品：{product_title}\n问题：{question}\n请生成一条50字以内、自然真实、不夸大功效的淘宝问大家回答。只输出回答正文。"
|
||
request_body = {"model": settings.llm_model, "temperature": 0.7, "messages": [{"role": "system", "content": "你是电商问大家回答助手。"}, {"role": "user", "content": prompt}]}
|
||
request = urllib.request.Request(url, data=json.dumps(request_body).encode("utf-8"), headers={"Content-Type": "application/json", "Authorization": f"Bearer {settings.llm_api_key}"}, method="POST")
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=settings.llm_timeout_seconds) as response:
|
||
result = json.loads(response.read().decode("utf-8"))
|
||
answer = str(result["choices"][0]["message"]["content"]).strip()
|
||
return answer[:300]
|
||
except Exception:
|
||
logger.exception("Answer generation failed for simulated OCR question")
|
||
return f"结合{product_title}的使用体验来看，建议按自身肤质先少量试用，确认合适后再正常使用。"
|
||
|
||
|
||
def _parse_ocr_content(content: str) -> list[dict]:
|
||
if isinstance(content, list):
|
||
content = "".join(str(item.get("text", "")) for item in content if isinstance(item, dict))
|
||
cleaned = str(content or "").strip()
|
||
if cleaned.startswith("```"):
|
||
cleaned = re.sub(r"^```(?:json)?\s*|\s*```$", "", cleaned, flags=re.I | re.S).strip()
|
||
try:
|
||
parsed = json.loads(cleaned)
|
||
except json.JSONDecodeError:
|
||
start, end = cleaned.find("{"), cleaned.rfind("}")
|
||
if start < 0 or end <= start:
|
||
return []
|
||
try:
|
||
parsed = json.loads(cleaned[start:end + 1])
|
||
except json.JSONDecodeError:
|
||
return []
|
||
raw_items = parsed.get("items") if isinstance(parsed, dict) else parsed
|
||
if not isinstance(raw_items, list):
|
||
return []
|
||
items = []
|
||
seen = set()
|
||
for raw in raw_items:
|
||
if not isinstance(raw, dict):
|
||
continue
|
||
question = re.sub(r"^[\s\d.#、.)（()]+", "", str(raw.get("question", ""))).strip()
|
||
if len(question) < 4 or question in seen:
|
||
continue
|
||
if any(token in question for token in ("回答", "规则", "待回答", "已回答", "已提问")):
|
||
continue
|
||
seen.add(question)
|
||
items.append({
|
||
"product_title": str(raw.get("product_title", "待识别商品")).strip() or "待识别商品",
|
||
"question": question[:300],
|
||
})
|
||
return items[:100]
|
||
|
||
|
||
def extract_questions_from_images(images: list[tuple[bytes, str]]) -> list[dict]:
|
||
"""Extract consumer questions from one or more Taobao Q&A screenshots.
|
||
|
||
The configured OpenAI-compatible model must support image input. The
|
||
caller keeps a deterministic fallback when no vision model/API key is
|
||
configured, so uploading screenshots never blocks the task workflow.
|
||
"""
|
||
if not settings.llm_api_key or not images:
|
||
return []
|
||
content = [{
|
||
"type": "text",
|
||
"text": (
|
||
"请识别这些淘宝‘问答/待回答’页面截图中的消费者提问。只提取真正的问题句子，"
|
||
"不要提取导航、按钮、商品名、用户昵称、规则文字或页面标题。去除序号并保持原文。"
|
||
"多张截图合并去重。只返回严格 JSON："
|
||
'{"items":[{"product_title":"商品名或待识别商品","question":"问题原文"}]}'
|
||
),
|
||
}]
|
||
for index, (image_data, content_type) in enumerate(images, start=1):
|
||
encoded = base64.b64encode(image_data).decode("ascii")
|
||
content.append({"type": "text", "text": f"截图 {index}"})
|
||
content.append({
|
||
"type": "image_url",
|
||
"image_url": {"url": f"data:{content_type};base64,{encoded}"},
|
||
})
|
||
request_body = {
|
||
"model": settings.llm_vision_model,
|
||
"temperature": 0,
|
||
"messages": [
|
||
{"role": "system", "content": "你是严谨的中文电商截图 OCR 提取器，只输出合法 JSON。"},
|
||
{"role": "user", "content": content},
|
||
],
|
||
}
|
||
url = settings.llm_base_url.rstrip("/") + "/chat/completions"
|
||
request = urllib.request.Request(
|
||
url,
|
||
data=json.dumps(request_body).encode("utf-8"),
|
||
headers={"Content-Type": "application/json", "Authorization": f"Bearer {settings.llm_api_key}"},
|
||
method="POST",
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=settings.llm_timeout_seconds) as response:
|
||
result = json.loads(response.read().decode("utf-8"))
|
||
content = result["choices"][0]["message"]["content"]
|
||
return _parse_ocr_content(content)
|
||
except Exception:
|
||
logger.exception("Question OCR failed")
|
||
return []
|