53e0065e3e
配置 SSOT（TOML + .env） 统一错误契约 Auth 与事务边界 Redis / Celery 可靠性:业务 Redis（DB/0）与 Celery broker/backend（DB/1）显式拆分；连接池、sync client 可观测性（OpenTelemetry + LGTM）
362 lines
13 KiB
Python
362 lines
13 KiB
Python
import json
|
||
import re
|
||
from typing import Any, Optional
|
||
|
||
from app.core.config import settings
|
||
from app.core.json_utils import extract_json_payload
|
||
from app.core.langchain_llm import invoke_json_object
|
||
from app.core.logging import get_logger
|
||
from app.features.memoir.constants import memoir
|
||
|
||
from .settings import MemoirImageSettings
|
||
|
||
logger = get_logger(__name__)
|
||
_CJK_RE = re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff\uf900-\ufaff]")
|
||
|
||
|
||
class MemoirImagePromptService:
|
||
CATEGORY_STYLE_MAP = {
|
||
"childhood": "watercolor",
|
||
"family": "watercolor",
|
||
"career_early": "realistic",
|
||
"career_achievement": "realistic",
|
||
"career_challenge": "realistic",
|
||
"beliefs": "editorial illustration",
|
||
"summary": "editorial illustration",
|
||
}
|
||
CATEGORY_FALLBACK_SUBJECT_MAP = {
|
||
"childhood": "childhood memory",
|
||
"education": "school memory",
|
||
"career_early": "early career memory",
|
||
"career_achievement": "career achievement memory",
|
||
"career_challenge": "career challenge memory",
|
||
"family": "family memory",
|
||
"beliefs": "reflective life memory",
|
||
"summary": "memoir summary scene",
|
||
}
|
||
|
||
def __init__(self, llm: Optional[Any], settings: MemoirImageSettings):
|
||
self.llm = llm
|
||
self.settings = settings
|
||
|
||
def build_prompt(
|
||
self,
|
||
chapter_title: str,
|
||
chapter_category: str,
|
||
description: str,
|
||
context_excerpt: str,
|
||
) -> dict[str, str]:
|
||
style = self.CATEGORY_STYLE_MAP.get(
|
||
chapter_category, self.settings.default_style
|
||
)
|
||
prompt_context = f"{chapter_category}: {chapter_title}"
|
||
|
||
llm_input = {
|
||
"chapter_title": chapter_title,
|
||
"chapter_category": chapter_category,
|
||
"description": description,
|
||
"context_excerpt": context_excerpt,
|
||
"default_style": style,
|
||
"default_size": self.settings.default_size,
|
||
}
|
||
|
||
if self.llm:
|
||
raw_response = None
|
||
try:
|
||
prompt_text = (
|
||
"Return JSON only with keys prompt, style, size. "
|
||
"Convert the memoir scene into an image-generation prompt. "
|
||
"The API uses response_format=json_object.\n"
|
||
+ json.dumps(llm_input, ensure_ascii=False)
|
||
)
|
||
raw_response = invoke_json_object(
|
||
self.llm,
|
||
prompt_text,
|
||
max_tokens=512,
|
||
agent="MemoirImagePromptService.build_prompt",
|
||
)
|
||
parsed = json.loads(extract_json_payload(raw_response))
|
||
return {
|
||
"prompt": _ensure_style_in_prompt(
|
||
parsed["prompt"], parsed.get("style", style)
|
||
),
|
||
"style": parsed.get("style", style),
|
||
"size": parsed.get("size", self.settings.default_size),
|
||
"prompt_context": prompt_context,
|
||
}
|
||
except Exception as exc:
|
||
if memoir.image_prompt_fallback_disabled:
|
||
raise
|
||
logger.warning(
|
||
"图片 prompt 生成回退到默认模板: chapter_category={}, title={}, error={}",
|
||
chapter_category,
|
||
chapter_title,
|
||
exc,
|
||
)
|
||
elif memoir.image_prompt_fallback_disabled:
|
||
raise RuntimeError(
|
||
"MemoirImagePromptService.build_prompt requires LLM when "
|
||
"image_prompt_fallback_disabled is True"
|
||
)
|
||
|
||
return {
|
||
"prompt": _ensure_style_in_prompt(
|
||
self._build_fallback_prompt(
|
||
chapter_category=chapter_category,
|
||
description=description,
|
||
context_excerpt=context_excerpt,
|
||
style=style,
|
||
),
|
||
style,
|
||
),
|
||
"style": style,
|
||
"size": self.settings.default_size,
|
||
"prompt_context": prompt_context,
|
||
}
|
||
|
||
def build_cover_prompt(
|
||
self,
|
||
chapter_title: str,
|
||
chapter_category: str,
|
||
context_excerpt: str,
|
||
) -> dict[str, str]:
|
||
"""生成章节封面图的 image-generation prompt。"""
|
||
excerpt = (context_excerpt or "").strip()
|
||
if memoir.image_prompt_fallback_disabled and not excerpt:
|
||
raise RuntimeError(
|
||
"Chapter cover prompt requires non-empty context_excerpt when "
|
||
"image_prompt_fallback_disabled is True"
|
||
)
|
||
|
||
style = self.CATEGORY_STYLE_MAP.get(
|
||
chapter_category, self.settings.default_style
|
||
)
|
||
prompt_context = f"{chapter_category}: {chapter_title}"
|
||
|
||
llm_input = {
|
||
"chapter_title": chapter_title,
|
||
"chapter_category": chapter_category,
|
||
"context_excerpt": excerpt,
|
||
"default_style": style,
|
||
"default_size": self.settings.default_size,
|
||
}
|
||
|
||
if self.llm:
|
||
try:
|
||
prompt_text = (
|
||
"Return JSON only with keys prompt, style, size. "
|
||
"Create an image-generation prompt for a memoir chapter COVER. "
|
||
"Emphasize: hero composition, evocative scene, chapter cover aesthetic. "
|
||
"The API uses response_format=json_object.\n"
|
||
+ json.dumps(llm_input, ensure_ascii=False)
|
||
)
|
||
raw = invoke_json_object(
|
||
self.llm,
|
||
prompt_text,
|
||
max_tokens=512,
|
||
agent="MemoirImagePromptService.build_cover_prompt",
|
||
)
|
||
parsed = json.loads(extract_json_payload(raw))
|
||
return {
|
||
"prompt": _ensure_style_in_prompt(
|
||
parsed["prompt"], parsed.get("style", style)
|
||
),
|
||
"style": parsed.get("style", style),
|
||
"size": parsed.get("size", self.settings.default_size),
|
||
"prompt_context": prompt_context,
|
||
}
|
||
except Exception as exc:
|
||
if memoir.image_prompt_fallback_disabled:
|
||
raise
|
||
logger.warning(
|
||
"封面 prompt 生成回退到默认模板: chapter_category={}, title={}, error={}",
|
||
chapter_category,
|
||
chapter_title,
|
||
exc,
|
||
)
|
||
elif memoir.image_prompt_fallback_disabled:
|
||
raise RuntimeError(
|
||
"MemoirImagePromptService.build_cover_prompt requires LLM when "
|
||
"image_prompt_fallback_disabled is True"
|
||
)
|
||
|
||
return {
|
||
"prompt": _ensure_style_in_prompt(
|
||
self._build_cover_fallback_prompt(
|
||
chapter_category=chapter_category,
|
||
context_excerpt=excerpt,
|
||
style=style,
|
||
),
|
||
style,
|
||
),
|
||
"style": style,
|
||
"size": self.settings.default_size,
|
||
"prompt_context": prompt_context,
|
||
}
|
||
|
||
def build_story_primary_prompt(
|
||
self,
|
||
story_title: str,
|
||
story_stage: str | None,
|
||
prompt_brief: str,
|
||
style_profile: str | None,
|
||
) -> dict[str, str]:
|
||
"""生成 story 主插图的 image-generation prompt（LLM / fallback 策略与章节配图一致）。
|
||
|
||
`story_stage` 与 `Story.stage` 一致：通常为章节 category（如 career_early），也可能为
|
||
访谈五阶段名（childhood/career/…）；二者都参与默认画风推断。
|
||
"""
|
||
from app.agents.stage_constants import STAGE_TO_DEFAULT_CATEGORY
|
||
|
||
brief = (prompt_brief or "").strip()
|
||
if memoir.image_prompt_fallback_disabled and not brief:
|
||
raise RuntimeError(
|
||
"Story image prompt requires non-empty prompt_brief when "
|
||
"image_prompt_fallback_disabled is True"
|
||
)
|
||
|
||
stage_key = (story_stage or "").strip()
|
||
if stage_key in self.CATEGORY_STYLE_MAP:
|
||
cat = stage_key
|
||
else:
|
||
cat = STAGE_TO_DEFAULT_CATEGORY.get(stage_key, "summary")
|
||
explicit_style = (style_profile or "").strip()
|
||
style = explicit_style or self.CATEGORY_STYLE_MAP.get(
|
||
cat, self.settings.default_style
|
||
)
|
||
prompt_context = f"story:{stage_key}:{story_title}"
|
||
|
||
llm_input = {
|
||
"story_title": story_title,
|
||
"story_stage": stage_key,
|
||
"prompt_brief": brief,
|
||
"default_style": style,
|
||
"default_size": self.settings.default_size,
|
||
}
|
||
|
||
if self.llm:
|
||
try:
|
||
prompt_text = (
|
||
"Return JSON only with keys prompt, style, size. "
|
||
"Convert into an image-generation prompt for the PRIMARY hero illustration "
|
||
"of a personal memoir story (one focal scene, emotionally resonant). "
|
||
"The API uses response_format=json_object.\n"
|
||
+ json.dumps(llm_input, ensure_ascii=False)
|
||
)
|
||
raw_response = invoke_json_object(
|
||
self.llm,
|
||
prompt_text,
|
||
max_tokens=512,
|
||
agent="MemoirImagePromptService.build_story_primary_prompt",
|
||
)
|
||
parsed = json.loads(extract_json_payload(raw_response))
|
||
return {
|
||
"prompt": _ensure_style_in_prompt(
|
||
parsed["prompt"], parsed.get("style", style)
|
||
),
|
||
"style": parsed.get("style", style),
|
||
"size": parsed.get("size", self.settings.default_size),
|
||
"prompt_context": prompt_context,
|
||
}
|
||
except Exception as exc:
|
||
if memoir.image_prompt_fallback_disabled:
|
||
raise
|
||
logger.warning(
|
||
"story 主图 prompt 生成回退到默认模板: stage={}, title={}, error={}",
|
||
story_stage,
|
||
story_title,
|
||
exc,
|
||
)
|
||
elif memoir.image_prompt_fallback_disabled:
|
||
raise RuntimeError(
|
||
"MemoirImagePromptService.build_story_primary_prompt requires LLM when "
|
||
"image_prompt_fallback_disabled is True"
|
||
)
|
||
|
||
from .image_placeholder_template import IMAGE_PLACEHOLDER_TEMPLATE
|
||
|
||
if brief:
|
||
body = brief
|
||
else:
|
||
joined = "，".join(
|
||
filter(
|
||
None,
|
||
[(story_title or "").strip(), stage_key or None],
|
||
)
|
||
)
|
||
body = joined or "人生故事"
|
||
prompt_line = f"{IMAGE_PLACEHOLDER_TEMPLATE}。{body}"
|
||
return {
|
||
"prompt": _ensure_style_in_prompt(prompt_line, style),
|
||
"style": style,
|
||
"size": self.settings.default_size,
|
||
"prompt_context": prompt_context,
|
||
}
|
||
|
||
def _build_cover_fallback_prompt(
|
||
self,
|
||
chapter_category: str,
|
||
context_excerpt: str,
|
||
style: str,
|
||
) -> str:
|
||
subject = self.CATEGORY_FALLBACK_SUBJECT_MAP.get(
|
||
chapter_category, "memoir scene"
|
||
)
|
||
if _contains_cjk(context_excerpt):
|
||
return (
|
||
f"A {style} chapter cover illustration of a {subject}, "
|
||
"hero composition, evocative scene, emotionally resonant, "
|
||
"cinematic framing, natural lighting, no text overlay."
|
||
)
|
||
details = (context_excerpt or "").strip()[:500]
|
||
if not details:
|
||
details = "A personal life story scene with authentic emotional detail"
|
||
return (
|
||
f"A {style} chapter cover illustration of a {subject}. "
|
||
f"Scene hint: {details}. "
|
||
"Hero composition, evocative scene, cinematic framing, no text overlay."
|
||
)
|
||
|
||
def _build_fallback_prompt(
|
||
self,
|
||
chapter_category: str,
|
||
description: str,
|
||
context_excerpt: str,
|
||
style: str,
|
||
) -> str:
|
||
subject = self.CATEGORY_FALLBACK_SUBJECT_MAP.get(
|
||
chapter_category, "memoir scene"
|
||
)
|
||
if _contains_cjk(description) or _contains_cjk(context_excerpt):
|
||
return (
|
||
f"A {style} illustration of a {subject}, emotionally resonant, cinematic composition, "
|
||
"authentic everyday details, natural lighting, expressive environment, no text overlay."
|
||
)
|
||
|
||
details = ". ".join(
|
||
part.strip() for part in (description, context_excerpt) if part.strip()
|
||
)
|
||
if not details:
|
||
details = "A personal life story scene with authentic emotional detail"
|
||
return (
|
||
f"A {style} illustration of a {subject}. "
|
||
f"Scene details: {details}. "
|
||
"Cinematic composition, authentic emotions, natural lighting, no text overlay."
|
||
)
|
||
|
||
|
||
def _contains_cjk(value: str) -> bool:
|
||
return bool(_CJK_RE.search(value or ""))
|
||
|
||
|
||
def _ensure_style_in_prompt(prompt: str, style: str) -> str:
|
||
cleaned_prompt = (prompt or "").strip()
|
||
cleaned_style = (style or "").strip()
|
||
if not cleaned_style:
|
||
return cleaned_prompt
|
||
if cleaned_style.lower() in cleaned_prompt.lower():
|
||
return cleaned_prompt
|
||
if not cleaned_prompt:
|
||
return cleaned_style
|
||
return f"{cleaned_style}, {cleaned_prompt}"
|