fix: 收紧跑题识别，避免股市等闲聊写入槽位

Co-authored-by: haijd <haijd@163.com>

.gitignore

@@ -0,0 +1,7 @@
+__pycache__/
+.venv/
+venv/
+.env
+data/
+*.pyc
+.DS_Store

README.md

@@ -1,32 +1,44 @@
-# 鲜境物联 · AI 智能客服｜技术路线建议
+# 鲜境物联 · AI 智能客服
 
-综合建议书、验收标准、**委托开发合同**与**附件二进度表**后的实施路线结论。
+用 **Ragflow** 做知识库检索，用本仓库的方案引擎做「追问 → 判定 → 出方案 → 留资」。
 
-## 结论
+线上咨询台：[鲜境智能客服](http://127.0.0.1:3847/)  
+Ragflow 控制台：[Ragflow](http://127.0.0.1:18080/)
 
-**推荐：开源自托管二开 + 核心业务定制。不建议纯自研，不建议纯公有云 SaaS。**
+示例产品资料在 `knowledge/corpus/`，型号是占位，等甲方资料替换。判定规则在 `knowledge/decision_tree.yaml`，由业务方定对错。
 
-| 路线 | 结论 |
+## 本机已经怎么跑
+
+1. Ragflow v0.27.1 用 Docker 起在 `/opt/ragflow/docker`（内存已调低，适配约 16GB 以下的机器）。
+2. `scripts/bootstrap_ragflow.py` 注册管理员、建 `products` / `solutions` / `faqs`，灌入语料并写 `.env`。
+3. 方案引擎读 `.env` 里的 API Key 与数据集 ID，检索走 `POST /api/v1/retrieval`。
+
+## 自己重来一遍
+
+```bash
+# 1. 起 Ragflow（首次会拉镜像）
+cd /opt/ragflow/docker
+sudo docker compose -f docker-compose.yml up -d
+
+# 2. 若容器互相连不上（日志里 MySQL timeout），补网桥转发后重启 ragflow
+bash scripts/fix_docker_forward.sh
+cd /opt/ragflow/docker && sudo docker restart docker-ragflow-cpu-1
+cd /workspace
+source .venv/bin/activate
+pip install -r engine/requirements.txt
+python scripts/bootstrap_ragflow.py
+
+# 3. 咨询台
+uvicorn engine.app.main:app --host 0.0.0.0 --port 3847
+```
+
+没有 Docker 时可以把 `.env` 里的 `RETRIEVAL_BACKEND` 改成 `mock`，只用本地示例片段演示对话，不代表知识库已接通。
+
+## 目录
+
+| 路径 | 作用 |
 |------|------|
-| 纯自研 | 不推荐首期（撞上 **7 周 / ¥5 万** 硬顶） |
-| 纯 SaaS（扣子 / 阿里云客服等） | 不推荐（数据自主可控冲突） |
-| **Dify 自托管二开 + 方案判定定制** | **推荐** |
-
-**二开首选：Dify**；检索吃力时加 **Ragflow**；人工席位 **Chatwoot**（或企微/飞书）。
-
-合同已锁：实施 **¥50,000**、订阅 **¥30,000/年**、周期 **7 周**、线上场景 **只做一个高频渠道**。详见完整建议书。
-
-## 文档
-
-| 文件 | 说明 |
-|------|------|
-| [docs/技术方案建议书.md](./docs/技术方案建议书.md) | 决策依据、平台对比、架构、7 周映射、风险 |
-| [uploads/](./uploads/) | 原始合同、进度表、建议书、验收标准 |
-
-## 建议立刻确认
-
-1. 首期唯一渠道：网站还是 Facebook？  
-2. 方案判定树的业务拍板人  
-3. 模型：智谱 API 是否可上生产 / 是否必须私有化  
-4. 场景二（录音辅助）是否与场景一同期验收  
-5. 部署：甲方机房还是乙方 VPC
+| `engine/` | 咨询 API 与页面 |
+| `knowledge/` | 槽位、判定树、灌库语料 |
+| `scripts/bootstrap_ragflow.py` | 对接 Ragflow 建库 |
+| `docs/` | 选型与实施说明 |

docs/Ragflow实施方案.md

@@ -0,0 +1,241 @@
+# 鲜境 AI 客服 · Ragflow 搭建方案
+
+> 前提：底座选 **Ragflow（自托管，Apache-2.0）**；「意图 → 追问 → 方案生成 → 转人工」用**自研薄方案引擎**补齐，不把业务逻辑只画在厂商画布里。
+
+---
+
+## 1. 总体架构
+
+```
+┌─────────────────────────────────────────────────────────────┐
+│ 渠道层                                                        │
+│  网站插件 / FB（首期只做一个） │ 销售端：录音上传或实时转写入口      │
+└───────────────┬─────────────────────────────┬───────────────┘
+                │                             │
+                ▼                             ▼
+┌──────────────────────────┐    ┌─────────────────────────────┐
+│ 方案引擎 API（自研）        │    │ ASR 服务（场景二）             │
+│ · 会话状态 / 追问槽位       │◄───│ 云 ASR 或私有 Whisper          │
+│ · 意图分类                 │    └─────────────────────────────┘
+│ · 方案判定树 + 产品组合     │
+│ · 置信度 / 敏感词 → 转人工  │
+│ · 方案文本模板              │
+└──────────┬───────────────┘
+           │ 检索 API（query → chunks + citation）
+           ▼
+┌──────────────────────────┐    ┌─────────────────────────────┐
+│ Ragflow（自托管）          │    │ 人工席                       │
+│ · Dataset 知识库           │    │ Chatwoot / 企微 / 飞书        │
+│ · 解析·切块·Embedding      │    │ 低置信度工单 + 对话记录推送     │
+│ · 召回·重排·引用           │    └─────────────────────────────┘
+│ ·（可选）内置 Chat 仅调试用  │
+└──────────────────────────┘
+           │
+           ▼
+┌──────────────────────────┐
+│ 模型层                     │
+│ LLM：智谱等（可切私有化）    │
+│ Embedding：与库同环境       │
+└──────────────────────────┘
+```
+
+**职责切分（务必守住）：**
+
+| 组件 | 负责 | 不负责 |
+|------|------|--------|
+| Ragflow | 资料入库、检索、引用片段 | 业务判定树、渠道 UI、人工工单主流程 |
+| 方案引擎 | 多轮状态、追问、匹配产品组合、出方案、转人工策略 | 重做一套向量库 |
+| 渠道/ASR/人工席 | 触达与交接 | 知识解析 |
+
+---
+
+## 2. 环境与部署
+
+### 2.1 建议拓扑（首期）
+
+- 单机或一小集群：`Docker Compose` 部署 Ragflow（含其依赖的 ES/Infinity、MySQL、MinIO、Redis 等，以官方 compose 为准）。
+- 方案引擎：独立服务（建议 **Python FastAPI** 或 **Node/Hono**），与 Ragflow 同 VPC。
+- 对象存储：产品原文 PDF/Word 放 MinIO 或客户 OSS；Ragflow 只存解析结果与向量。
+- 配置：模型 Key、Ragflow API Key 进环境变量，不进仓库。
+
+### 2.2 模型
+
+| 用途 | 建议 | 说明 |
+|------|------|------|
+| 对话 / 方案生成 | 智谱 GLM 等国产 API | 合同要求数据可控；可按成本换模型 |
+| Embedding | Ragflow 推荐且已验证的中英模型 | **一经选定勿随意换**（换则全量重建） |
+| ASR | 先云端；涉密再私有化 Whisper | 场景二 |
+
+### 2.3 许可与交付物
+
+- 开源清单写明：Ragflow = Apache-2.0；方案引擎为定制代码（归甲方）；列出传递依赖。
+- 保留 Ragflow 版权声明即可，无 Dify 类「去 Logo / 多租户」附加条款问题。
+
+---
+
+## 3. 知识库怎么建（Ragflow 内）
+
+### 3.1 Dataset 划分（建议）
+
+不要做成一个「大杂烩库」：
+
+| Dataset | 内容 | 用途 |
+|---------|------|------|
+| `products` | 型号、参数、适用工况、禁忌 | 产品组合匹配时的硬事实 |
+| `solutions` | 历史方案、两大类技术选型说明 | 方案话术与结构 |
+| `faqs` | 高频询盘 Q&A | 降重复劳动 |
+| `policies` | 敏感词、不可承诺条款（可选） | 护栏 |
+
+### 3.2 入库规范（W1 和甲方定）
+
+- 每份文档元数据：`产品线`、`技术大类(A/B)`、`型号`、`语言`、`版本日期`。
+- 优先：**结构化表（型号×参数）+ 方案范文**；纯长 PDF 靠解析，但关键参数尽量表结构化。
+- 切块：参数表用「按行/按型号」策略；方案用「按章节」；入库后做 **Recall 测试**（人工出 30 个真实问法看片段对不对）。
+
+### 3.3 检索调用方式
+
+方案引擎只调 Ragflow **检索 API**（或官方 SDK），拿到：
+
+- `content` / `score` / `document` / 引用定位  
+- 再交给 LLM **仅基于引用**生成；无命中则追问或转人工，禁止空编参数。
+
+---
+
+## 4. 方案引擎怎么搭（自研核心）
+
+### 4.1 会话状态机（最小闭环）
+
+```
+新会话
+  → 意图识别（咨询/比价/售后/无关…）
+  → 若「方案咨询」：进入技术大类判定
+       → 追问槽位未满：生成下一个问题
+       → 槽位已满：检索 + 判定树 → 产品组合
+       → 渲染方案模板 → 返回用户
+  → 低置信 / 敏感 / 用户要人工 → 转人工 + 留资
+```
+
+### 4.2 追问槽位（示例，W1 由业务专家定稿）
+
+按鲜境「两大类技术」分别设必问项，例如：
+
+- 应用场景 / 工况环境  
+- 处理对象或指标范围  
+- 产能或规模  
+- 现场约束（空间、已有设备、预算区间）  
+- 交付区域/语言（若接海外询盘）
+
+槽位存会话表（Postgres/SQLite 首期均可），不要只靠模型「记忆」。
+
+### 4.3 判定树 + 产品组合
+
+- 专家规则优先：`IF 大类=A AND 工况=X AND 指标∈[…] THEN 推荐型号组合`  
+- RAG 补充：方案说明、话术、注意事项从 `solutions`/`products` 检索填空  
+- 输出结构固定：`推荐组合` + `理由（带引用）` + `需人工确认项` + `下一步留资`
+
+### 4.4 接口草图
+
+```
+POST /chat          # 渠道统一入口：session_id + message
+POST /sessions/:id/handoff
+POST /ingest/notify # 可选：资料更新后重建/增量索引钩子
+GET  /admin/evals   # 评测任务与结果
+```
+
+渠道插件、FB Webhook、销售录音页都只打 `/chat`。
+
+---
+
+## 5. 两场景落地
+
+### 场景一：线上 24h 咨询（合同：只做一个高频渠道）
+
+1. 选定网站 **或** FB。  
+2. 嵌入聊天窗 / Webhook → 方案引擎。  
+3. UI 展示：回复 +（可选）引用折叠；「转人工」「留电话」。  
+4. 转人工：创建 Chatwoot 会话或推企微，附带槽位摘要与最近检索命中。
+
+### 场景二：销售辅助
+
+1. 上传录音或对接会议转写 → ASR 文本。  
+2. 文本进入同一状态机（可跳过寒暄意图，直接抽槽位）。  
+3. 输出「可发给客户的方案草稿」+ 内部注意点；由销售确认后发送。
+
+---
+
+## 6. 人工接管与后台
+
+| 能力 | 做法 |
+|------|------|
+| 转人工 | 引擎规则触发 → Chatwoot/企微；超时未接则留资入库 |
+| 对话追溯 | 引擎写 `messages` 表；可同步一份到人工席 |
+| 知识维护 | 业务在 Ragflow 控制台更新 Dataset；重大变更跑 Recall 抽样 |
+| 效果看板 | 首期：导出日志 + 简单页（任务完成率、转人工率、Eval 分数）；勿一上来做大屏 |
+
+---
+
+## 7. AI Eval（对着附件三）
+
+| 指标 | 做法 |
+|------|------|
+| 意图识别 ≥85% | 200 条标注；引擎意图模块离线测 |
+| 方案匹配 ≥80% | 专家盲测「引擎方案 vs 专家方案」 |
+| 任务完成 ≥70% | 日志：未转人工且用户达方案/留资 |
+| 幻觉 ≤5% | 抽检：断言中参数是否出现在引用 chunk |
+| 敏感泄露 =0% | 正则+词表扫描回复 |
+
+评测集与判定树一样：**外置文件**，不锁在 Ragflow 里。
+
+---
+
+## 8. 推荐实施顺序（可不绑死 7 周）
+
+| 步 | 做什么 | 完成标准 |
+|----|--------|----------|
+| 1 | Compose 拉起 Ragflow；接通 LLM/Embedding | 控制台可聊「基于空库」 |
+| 2 | 灌入一小撮真实产品+方案；Recall 测试 | 业务说「找得对」 |
+| 3 | 定稿判定树与槽位；引擎跑通 API | Postman 多轮能出方案结构 |
+| 4 | 护栏：无检索不编参数；转人工；留资 | 故意问库外问题会转人工 |
+| 5 | 接一个渠道 + 基础聊天 UI | 真实访客可走通 |
+| 6 | 场景二 ASR 联调 | 一段销售录音能出草稿 |
+| 7 | Eval 全量 + UAT | 附件三阈值或书面差距说明 |
+| 8 | 生产部署、备份、监控、开源清单 | 可交接 |
+
+---
+
+## 9. 仓库/工程建议（乙方交付形态）
+
+```
+/infra          # docker-compose、备份脚本
+/engine         # 方案引擎（状态机、判定、调 Ragflow）
+/web-widget     # 网站聊天插件（若渠道是网站）
+/eval           # 评测集、脚本
+/knowledge      # 元数据说明、判定树 YAML、产品表（原文大文件可外链）
+/docs           # 部署说明、开源清单、运维手册
+```
+
+定制成果在 `engine` / `knowledge` / `eval`；Ragflow 用官方镜像，**尽量不 Fork 魔改核心**（升级痛苦）。二开以「配置 + 外围服务」为主。
+
+---
+
+## 10. 风险与戒条
+
+1. 不要用 Ragflow 默认 Chat 直接当生产客服终态——缺槽位与判定，难稳过方案匹配。  
+2. 不要把判定树只写在 Prompt 里——要结构化规则 + 检索填空。  
+3. 换 Embedding / 大改切块 = 重建索引 + 重做 Recall。  
+4. 首期渠道只做一个；双渠道等订阅期再加。  
+5. 双栈再挂 Dify：仅当自研编排人力不够时临时用，默认不必。
+
+---
+
+## 11. 和合同交付的对应
+
+| 合同/验收项 | 落在哪 |
+|-------------|--------|
+| 知识库可检索、可维护 | Ragflow Dataset + 权限 |
+| 智能问答多轮 | 方案引擎状态机 |
+| 方案自动生成 | 判定树 + 检索 + 模板 |
+| 人工接管 | 引擎策略 + Chatwoot/企微 |
+| 对话记录/看板 | 引擎库表 + 轻量看板 |
+| AI Eval 阈值 | `/eval` + 专家盲测 |
+| 开源清单 | Ragflow Apache-2.0 + 依赖列表 |

docs/原型搭建步骤.md

@@ -0,0 +1,18 @@
+# 需求未定稿时：先搭通用原型的步骤
+
+## 要不要先搭原型？
+
+**要。** 做成「配置驱动的通用骨架」，不要等需求 100% 精确再动手。
+
+- 先验证：追问 → 判定 → 检索引用 → 出方案草稿  
+- 业务变了：主要改 `knowledge/*.yaml`  
+- 后台/渠道：闭环跑通后再补  
+
+## 具体步骤
+
+1. **拉起本原型**（不必先装 Ragflow）  
+2. **和业务改槽位/判定树 YAML**（边聊边改，`POST /api/admin/reload-config`）  
+3. **灌真实小样本到 Ragflow**，切 `RETRIEVAL_BACKEND=ragflow`  
+4. **再补**转人工、单渠道、Eval、薄后台  
+
+不要颠倒：先做大后台或双渠道。

docs/技术方案建议书.md

@@ -152,6 +152,35 @@ Dify（意图分类 → 追问槽位 → 调 Ragflow → 方案生成 → 转人
 - 只求本单快验收、单租户可留 Logo → Dify 仍可行，但把判定树/评测集/原始知识**外置**，当保险。  
 - 不要假设「二开完还能廉价换家」——能换的是资产，不是画布。
 
+#### 若不纯赶工期，怎么选？
+
+时间让位之后，排序应变成：**许可干净 > 检索/幻觉可过验收 > 对话编排体验 > 上线速度**。
+
+**推荐默认改成：Ragflow 为主底座 + 自研薄「方案引擎」（判定树 / 追问槽位 / 转人工策略）。**
+
+| 理由 | 说明 |
+|------|------|
+| 许可 | 纯 Apache，乙方后续多客户复用、招标合规更省心 |
+| 验收 | 鲜境是资料驱动的「卖方案」，命中率/幻觉更吃检索，Ragflow 更对症 |
+| 锁定 | 首期就站在以后更可能留下的底座上，避免 Dify→Ragflow 重配画布 |
+| 编排缺口 | 用自研 API/状态机补「意图→追问→出方案」，业务逻辑落在你们代码与配置里，不绑死厂商画布 |
+
+**这时还要不要 Dify？**
+
+- **一般不必再引入 Dify**（双栈运维、两套权限/日志，长期总拥有成本更高）。  
+- 仅当方案引擎自研人力明显不足、又强依赖可视化编排时，可短期用 Dify 只做编排、检索仍走 Ragflow——但要接受多一套依赖。  
+- Flowise 同理：许可尚可，却更容易把逻辑锁在画布上；时间充裕时不如 **Ragflow + 自研编排** 干净。
+
+**时间充裕时的分期（示意，非合同附件二）：**
+
+1. 外置资产：资料仓、产品结构表、判定树、Eval 集  
+2. Ragflow 灌库 + 召回/幻觉打穿附件三阈值  
+3. 自研对话状态（槽位追问、低置信转人工、方案模板）  
+4. 单渠道接入 + 场景二 ASR  
+5. 再考虑是否要可视化编排器（多数情况不再需要）
+
+**一句话：不赶时间 → 选 Ragflow 路线更安全、更好；Dify 的优势主要是快，不是长期更优。**
+
 ---
 ### 为何首选 Dify（而不是 Flowise）
 

engine/__init__.py

@@ -0,0 +1 @@
+# Package

engine/app/__init__.py

@@ -0,0 +1 @@
+# Package marker

engine/app/config_loader.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from typing import Any
+
+import yaml
+
+from .settings import settings
+
+
+def _load_yaml(name: str) -> dict[str, Any]:
+    path = settings.knowledge_dir / name
+    with path.open(encoding="utf-8") as f:
+        data = yaml.safe_load(f) or {}
+    if not isinstance(data, dict):
+        raise ValueError(f"{name} must be a mapping")
+    return data
+
+
+class BusinessConfig:
+    def __init__(self) -> None:
+        self.reload()
+
+    def reload(self) -> None:
+        self.slots_doc = _load_yaml("slots.yaml")
+        self.tree_doc = _load_yaml("decision_tree.yaml")
+        self.chunks_doc = _load_yaml("sample_chunks.yaml")
+
+    @property
+    def slots(self) -> list[dict[str, Any]]:
+        return list(self.slots_doc.get("slots") or [])
+
+    @property
+    def handoff_phrases(self) -> list[str]:
+        return list(self.slots_doc.get("handoff_phrases") or [])
+
+    @property
+    def closing_prompt(self) -> str:
+        return str(self.slots_doc.get("closing_prompt") or "正在生成方案…")
+
+    @property
+    def rules(self) -> list[dict[str, Any]]:
+        return sorted(self.tree_doc.get("rules") or [], key=lambda r: r.get("order", 100))
+
+    @property
+    def handoff_confidence_below(self) -> float:
+        return float(self.tree_doc.get("handoff_confidence_below", 0.55))
+
+    @property
+    def chunks(self) -> list[dict[str, Any]]:
+        return list(self.chunks_doc.get("chunks") or [])
+
+
+business_config = BusinessConfig()

engine/app/decision.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+def _match_condition(cond: dict[str, Any], slots: dict[str, str]) -> bool:
+    ctype = cond.get("type")
+    if ctype == "always":
+        return True
+    slot = str(cond.get("slot") or "")
+    value = str(cond.get("value") or "")
+    actual = (slots.get(slot) or "").strip()
+    if ctype == "slot_equals":
+        return actual == value
+    if ctype == "slot_contains":
+        return value.lower() in actual.lower()
+    return False
+
+
+def evaluate_tree(rules: list[dict[str, Any]], slots: dict[str, str]) -> dict[str, Any]:
+    for rule in rules:
+        conditions = rule.get("when_all") or []
+        if all(_match_condition(c, slots) for c in conditions):
+            then = dict(rule.get("then") or {})
+            return {
+                "rule_id": rule.get("id"),
+                "description": rule.get("description"),
+                **then,
+            }
+    return {
+        "rule_id": "empty",
+        "product_bundle": ["待专家确认"],
+        "tech_line": "待判定",
+        "summary": "未匹配到规则",
+        "confidence": 0.0,
+        "handoff": True,
+    }

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) >= 4 and not is_domain_related(t) and looks_like_question(t):
+        return True
+    if len(t) >= 10 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

engine/app/leads.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import sqlite3
+from datetime import datetime, timezone
+from pathlib import Path
+
+from .settings import ROOT
+
+DB_PATH = ROOT / "data" / "leads.sqlite"
+
+
+def _conn() -> sqlite3.Connection:
+    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    conn.execute(
+        """
+        CREATE TABLE IF NOT EXISTS leads (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            session_id TEXT,
+            name TEXT,
+            phone TEXT,
+            note TEXT,
+            slots_json TEXT,
+            scheme_json TEXT,
+            created_at TEXT
+        )
+        """
+    )
+    return conn
+
+
+def add_lead(
+    *,
+    session_id: str,
+    name: str,
+    phone: str,
+    note: str,
+    slots_json: str,
+    scheme_json: str,
+) -> int:
+    now = datetime.now(timezone.utc).isoformat()
+    with _conn() as conn:
+        cur = conn.execute(
+            """
+            INSERT INTO leads (session_id, name, phone, note, slots_json, scheme_json, created_at)
+            VALUES (?, ?, ?, ?, ?, ?, ?)
+            """,
+            (session_id, name, phone, note, slots_json, scheme_json, now),
+        )
+        return int(cur.lastrowid)
+
+
+def list_leads(limit: int = 50) -> list[dict]:
+    with _conn() as conn:
+        rows = conn.execute(
+            "SELECT id, session_id, name, phone, note, created_at FROM leads ORDER BY id DESC LIMIT ?",
+            (limit,),
+        ).fetchall()
+    return [dict(row) for row in rows]

engine/app/main.py

@@ -0,0 +1,152 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from fastapi import FastAPI, HTTPException
+from fastapi.responses import FileResponse
+from fastapi.staticfiles import StaticFiles
+from pydantic import BaseModel, Field
+
+from .config_loader import business_config
+from .leads import add_lead, list_leads
+from .ragflow_client import RagflowClient, RagflowError
+from .scheme_engine import create_session, handle_message
+from .session_store import store
+from .settings import ROOT, settings
+
+STATIC_DIR = Path(__file__).resolve().parent / "static"
+CORPUS_DIR = ROOT / "knowledge" / "corpus"
+
+app = FastAPI(title=settings.app_name, version="0.2.0")
+app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
+
+
+class ChatRequest(BaseModel):
+    message: str = Field(..., min_length=1)
+    session_id: str | None = None
+
+
+class SessionResponse(BaseModel):
+    session_id: str
+
+
+class HandoffRequest(BaseModel):
+    session_id: str
+    name: str = Field(..., min_length=1, max_length=40)
+    phone: str = Field(..., min_length=5, max_length=30)
+    note: str = ""
+
+
+@app.get("/")
+def index() -> FileResponse:
+    return FileResponse(STATIC_DIR / "index.html")
+
+
+@app.get("/api/health")
+def health() -> dict:
+    rag = RagflowClient().health()
+    return {
+        "ok": True,
+        "app": settings.app_name,
+        "retrieval_backend": settings.retrieval_backend,
+        "slots": len(business_config.slots),
+        "rules": len(business_config.rules),
+        "ragflow": rag,
+        "ragflow_web": settings.ragflow_web_url,
+        "dataset_ids_configured": bool(settings.ragflow_dataset_ids.strip()),
+    }
+
+
+@app.get("/api/config/slots")
+def get_slots() -> dict:
+    return business_config.slots_doc
+
+
+@app.get("/api/config/decision-tree")
+def get_tree() -> dict:
+    return business_config.tree_doc
+
+
+@app.post("/api/admin/reload-config")
+def reload_config() -> dict:
+    business_config.reload()
+    return {"reloaded": True, "slots": len(business_config.slots), "rules": len(business_config.rules)}
+
+
+@app.get("/api/knowledge/files")
+def knowledge_files() -> dict:
+    files = []
+    if CORPUS_DIR.exists():
+        for path in sorted(CORPUS_DIR.glob("*.md")):
+            files.append({"name": path.name, "bytes": path.stat().st_size})
+    return {"files": files}
+
+
+@app.get("/api/ragflow/status")
+def ragflow_status() -> dict:
+    client = RagflowClient()
+    health = client.health()
+    datasets: list[dict] = []
+    error = ""
+    if health.get("ok") and settings.ragflow_api_key:
+        try:
+            for item in client.list_datasets():
+                datasets.append(
+                    {
+                        "id": item.get("id"),
+                        "name": item.get("name"),
+                        "document_count": item.get("document_count"),
+                        "chunk_count": item.get("chunk_count"),
+                    }
+                )
+        except Exception as exc:
+            error = str(exc)
+    return {
+        "health": health,
+        "web": settings.ragflow_web_url,
+        "api_key_set": bool(settings.ragflow_api_key),
+        "dataset_ids": [x for x in settings.ragflow_dataset_ids.split(",") if x.strip()],
+        "datasets": datasets,
+        "error": error,
+    }
+
+
+@app.get("/api/leads")
+def api_leads() -> dict:
+    return {"leads": list_leads()}
+
+
+@app.post("/api/sessions", response_model=SessionResponse)
+def api_create_session() -> SessionResponse:
+    session = create_session()
+    return SessionResponse(session_id=session.id)
+
+
+@app.post("/api/chat")
+def api_chat(body: ChatRequest) -> dict:
+    session_id = body.session_id
+    if not session_id:
+        session_id = create_session().id
+    try:
+        return handle_message(session_id, body.message)
+    except KeyError:
+        raise HTTPException(status_code=404, detail="session_not_found") from None
+
+
+@app.post("/api/handoff")
+def api_handoff(body: HandoffRequest) -> dict:
+    session = store.get(body.session_id)
+    slots = session.slots if session else {}
+    scheme = session.last_scheme if session else None
+    lead_id = add_lead(
+        session_id=body.session_id,
+        name=body.name.strip(),
+        phone=body.phone.strip(),
+        note=body.note.strip(),
+        slots_json=json.dumps(slots, ensure_ascii=False),
+        scheme_json=json.dumps(scheme or {}, ensure_ascii=False),
+    )
+    if session:
+        session.phase = "handoff"
+    return {"ok": True, "lead_id": lead_id}

engine/app/ragflow_client.py

@@ -0,0 +1,217 @@
+from __future__ import annotations
+
+import base64
+from pathlib import Path
+from typing import Any
+
+import httpx
+try:
+    from Cryptodome.Cipher import PKCS1_v1_5
+    from Cryptodome.PublicKey import RSA
+except ImportError:
+    from Crypto.Cipher import PKCS1_v1_5
+    from Crypto.PublicKey import RSA
+
+from .settings import ROOT, settings
+
+PUBLIC_PEM = ROOT / "infra" / "ragflow-public.pem"
+
+
+class RagflowError(RuntimeError):
+    pass
+
+
+def encrypt_password(password: str) -> str:
+    """Match Ragflow crypt(): RSA-encrypt base64(password)."""
+    rsa_key = RSA.import_key(PUBLIC_PEM.read_text())
+    cipher = PKCS1_v1_5.new(rsa_key)
+    password_b64 = base64.b64encode(password.encode("utf-8"))
+    encrypted = cipher.encrypt(password_b64)
+    return base64.b64encode(encrypted).decode("utf-8")
+
+
+class RagflowClient:
+    def __init__(self, base_url: str | None = None, api_key: str | None = None) -> None:
+        self.base_url = (base_url or settings.ragflow_base_url).rstrip("/")
+        self.api_key = api_key if api_key is not None else settings.ragflow_api_key
+
+    def _headers(self, token: str | None = None) -> dict[str, str]:
+        key = token or self.api_key
+        if not key:
+            return {}
+        if key.lower().startswith("bearer "):
+            return {"Authorization": key}
+        return {"Authorization": f"Bearer {key}"}
+
+    def health(self) -> dict[str, Any]:
+        url = f"{self.base_url}/api/v1/system/healthz"
+        try:
+            with httpx.Client(timeout=8.0) as client:
+                resp = client.get(url)
+            ok = resp.status_code < 500
+            return {"ok": ok, "status_code": resp.status_code, "url": url}
+        except Exception as exc:
+            return {"ok": False, "error": str(exc), "url": url}
+
+    def register(self, email: str, password: str, nickname: str = "鲜境管理员") -> str:
+        payload = {
+            "nickname": nickname,
+            "email": email,
+            "password": encrypt_password(password),
+        }
+        with httpx.Client(timeout=30.0) as client:
+            resp = client.post(f"{self.base_url}/api/v1/users", json=payload)
+        if resp.status_code >= 400:
+            raise RagflowError(f"注册失败 {resp.status_code}: {resp.text[:400]}")
+        auth = resp.headers.get("Authorization") or ""
+        body = resp.json() if resp.content else {}
+        if body.get("code") not in (0, None) and body.get("code") != 0:
+            # already registered or validation error
+            message = str(body.get("message") or body)
+            if "already" in message.lower() or "已经" in message:
+                return self.login(email, password)
+            raise RagflowError(message)
+        if not auth:
+            data = body.get("data") or {}
+            auth = str(data.get("id") or "")
+        if not auth:
+            raise RagflowError(f"注册成功但没有登录凭证: {body}")
+        return auth
+
+    def login(self, email: str, password: str) -> str:
+        payload = {"email": email, "password": encrypt_password(password)}
+        with httpx.Client(timeout=30.0) as client:
+            resp = client.post(f"{self.base_url}/api/v1/auth/login", json=payload)
+        if resp.status_code >= 400:
+            raise RagflowError(f"登录失败 {resp.status_code}: {resp.text[:400]}")
+        auth = resp.headers.get("Authorization") or ""
+        if not auth:
+            raise RagflowError("登录没有返回 Authorization")
+        return auth
+
+    def ensure_api_token(self, session_auth: str) -> str:
+        headers = {"Authorization": session_auth}
+        with httpx.Client(timeout=30.0) as client:
+            listed = client.get(f"{self.base_url}/api/v1/system/tokens", headers=headers)
+            if listed.status_code < 400:
+                data = (listed.json() or {}).get("data") or []
+                if isinstance(data, list) and data:
+                    token = data[0].get("token")
+                    if token:
+                        return token
+            created = client.post(f"{self.base_url}/api/v1/system/tokens", headers=headers)
+        if created.status_code >= 400:
+            raise RagflowError(f"创建 API Token 失败: {created.text[:400]}")
+        body = created.json()
+        token = (body.get("data") or {}).get("token")
+        if not token:
+            raise RagflowError(f"Token 响应异常: {body}")
+        return token
+
+    def list_datasets(self) -> list[dict[str, Any]]:
+        with httpx.Client(timeout=20.0) as client:
+            resp = client.get(f"{self.base_url}/api/v1/datasets", headers=self._headers(), params={"page": 1, "page_size": 50})
+        resp.raise_for_status()
+        data = (resp.json() or {}).get("data") or []
+        if isinstance(data, dict):
+            data = data.get("datasets") or data.get("kbs") or []
+        return list(data)
+
+    def create_dataset(self, name: str, description: str) -> dict[str, Any]:
+        payload = {
+            "name": name,
+            "description": description,
+            "chunk_method": "naive",
+            "permission": "me",
+            "embedding_model": "BAAI/bge-small-zh-v1.5@Builtin",
+            "parser_config": {
+                "graphrag": {"use_graphrag": False},
+                "raptor": {"use_raptor": False},
+            },
+        }
+        with httpx.Client(timeout=30.0) as client:
+            resp = client.post(f"{self.base_url}/api/v1/datasets", headers=self._headers(), json=payload)
+        if resp.status_code >= 400:
+            raise RagflowError(resp.text[:400])
+        body = resp.json()
+        if body.get("code") not in (0, None):
+            raise RagflowError(str(body.get("message") or body))
+        return body.get("data") or {}
+
+    def update_dataset(self, dataset_id: str, payload: dict[str, Any]) -> dict[str, Any]:
+        with httpx.Client(timeout=30.0) as client:
+            resp = client.put(
+                f"{self.base_url}/api/v1/datasets/{dataset_id}",
+                headers=self._headers(),
+                json=payload,
+            )
+        if resp.status_code >= 400:
+            raise RagflowError(resp.text[:400])
+        body = resp.json()
+        if body.get("code") not in (0, None):
+            raise RagflowError(str(body.get("message") or body))
+        return body.get("data") or {}
+
+    def upload_document(self, dataset_id: str, path: Path) -> list[dict[str, Any]]:
+        with path.open("rb") as fh:
+            files = {"file": (path.name, fh, "text/markdown")}
+            with httpx.Client(timeout=60.0) as client:
+                resp = client.post(
+                    f"{self.base_url}/api/v1/datasets/{dataset_id}/documents",
+                    headers=self._headers(),
+                    files=files,
+                )
+        if resp.status_code >= 400:
+            raise RagflowError(resp.text[:500])
+        body = resp.json()
+        data = body.get("data") or []
+        if isinstance(data, dict):
+            data = [data]
+        return data
+
+    def parse_documents(self, dataset_id: str, document_ids: list[str]) -> None:
+        with httpx.Client(timeout=30.0) as client:
+            resp = client.post(
+                f"{self.base_url}/api/v1/datasets/{dataset_id}/documents/parse",
+                headers=self._headers(),
+                json={"document_ids": document_ids},
+            )
+        if resp.status_code >= 400:
+            raise RagflowError(resp.text[:400])
+
+    def list_documents(self, dataset_id: str) -> list[dict[str, Any]]:
+        with httpx.Client(timeout=20.0) as client:
+            resp = client.get(
+                f"{self.base_url}/api/v1/datasets/{dataset_id}/documents",
+                headers=self._headers(),
+                params={"page": 1, "page_size": 50},
+            )
+        resp.raise_for_status()
+        data = (resp.json() or {}).get("data") or []
+        if isinstance(data, dict):
+            data = data.get("docs") or data.get("documents") or []
+        return list(data)
+
+    def retrieve(self, question: str, dataset_ids: list[str], page_size: int = 4) -> list[dict[str, Any]]:
+        if not self.api_key:
+            raise RagflowError("未配置 RAGFLOW_API_KEY")
+        if not dataset_ids:
+            raise RagflowError("未配置 RAGFLOW_DATASET_IDS")
+        payload = {
+            "question": question,
+            "dataset_ids": dataset_ids,
+            "page": 1,
+            "page_size": page_size,
+            "similarity_threshold": 0.15,
+            "vector_similarity_weight": 0.3,
+        }
+        with httpx.Client(timeout=30.0) as client:
+            resp = client.post(f"{self.base_url}/api/v1/retrieval", headers=self._headers(), json=payload)
+        if resp.status_code >= 400:
+            raise RagflowError(f"检索失败 {resp.status_code}: {resp.text[:300]}")
+        body = resp.json()
+        if body.get("code") not in (0, None):
+            raise RagflowError(str(body.get("message") or body))
+        data = body.get("data") or {}
+        chunks = data.get("chunks") if isinstance(data, dict) else data
+        return list(chunks or [])

engine/app/retrieval.py

@@ -0,0 +1,142 @@
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from .config_loader import business_config
+from .ragflow_client import RagflowClient, RagflowError
+from .settings import settings
+
+
+class RetrievalResult:
+    def __init__(self, chunks: list[dict[str, Any]], backend: str, warning: str = ""):
+        self.chunks = chunks
+        self.backend = backend
+        self.warning = warning
+
+
+def retrieve_for_scheme(slots: dict[str, str], bundle: list[str]) -> RetrievalResult:
+    backend = (settings.retrieval_backend or "ragflow").lower()
+    if backend == "mock":
+        return _retrieve_mock(slots, bundle)
+    try:
+        return _retrieve_ragflow(slots, bundle)
+    except RagflowError as exc:
+        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]] = []
+    bundle_set = set(bundle)
+    for chunk in chunks:
+        if chunk.get("product") in bundle_set:
+            picked.append(chunk)
+    blob = " ".join(slots.values())
+    for chunk in chunks:
+        if chunk in picked:
+            continue
+        tags = chunk.get("tags") or []
+        if any(str(tag) in blob for tag in tags):
+            picked.append(chunk)
+        if len(picked) >= 4:
+            break
+    if not picked:
+        picked = chunks[:2]
+    return RetrievalResult(picked[:4], "mock")
+
+
+def _retrieve_ragflow(slots: dict[str, str], bundle: list[str]) -> RetrievalResult:
+    client = RagflowClient()
+    query = " ".join(
+        [
+            slots.get("tech_category", ""),
+            slots.get("scenario", ""),
+            slots.get("key_metric", ""),
+            slots.get("constraints", ""),
+            " ".join(bundle),
+            "产品组合 适用场景 注意事项",
+        ]
+    ).strip()
+    dataset_ids = [item.strip() for item in settings.ragflow_dataset_ids.split(",") if item.strip()]
+    raw = client.retrieve(query, dataset_ids, page_size=4)
+    normalized = []
+    for index, item in enumerate(raw[:4]):
+        if not isinstance(item, dict):
+            continue
+        normalized.append(
+            {
+                "id": item.get("id") or f"rf-{index}",
+                "title": item.get("document_keyword")
+                or item.get("document_name")
+                or item.get("docnm_kwd")
+                or "知识片段",
+                "product": item.get("document_id") or "",
+                "content": item.get("content") or item.get("content_with_weight") or "",
+                "tags": [],
+                "score": item.get("similarity"),
+            }
+        )
+    warning = "" if normalized else "知识库没有召回到可用片段，方案仅供草稿，建议转人工。"
+    return RetrievalResult(normalized, "ragflow", warning=warning)

