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

from typing import TypedDict, Optional, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from app.tools.java_client import JavaClient
from app.tools.product_tools import (
    search_product_by_keyword,
    search_article_by_keyword,
)
from app.memory.store import MemoryManager
from app.config import settings
from app.prompt_service import get_prompt
from app.portrait_service import get_user_portrait_text
import logging
import json

logger = logging.getLogger(__name__)


class HealthButlerState(TypedDict):
    query: str
    user_id: int
    conversation_id: str
    family_id: int | None
    child_id: int | None
    report_id: int | None
    focus: str | None          # gut / nutrition / overall / chronic
    kb_context: dict | None
    knowledge_results: list[dict]
    answer: str | None
    tasks: list[dict]
    sources: list[dict]
    messages: list | None


DEFAULT_BUTLER_PROMPT = """你是一个家庭AI健康管家, 为用户提供专业、温暖的个性化健康建议。

你的能力:
1. 基于健康知识库（菌属功能、指标解读、营养素、食物）提供科学建议
2. 解读健康报告数据, 结合知识库给出改善方案
3. 根据菌群检测结果推荐精准营养方案
4. 提供可执行的健康微行动

回答原则:
1. 用中文, 像家庭医生一样亲切但专业
2. 引用知识库中的菌属/指标/食物信息作为证据
3. 优先使用知识库数据, 其次使用通用健康知识
4. 需要产品推荐时使用搜索工具
5. 生成 [TASK: {"title": "任务名", "dimension": "身", "points": 10, "frequency": "每天"}] 标记创建健康行动任务
6. 超出知识库范围的慢性病/急症问题, 建议咨询专业医生
7. 不要编造医学建议, 不夸大效果
"""


