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

"""
健康管家 LangGraph - 专为儿童健康咨询设计的对话图
Uses RAG retrieval from health knowledge base.
"""
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.messages import SystemMessage, HumanMessage
from app.rag.retriever import RagRetriever
from app.memory.store import MemoryManager
from app.config import settings
import logging

logger = logging.getLogger(__name__)


class HealthButlerState(TypedDict):
    query: str
    user_id: int
    child_id: int | None
    conversation_id: str | None
    context: dict | None
    answer: str | None
    sources: list[dict]
    memory_messages: list | None


HEALTH_BUTLER_PROMPT = """你是一个专业、温暖的家庭健康顾问，专注于儿童健康成长。

你的职责：
1. 儿童体质调理咨询（如：挑食、瘦弱、易疲劳、反复生病）
2. 科学喂养和营养建议（辅食搭配、食欲不振、营养补充）
3. 日常保健知识（科学睡眠、运动建议、季节防护）
4. 体质辨识基础咨询（气虚/阳虚/阴虚/痰湿等体质特征解读）
5. 健康问题初步判断和就医建议

回答原则：
1. 用中文，语气温暖专业，像一个关心孩子的家庭医生
2. 涉及具体健康问题，先问清年龄、性别、症状持续时间等关键信息
3. 不确定时明确说"建议咨询专业医生"，绝不臆断严重疾病
4. 回答结构：问题分析 → 建议方案 → 日常调养tip
5. 涉及儿童安全（性早熟、严重外伤、急性症状等）立刻建议就医

你可以通过检索健康知识库来获取依据。
"""


def create_health_butler_graph():
    """创建健康管家 StateGraph"""
    # LLM
    llm = ChatOpenAI(
        model=settings.llm_model,
        api_key=settings.llm_api_key,
        base_url=settings.llm_base_url,
        temperature=settings.llm_temperature,
    )

    # 嵌入和检索
    embeddings = OpenAIEmbeddings(
        model=settings.embedding_model,
        api_key=settings.effective_embedding_api_key,
        base_url=settings.effective_embedding_base_url,
    )
    retriever = RagRetriever(collection_name="health_knowledge")
    memory_mgr = MemoryManager(embeddings)

    # 带工具的 LLM（健康知识库检索）
    async def health_retrieve(query: str, user_id: int) -> list[dict]:
        try:
            results = await retriever.retrieve(
                query,
                filters={"user_id": user_id},
                k=3,
            )
            return results
        except Exception as e:
            logger.warning("健康知识检索失败: %s", e)
            return []

    async def generate_answer(state: HealthButlerState) -> dict:
        """带健康知识检索的 LLM 生成"""
        messages = [SystemMessage(content=HEALTH_BUTLER_PROMPT)]

        # 家庭上下文
        ctx = state.get("context") or {}
        if ctx:
            child_info = []
            if ctx.get("child_id"):
                child_info.append(f"孩子ID: {ctx['child_id']}")
            if ctx.get("family_id"):
                child_info.append(f"家庭ID: {ctx['family_id']}")
            if child_info:
                messages.append(SystemMessage(
                    content="用户背景信息: " + ", ".join(child_info)
                ))

        # 健康知识检索
        retrieved = await health_retrieve(state["query"], state["user_id"])
        sources = []
        if retrieved:
            context_texts = []
            for i, r in enumerate(retrieved[:3]):
                src = r.get("metadata", {})
                title = src.get("title", "健康知识")
                content = r["content"][:300]
                context_texts.append(f"[{i+1}] {title}:\n{content}")
                sources.append({
                    "type": "knowledge",
                    "title": title,
                    "content": content,
                })
            knowledge_context = "\n\n---\n".join(context_texts)
            messages.append(SystemMessage(
                content=f"以下是与问题相关的健康知识参考：\n{knowledge_context}"
            ))

        # 长期记忆
        try:
            memories = await memory_mgr.recall(state["user_id"], state["query"])
            if memories:
                mem_text = "\n".join([f"- {m}" for m in memories[:3]])
                messages.append(SystemMessage(
                    content=f"该用户的历史健康咨询记录:\n{mem_text}"
                ))
        except Exception as e:
            logger.warning("召回健康记忆失败: %s", e)

        # 用户问题
        messages.append(HumanMessage(content=state["query"]))

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

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

    async def save_memory(state: HealthButlerState) -> dict:
        """保存对话到长期记忆"""
        try:
            if state.get("memory_messages"):
                await memory_mgr.save_conversation(
                    state["user_id"],
                    state.get("conversation_id") or f"health_{state['user_id']}",
                    state["memory_messages"],
                )
        except Exception as e:
            logger.warning("保存健康管家记忆失败: %s", e)
        return {}

    # ── 构建图 ──

    builder = StateGraph(HealthButlerState)

    builder.add_node("generate_answer", generate_answer)
    builder.add_node("save_memory", save_memory)

    builder.add_edge(START, "generate_answer")
    builder.add_edge("generate_answer", "save_memory")
    builder.add_edge("save_memory", END)

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

    return graph