cd D:\workspace\knowforge-rag-platform\codealong\chapters\ch05_intent_classification
python -m unittest discover -s tests

cd D:\workspace\knowforge-rag-platform
pytest tests/test_intent_and_scenarios.py tests/test_retrieval_and_prompt.py -q

用户说："入职流程有哪些步骤"
意图：知识咨询(KNOWLEDGE_QUERY)
→ 需要检索文档 + LLM 生成答案

用户说："你好"
意图：问候(GREETING)
→ 直接返回问候语，不需要检索

意图识别结果
├─ 是否直接返回答案(跳过检索和生成)
├─ 是否需要结合历史改写追问
├─ FAQ 和文档各召回多少条
├─ FAQ 直出阈值应该高还是低
├─ 是否能自动推断业务分类过滤项
└─ 使用哪套 Prompt Profile 生成答案

Intent = Literal[
    "GREETING",        # 问候 → 你好、在吗、你是谁
    "FOLLOW_UP",       # 追问 → 那审批呢、费用呢
    "KNOWLEDGE_QUERY", # 知识咨询 → 入职流程有哪些步骤
    "FAQ_QUERY",       # 标准问答 → API 限流怎么办
    "HUMAN_SERVICE",   # 人工客服 → 客服电话、转人工
    "OUT_OF_SCOPE",    # 越界 → 彩票、赌博、攻击
]

def classify_intent(
    query: str,
    history: list[BaseMessage],
    scenario: ScenarioDefinition | None = None
) -> IntentResult:
    normalized = query.strip().lower()
    suggested_source = infer_source(query, scenario)

    # 第一步：问候
    greeting_answers = [
        f"你好！我是{assistant_name}，可以帮你查询{business_domain}中的...",
        f"我是{assistant_name}，负责解答{business_domain}相关问题。",
    ]
    for pattern, answer in zip(GREETING_PATTERNS, greeting_answers):
        if re.match(pattern, normalized, re.IGNORECASE):
            return IntentResult(
                intent="GREETING",
                direct_answer=answer,
                confidence=1.0,
                reason="greeting_rule"
            )

    # 第二步：安全拦截
    if OFF_TOPIC_HINTS.search(normalized):
        return IntentResult(
            intent="OUT_OF_SCOPE",
            direct_answer=f"这个问题超出了{business_domain}的问答范围，我无法提供帮助。",
            confidence=0.95,
            reason="safety_rule"
        )

    # 第三步：人工客服
    if HUMAN_SERVICE_HINTS.search(normalized) and len(normalized) <= 18:
        return IntentResult(
            intent="HUMAN_SERVICE",
            direct_answer=f"可以联系人工支持，联系方式：{support_contact}。",
            confidence=0.9,
            reason="human_service_rule",
        )

    # 第四步：追问识别
    if history and (
        FOLLOW_UP_HINTS.search(query.strip()) or len(query.strip()) <= 8
    ):
        return IntentResult(
            intent="FOLLOW_UP",
            confidence=0.8,
            reason="follow_up_rule",
            requires_rewrite=True,
        )

    # 第五步：强规则覆盖高频业务场景
    strong_rule_intent = _strong_rule_domain_intent(query, suggested_source)
    if strong_rule_intent is not None:
        return strong_rule_intent

    # 第六步：仍不确定才调用 LLM
    return _classify_with_llm(query, history, suggested_source, scenario)

def _strong_rule_domain_intent(
    query: str, suggested_source: str | None
) -> IntentResult | None:
    normalized = query.strip().lower()

    # 规则 1：FAQ 关键词 → FAQ_QUERY
    if FAQ_HINTS.search(normalized):
        return IntentResult(
            intent="FAQ_QUERY", confidence=0.82,
            reason="strong_faq_rule", suggested_source=suggested_source
        )

    # 规则 2：source 已推断 + 短问题 + 标准问法 → FAQ_QUERY
    if (suggested_source and len(normalized) <= 32
        and FAQ_QUESTION_SHAPE_HINTS.search(normalized)):
        return IntentResult(
            intent="FAQ_QUERY", confidence=0.83,
            reason="source_question_shape_rule", suggested_source=suggested_source
        )

    # 规则 3：source 已推断 + 口语化短问法 → FAQ_QUERY
    if (suggested_source and len(normalized) <= 36
        and DIRECT_FAQ_SHAPE_HINTS.search(normalized)):
        return IntentResult(
            intent="FAQ_QUERY", confidence=0.84,
            reason="direct_faq_shape_rule", suggested_source=suggested_source
        )

    # 规则 4：知识库关键词 + 有 source 或短问题 → KNOWLEDGE_QUERY
    if (KNOWLEDGE_HINTS.search(normalized)
        and (suggested_source or len(normalized) <= 24)):
        return IntentResult(
            intent="KNOWLEDGE_QUERY", confidence=0.84,
            reason="strong_knowledge_rule", suggested_source=suggested_source
        )

    return None  # 规则无法确定，交给 LLM

