cfc/cfc-langgraph/tests/qna/test_graph.py

"""qna 引擎单测：用 fake LLM 返回固定 JSON，验证图节点输出"""
import json
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))  # cfc-langgraph 根

from src.qna import schemas, graph, prompts  # noqa: E402


class FakeLLM:
    """返回固定决策 JSON 的假 LLM"""

    def __init__(self, decisions):
        self.decisions = list(decisions)
        self.calls = []

    def invoke(self, messages):
        self.calls.append(messages)
        d = self.decisions.pop(0) if len(self.decisions) > 1 else self.decisions[0]
        return type("R", (), {"content": json.dumps(d, ensure_ascii=False)})()


def make_scene(**kw):
    base = {
        "scene_key": "microbiome",
        "opening_prompt": "了解您的肠道健康状况",
        "dimensions_json": {"user": ["肠道状态"], "need": ["营养需求"]},
        "kb_scope": ["microbiome"],
        "max_questions": 12,
    }
    base.update(kw)
    return base


def test_decide_next_ask():
    llm = FakeLLM([{"action": "ask", "question": {
        "id": "q1", "type": "single", "text": "您多久吃一次蔬菜？",
        "options": [{"id": "a", "label": "每天"}]}, "reason": "了解饮食"}])
    state = graph.decide_next(llm, make_scene(), [], prompt_fn=prompts.build_decide_prompt)
    assert state["action"] == "ask"
    assert state["question"]["text"].startswith("您多久")


def test_decide_next_force_finish_when_max_reached():
    llm = FakeLLM([{"action": "ask", "question": {"id": "q9", "type": "text", "text": "x"}}])
    history = [{"question": {"text": f"q{i}"}, "answer": "a"} for i in range(12)]
    state = graph.decide_next(llm, make_scene(max_questions=12), history, prompt_fn=prompts.build_decide_prompt)
    assert state["action"] == "finish"


def test_decide_next_invalid_json_retries_once():
    llm = FakeLLM(["not json", {"action": "finish", "reason": "信息足够"}])
    state = graph.decide_next(llm, make_scene(), [], prompt_fn=prompts.build_decide_prompt)
    assert state["action"] == "finish"
    assert len(llm.calls) == 2  # 重试了一次


def test_generate_profile_structure():
    llm = FakeLLM([{"user_profile": [{"dimension": "肠道状态", "score": 70, "description": "偏健康",
                                       "evidence": ["答1"]}],
                    "need_profile": [{"dimension": "营养需求", "description": "补纤维",
                                      "evidence": ["答1"], "suggestion": "多吃粗粮"}]}])
    history = [{"question": {"text": "q1"}, "answer": "a"}]
    profile, kb_used = graph.generate_profile(llm, make_scene(), history, kb_context=[{"content": "菌属知识"}],
                                              prompt_fn=prompts.build_profile_prompt)
    assert "user_profile" in profile and "need_profile" in profile
    assert profile["user_profile"][0]["dimension"] == "肠道状态"
    assert 0 <= profile["user_profile"][0]["score"] <= 100
    assert kb_used is True