engine/app/scheme_engine.py

@@ -0,0 +1,181 @@
+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,
+    }

engine/app/session_store.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+import re
+import uuid
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+@dataclass
+class Session:
+    id: str
+    slots: dict[str, str] = field(default_factory=dict)
+    messages: list[dict[str, Any]] = field(default_factory=list)
+    phase: str = "collecting"  # collecting | scheme_ready | handoff
+    last_scheme: dict[str, Any] | None = None
+    created_at: str = field(default_factory=_now)
+    updated_at: str = field(default_factory=_now)
+
+
+class SessionStore:
+    def __init__(self) -> None:
+        self._sessions: dict[str, Session] = {}
+
+    def create(self) -> Session:
+        session = Session(id=str(uuid.uuid4()))
+        self._sessions[session.id] = session
+        return session
+
+    def get(self, session_id: str) -> Session | None:
+        return self._sessions.get(session_id)
+
+    def touch(self, session: Session) -> None:
+        session.updated_at = _now()
+
+
+store = SessionStore()
+
+
+def looks_like_handoff(text: str, phrases: list[str]) -> bool:
+    t = text.strip().lower()
+    return any(p.lower() in t for p in phrases)
+
+
+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 or not should_write_slot(text):
+        return updated
+
+    touched = False
+
+    def set_slot(slot_id: str, value: str) -> None:
+        nonlocal touched
+        if not updated.get(slot_id):
+            updated[slot_id] = value
+            touched = True
+
+    if re.search(r"TagPlus|USB|PDF", text, re.I):
+        set_slot("tech_category", "TagPlus USB多次记录")
+    elif re.search(r"BlueTag|蓝牙|不开箱", text, re.I):
+        set_slot("tech_category", "BlueTag蓝牙手机读数")
+    elif re.search(r"GAEA|网关|实时", text, re.I):
+        set_slot("tech_category", "GAEA网关实时仓储")
+    elif "不确定" in text:
+        set_slot("tech_category", "不确定")
+
+    if "产线" in text:
+        set_slot("scenario", "产线")
+    elif "户外" in text:
+        set_slot("scenario", "户外监测")
+    elif "实验室" in text:
+        set_slot("scenario", "实验室小仓")
+    elif any(k in text for k in ("冷库", "仓储", "仓库")):
+        set_slot("scenario", "冷库仓储")
+    elif any(k in text for k in ("冷链", "药品", "疫苗", "生鲜", "冷冻", "保温箱", "运输")):
+        set_slot("scenario", text if len(text) <= 40 else "冷链运输")
+    if "Fresh Tag" in text or "fresh tag" in text.lower():
+        set_slot("scenario", "冷链运输")
+
+
+    # 本轮若已通过关键词写入至少一个槽，就不再把整句灌进其它空槽
+    # 但允许多个关键词槽在同一轮同时写入（如「冷库+实时」）
+    if touched:
+        return updated
+
+
+    pending = sorted(
+        [s for s in slots_meta if s.get("required") and not updated.get(s["id"])],
+        key=lambda s: s.get("priority", 100),
+    )
+    if pending:
+        updated[pending[0]["id"]] = text
+        return updated
+
+    optional = sorted(
+        [s for s in slots_meta if not s.get("required") and not updated.get(s["id"])],
+        key=lambda s: s.get("priority", 100),
+    )
+    if optional:
+        updated[optional[0]["id"]] = text
+    return updated
+
+
+def next_slot_prompt(slots_meta: list[dict[str, Any]], filled: dict[str, str]) -> dict[str, str] | None:
+    pending = sorted(
+        [s for s in slots_meta if s.get("required") and not (filled.get(s["id"]) or "").strip()],
+        key=lambda s: s.get("priority", 100),
+    )
+    if not pending:
+        # optional slots: ask once if empty
+        optional = sorted(
+            [s for s in slots_meta if not s.get("required") and not (filled.get(s["id"]) or "").strip()],
+            key=lambda s: s.get("priority", 100),
+        )
+        if optional:
+            s = optional[0]
+            return {"id": s["id"], "label": s["label"], "prompt": s["prompt"]}
+        return None
+    s = pending[0]
+    return {"id": s["id"], "label": s["label"], "prompt": s["prompt"]}

