from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage from app.config import settings from app.monitoring import monitor_agent from app.prompt_service import get_prompt import logging logger = logging.getLogger(__name__) SYSTEM_PROMPT = """你是一位融合中国传统文化与现代心理学的家庭成长解读师。 请根据用户的先天画像数据,生成一段温情、专业、可执行的成长解读文案。 要求: 1. 用第二人称"你"称呼 2. 先认可先天特质优势,再给出1-2条心维度成长建议 3. 语气温暖克制,避免玄学恐吓或过度承诺 4. 800字以内,用中文,分2-3段 5. 以"心能量小贴士"收尾,给一个当天就能做的行动 """ class InnatePortraitAgent: def __init__(self): self.llm = ChatOpenAI( model=settings.llm_model, api_key=settings.llm_api_key, base_url=settings.llm_base_url, temperature=settings.llm_temperature, ) @monitor_agent("innate_portrait") async def run(self, portrait: dict) -> dict: """生成先天画像解读,失败返回空 reading(调用方降级模板)""" try: source_summary = self._summarize(portrait) messages = [ SystemMessage(content=await get_prompt("innate_portrait") or SYSTEM_PROMPT), HumanMessage(content=f"先天画像数据:\n{source_summary}"), ] response = await self.llm.ainvoke(messages) return {"reading": response.content.strip()} except Exception as e: logger.warning("先天画像解读生成失败: %s", e) return {"reading": ""} @staticmethod def _summarize(portrait: dict) -> str: parts = [] if portrait.get("zodiac"): parts.append(f"生肖: {portrait['zodiac']}") if portrait.get("bloodType"): parts.append(f"血型: {portrait['bloodType']}") if portrait.get("wuxingElements"): parts.append(f"五行: {portrait['wuxingElements']}") if portrait.get("mindBaseScore") is not None: parts.append(f"心先天基础分: {portrait['mindBaseScore']}") if portrait.get("wisdomBaseScore") is not None: parts.append(f"智先天基础分: {portrait['wisdomBaseScore']}") if portrait.get("eightCharacters"): parts.append(f"八字四柱: {portrait['eightCharacters']}") ns = portrait.get("numSoul") or {} if ns.get("lifePath") is not None: parts.append( f"生命灵数: {ns['lifePath']}" + (f"({ns['lifePathTitle']})" if ns.get("lifePathTitle") else "") ) return "\n".join(parts) if parts else "暂无画像数据"