xianjing-ai/engine/app/scheme_engine.py

from __future__ import annotations

from typing import Any

from .config_loader import business_config
from .decision import evaluate_tree
from .intent import (
    greeting_reply,
    has_offtopic_topic,
    is_domain_related,
    is_greeting,
    looks_like_question,
    looks_off_topic,
    refuse_reply,
    should_write_slot,
)
from .retrieval import retrieve_for_scheme, retrieve_question
from .session_store import (
    Session,
    looks_like_handoff,
    naive_fill_slots,
    next_slot_prompt,
    store,
)


def create_session() -> Session:
    return store.create()


def handle_message(session_id: str, message: str) -> dict[str, Any]:
    session = store.get(session_id)
    if not session:
        raise KeyError("session_not_found")

    text = (message or "").strip()
    session.messages.append({"role": "user", "content": text})
    nxt = next_slot_prompt(business_config.slots, session.slots)
    next_prompt = nxt["prompt"] if nxt else None

    if looks_like_handoff(text, business_config.handoff_phrases):
        session.phase = "handoff"
        reply = "已为您标记转人工。请留下称呼与联系方式，销售同事会跟进。（原型：仅记录意图，未接 Chatwoot）"
        return _finish(session, reply, need_handoff=True)

    # 寒暄：回问候 + 合理追问，绝不把 hello 写进槽位
    if is_greeting(text):
        return _finish(session, greeting_reply(next_prompt), asking_slot=nxt["id"] if nxt else None)

    # 明显跑题：拒答并拉回当前追问
    if looks_off_topic(text):
        return _finish(
            session,
            refuse_reply("off_topic", next_prompt),
            asking_slot=nxt["id"] if nxt else None,
        )

    # 业务相关自由提问：先查知识库；没命中则拒答，不强行灌槽
    # 同时夹带明显跑题话题（如「记录仪能炒股吗」）→ 直接拒答
    if looks_like_question(text) and is_domain_related(text):
        if has_offtopic_topic(text):
            return _finish(
                session,
                refuse_reply("off_topic", next_prompt),
                asking_slot=nxt["id"] if nxt else None,
            )
        qa = retrieve_question(text)
        if not qa.chunks:
            return _finish(
                session,
                refuse_reply("kb_miss", next_prompt),
                asking_slot=nxt["id"] if nxt else None,
            )
        lines = ["根据知识库，简要说明如下：", ""]
        for citation in qa.chunks[:3]:
            lines.append(f"- {citation.get('title')}：{citation.get('content')}")
        if next_prompt and session.phase == "collecting":
            lines.append("")
            lines.append(f"若要继续匹配方案：{next_prompt}")
        return _finish(session, "\n".join(lines), asking_slot=nxt["id"] if nxt else None)

    # 正常收集槽位
    if session.phase in ("collecting", "scheme_ready") and should_write_slot(text):
        session.slots = naive_fill_slots(text, business_config.slots, session.slots)

    nxt = next_slot_prompt(business_config.slots, session.slots)
    if nxt and session.phase == "collecting":
        reply = nxt["prompt"]
        return _finish(session, reply, asking_slot=nxt["id"])

    # 槽位齐了 → 出方案；知识库完全无依据时明确说明无法背书
    decision = evaluate_tree(business_config.rules, session.slots)
    confidence = float(decision.get("confidence") or 0)
    need_handoff = bool(decision.get("handoff")) or confidence < business_config.handoff_confidence_below
    bundle = list(decision.get("product_bundle") or [])
    retrieval = retrieve_for_scheme(session.slots, bundle)

    # 槽位内容本身不像业务需求，且检索也空 → 拒答
    slot_blob = " ".join(str(v) for v in session.slots.values() if v)
    if not is_domain_related(slot_blob) and not retrieval.chunks:
        reply = refuse_reply("kb_miss", None)
        session.phase = "collecting"
        # 清空无效槽，避免一直卡在坏状态
        session.slots = {}
        first = next_slot_prompt(business_config.slots, session.slots)
        if first:
            reply = refuse_reply("off_topic", first["prompt"])
            return _finish(session, reply, asking_slot=first["id"])
        return _finish(session, reply)

    scheme = {
        "tech_line": decision.get("tech_line"),
        "product_bundle": bundle,
        "summary": decision.get("summary"),
        "confidence": confidence,
        "rule_id": decision.get("rule_id"),
        "retrieval_backend": retrieval.backend,
        "retrieval_warning": retrieval.warning,
        "citations": [
            {
                "id": c.get("id"),
                "title": c.get("title"),
                "content": c.get("content"),
                "score": c.get("score"),
            }
            for c in retrieval.chunks
        ],
        "slots": dict(session.slots),
    }
    session.last_scheme = scheme
    session.phase = "handoff" if need_handoff else "scheme_ready"

    lines = [
        business_config.closing_prompt,
        "",
        f"技术路线：{scheme['tech_line']}",
        f"推荐组合：{', '.join(bundle)}",
        f"说明：{scheme['summary']}",
        f"置信度：{confidence:.0%}",
        f"检索：{retrieval.backend}",
    ]
    if retrieval.warning:
        lines.append(f"注意：{retrieval.warning}")
    if scheme["citations"]:
        lines.append("")
        lines.append("知识依据：")
        for citation in scheme["citations"][:3]:
            lines.append(f"- {citation['title']}：{citation['content']}")
    else:
        lines.append("")
        lines.append("知识库未给出可引用片段，请勿把上述组合当成已确认报价。若问题超出资料范围，我无法保证回答准确。")
    if need_handoff or retrieval.warning:
        lines.append("")
        lines.append("建议转人工确认后再发给客户。可在右侧留下称呼和电话。")
    else:
        lines.append("")
        lines.append("可继续补充约束，或留下联系方式转销售确认。")

    reply = "\n".join(lines)
    return _finish(session, reply, scheme=scheme, need_handoff=need_handoff)


def _finish(
    session: Session,
    reply: str,
    *,
    asking_slot: str | None = None,
    scheme: dict[str, Any] | None = None,
    need_handoff: bool = False,
) -> dict[str, Any]:
    session.messages.append({"role": "assistant", "content": reply})
    store.touch(session)
    return {
        "session_id": session.id,
        "reply": reply,
        "phase": session.phase,
        "slots": session.slots,
        "asking_slot": asking_slot,
        "scheme": scheme or session.last_scheme,
        "need_handoff": need_handoff,
    }