FAQ_HINTS = re.compile(
    r"(费用|价格|安装|环境|失败|报错|地址|时间|退费|优惠|"
    r"发票|账号|登录|权限|审批|合同|隐私|账单|支付|开票|工单|售后)"
)

FAQ_QUESTION_SHAPE_HINTS = re.compile(
    r"(怎么办|如何处理|怎么处理|需要什么|需要哪些|"
    r"需要准备哪些|有哪些|为什么|什么时候|由谁|能不能|会不会)"
)

DIRECT_FAQ_SHAPE_HINTS = re.compile(
    r"(资料呢|材料呢|是什么|如何回收|怎么排查|怎么处理|能不能|可以吗|要看什么)"
)

KNOWLEDGE_HINTS = re.compile(
    r"(知识库|文档|手册|流程|制度|规范|说明|配置|接口|功能|"
    r"排查|故障|步骤|sop|告警|巡检|设备|合规|条款|入职|"
    r"审批|合同|隐私|webhook|回调|发票|账单)"
)

def _classify_with_llm(
    query: str, history: list[BaseMessage],
    suggested_source: str | None,
    scenario: ScenarioDefinition | None = None,
) -> IntentResult:
    # 格式化最近 6 条历史消息
    history_text = format_messages(history[-6:]) if history else "无"

    # 创建带结构化输出的模型
    model = get_chat_model(streaming=False).with_structured_output(
        IntentLLMDecision
    )

    # 调用 LLM
    decision = model.invoke([
        SystemMessage(content=INTENT_SYSTEM_PROMPT),
        HumanMessage(content=f"对话历史：\n{history_text}\n\n当前问题：{query}"),
    ])

    # 校验 LLM 返回的 source 是否在场景白名单中
    source = decision.suggested_source or suggested_source
    valid_sources = scenario.valid_sources if scenario else []
    if source and source not in valid_sources:
        # 模型可能输出不在白名单中的分类名，必须丢弃
        source = suggested_source

    return IntentResult(
        intent=decision.intent,
        confidence=decision.confidence,
        reason=decision.reason or "llm_structured",
        requires_rewrite=(
            decision.requires_rewrite or decision.intent == "FOLLOW_UP"
        ),
        suggested_source=source,
    )

if source and source not in valid_sources:
    source = suggested_source  # 丢弃 LLM 的无效输出

# 注意：LLM 失败时直接抛异常，而不是悄悄改用规则
# 规则路径只在 LLM 之前运行，不作为 LLM 失败时的降级方案

def infer_source(query: str, scenario: ScenarioDefinition | None = None) -> str | None:
    """只根据当前业务场景配置推断可能的业务分类过滤项。

    该结果只是建议值，不会覆盖前端显式选择；如果没有传入场景，直接返回 None。
    """
    if scenario is None:
        return None
    best_source, _ = score_source_matches(query, scenario)
    return best_source

# qa_core/scenarios/boundary.py
def score_source_map(query: str, scenario: ScenarioDefinition) -> dict[str, int]:
    """计算问题在某个场景内命中各 source 的分数。

    分数来自场景 TOML 里的 source_patterns。命中次数越多、命中文本越长，
    分数越高；source 配置顺序用于平局时的优先级(后配置的 index 小、惩罚少)。
    """
    normalized = query.strip().lower()
    scores: dict[str, int] = {}
    for index, (source, pattern) in enumerate(scenario.compiled_source_patterns().items()):
        matches = list(pattern.finditer(query)) + list(pattern.finditer(normalized))
        if not matches:
            continue
        scores[source] = len(matches) * 10 + sum(len(match.group(0)) for match in matches) - index
    return scores

def score_source_matches(query: str, scenario: ScenarioDefinition) -> tuple[str | None, int]:
    """返回问题在某个场景内最像的 source 及其加权分数。"""
    scores = score_source_map(query, scenario)
    best_source = None
    best_score = 0
    for source, score in scores.items():
        if score > best_score:
            best_source = source
            best_score = score
    return best_source, best_score