engine/app/settings.py

@@ -0,0 +1,24 @@
+from pathlib import Path
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+ROOT = Path(__file__).resolve().parents[2]
+KNOWLEDGE_DIR = ROOT / "knowledge"
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_file=str(ROOT / ".env"), extra="ignore")
+
+    app_name: str = "鲜境智能客服"
+    knowledge_dir: Path = KNOWLEDGE_DIR
+    retrieval_backend: str = "ragflow"
+    ragflow_base_url: str = "http://127.0.0.1:9380"
+    ragflow_api_key: str = ""
+    ragflow_dataset_ids: str = ""
+    ragflow_web_url: str = "http://127.0.0.1:18080"
+    ragflow_email: str = "admin@xianjing.local"
+    ragflow_password: str = "XianjingDemo2026"
+
+
+settings = Settings()

engine/app/static/app.js

@@ -0,0 +1,144 @@
+let sessionId = null;
+
+const logEl = document.getElementById("log");
+const slotsEl = document.getElementById("slots");
+const schemeEl = document.getElementById("scheme");
+const metaEl = document.getElementById("meta");
+const form = document.getElementById("form");
+const input = document.getElementById("input");
+const statusEl = document.getElementById("status");
+
+function addBubble(role, text) {
+  const div = document.createElement("div");
+  div.className = `bubble ${role}`;
+  div.textContent = text;
+  logEl.appendChild(div);
+  logEl.scrollTop = logEl.scrollHeight;
+}
+
+async function ensureSession() {
+  if (sessionId) return sessionId;
+  const res = await fetch("/api/sessions", { method: "POST" });
+  const data = await res.json();
+  sessionId = data.session_id;
+  metaEl.textContent = `会话 ${sessionId.slice(0, 8)}`;
+  return sessionId;
+}
+
+async function sendMessage(message) {
+  await ensureSession();
+  addBubble("user", message);
+  const res = await fetch("/api/chat", {
+    method: "POST",
+    headers: { "Content-Type": "application/json" },
+    body: JSON.stringify({ session_id: sessionId, message }),
+  });
+  if (!res.ok) {
+    addBubble("bot", `请求失败：${res.status}`);
+    return;
+  }
+  const data = await res.json();
+  addBubble("bot", data.reply);
+  slotsEl.textContent = JSON.stringify(data.slots || {}, null, 2);
+  schemeEl.textContent = data.scheme ? JSON.stringify(data.scheme, null, 2) : "还没到出方案这一步";
+  metaEl.textContent = `会话 ${sessionId.slice(0, 8)} · ${data.phase}`;
+}
+
+form.addEventListener("submit", async (event) => {
+  event.preventDefault();
+  const text = input.value.trim();
+  if (!text) return;
+  input.value = "";
+  form.querySelector("button").disabled = true;
+  try {
+    await sendMessage(text);
+  } finally {
+    form.querySelector("button").disabled = false;
+    input.focus();
+  }
+});
+
+document.getElementById("reset").addEventListener("click", () => {
+  sessionId = null;
+  logEl.innerHTML = "";
+  slotsEl.textContent = "{}";
+  schemeEl.textContent = "还没到出方案这一步";
+  metaEl.textContent = "";
+  addBubble("bot", "新会话。可以直接说：「产线检测，偏向 B 类」。");
+  input.focus();
+});
+
+document.getElementById("handoff").addEventListener("submit", async (event) => {
+  event.preventDefault();
+  const msg = document.getElementById("handoff-msg");
+  if (!sessionId) {
+    msg.textContent = "请先开始一段咨询。";
+    return;
+  }
+  const body = Object.fromEntries(new FormData(event.target).entries());
+  const res = await fetch("/api/handoff", {
+    method: "POST",
+    headers: { "Content-Type": "application/json" },
+    body: JSON.stringify({ session_id: sessionId, ...body }),
+  });
+  const data = await res.json();
+  msg.textContent = res.ok ? `已记录，编号 ${data.lead_id}` : "提交失败";
+  if (res.ok) loadLeads();
+});
+
+document.querySelectorAll(".tab").forEach((button) => {
+  button.addEventListener("click", () => {
+    document.querySelectorAll(".tab").forEach((item) => item.classList.remove("active"));
+    document.querySelectorAll(".pane").forEach((item) => item.classList.remove("active"));
+    button.classList.add("active");
+    document.getElementById(button.dataset.tab).classList.add("active");
+    if (button.dataset.tab === "knowledge") loadKnowledge();
+    if (button.dataset.tab === "leads") loadLeads();
+  });
+});
+
+async function loadHealth() {
+  try {
+    const data = await (await fetch("/api/health")).json();
+    const ragOk = data.ragflow && data.ragflow.ok && data.dataset_ids_configured;
+    statusEl.textContent = ragOk ? "Ragflow 已接入" : "Ragflow 未就绪，出方案时会标明检索失败";
+    statusEl.className = `status ${ragOk ? "ok" : "bad"}`;
+    if (data.ragflow_web) document.getElementById("ragflow-link").href = data.ragflow_web;
+  } catch {
+    statusEl.textContent = "咨询服务未连通";
+    statusEl.className = "status bad";
+  }
+}
+
+async function loadKnowledge() {
+  const box = document.getElementById("ragflow-status");
+  const files = document.getElementById("files");
+  const status = await (await fetch("/api/ragflow/status")).json();
+  box.textContent = JSON.stringify(status, null, 2);
+  const list = await (await fetch("/api/knowledge/files")).json();
+  files.innerHTML = "";
+  (list.files || []).forEach((file) => {
+    const li = document.createElement("li");
+    li.textContent = `${file.name} · ${file.bytes} 字节`;
+    files.appendChild(li);
+  });
+}
+
+async function loadLeads() {
+  const box = document.getElementById("lead-list");
+  const data = await (await fetch("/api/leads")).json();
+  if (!data.leads || !data.leads.length) {
+    box.textContent = "暂无";
+    return;
+  }
+  box.innerHTML = "";
+  data.leads.forEach((lead) => {
+    const div = document.createElement("div");
+    div.className = "lead";
+    div.textContent = `#${lead.id} ${lead.name} ${lead.phone} · ${lead.note || "无备注"} · ${lead.created_at}`;
+    box.appendChild(div);
+  });
+}
+
+addBubble("bot", "你好，我是鲜境方案客服。先问清技术路线和场景，再给出产品组合草稿，并附上知识库依据。\n试试：「户外监测，指标还不确定」。");
+loadHealth();

