feat(qna): 动态问卷引擎核心（schemas/prompts/decide_next/generate_profile）

cfc-langgraph/.gitignore

@@ -0,0 +1,12 @@
+__pycache__/
+*.py[cod]
+.env
+data/
+*.egg-info/
+dist/
+.ruff_cache/
+.pytest_cache/
+
+# Un-ignore .py files (root .gitignore has *.py)
+!*.py
+.venv-bin.tar.gz\n.venv-packages.tar.gz

cfc-langgraph/src/qna/graph.py

@@ -0,0 +1,76 @@
+"""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):
+        resp = llm.invoke([{"role": "human", "content": prompt}])
+        try:
+            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)
+    resp = llm.invoke([{"role": "human", "content": prompt}])
+    try:
+        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

cfc-langgraph/src/qna/prompts.py

@@ -0,0 +1,57 @@
+"""qna 引擎 - prompt 模板（场景可配维度，引擎不硬编码）"""
+import json
+
+
+def build_decide_prompt(scene: dict, knowledge_text: str, history: list) -> str:
+    kb = knowledge_text or "(知识库未命中，请基于通用健康知识回答)"
+    hist = "\n".join(
+        f"Q{i+1}: {h['question']['text']} → 答: {h['answer']}" for i, h in enumerate(history)
+    ) or "(问卷刚开始)"
+    dims = json.dumps(scene.get("dimensions_json", {}), ensure_ascii=False)
+    return f"""你是智能健康问卷助手。根据用户已答内容，动态生成下一题，用于最终生成用户画像与需求画像。
+
+【场景】{scene.get('scene_name', scene['scene_key'])}
+【开场引导】{scene.get('opening_prompt', '')}
+【画像维度】(出题时请围绕这些维度收集信息) {dims}
+【知识库参考】
+{kb}
+
+【用户历史回答】
+{hist}
+
+【规则】
+1. 若信息已足够覆盖画像维度，输出 finish；否则输出 ask 出一题。
+2. 题目类型 single=单选(带options) / multi=多选(带options) / scale=量表(带scale) / text=自由文本。
+3. 问题需结合知识库内容与用户回答，有针对性；不要问与已答重复的信息。
+4. 严格输出 JSON，不要输出其他内容：
+{{"action": "ask"|"finish", "question": {{"id": "q1", "type": "single", "text": "题目", "options": [{{"id": "a", "label": "选项"}}]}}, "reason": "简短说明"}}
+"""
+
+
+def build_profile_prompt(scene: dict, knowledge_text: str, history: list) -> str:
+    kb = knowledge_text or "(知识库未命中)"
+    hist = "\n".join(
+        f"Q{i+1}: {h['question']['text']} → 答: {h['answer']}" for i, h in enumerate(history)
+    )
+    dims = json.dumps(scene.get("dimensions_json", {}), ensure_ascii=False)
+    return f"""你是智能健康分析助手。基于用户问卷回答与知识库，生成用户画像与需求画像。
+
+【画像维度定义】(维度名称由场景配置，不得新增未定义维度)
+{dims}
+
+【知识库参考】(引用其中的菌属/营养素/指标知识作为依据)
+{kb}
+
+【用户问卷历史】
+{hist}
+
+【输出要求】严格 JSON：
+{{
+  "user_profile": [
+    {{"dimension": "维度名", "score": 0-100, "description": "分析", "evidence": ["证据1"]}}
+  ],
+  "need_profile": [
+    {{"dimension": "维度名", "description": "需求分析", "evidence": ["证据"], "suggestion": "建议"}}
+  ]
+}}
+"""

cfc-langgraph/src/qna/schemas.py

@@ -0,0 +1,61 @@
+"""qna 动态问卷引擎 - Pydantic 模型"""
+from typing import List, Optional, Union
+from pydantic import BaseModel, Field
+
+
+class Option(BaseModel):
+    id: str
+    label: str
+
+
+class Scale(BaseModel):
+    min: int = 0
+    max: int = 10
+    minLabel: str = "从不"
+    maxLabel: str = "每天"
+
+
+class Question(BaseModel):
+    id: str
+    type: str = Field(pattern="^(single|multi|scale|text)$")
+    text: str
+    options: Optional[List[Option]] = None
+    scale: Optional[Scale] = None
+
+    def is_valid(self) -> bool:
+        if self.type in ("single", "multi"):
+            return bool(self.options) and not self.scale
+        if self.type == "scale":
+            return self.scale is not None
+        return True  # text
+
+
+class HistoryItem(BaseModel):
+    question: Question
+    answer: str
+
+
+class SceneConfig(BaseModel):
+    scene_key: str
+    opening_prompt: str = ""
+    dimensions_json: dict = {}
+    kb_scope: List[str] = ["microbiome"]
+    max_questions: int = 12
+    system_prompt: Optional[str] = None
+
+
+class QnaRequest(BaseModel):
+    scene: SceneConfig
+    history: List[HistoryItem] = []
+
+
+class QnaResponse(BaseModel):
+    action: str  # ask | finish
+    question: Optional[Question] = None
+    reason: Optional[str] = None
+    finished: bool = False
+
+
+class ProfileResponse(BaseModel):
+    profile: dict
+    kb_used: bool = False

cfc-langgraph/tests/qna/test_graph.py

@@ -0,0 +1,69 @@
+"""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