""" NarrativeAgent:生成创意标题和叙事改写。 叙事正文走 `get_narrative_json_prompt` / `get_narrative_merge_json_prompt`(传记作家式书面语 + 事实边界)。 """ from __future__ import annotations from typing import Any, Dict, Optional from app.agents.memoir.prompts import ( get_creative_title_json_prompt, get_narrative_json_prompt, get_narrative_merge_json_prompt, ) from app.agents.memoir.schemas import MemoirTitleOutput from app.agents.stage_constants import CHAPTER_CATEGORIES, chapter_category_display from app.core.config import settings from app.core.langchain_llm import invoke_json_object from app.core.llm_call import llm_json_call from app.core.logging import get_logger logger = get_logger(__name__) def _default_title_for(stage: str, language: str) -> str: if language == "en": cat = chapter_category_display(stage, language="en") or stage return f"{cat} Memory" return f"{CHAPTER_CATEGORIES.get(stage, stage)} 回忆" class NarrativeAgent: """生成章节标题和叙事正文""" def generate_title( self, stage: str, emotion: str, slots: Dict[str, str], user_profile: str = "", birth_year: Optional[int] = None, llm: Any = None, language: str = "zh", ) -> str: """生成创意标题。若无 LLM 则返回默认标题""" if not llm: return _default_title_for(stage, language) try: prompt = get_creative_title_json_prompt( stage=stage, emotion=emotion, slots=slots, user_profile=user_profile, birth_year=birth_year, language=language, ) default_title = _default_title_for(stage, language) def _title_fallback() -> MemoirTitleOutput: return MemoirTitleOutput(title=default_title) out = llm_json_call( llm, prompt, MemoirTitleOutput, max_tokens=settings.memoir_title_max_tokens, agent="NarrativeAgent.generate_title", fallback_factory=_title_fallback, ) title = (out.title or "").strip() if title: return title.strip('"') return default_title except Exception as e: logger.warning("NarrativeAgent 生成标题失败: {}", e) return _default_title_for(stage, language) def generate_narrative( self, stage: str, slots: Dict[str, str], new_content: str, existing_content: str = "", user_profile: str = "", birth_year: Optional[int] = None, llm: Any = None, background_voice: str = "default", occupation: str = "", *, fallback_plain_oral: str = "", language: str = "zh", ) -> str: """将新对话改写为叙述。若无 LLM 则直接拼接。 若 `existing_content` 非空(append 路径),使用整篇合并提示,输出覆盖全篇的有序段落。 `fallback_plain_oral`:仅含本段口述(勿传含 evidence 的组装串)。LLM 异常时只回退到 口述/旧正文拼接,避免把「本段用户口述+摘录」整包写入 story。 """ oral_fb = (fallback_plain_oral or "").strip() if not llm: if existing_content: if oral_fb: return f"{existing_content}\n\n{oral_fb}" return f"{existing_content}\n\n{new_content}" return oral_fb or new_content try: merge_mode = bool((existing_content or "").strip()) if merge_mode: prompt = get_narrative_merge_json_prompt( stage=stage, slots=slots, new_content=new_content, existing_content=existing_content, user_profile=user_profile, birth_year=birth_year, background_voice=background_voice, occupation=occupation, language=language, ) max_tokens = int(settings.memoir_narrative_merge_max_tokens) agent_name = "NarrativeAgent.generate_narrative_merge" else: prompt = get_narrative_json_prompt( stage=stage, slots=slots, new_content=new_content, existing_content=existing_content, user_profile=user_profile, birth_year=birth_year, background_voice=background_voice, occupation=occupation, language=language, ) max_tokens = int(settings.memoir_narrative_max_tokens) agent_name = "NarrativeAgent.generate_narrative" return invoke_json_object( llm, prompt, max_tokens=max_tokens, agent=agent_name, ).strip() except Exception as e: logger.warning("NarrativeAgent 生成叙事失败: {}", e) ex = (existing_content or "").strip() if ex and oral_fb: return f"{existing_content}\n\n{oral_fb}" if oral_fb: return oral_fb if ex: return str(existing_content) return ""