engine/app/static/index.html

@@ -0,0 +1,79 @@
+<!DOCTYPE html>
+<html lang="zh-CN">
+<head>
+  <meta charset="UTF-8" />
+  <meta name="viewport" content="width=device-width, initial-scale=1" />
+  <title>鲜境智能客服</title>
+  <link rel="preconnect" href="https://fonts.googleapis.com" />
+  <link rel="preconnect" href="https://fonts.gstatic.com" crossorigin />
+  <link href="https://fonts.googleapis.com/css2?family=DM+Sans:wght@500;700&family=Noto+Sans+SC:wght@400;500;700&display=swap" rel="stylesheet" />
+  <link rel="stylesheet" href="/static/styles.css" />
+</head>
+<body>
+  <div class="shell">
+    <header class="top">
+      <div>
+        <p class="brand">鲜境物联</p>
+        <h1>方案型智能客服</h1>
+      </div>
+      <div class="status" id="status">正在检查知识库…</div>
+    </header>
+
+    <nav class="tabs">
+      <button type="button" class="tab active" data-tab="consult">线上咨询</button>
+      <button type="button" class="tab" data-tab="knowledge">知识库</button>
+      <button type="button" class="tab" data-tab="leads">留资</button>
+    </nav>
+
+    <section id="consult" class="pane active">
+      <div class="layout">
+        <section class="panel chat">
+          <div id="log" class="log" aria-live="polite"></div>
+          <form id="form" class="composer">
+            <input id="input" type="text" placeholder="描述现场需求，或回答追问" autocomplete="off" required />
+            <button type="submit">发送</button>
+          </form>
+          <div class="toolbar">
+            <button type="button" id="reset" class="ghost">新会话</button>
+            <span id="meta" class="meta"></span>
+          </div>
+        </section>
+        <aside class="panel side">
+          <h2>已收集信息</h2>
+          <pre id="slots">{}</pre>
+          <h2>方案草稿</h2>
+          <pre id="scheme">还没到出方案这一步</pre>
+          <h2>转人工留资</h2>
+          <form id="handoff" class="stack">
+            <input name="name" placeholder="称呼" required />
+            <input name="phone" placeholder="电话" required />
+            <textarea name="note" rows="2" placeholder="补充说明，可选"></textarea>
+            <button type="submit">提交给销售</button>
+            <p id="handoff-msg" class="meta"></p>
+          </form>
+        </aside>
+      </div>
+    </section>
+
+    <section id="knowledge" class="pane">
+      <div class="panel pad">
+        <h2>Ragflow 知识库</h2>
+        <p class="lede">检索走自托管 Ragflow。控制台可改文档、看召回；这里只看接入状态。示例资料待甲方替换。</p>
+        <p><a id="ragflow-link" href="http://127.0.0.1:18080" target="_blank" rel="noreferrer">打开 Ragflow 控制台</a></p>
+        <pre id="ragflow-status">加载中</pre>
+        <h2>本地语料（灌库来源）</h2>
+        <ul id="files"></ul>
+      </div>
+    </section>
+
+    <section id="leads" class="pane">
+      <div class="panel pad">
+        <h2>已留资</h2>
+        <p class="lede">演示库存在本机 SQLite，不是正式 CRM。</p>
+        <div id="lead-list" class="lead-list">暂无</div>
+      </div>
+    </section>
+  </div>
+  <script src="/static/app.js"></script>
+</body>
+</html>