@dataclass(frozen=True)
class IntentResult:
    intent: Intent                          # 意图枚举值
    direct_answer: str | None = None        # 直接答案(非空时跳过检索)
    confidence: float = 0.6                 # 置信度
    reason: str = "rule"                    # 判断原因(可解释性)
    requires_rewrite: bool = False          # 是否需要改写追问
    suggested_source: str | None = None     # 建议的业务分类

# intent → 检索计划参数
if intent.intent == "FAQ_QUERY":
    doc_top_k = doc_top_k // 2        # 减少文档候选
    direct_threshold = 0.62           # 降低直出阈值

elif intent.intent == "KNOWLEDGE_QUERY":
    doc_top_k *= 2                    # 增加文档上下文

# direct_answer → 是否跳过检索和生成
if intent.direct_answer:
    return direct_answer  # 不走 RAG 流程

# requires_rewrite → 是否调用改写模型
if intent.requires_rewrite:
    query = rewrite_query_if_needed(query, history)

# suggested_source → Milvus 过滤表达式
if suggested_source and not source_filter:
    source_filter = suggested_source

@dataclass(frozen=True)
class IntentResult:
    intent: Intent
    confidence: float = 0.6
    reason: str = ""
    direct_answer: str | None = None
    requires_rewrite: bool = False
    suggested_source: str | None = None

def classify_intent(query, history, scenario):
    normalized = query.strip().lower()
    suggested_source = infer_source(query, scenario)

    # 规则 1：问候/自我介绍，直接返回
    if GREETING_PATTERNS.match(normalized):
        return IntentResult("GREETING", direct_answer=greeting_text, confidence=1.0)

    # 规则 2：越界/风险话题，直接拒答
    if OFF_TOPIC_HINTS.search(normalized):
        return IntentResult("OUT_OF_SCOPE", direct_answer=refuse_text, confidence=0.95)

    # 规则 3：短句转人工，避免长文本偶然包含“客服”被误判
    if HUMAN_SERVICE_HINTS.search(normalized) and len(normalized) <= 18:
        return IntentResult("HUMAN_SERVICE", direct_answer=support_text, confidence=0.9)

    # 规则 4：有历史 + 省略表达，判定为追问，需要后续改写
    if history and (FOLLOW_UP_HINTS.search(query.strip()) or len(query.strip()) <= 8):
        return IntentResult("FOLLOW_UP", requires_rewrite=True, confidence=0.8)

    # 规则 5：强领域规则，识别 FAQ_QUERY / KNOWLEDGE_QUERY
    strong_rule_intent = _strong_rule_domain_intent(query, suggested_source)
    if strong_rule_intent is not None:
        return strong_rule_intent

    # 规则 6：前 5 条都不命中，再调用 LLM 结构化兜底
    return _classify_with_llm(query, history, suggested_source, scenario)

if FAQ_HINTS.search(normalized):
    return IntentResult("FAQ_QUERY", confidence=0.82, reason="strong_faq_rule")

if suggested_source and len(normalized) <= 32 and FAQ_QUESTION_SHAPE_HINTS.search(normalized):
    return IntentResult("FAQ_QUERY", confidence=0.83, reason="source_question_shape_rule")

if suggested_source and len(normalized) <= 36 and DIRECT_FAQ_SHAPE_HINTS.search(normalized):
    return IntentResult("FAQ_QUERY", confidence=0.84, reason="direct_faq_shape_rule")

if KNOWLEDGE_HINTS.search(normalized) and (suggested_source or len(normalized) <= 24):
    return IntentResult("KNOWLEDGE_QUERY", confidence=0.84, reason="strong_knowledge_rule")

def infer_source(query: str, scenario: ScenarioDefinition | None) -> str | None:
    if scenario is None:
        return None
    for source, patterns in scenario.source_patterns.items():
        if any(pattern in query for pattern in patterns):
            return source
    return None

model = get_chat_model(streaming=False).with_structured_output(IntentLLMDecision)
decision = model.invoke(messages)

return IntentResult(
    intent=decision.intent,
    confidence=decision.confidence,
    reason=decision.reason,
    requires_rewrite=decision.requires_rewrite or decision.intent == "FOLLOW_UP",
    suggested_source=validated_source,
)

cd D:\workspace\knowforge-rag-platform\codealong\chapters\ch05_intent_classification
python -m unittest discover -s tests

cd D:\workspace\knowforge-rag-platform
python -m pytest tests/test_intent_and_scenarios.py -q