From 625a6a3a9df696ce9a1c87307d9b43f47d9ec433 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 20 Sep 2026 05:52:22 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=A2=E6=9C=8D=E5=AF=B9=E8=AF=9D?= =?UTF-8?q?=E6=94=AF=E6=8C=81=E5=AF=92=E6=9A=84=E5=9B=9E=E5=A4=8D=E4=B8=8E?= =?UTF-8?q?=E8=B7=91=E9=A2=98=E6=8B=92=E7=AD=94?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 问候不再写入槽位；明显无关问题直接拒绝，知识库未命中时明确说明无法回答，并继续合理追问。 Co-authored-by: haijd --- engine/app/intent.py | 116 ++++++++++++++++++++++++++++++++++++ engine/app/retrieval.py | 62 +++++++++++++++++++ engine/app/scheme_engine.py | 81 +++++++++++++++++++++---- engine/app/session_store.py | 4 +- 4 files changed, 249 insertions(+), 14 deletions(-) create mode 100644 engine/app/intent.py diff --git a/engine/app/intent.py b/engine/app/intent.py new file mode 100644 index 0000000..7acc8e4 --- /dev/null +++ b/engine/app/intent.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +import re + +# 寒暄：不写槽、先回问候再追问 +_GREETING_RE = re.compile( + r"^\s*(hi|hello|hey|hola|yo|你好|您好|哈喽|嗨|早上好|上午好|中午好|下午好|晚上好|早|在吗|在不在)[\s!！.。~～？?]*$", + re.I, +) + +# 跳过/暂无类有效回答 +_SKIP_RE = re.compile(r"暂无|没有|无|不清楚|不确定|随便|都可以|未知|n/?a|none|no\b", re.I) + +# 鲜境业务相关词：冷链温控 / 官网产品线 +_DOMAIN_RE = re.compile( + r"冷链|冷库|仓储|仓库|运输|配送|保温|冷冻|冷藏|温控|温度|湿度|" + r"药品|疫苗|生鲜|实验室|产线|户外|监测|告警|报警|" + r"TagPlus|BlueTag|GAEA|Fresh\s*Tag|USB|PDF|蓝牙|网关|记录仪|" + r"不开箱|行程|温区|WiFi|4G|LAN|API|方案|产品|报价|选型|" + r"实时|读数|报告|传感器|物联网|物联|预算|网络|对接|系统|次数|可重复", + re.I, +) + +# 明显跑题话题 +_OFFTOPIC_RE = re.compile( + r"天气|气温|下雨|新闻|股票|基金|彩票|足球|篮球|电影|电视剧|游戏|" + r"笑话|段子|星座|运势|做饭|菜谱|旅游|机票|酒店|恋爱|八卦|" + r"政治|选举|战争|加密货币|比特币|chatgpt|写诗|作文|数学题|" + r"编程作业|leetcode|今天吃什么", + re.I, +) + +_QUESTION_RE = re.compile(r"[?？]|什么是|怎么|如何|能否|可以吗|多少|哪[个些]|是否|介绍一下|有没有") + + +def is_greeting(text: str) -> bool: + return bool(_GREETING_RE.match((text or "").strip())) + + +def is_skip_answer(text: str) -> bool: + t = (text or "").strip() + if not t: + return False + if _SKIP_RE.fullmatch(t): + return True + return t in {"没有要求", "无要求", "没什么", "先不确定"} + + +def is_domain_related(text: str) -> bool: + return bool(_DOMAIN_RE.search(text or "")) + + +def looks_like_question(text: str) -> bool: + return bool(_QUESTION_RE.search(text or "")) + + +def has_offtopic_topic(text: str) -> bool: + return bool(_OFFTOPIC_RE.search(text or "")) + + +def looks_off_topic(text: str) -> bool: + """明显与冷链方案无关的输入；寒暄单独处理，不算跑题。""" + t = (text or "").strip() + if not t or is_greeting(t) or is_skip_answer(t): + return False + if is_domain_related(t): + return False + if _OFFTOPIC_RE.search(t): + return True + # 纯英文闲聊 / 极短无意义且不像在答槽 + if re.fullmatch(r"[a-zA-Z\s!.?~]{1,40}", t) and not is_domain_related(t): + lower = t.lower() + if any(w in lower for w in ("hello", "hi", "hey", "thanks", "bye", "ok", "test")): + return False # greeting/ack 另判 + if looks_like_question(t): + return True + # 较长文本却无任何业务词 → 视为跑题 + if len(t) >= 8 and not is_domain_related(t) and looks_like_question(t): + return True + if len(t) >= 16 and not is_domain_related(t): + return True + return False + + +def should_write_slot(text: str) -> bool: + """是否允许本轮用户话写入槽位。""" + t = (text or "").strip() + if not t or is_greeting(t): + return False + if looks_off_topic(t): + return False + return True + + +def greeting_reply(next_prompt: str | None) -> str: + base = "你好！我是鲜境方案客服，可以帮您梳理冷链温控场景并匹配官网产品组合。" + if next_prompt: + return f"{base}\n\n{next_prompt}" + return f"{base}\n\n您可以直接描述场景，例如：「冷库仓储，要实时告警」。" + + +def refuse_reply(reason: str = "off_topic", next_prompt: str | None = None) -> str: + if reason == "kb_miss": + msg = ( + "抱歉，我无法回答这个问题。" + "当前知识库里没有足够相关依据，请换一个与鲜境冷链温控产品/场景相关的问题，" + "或描述您的应用现场，我再帮您追问选型。" + ) + else: + msg = ( + "抱歉，我无法回答这类问题。" + "我主要协助鲜境物联的冷链温控方案咨询（运输记录、蓝牙读数、仓储网关等）。" + ) + if next_prompt: + return f"{msg}\n\n我们继续：{next_prompt}" + return msg diff --git a/engine/app/retrieval.py b/engine/app/retrieval.py index 287d3b8..486f8d7 100644 --- a/engine/app/retrieval.py +++ b/engine/app/retrieval.py @@ -1,5 +1,6 @@ from __future__ import annotations +import re from typing import Any from .config_loader import business_config @@ -24,6 +25,67 @@ def retrieve_for_scheme(slots: dict[str, str], bundle: list[str]) -> RetrievalRe return RetrievalResult([], "ragflow", warning=str(exc)) +def retrieve_question(question: str, *, page_size: int = 3, min_score: float = 0.22) -> RetrievalResult: + """自由问答检索；分数过低视为知识库未命中。""" + backend = (settings.retrieval_backend or "ragflow").lower() + q = (question or "").strip() + if not q: + return RetrievalResult([], backend or "none", warning="empty_query") + if backend == "mock": + return _retrieve_mock_question(q, min_score=min_score) + try: + client = RagflowClient() + dataset_ids = [item.strip() for item in settings.ragflow_dataset_ids.split(",") if item.strip()] + raw = client.retrieve(q, dataset_ids, page_size=page_size) + chunks = [] + for index, item in enumerate(raw[:page_size]): + if not isinstance(item, dict): + continue + score = item.get("similarity") + try: + score_f = float(score) if score is not None else 0.0 + except (TypeError, ValueError): + score_f = 0.0 + if score_f < min_score: + continue + chunks.append( + { + "id": item.get("id") or f"rf-q-{index}", + "title": item.get("document_keyword") + or item.get("document_name") + or item.get("docnm_kwd") + or "知识片段", + "content": item.get("content") or item.get("content_with_weight") or "", + "score": score_f, + } + ) + warning = "" if chunks else "知识库未召回到足够相关的内容" + return RetrievalResult(chunks, "ragflow", warning=warning) + except RagflowError as exc: + return RetrievalResult([], "ragflow", warning=str(exc)) + + +def _retrieve_mock_question(question: str, *, min_score: float = 0.22) -> RetrievalResult: + picked: list[dict[str, Any]] = [] + for chunk in business_config.chunks: + blob = f"{chunk.get('title', '')} {chunk.get('content', '')} {' '.join(chunk.get('tags') or [])}" + hits = sum(1 for token in re_split_tokens(question) if token and token in blob) + if hits: + score = min(0.9, 0.2 + 0.15 * hits) + if score >= min_score: + item = dict(chunk) + item["score"] = score + picked.append(item) + if len(picked) >= 3: + break + warning = "" if picked else "知识库未召回到足够相关的内容" + return RetrievalResult(picked, "mock", warning=warning) + + +def re_split_tokens(text: str) -> list[str]: + return [t for t in re.split(r"[\s,，。；;？?！!/|]+", text) if len(t) >= 2] + + def _retrieve_mock(slots: dict[str, str], bundle: list[str]) -> RetrievalResult: chunks = business_config.chunks picked: list[dict[str, Any]] = [] diff --git a/engine/app/scheme_engine.py b/engine/app/scheme_engine.py index 1880842..6d80817 100644 --- a/engine/app/scheme_engine.py +++ b/engine/app/scheme_engine.py @@ -4,7 +4,17 @@ from typing import Any from .config_loader import business_config from .decision import evaluate_tree -from .retrieval import retrieve_for_scheme +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, @@ -25,34 +35,79 @@ def handle_message(session_id: str, message: str) -> dict[str, Any]: 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) - # fill slots while collecting - if session.phase in ("collecting", "scheme_ready"): + # 寒暄：回问候 + 合理追问，绝不把 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": - # first turn with empty slots: greet + first question - if len(session.messages) == 1 and not any(session.slots.values()): - # put first user utterance into first slot if they already answered something - session.slots = naive_fill_slots(text, business_config.slots, session.slots) - nxt = next_slot_prompt(business_config.slots, session.slots) - if nxt: - reply = nxt["prompt"] - return _finish(session, reply, asking_slot=nxt["id"]) + reply = nxt["prompt"] + return _finish(session, reply, asking_slot=nxt["id"]) - # all required (+ optional asked or filled) → scheme + # 槽位齐了 → 出方案；知识库完全无依据时明确说明无法背书 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, @@ -93,7 +148,7 @@ def handle_message(session_id: str, message: str) -> dict[str, Any]: lines.append(f"- {citation['title']}：{citation['content']}") else: lines.append("") - lines.append("知识库未给出可引用片段，请勿把上述组合当成已确认报价。") + lines.append("知识库未给出可引用片段，请勿把上述组合当成已确认报价。若问题超出资料范围，我无法保证回答准确。") if need_handoff or retrieval.warning: lines.append("") lines.append("建议转人工确认后再发给客户。可在右侧留下称呼和电话。") diff --git a/engine/app/session_store.py b/engine/app/session_store.py index ee3a623..17c0267 100644 --- a/engine/app/session_store.py +++ b/engine/app/session_store.py @@ -48,9 +48,11 @@ def looks_like_handoff(text: str, phrases: list[str]) -> bool: def naive_fill_slots(message: str, slots_meta: list[dict[str, Any]], current: dict[str, str]) -> dict[str, str]: """每轮通常只推进一个槽：先做明确关键词，再把整句写入当前优先空槽。""" + from .intent import should_write_slot + updated = dict(current) text = message.strip() - if not text: + if not text or not should_write_slot(text): return updated touched = False