engine/app/static/styles.css

@@ -0,0 +1,75 @@
+:root {
+  --bg0: #10211a;
+  --bg1: #1b3b2c;
+  --panel: #f6f8f4;
+  --ink: #14231c;
+  --muted: #5d7368;
+  --accent: #1d7a4d;
+  --line: rgba(20, 35, 28, 0.12);
+  --bot: #e7f3eb;
+}
+
+* { box-sizing: border-box; }
+body {
+  margin: 0;
+  min-height: 100vh;
+  color: var(--ink);
+  font-family: "Noto Sans SC", "DM Sans", sans-serif;
+  background:
+    radial-gradient(900px 480px at 0% 0%, #3d7a58 0%, transparent 55%),
+    linear-gradient(165deg, var(--bg0), var(--bg1));
+}
+.shell { max-width: 1120px; margin: 0 auto; padding: 24px 16px 40px; }
+.top { display: flex; justify-content: space-between; gap: 16px; align-items: flex-end; color: #f4faf6; }
+.brand { margin: 0; font-family: "DM Sans", sans-serif; letter-spacing: 0.08em; font-size: 13px; color: #c6ef7a; }
+.top h1 { margin: 4px 0 0; font-size: clamp(1.5rem, 3vw, 2rem); }
+.status { font-size: 13px; padding: 8px 12px; border-radius: 999px; background: rgba(0,0,0,0.25); }
+.status.ok { background: #1d7a4d; }
+.status.bad { background: #8a3b2d; }
+
+.tabs { display: flex; gap: 8px; margin: 18px 0; }
+.tab {
+  border: 0; background: rgba(255,255,255,0.12); color: #f4faf6;
+  border-radius: 999px; padding: 8px 14px; font: inherit; cursor: pointer;
+}
+.tab.active { background: #f6f8f4; color: var(--ink); }
+
+.pane { display: none; }
+.pane.active { display: block; }
+.layout { display: grid; grid-template-columns: 1.45fr 0.9fr; gap: 14px; }
+@media (max-width: 860px) { .layout { grid-template-columns: 1fr; } }
+
+.panel {
+  background: var(--panel);
+  border-radius: 18px;
+  box-shadow: 0 16px 40px rgba(0,0,0,0.18);
+}
+.chat { display: flex; flex-direction: column; min-height: 560px; }
+.log { flex: 1; overflow: auto; padding: 16px; display: flex; flex-direction: column; gap: 10px; }
+.bubble { max-width: 92%; padding: 10px 12px; border-radius: 12px; white-space: pre-wrap; line-height: 1.5; font-size: 14px; }
+.bubble.bot { align-self: flex-start; background: var(--bot); }
+.bubble.user { align-self: flex-end; background: var(--accent); color: white; }
+.composer { display: flex; gap: 8px; padding: 12px; border-top: 1px solid var(--line); }
+.composer input, .stack input, .stack textarea {
+  width: 100%; border: 1px solid var(--line); border-radius: 10px; padding: 11px 12px; font: inherit;
+}
+.composer button, .stack button, .ghost {
+  border: 0; border-radius: 10px; padding: 11px 14px; font: inherit; font-weight: 600; cursor: pointer;
+}
+.composer button, .stack button { background: var(--accent); color: white; }
+.ghost { background: transparent; color: var(--accent); border: 1px solid var(--line); }
+.toolbar { display: flex; justify-content: space-between; align-items: center; padding: 0 12px 12px; }
+.meta, .lede { color: var(--muted); font-size: 13px; }
+.side, .pad { padding: 16px; }
+.side h2, .pad h2 { margin: 14px 0 8px; font-size: 13px; letter-spacing: 0.04em; color: var(--muted); }
+.side h2:first-child, .pad h2:first-child { margin-top: 0; }
+pre {
+  margin: 0; padding: 10px; background: #eef5f0; border-radius: 10px;
+  font-size: 12px; overflow: auto; max-height: 180px; white-space: pre-wrap;
+}
+.stack { display: grid; gap: 8px; }
+.lead-list { display: grid; gap: 8px; }
+.lead {
+  padding: 10px 12px; border-radius: 10px; background: #eef5f0; font-size: 14px;
+}
+a { color: var(--accent); }

engine/requirements.txt

@@ -0,0 +1,8 @@
+fastapi==0.115.6
+uvicorn[standard]==0.32.1
+pydantic==2.10.3
+pydantic-settings==2.6.1
+pyyaml==6.0.2
+httpx==0.28.1
+python-multipart==0.0.20
+pycryptodome==3.21.0

infra/ragflow-public.pem

@@ -0,0 +1,9 @@
+-----BEGIN PUBLIC KEY-----
+MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEArq9XTUSeYr2+N1h3Afl/
+z8Dse/2yD0ZGrKwx+EEEcdsBLca9Ynmx3nIB5obmLlSfmskLpBo0UACBmB5rEjBp
+2Q2f3AG3Hjd4B+gNCG6BDaawuDlgANIhGnaTLrIqWrrcm4EMzJOnAOI1fgzJRsOO
+UEfaS318Eq9OVO3apEyCCt0lOQK6PuksduOjVxtltDav+guVAA068NrPYmRNabVK
+RNLJpL8w4D44sfth5RvZ3q9t+6RTArpEtc5sh5ChzvqPOzKGMXW83C95TxmXqpbK
+6olN4RevSfVjEAgCydH6HN6OhtOQEcnrU97r9H0iZOWwbw3pVrZiUkuRD1R56Wzs
+2wIDAQAB
+-----END PUBLIC KEY-----

knowledge/corpus/faq.md

@@ -0,0 +1,17 @@
+# 鲜境示例资料 · 常见询盘
+
+## 下班后没人回怎么办
+
+线上咨询由本系统 7×24 接待。答不上、置信度低、或客户要求人工时，留下称呼和电话，销售在工作时间跟进。
+
+## 你们是卖设备还是卖方案
+
+鲜境是卖方案。先判断走传感监测还是控制联动，再给产品组合，而不是一问一答式报价单。
+
+## 能不能先给个大概价格
+
+演示知识库没有价格。价格、折扣、交期一律转人工，不得估算。
+
+## 资料里没有的参数
+
+回复「资料未覆盖，需要工程师确认」，不要用常识补全型号参数。

knowledge/corpus/freshliance-homepage-company.md

@@ -0,0 +1,54 @@
+# 鲜境物联 · 公司与首页主推产品（官网样例）
+
+> 来源：https://www.freshliance.com/  
+> 公司：郑州鲜境电子有限公司（Zhengzhou Freshliance Electronics Corp., Ltd.）  
+> 成立：2015 年；质量管理：ISO9000 体系认证企业（官网表述）
+
+## 公司定位
+
+专注物联网温湿度数据记录仪，以及冷链在线环境监测与追踪方案的研发、制造与供应。重点服务生命科学、医药、果蔬、食品等冷链运输与仓储场景。
+
+官网承诺方向：感知、监测、记录、连接（sensing / monitoring / logging），为冷链环境变化提供可追溯记录。
+
+联系邮箱（官网）：inquiry@freshliance.com
+
+## 首页主推产品线
+
+### 1. TagPlus-T Temperature Data Loggers
+
+- 多次使用（multi-use）USB 温度记录仪  
+- 带 LCD，便于现场查看  
+- 参数可配置；行程结束后插电脑生成加密报告（含摘要、曲线、表格）  
+- 支持临时读取模式：生成临时报告后仍可继续记录  
+- 适合运输途中对温敏货物做温度变化记录  
+
+### 2. BlueTag T10 Bluetooth Temperature Data Logger
+
+- 蓝牙温度记录仪，体积小、便携  
+- 通过手机 APP 远程配置、查看与导出，无需开箱/开柜/开集装箱  
+- 数据可同步云端，支持多用户同时在线查看  
+- APP 提供分析、筛选、曲线、报警等能力  
+- 适合冷链物流、疫苗运输、生鲜配送、冷藏运输、保温箱运输等  
+
+### 3. GAEA Series G1000 4G/WiFi/LAN Bluetooth Gateway
+
+- 面向小面积仓储等场景的实时温湿度监测网关  
+- 网络接口可选 4G / WiFi / LAN（下单时按需求配置）  
+- 通过蓝牙汇聚现场记录仪数据，再上传云平台  
+- 可与 BlueTag 系列配合，构成实时监测系统；官网称可连接多达约 30 个传感器（G1000-S 相关页）  
+- 断网时终端可继续本地记录，恢复网络后补传  
+
+## 官网服务能力（SERVICE）
+
+1. **API service**  
+   简单易用；可与 IoT 平台集成；可把实时与历史传感数据拉到自有网站或看板做分析与集中管理。
+
+2. **Tailored Solutions（定制方案）**  
+   IoT 平台集成、远程实时洞察与告警、自动数据处理与报告、多地点设备集中管理、基于角色的多级权限；官网提及符合 FDA 21 CFR Part 11。
+
+3. **Configuration Download**  
+   提供记录仪对应的软件、手机 APP、云平台；用于配置与激活、数据可视化管理、PDF/CSV 报告生成与下载。
+
+## 对外边界
+
+本样例仅整理官网首页及首页关联产品公开信息。价格、交期、折扣、具体认证报告编号不在此承诺，客户追问转销售（inquiry@freshliance.com）。

knowledge/corpus/freshliance-homepage-solutions.md

@@ -0,0 +1,61 @@
+# 鲜境物联 · 首页产品选型说明（样例）
+
+> 依据 https://www.freshliance.com/ 首页主推的 TagPlus-T、BlueTag T10、GAEA G1000，整理咨询时如何判定路线。  
+> 目标：先判断「运输记录 / 蓝牙查数 / 仓储实时」，再推荐对应产品组合。
+
+## 三条主路线
+
+| 客户需求关键词 | 推荐方向 | 首页对应产品 |
+|----------------|----------|--------------|
+| USB 插电脑出报告、可重复用、运输途中看数/到货出 PDF | 多次使用 USB 记录 | **TagPlus-T** |
+| 不开箱用手机看温度、蓝牙配置、多用户同步 | 蓝牙记录 + APP | **BlueTag T10** |
+| 冷库/小仓实时看数、云端告警、多探头汇聚 | 网关 + 蓝牙终端 | **GAEA G1000 + BlueTag 系列** |
+
+## 追问槽位（客服必问）
+
+1. 场景：运输途中，还是库房/冷柜长期监测？  
+2. 取数方式：到货插电脑出报告，还是手机/云端随时看？  
+3. 是否需要实时告警（短信/邮件/APP）？  
+4. 是否多次重复使用，还是单次行程一次性？  
+5. 网络条件：现场有 WiFi/LAN，还是需要 4G？  
+6. 是否要对接自有系统（API / 私有云 / ERP/WMS）？
+
+## 判定建议
+
+### A. 推荐 TagPlus-T
+
+- 客户说：冷链运输、到货要 PDF/CSV、希望可配置、可多次用、有屏幕看当前温度  
+- 强调：加密报告、临时读取后可继续记、适合温敏货在途记录  
+
+### B. 推荐 BlueTag T10
+
+- 客户说：不想开箱读数、手机 APP 配置、疫苗/生鲜/保温箱运输、多人同时看数据  
+- 强调：蓝牙无线读数、云端同步、分析与报警  
+
+### C. 推荐 GAEA G1000（常与 BlueTag 组合）
+
+- 客户说：冷库、实验室小仓、小面积仓储要实时温湿度、要远程告警、要集中管理多点  
+- 强调：4G/WiFi/LAN 可选、蓝牙汇聚多终端、断点续传、APP FreshLink / 云平台  
+
+### D. 服务加项
+
+- 要对接自有看板/系统 → 介绍 **API service**  
+- 要多仓权限、自动报告、Part 11 合规诉求 → 介绍 **Tailored Solutions**  
+- 只要软件/APP/配置工具 → 介绍 **Configuration Download**
+
+## 不要硬推的情况
+
+- 只要一次性极低成本 USB、不关心重复使用 → 不要只推 TagPlus-T，转人工核对是否有一次性系列  
+- 大面积厂区复杂组网、无官网明确型号覆盖 → 转人工做定制评估  
+- 价格、交期、合同条款 → 一律转销售，邮箱 inquiry@freshliance.com  
+
+## 咨询话术示例
+
+客户：「我们药品冷链运输，到货要出报告。」  
+→ 先确认是否可重复使用、要不要屏幕；倾向 **TagPlus-T**。  
+
+客户：「保温箱运输，不能开箱，司机用手机看温度。」  
+→ 倾向 **BlueTag T10**。  
+
+客户：「小冷库要 7×24 实时看，断网还能补数据。」  
+→ 倾向 **GAEA G1000 + BlueTag**，并确认 WiFi/4G/LAN。

knowledge/corpus/policies.md

@@ -0,0 +1,13 @@
+# 鲜境示例资料 · 对外边界
+
+## 敏感与承诺
+
+- 不报未入库的精度、认证、交期、折扣、质保年限。
+- 客户留下的电话只用于本次跟进，回复中不得复述其他客户信息。
+- 价格一律转人工。
+
+## 转人工触发
+
+- 客户明确说转人工、找销售、要真人。
+- 判定置信度低于 0.55。
+- 知识库没有命中依据却被追问具体参数。

knowledge/corpus/products-a.md

@@ -0,0 +1,28 @@
+# 鲜境示例资料 · A 类传感监测
+
+> 本文件是演示知识，型号与参数为占位。正式上线前替换为甲方产品手册。未写入本库的参数，客服不得对外承诺。
+
+## 技术路线
+
+A 类面向监测与数据采集：现场传感器采集、网关汇聚、远程查看。典型客户说的是「要看数据、要报警、要远程」，而不是「要联动产线设备」。
+
+## 推荐组合
+
+- 主机：XJ-A100
+- 网关：XJ-A-GW
+
+XJ-A100 适用于户外与环境监测，宽温工作，需配合 XJ-A-GW 做多节点汇聚与协议转换。不要单独报主机而不报网关。
+
+## 适用场景
+
+- 户外监测
+- 环境采集
+- 实验室标定（轻量）
+
+## 必问再确认
+
+防护等级、供电方式、是否已有采集平台。这三项资料未齐时，只能给草稿，建议转销售确认。
+
+## 不适合
+
+产线设备联动、大量 IO 点位、本地逻辑控制。这些走 B 类控制路线。

knowledge/corpus/products-b.md

@@ -0,0 +1,28 @@
+# 鲜境示例资料 · B 类控制联动
+
+> 本文件是演示知识，型号与参数为占位。正式上线前替换为甲方产品手册。未写入本库的参数，客服不得对外承诺。
+
+## 技术路线
+
+B 类面向产线与设备控制：本地逻辑、IO 扩展、与已有设备联动。客户常说「要控设备、要对接点位、要跟产线走」。
+
+## 推荐组合
+
+- 控制主机：XJ-B200
+- IO 扩展：XJ-B-IO
+
+XJ-B200 支持本地逻辑与扩展 IO。XJ-B-IO 用于增加数字量或模拟量通道，应与 XJ-B200 配套报价，不要只报扩展模块。
+
+## 适用场景
+
+- 产线检测
+- 设备联动
+- 现场控制柜改造
+
+## 必问再确认
+
+点位数量、已有 PLC 或上位机、安装空间、是否需要改造停电窗口。
+
+## 不适合
+
+纯远程环境监测、只要看数不要控制。这些走 A 类传感路线。

knowledge/corpus/solutions.md

@@ -0,0 +1,22 @@
+# 鲜境示例资料 · 方案判定说明
+
+> 演示用判定说明，与 knowledge/decision_tree.yaml 一致。业务专家确认后以 YAML 为准。
+
+## 先问再出方案
+
+1. 技术大类倾向：A 类传感 / B 类控制 / 不确定
+2. 应用场景：产线、户外、实验室等
+3. 关键指标：量程、精度、点位或响应时间；没有就记「暂无」
+4. 现场约束：空间、已有设备、预算、交付区域；没有就记「暂无」
+
+## 判定顺序
+
+1. 客户明确要 A 类或传感：推荐 XJ-A100 + XJ-A-GW
+2. 客户明确要 B 类或控制：推荐 XJ-B200 + XJ-B-IO
+3. 未指定大类但场景是产线：先按 B 类给草稿，并写明需人工确认
+4. 未指定大类但场景是户外监测：先按 A 类给草稿，并写明需确认防护与供电
+5. 以上都不像：不要硬推型号，转人工
+
+## 对外话术边界
+
+方案正文只能复述知识库里出现过的事实。客户追问具体认证、交期、折扣时，一律转销售，不编造。

knowledge/decision_tree.yaml

@@ -0,0 +1,147 @@
+# 判定树：对齐 freshliance.com 首页主推产品
+version: 1
+
+rules:
+  - id: tagplus_explicit
+    order: 5
+    description: 明确要 TagPlus / USB 多次记录 / 到货 PDF
+    when_all:
+      - type: slot_contains
+        slot: tech_category
+        value: TagPlus
+    then:
+      product_bundle: ["TagPlus-T"]
+      tech_line: 多次使用USB温度记录
+      summary: 按官网首页主推匹配 TagPlus-T：可配置、LCD、USB 加密 PDF/CSV，适合运输途中温敏货记录。
+      confidence: 0.88
+
+  - id: usb_pdf_transport
+    order: 8
+    description: USB/PDF/到货报告 + 运输
+    when_all:
+      - type: slot_contains
+        slot: key_metric
+        value: PDF
+    then:
+      product_bundle: ["TagPlus-T"]
+      tech_line: 多次使用USB温度记录
+      summary: 需要到货 PDF/报告时，优先推荐 TagPlus-T（首页主推 USB 温度记录仪）。
+      confidence: 0.82
+
+  - id: bluetag_explicit
+    order: 10
+    description: 明确要蓝牙 / BlueTag / 不开箱
+    when_all:
+      - type: slot_contains
+        slot: tech_category
+        value: BlueTag
+    then:
+      product_bundle: ["BlueTag T10"]
+      tech_line: 蓝牙温度记录
+      summary: 按官网首页主推匹配 BlueTag T10：蓝牙+APP 配置与读数，适合不开箱查看。
+      confidence: 0.88
+
+  - id: no_open_box
+    order: 12
+    description: 不开箱 / 手机读数
+    when_all:
+      - type: slot_contains
+        slot: key_metric
+        value: 不开箱
+    then:
+      product_bundle: ["BlueTag T10"]
+      tech_line: 蓝牙温度记录
+      summary: 需要不开箱读数时推荐 BlueTag T10；若还要仓储实时汇聚，可再评估加 GAEA 网关。
+      confidence: 0.84
+
+  - id: gaea_explicit
+    order: 14
+    description: 明确要网关 / GAEA / 实时仓储
+    when_all:
+      - type: slot_contains
+        slot: tech_category
+        value: GAEA
+    then:
+      product_bundle: ["GAEA G1000", "BlueTag 系列"]
+      tech_line: 网关实时监测
+      summary: 按官网首页主推匹配 GAEA G1000：4G/WiFi/LAN 网关汇聚蓝牙终端，适合小面积仓储实时监测。
+      confidence: 0.86
+
+  - id: realtime_storage
+    order: 16
+    description: 冷库/仓储实时告警
+    when_all:
+      - type: slot_contains
+        slot: scenario
+        value: 冷库
+    then:
+      product_bundle: ["GAEA G1000", "BlueTag 系列"]
+      tech_line: 网关实时监测
+      summary: 冷库/仓储实时监测优先 GAEA G1000 + BlueTag；需确认现场网络（4G/WiFi/LAN）。
+      confidence: 0.83
+
+  - id: warehouse
+    order: 17
+    description: 仓储场景
+    when_all:
+      - type: slot_contains
+        slot: scenario
+        value: 仓储
+    then:
+      product_bundle: ["GAEA G1000", "BlueTag 系列"]
+      tech_line: 网关实时监测
+      summary: 仓储环境监测按首页方案推荐 GAEA 网关组合；点数与组网方式需人工确认。
+      confidence: 0.8
+
+  - id: vaccine_bluetag
+    order: 20
+    description: 疫苗/药品运输偏蓝牙查数
+    when_all:
+      - type: slot_contains
+        slot: scenario
+        value: 疫苗
+    then:
+      product_bundle: ["BlueTag T10"]
+      tech_line: 蓝牙温度记录
+      summary: 疫苗运输场景官网示例适配 BlueTag T10；若只要到货 USB 报告也可评估 TagPlus-T，请再确认取数方式。
+      confidence: 0.8
+
+  - id: cold_chain_transport
+    order: 25
+    description: 冷链运输默认 USB 多次记录
+    when_all:
+      - type: slot_contains
+        slot: scenario
+        value: 冷链
+    then:
+      product_bundle: ["TagPlus-T"]
+      tech_line: 多次使用USB温度记录
+      summary: 冷链运输通用场景先推荐首页主推 TagPlus-T；若强调手机不开箱读数则改 BlueTag T10。
+      confidence: 0.76
+
+  - id: fresh_food
+    order: 26
+    description: 生鲜配送
+    when_all:
+      - type: slot_contains
+        slot: scenario
+        value: 生鲜
+    then:
+      product_bundle: ["BlueTag T10"]
+      tech_line: 蓝牙温度记录
+      summary: 生鲜配送官网示例适配 BlueTag T10；需要到货加密 PDF 时可改推 TagPlus-T。
+      confidence: 0.78
+
+  - id: fallback_uncertain
+    order: 1000
+    description: 兜底
+    when_all:
+      - type: always
+    then:
+      product_bundle: ["待专家确认"]
+      tech_line: 待判定
+      summary: 信息不足，建议补充「运输还是仓储、USB报告还是手机/云端实时」，或转人工。联系 inquiry@freshliance.com。
+      confidence: 0.4
+      handoff: true
+
+handoff_confidence_below: 0.55

knowledge/sample_chunks.yaml

@@ -0,0 +1,31 @@
+# Mock 检索语料：无 Ragflow 时用于演示「带引用的方案」
+chunks:
+  - id: c1
+    title: XJ-A100 产品要点
+    product: XJ-A100
+    content: XJ-A100 适用于户外与环境监测类场景，支持宽温工作，需配合 XJ-A-GW 网关做数据汇聚。
+    tags: [A, 户外, 传感]
+
+  - id: c2
+    title: XJ-A-GW 网关
+    product: XJ-A-GW
+    content: XJ-A-GW 提供多节点接入与协议转换，常与 XJ-A100 组成监测方案。
+    tags: [A, 网关]
+
+  - id: c3
+    title: XJ-B200 控制主机
+    product: XJ-B200
+    content: XJ-B200 面向产线联动与设备控制，支持本地逻辑与扩展 IO。
+    tags: [B, 产线, 控制]
+
+  - id: c4
+    title: XJ-B-IO 扩展
+    product: XJ-B-IO
+    content: XJ-B-IO 用于扩展数字量/模拟量通道，与 XJ-B200 配套。
+    tags: [B, IO]
+
+  - id: c5
+    title: 方案注意事项
+    product: GENERAL
+    content: 正式报价前需确认现场供电、安装空间与是否已有同类设备，避免重复建设。
+    tags: [通用, 约束]

knowledge/slots.yaml

@@ -0,0 +1,36 @@
+# 追问槽位：需求未定稿时可随时增删字段；引擎按 priority 顺序问未填项
+version: 1
+intent: solution_consult
+
+slots:
+  - id: tech_category
+    label: 方案路线倾向
+    prompt: 更倾向哪一类？（TagPlus USB多次记录 / BlueTag蓝牙手机读数 / GAEA网关实时仓储 / 不确定，希望推荐）
+    required: true
+    priority: 10
+    extract_hint: USB、蓝牙、网关实时、不确定
+
+  - id: scenario
+    label: 应用场景
+    prompt: 主要用在什么场景？（冷链运输、药品/疫苗、生鲜配送、冷库仓储、实验室小仓、保温箱等）
+    required: true
+    priority: 20
+
+  - id: key_metric
+    label: 关键要求
+    prompt: 有没有必须满足的要求？（实时告警、到货PDF报告、不开箱读数、行程天数、温区等；没有可说「暂无」）
+    required: true
+    priority: 30
+
+  - id: constraints
+    label: 现场约束
+    prompt: 现场有哪些约束？（网络：WiFi/4G/LAN、是否对接自有系统API、预算区间等；没有可说「暂无」）
+    required: false
+    priority: 40
+
+closing_prompt: 信息已较完整，正在按鲜境官网产品线匹配方案…
+handoff_phrases:
+  - 转人工
+  - 人工客服
+  - 找销售
+  - 真人

scripts/bootstrap_ragflow.py

@@ -0,0 +1,150 @@
+#!/usr/bin/env python3
+"""注册 Ragflow 账号、建库、灌入 knowledge/corpus，并写回仓库 .env。"""
+
+from __future__ import annotations
+
+import sys
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from engine.app.ragflow_client import RagflowClient, RagflowError  # noqa: E402
+from engine.app.settings import settings  # noqa: E402
+
+CORPUS = {
+    "products": ["products-a.md", "products-b.md"],
+    "solutions": ["solutions.md"],
+    "faqs": ["faq.md", "policies.md"],
+}
+DESCRIPTIONS = {
+    "products": "鲜境示例产品资料：A 类传感、B 类控制",
+    "solutions": "鲜境示例方案判定说明",
+    "faqs": "鲜境示例询盘与对外边界",
+}
+
+
+def wait_healthy(client: RagflowClient, timeout: int = 180) -> None:
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        health = client.health()
+        if health.get("ok"):
+            print("ragflow health ok", health)
+            return
+        print("waiting ragflow", health)
+        time.sleep(5)
+    raise SystemExit("Ragflow 未在时限内就绪")
+
+
+def auth_or_register(client: RagflowClient) -> str:
+    try:
+        return client.register(settings.ragflow_email, settings.ragflow_password, "鲜境管理员")
+    except RagflowError as exc:
+        text = str(exc)
+        if "already" in text.lower() or "已经" in text or "registered" in text.lower():
+            return client.login(settings.ragflow_email, settings.ragflow_password)
+        raise
+
+
+def upsert_datasets(client: RagflowClient) -> dict[str, str]:
+    existing = {item.get("name"): item for item in client.list_datasets()}
+    ids: dict[str, str] = {}
+    for name, description in DESCRIPTIONS.items():
+        found = existing.get(name)
+        if found:
+            ids[name] = found["id"]
+            print("reuse dataset", name, found["id"])
+            continue
+        created = client.create_dataset(name, description)
+        dataset_id = created.get("id")
+        if not dataset_id:
+            raise SystemExit(f"创建数据集失败: {created}")
+        ids[name] = dataset_id
+        print("created dataset", name, dataset_id)
+    embed = {
+        "embedding_model": "BAAI/bge-small-zh-v1.5@Builtin",
+        "parser_config": {
+            "graphrag": {"use_graphrag": False},
+            "raptor": {"use_raptor": False},
+        },
+    }
+    for dataset_id in ids.values():
+        client.update_dataset(dataset_id, embed)
+    return ids
+
+
+def ingest(client: RagflowClient, dataset_ids: dict[str, str]) -> None:
+    corpus_dir = ROOT / "knowledge" / "corpus"
+    for name, files in CORPUS.items():
+        dataset_id = dataset_ids[name]
+        docs = client.list_documents(dataset_id)
+        present = {item.get("name") or item.get("location") for item in docs}
+        new_ids = []
+        for filename in files:
+            if filename in present:
+                print("skip existing", filename)
+                continue
+            uploaded = client.upload_document(dataset_id, corpus_dir / filename)
+            for item in uploaded:
+                doc_id = item.get("id")
+                if doc_id:
+                    new_ids.append(doc_id)
+            print("uploaded", filename, new_ids[-1] if new_ids else uploaded)
+        if new_ids:
+            client.parse_documents(dataset_id, new_ids)
+            print("parse queued", name, new_ids)
+
+
+def wait_chunks(client: RagflowClient, dataset_ids: dict[str, str], timeout: int = 300) -> None:
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        pending = 0
+        for dataset_id in dataset_ids.values():
+            for doc in client.list_documents(dataset_id):
+                run = str(doc.get("run") or "")
+                chunks = int(doc.get("chunk_count") or 0)
+                if chunks > 0 or run.upper() in {"DONE", "3", "SUCCESS"}:
+                    continue
+                if run.upper() in {"FAIL", "4", "FAILED"}:
+                    raise SystemExit(f"解析失败 {doc.get('name')}: {(doc.get('progress_msg') or '')[-300:]}")
+                pending += 1
+                print("parsing", doc.get("name"), "run=", run, "chunks=", chunks)
+        if pending == 0:
+            print("all documents parsed")
+            return
+        time.sleep(6)
+    print("warn: parse still running, continue anyway")
+
+
+def write_env(dataset_ids: dict[str, str], api_key: str) -> None:
+    env_path = ROOT / ".env"
+    lines = [
+        "RETRIEVAL_BACKEND=ragflow",
+        "RAGFLOW_BASE_URL=http://127.0.0.1:9380",
+        f"RAGFLOW_API_KEY={api_key}",
+        "RAGFLOW_DATASET_IDS=" + ",".join(dataset_ids[name] for name in ("products", "solutions", "faqs")),
+        "RAGFLOW_WEB_URL=http://127.0.0.1:18080",
+        f"RAGFLOW_EMAIL={settings.ragflow_email}",
+        f"RAGFLOW_PASSWORD={settings.ragflow_password}",
+        "",
+    ]
+    env_path.write_text("\n".join(lines), encoding="utf-8")
+    print("wrote", env_path)
+
+
+def main() -> None:
+    client = RagflowClient(api_key="")
+    wait_healthy(client)
+    session_auth = auth_or_register(client)
+    token = client.ensure_api_token(session_auth)
+    client.api_key = token
+    dataset_ids = upsert_datasets(client)
+    ingest(client, dataset_ids)
+    wait_chunks(client, dataset_ids)
+    write_env(dataset_ids, token)
+    print("bootstrap done")
+
+
+if __name__ == "__main__":
+    main()

scripts/fix_docker_forward.sh

@@ -0,0 +1,12 @@
+#!/usr/bin/env bash
+# Docker 在部分环境里只写了 nft 规则，内核仍走 iptables-legacy 且 FORWARD 为 DROP，
+# 容器之间会连不上 MySQL。本脚本给 Ragflow 网桥补上转发放行。
+set -euo pipefail
+bridge=$(ip -br link | awk '/^br-/ && $2 ~ /UP/ {print $1; exit}')
+if [[ -z "${bridge}" ]]; then
+  echo "no up bridge" >&2
+  exit 1
+fi
+sudo iptables-legacy -C FORWARD -i "$bridge" -j ACCEPT 2>/dev/null || sudo iptables-legacy -I FORWARD 1 -i "$bridge" -j ACCEPT
+sudo iptables-legacy -C FORWARD -o "$bridge" -j ACCEPT 2>/dev/null || sudo iptables-legacy -I FORWARD 1 -o "$bridge" -j ACCEPT
+echo "forward allowed on ${bridge}"