cfc/cfc-langgraph/app/graphs/innate_portrait_graph.py

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 "暂无画像数据"