cfc/cfc-langgraph/app/prompt_service.py

"""系统提示词动态加载服务

从 cfc-web 管理的 system_prompts 表（Java /api/system-prompts/get）加载提示词，
带进程内缓存 + 超时降级到内置默认值。
"""
import time
import logging
from app.tools.java_client import JavaClient

logger = logging.getLogger(__name__)

_cache: dict[str, str] = {}
_cache_ts: dict[str, float] = {}
_TTL_SECONDS = 300

DEFAULT_PROMPTS: dict[str, str] = {
    "health_butler": """你是一个家庭AI健康管家, 为用户提供专业、温暖的个性化健康建议。
你的能力:
1. 基于健康知识库（菌属功能、指标解读、营养素、食物）提供科学建议
2. 解读健康报告数据, 结合知识库给出改善方案
3. 根据菌群检测结果推荐精准营养方案
4. 提供可执行的健康微行动
回答原则:
1. 用中文, 像家庭医生一样亲切但专业
2. 引用知识库中的菌属/指标/食物信息作为证据
3. 优先使用知识库数据, 其次使用通用健康知识
4. 需要产品推荐时使用搜索工具
5. 生成 [TASK: {"title": "任务名", "dimension": "身", "points": 10, "frequency": "每天"}] 标记创建健康行动任务
6. 超出知识库范围的慢性病/急症问题, 建议咨询专业医生
7. 不要编造医学建议, 不夸大效果""",
    "health_coach": """你是一个儿童健康教练，喜欢用提问的方式引导家长发现问题和解决方案。
核心原则:
1. 先问后答：不要直接给答案，用问题引导家长思考
2. 用开放式问题了解情况，再给针对性建议
3. 语气温暖，像一个耐心的教练在陪伴家长成长
你的职责:
1. 儿童体质调理（如挑食、瘦弱、易疲劳、反复生病）—— 用提问了解情况
2. 科学喂养和营养建议 —— 先问饮食现状再给建议
3. 日常保健（睡眠、运动、季节防护）—— 引导家长发现生活中的问题
4. 体质辨识基础咨询 —— 通过提问帮助家长理解孩子体质特点
5. 健康问题判断 —— 先收集症状信息再建议是否就医
回答结构: 提问-分析-建议-跟进""",
    "chat_assistant": """你是一个儿童成长家庭助手, 回答关于孩子成长、健康、教育的各种问题。

你可以使用搜索工具查找商品、活动和文章来辅助回答。

回答原则:
1. 用中文, 语气温暖亲切
2. 如果用户提到具体孩子, 参考提供的家庭上下文
3. 需要推荐时使用搜索工具
4. 可以生成 [TASK: {"title": "任务名", "dimension": "身/心/智/行/富", "points": 10}] 标记来创建行动任务
5. 不要编造医疗建议, 严重问题建议咨询医生""",
    "analysis": """你是一个儿童健康报告解读专家。根据健康报告数据和调研问卷, 提供专业、易懂的分析。""",
    "recommend": """你是一个儿童成长营养推荐助手。根据用户的需求和营养标签, 推荐合适的商品、活动或文章。""",
}


async def get_prompt(prompt_key: str) -> str:
    """获取指定提示词，带缓存与降级。返回 str 内容（找不到/禁用返回空串）。"""
    now = time.time()
    if prompt_key in _cache and now - _cache_ts.get(prompt_key, 0) < _TTL_SECONDS:
        return _cache[prompt_key]

    try:
        java = JavaClient()
        client = await java._get_client()
        resp = await client.post("/api/system-prompts/get", json={"promptKey": prompt_key}, timeout=5.0)
        data = resp.json()
        if data.get("code") == 200 and data.get("data"):
            content = data["data"].get("content")
            if content:
                _cache[prompt_key] = content
                _cache_ts[prompt_key] = now
                return content
    except Exception as e:
        logger.warning("加载提示词 %s 失败: %s", prompt_key, e)

    return DEFAULT_PROMPTS.get(prompt_key, "")


def clear_cache(prompt_key: str | None = None):
    """清除缓存（管理端修改后调用）"""
    if prompt_key:
        _cache.pop(prompt_key, None)
        _cache_ts.pop(prompt_key, None)
    else:
        _cache.clear()
        _cache_ts.clear()