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

from typing import TypedDict, 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.agents.intent_classifier import IntentClassifier, Intent
from app.agents.chat_agent import ChatAgent
from app.tools.product_tools import (
    search_product_by_keyword,
    search_article_by_keyword,
    search_activity_by_keyword,
)
from app.memory.store import MemoryManager
from app.config import settings
from app.prompt_service import get_prompt
import logging

logger = logging.getLogger(__name__)


class ChatState(TypedDict):
    query: str
    user_id: int
    conversation_id: str
    child_id: int | None
    intent: Intent | None
    context: dict | None
    messages: list | None
    answer: str | None
    tasks: list[dict]
    sources: list[dict]


DEFAULT_CHAT_PROMPT = """你是一个儿童成长家庭助手, 回答关于孩子成长、健康、教育的各种问题。

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

回答原则:
1. 用中文, 语气温暖亲切
2. 如果用户提到具体孩子, 参考提供的家庭上下文
3. 需要推荐时使用搜索工具
4. 可以生成 [TASK: {"title": "任务名", "dimension": "身/心/智/行/富", "points": 10}] 标记来创建行动任务
5. 不要编造医疗建议, 严重问题建议咨询医生
"""


def create_chat_graph():
    """创建聊天 StateGraph"""
    agent = ChatAgent()
    classifier = IntentClassifier()

    llm = ChatOpenAI(
        model=settings.llm_model,
        api_key=settings.llm_api_key,
        base_url=settings.llm_base_url,
        temperature=settings.llm_temperature,
    )
    
    # 初始化嵌入模型用于记忆存储
    from langchain_openai import OpenAIEmbeddings
    embeddings = OpenAIEmbeddings(
        model=settings.embedding_model,
        api_key=settings.effective_embedding_api_key,
        base_url=settings.effective_embedding_base_url,
    )
    memory_mgr = MemoryManager(embeddings)

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

    builder = StateGraph(ChatState)

    # ── 节点 ──

    async def classify_intent(state: ChatState) -> dict:
        intent = await classifier.classify(
            state["query"],
            context=str(state.get("context", {})),
        )
        return {"intent": intent}

    async def load_context(state: ChatState) -> dict:
        ctx = await agent.load_context(state["user_id"], state.get("child_id"))
        return {"context": ctx}

    async def llm_call(state: ChatState) -> dict:
        """核心 LLM 调用 + Tool"""
        chat_prompt = await get_prompt("chat_assistant") or DEFAULT_CHAT_PROMPT
        messages = [SystemMessage(content=chat_prompt)]

        # 注入家庭上下文 + 用户健康现状
        ctx = state.get("context", {})
        if isinstance(ctx, dict):
            ctx_parts = []
            if ctx.get("child_id") is not None:
                ctx_parts.append(f"child_id: {ctx['child_id']}")
            if ctx.get("report_id") is not None:
                ctx_parts.append(f"report_id: {ctx['report_id']}")
            if ctx.get("family_id") is not None:
                ctx_parts.append(f"family_id: {ctx['family_id']}")
            if ctx.get("health_status"):
                ctx_parts.append(f"用户健康现状: {ctx['health_status']}")
            if ctx.get("diet_preferences"):
                ctx_parts.append(f"用户饮食偏好: {ctx['diet_preferences']}")
            if ctx_parts:
                messages.append(SystemMessage(content="\n家庭上下文:\n" + "\n".join(ctx_parts)))

        # 注入长期记忆
        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)

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

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

        # 提取任务
        tasks = await agent.extract_tasks(answer, state["user_id"], state["conversation_id"])

        # 提取来源
        sources = []
        if response.response_metadata.get("tool_calls"):
            for tc in response.response_metadata["tool_calls"]:
                sources.append({
                    "type": "tool",
                    "name": tc.get("name", ""),
                    "input": tc.get("args", {}),
                })

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

    async def save_memory(state: ChatState) -> 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_by_intent(state: ChatState) -> Literal["llm_call", END]:
        if state["intent"] in (
            Intent.RECOMMEND,
            Intent.ANALYSIS,
            Intent.HEALTH,
        ):
            # 这些意图需要更专业的 Agent (Phase 3 实现)
            # 当前先走通用 LLM
            pass
        return "llm_call"

    # ── 构建图 ──

    builder.add_node("classify_intent", classify_intent)
    builder.add_node("load_context", load_context)
    builder.add_node("llm_call", llm_call)
    builder.add_node("save_memory", save_memory)

    builder.add_edge(START, "classify_intent")
    builder.add_edge("classify_intent", "load_context")
    builder.add_conditional_edges("load_context", route_by_intent)
    builder.add_edge("llm_call", "save_memory")
    builder.add_edge("save_memory", END)

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

    return graph