cfc/cfc-langgraph/src/qna/graph.py

"""qna 引擎 - 无状态纯函数（可单测），供 router 调用"""
import json
import re
from typing import Optional

from . import schemas, prompts  # noqa: F401


def _extract_keywords(history: list) -> list:
    texts = []
    for h in history:
        texts.append(h.get("question", {}).get("text", ""))
        texts.append(h.get("answer", ""))
    joined = " ".join(texts)
    stopwords = {"请问", "帮我", "怎么", "什么", "如何", "是否", "一个", "这个", "那个"}
    words = re.findall(r"[\u4e00-\u9fff]{2,6}", joined)
    seen = []
    for w in words:
        w = w.strip()
        if w and w not in stopwords and w not in seen:
            seen.append(w)
    return seen[:8]


def _extract_json(text: str) -> dict:
    t = text.strip()
    if "```" in t:
        t = t.split("```json")[-1].split("```")[0].strip() if "```json" in t else t.split("```")[1].split("```")[0].strip()
    start, end = t.find("{"), t.rfind("}")
    if start == -1 or end == -1:
        raise ValueError("无 JSON 块")
    return json.loads(t[start:end + 1])


def decide_next(llm, scene: dict, history: list, prompt_fn) -> dict:
    """根据场景+知识+历史，输出 {action, question?, reason?}。LLM 输出非法自动重试 1 次。"""
    max_q = int(scene.get("max_questions", 12))
    if len(history) >= max_q:
        return {"action": "finish", "reason": "已达最大题目数"}
    prompt = prompt_fn(scene, "", history)  # 知识在 router 中检索后注入，见 retrieve_and_decide
    last_err = None
    for attempt in range(2):
        try:
            resp = llm.invoke([{"role": "human", "content": prompt}])
            data = _extract_json(resp.content)
            action = data.get("action")
            if action == "ask":
                q = data.get("question", {})
                qobj = schemas.Question(**q)
                if not qobj.is_valid():
                    raise ValueError(f"题目结构非法: {q}")
                return {"action": "ask", "question": qobj.model_dump(), "reason": data.get("reason", "")}
            if action == "finish":
                return {"action": "finish", "reason": data.get("reason", "信息已足够")}
            raise ValueError(f"未知 action: {action}")
        except Exception as e:
            last_err = str(e)
    return {"action": "ask", "fallback": True, "reason": f"LLM 输出异常: {last_err}",
            "question": {"id": f"fb{len(history)+1}", "type": "text", "text": "请简单描述您最近一周的饮食情况。"}}


def generate_profile(llm, scene: dict, history: list, kb_context: Optional[list], prompt_fn) -> tuple:
    kb_text = ""
    if kb_context:
        kb_text = "\n---\n".join(d.get("content", "") for d in kb_context[:5])
    prompt = prompt_fn(scene, kb_text, history)
    try:
        resp = llm.invoke([{"role": "human", "content": prompt}])
        data = _extract_json(resp.content)
        up = data.get("user_profile", [])
        np = data.get("need_profile", [])
        for item in up:
            item["score"] = max(0, min(100, int(item.get("score", 0))))
        return {"user_profile": up, "need_profile": np}, bool(kb_text)
    except Exception:
        return {"user_profile": [], "need_profile": []}, False


def retrieve_and_decide(llm, scene, history, kb_context, prompt_fn) -> dict:
    """知识注入版出题：把 kb_context 转文本后调用 LLM 决策（供 router 使用）"""
    kb_text = ""
    if kb_context:
        kb_text = "\n---\n".join(d.get("content", "") for d in kb_context[:5])
    prompt = prompt_fn(scene, kb_text, history)
    try:
        resp = llm.invoke([{"role": "human", "content": prompt}])
        data = _extract_json(resp.content)
        if data.get("action") == "finish":
            return {"action": "finish", "reason": data.get("reason", "信息已足够")}
        q = schemas.Question(**data["question"])
        return {"action": "ask", "question": q.model_dump(), "reason": data.get("reason", "")}
    except Exception as e:
        return {"fallback": True, "reason": f"出题失败: {e}",
                "question": {"id": f"fb{len(history)+1}", "type": "text",
                             "text": "请简单描述您最近一周的饮食和作息情况。"}}