def create_health_butler_graph():
    """创建 AI 健康管家 StateGraph — 知识检索 + LLM 建议 + 记忆"""
    llm = ChatOpenAI(
        model=settings.llm_model,
        api_key=settings.llm_api_key,
        base_url=settings.llm_base_url,
        temperature=settings.llm_temperature,
    )

    from app.rag.embeddings import get_embeddings
    embeddings = get_embeddings()
    memory_mgr = MemoryManager(embeddings)

    java = JavaClient()

    llm_with_tools = llm.bind_tools([
        search_product_by_keyword,
        search_article_by_keyword,
    ])

    builder = StateGraph(HealthButlerState)

    async def load_knowledge(state: HealthButlerState) -> dict:
        """阶段1: 从 Java 后端加载健康知识库上下文"""
        context = {}
        try:
            client = await java._get_client()
            resp = await client.post(
                "/api/health/knowledge/list",
                json={},
            )
            data = resp.json()
            if data.get("code") == 200:
                items = data.get("data", [])
                # 按 itemType 分组
                by_type = {}
                for item in items:
                    t = item.get("itemType", "unknown")
                    if t not in by_type:
                        by_type[t] = []
                    by_type[t].append(item)
                context["kb_items"] = len(items)
                context["kb_by_type"] = {k: len(v) for k, v in by_type.items()}
                context["kb_loaded"] = True
            else:
                context["kb_loaded"] = False
        except Exception as e:
            logger.warning("加载健康知识库失败: %s", e)
            context["kb_loaded"] = False

        return {"kb_context": context}

    async def search_knowledge(state: HealthButlerState) -> dict:
        """阶段2: 根据 query 意图精准搜索知识库"""
        query = state["query"]
        results = []

        # 提取 query 中可能的菌属名/指标名/食物名关键词
        # 策略: 对 query 做简单的关键词匹配, 调用 Java 知识库逐个查询
        keywords = _extract_health_keywords(query)
        if keywords:
            client = await java._get_client()
            for kw in keywords[:5]:  # 最多查询5个关键词
                # 优先查 bacteria 类型
                for item_type in ["bacteria", "indicator", "food", "nutrient"]:
                    try:
                        resp = await client.post(
                            "/api/health/knowledge/query",
                            json={"itemType": item_type, "itemName": kw},
                        )
                        data = resp.json()
                        if data.get("code") == 200 and data.get("data"):
                            results.append({
                                "type": item_type,
                                "keyword": kw,
                                "data": data["data"],
                            })
                            break  # 命中即跳出
                    except Exception:
                        continue

        return {"knowledge_results": results}

    async def llm_answer(state: HealthButlerState) -> dict:
        """阶段3: LLM 生成个性化健康建议"""
        messages = [SystemMessage(content=await get_prompt("health_butler") or DEFAULT_BUTLER_PROMPT)]

        # 画像注入
        member_id = state.get("child_id")
        portrait_text = await get_user_portrait_text(java, state["user_id"], member_id)
        if portrait_text:
            messages.insert(1, SystemMessage(content=portrait_text))

        # 注入健康知识库上下文
        kb_ctx = state.get("kb_context", {})
        knowledge_results = state.get("knowledge_results", [])
        if knowledge_results:
            kb_text_parts = []
            for r in knowledge_results:
                d = r.get("data", {})
                name = d.get("itemName", r.get("keyword", ""))
                cat = d.get("category", "")
                desc = d.get("description", "")
                suggestion = d.get("suggestion", "")
                normal_range = d.get("normalRange", "")

                parts = [f"【{r['type']}】{name}"]
                if cat:
                    parts.append(f"(分类: {cat})")
                if normal_range:
                    parts.append(f"正常范围: {normal_range}")
                if desc:
                    parts.append(f"{desc}")
                if suggestion:
                    parts.append(f"建议: {suggestion}")
                kb_text_parts.append(" ".join(parts))

            kb_text = "\n---\n".join(kb_text_parts)
            messages.append(SystemMessage(content=f"健康知识库参考:\n{kb_text}"))

        # 注入家庭上下文
        family_id = state.get("family_id")
        if family_id:
            try:
                client = await java._get_client()
                resp = await client.post(
                    "/api/ai/context",
                    json={
                        "user_id": str(state["user_id"]),
                        "params": {
                            "familyId": family_id,
                            **({"childId": state["child_id"]} if state.get("child_id") else {}),
                        },
                    },
                )
                data = resp.json()
                if data.get("code") == 200:
                    ctx = data.get("data", {})
                    if ctx:
                        messages.append(SystemMessage(content=f"家庭上下文:\n{json.dumps(ctx, ensure_ascii=False, indent=2)}"))
            except Exception as e:
                logger.debug("加载家庭上下文跳过: %s", e)

        # 注入长期记忆
        try:
            memories = await memory_mgr.recall(state["user_id"], state["query"])
            if memories:
                mem_text = "\n".join([f"- {m}" for m in memories])
                messages.append(SystemMessage(content=f"相关历史对话:\n{mem_text}"))
        except Exception as e:
            logger.warning("召回记忆失败: %s", e)

        # 用户消息
        focus = state.get("focus")
        query = state["query"]
        if focus:
            query = f"[重点关注: {focus}] {query}"
        messages.append(HumanMessage(content=query))

        response = await llm_with_tools.ainvoke(messages)
        answer = response.content

        # 提取任务
        import re
        tasks = []
        pattern = r'\[TASK:\s*\{([^}]+)\}\]'
        for match in re.findall(pattern, answer):
            task_info = {}
            for kv in match.split(","):
                if ":" in kv:
                    k, v = kv.split(":", 1)
                    task_info[k.strip()] = v.strip().strip('"').strip("'")
            if task_info.get("title"):
                tasks.append(task_info)

        sources = []
        for r in knowledge_results:
            d = r.get("data", {})
            sources.append({
                "type": "health_kb",
                "name": d.get("itemName", r.get("keyword", "")),
                "category": d.get("category", ""),
            })
        if response.response_metadata.get("tool_calls"):
            for tc in response.response_metadata["tool_calls"]:
                sources.append({
                    "type": "tool",
                    "name": tc.get("name", ""),
                })

        return {
            "answer": answer,
            "tasks": tasks,
            "sources": sources,
            "messages": [
                {"role": "user", "content": state["query"]},
                {"role": "assistant", "content": answer},
            ],
        }

    async def save_memory(state: HealthButlerState) -> dict:
        try:
            if state.get("messages"):
                await memory_mgr.save_conversation(
                    state["user_id"],
                    state["conversation_id"],
                    state["messages"],
                )
        except Exception as e:
            logger.warning("保存记忆失败: %s", e)
        return {}

    # ── 路由 ──

    def route_after_knowledge(state: HealthButlerState) -> Literal["search_knowledge", "llm_answer"]:
        kb_ctx = state.get("kb_context", {})
        if kb_ctx.get("kb_loaded"):
            return "search_knowledge"
        return "llm_answer"

    # ── 构建图 ──

    builder.add_node("load_knowledge", load_knowledge)
    builder.add_node("search_knowledge", search_knowledge)
    builder.add_node("llm_answer", llm_answer)
    builder.add_node("save_memory", save_memory)

    builder.add_edge(START, "load_knowledge")
    builder.add_conditional_edges("load_knowledge", route_after_knowledge)
    builder.add_edge("search_knowledge", "llm_answer")
    builder.add_edge("llm_answer", "save_memory")
    builder.add_edge("save_memory", END)

    checkpointer = MemorySaver()
    graph = builder.compile(checkpointer=checkpointer)

    return graph


def _extract_health_keywords(query: str) -> list[str]:
    """从 query 中提取健康相关关键词（菌属、指标、食物等名词）"""
    # 简单策略: 提取2字以上的中文词作为候选
    import re
    # 提取中文字符组成的连续词段
    chinese_words = re.findall(r'[\u4e00-\u9fff]{2,}', query)
    # 提取英文/拉丁名词组
    english_words = re.findall(r'[A-Za-z][A-Za-z\s]{2,20}[A-Za-z]', query)

    keywords = []
    # 过滤常见通用词
    stopwords = {"请问", "帮我", "能不能", "可不可以", "怎么样", "为什么",
                 "是什么", "怎么办", "什么意思", "好不好", "是否", "你好",
                 "谢谢", "请", "我想", "需要", "关于", "一个", "这个",
                 "那个", "哪些", "什么", "怎么", "如何", "多少", "是不是"}
    for w in chinese_words:
        if w not in stopwords and len(w) >= 2:
            keywords.append(w)

    for w in english_words:
        w = w.strip()
        if len(w) > 2:
            keywords.append(w)

    # 去重 + 限数量
    seen = set()
    deduped = []
    for kw in keywords:
        if kw not in seen:
            seen.add(kw)
            deduped.append(kw)

    return deduped[:10]