diff --git a/.gitignore b/.gitignore index 48c4f6f..61663bc 100644 --- a/.gitignore +++ b/.gitignore @@ -2,5 +2,6 @@ __pycache__/ .venv/ venv/ .env +data/ *.pyc .DS_Store diff --git a/README.md b/README.md index b0f649d..64d86e1 100644 --- a/README.md +++ b/README.md @@ -1,51 +1,44 @@ -# 鲜境方案引擎 · 通用原型（需求未定稿版） +# 鲜境物联 · AI 智能客服 -业务规则还在变时：**先搭可配置原型，再按真实需求改 YAML，而不是先写死后台。** +用 **Ragflow** 做知识库检索，用本仓库的方案引擎做「追问 → 判定 → 出方案 → 留资」。 -## 策略 +线上咨询台：[鲜境智能客服](http://127.0.0.1:3847/) +Ragflow 控制台：[Ragflow](http://127.0.0.1:18080/) -1. 先跑通：多轮追问（槽位）→ 判定树出组合 → 检索补依据 → 出方案草稿 -2. 槽位 / 判定树 / 示例知识 **全部外置配置**，需求变了主要改配置 -3. 检索默认 **Mock**（不依赖本机 Ragflow）；有 Ragflow 后改环境变量即可接真库 -4. 不做大而全后台；控制台能聊天验证闭环即可 +示例产品资料在 `knowledge/corpus/`，型号是占位，等甲方资料替换。判定规则在 `knowledge/decision_tree.yaml`，由业务方定对错。 -## 本仓库原型包含 +## 本机已经怎么跑 -| 路径 | 作用 | -|------|------| -| `engine/` | FastAPI 方案引擎 + 简易聊天页 | -| `knowledge/` | `slots.yaml` / `decision_tree.yaml` / 示例检索片段 | -| `docs/Ragflow实施方案.md` | 完整架构说明 | +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 -python3 -m venv .venv 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 ``` -打开：[方案引擎原型](http://127.0.0.1:3847/) +没有 Docker 时可以把 `.env` 里的 `RETRIEVAL_BACKEND` 改成 `mock`，只用本地示例片段演示对话，不代表知识库已接通。 -健康检查：`GET /api/health` -重载业务配置：`POST /api/admin/reload-config`（改完 YAML 不用重启进程时用） +## 目录 -## 接真 Ragflow（稍后） - -```bash -export RETRIEVAL_BACKEND=ragflow -export RAGFLOW_BASE_URL=http://127.0.0.1:9380 -export RAGFLOW_API_KEY=... -export RAGFLOW_DATASET_IDS=... -``` - -未配置时保持 `RETRIEVAL_BACKEND=mock`。 - -## 需求清晰后怎么补 - -1. 改 `knowledge/slots.yaml`（追问项） -2. 改 `knowledge/decision_tree.yaml`（推荐规则） -3. 灌真实资料进 Ragflow，切 `ragflow` 后端 -4. 再加渠道 Webhook / 转人工 / Eval，而不是推倒重来 +| 路径 | 作用 | +|------|------| +| `engine/` | 咨询 API 与页面 | +| `knowledge/` | 槽位、判定树、灌库语料 | +| `scripts/bootstrap_ragflow.py` | 对接 Ragflow 建库 | +| `docs/` | 选型与实施说明 | diff --git a/engine/app/leads.py b/engine/app/leads.py new file mode 100644 index 0000000..ac1a1d8 --- /dev/null +++ b/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] diff --git a/engine/app/main.py b/engine/app/main.py index e4b3b90..6a0584b 100644 --- a/engine/app/main.py +++ b/engine/app/main.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json from pathlib import Path from fastapi import FastAPI, HTTPException @@ -8,12 +9,16 @@ 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 .settings import settings +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.1.0") +app = FastAPI(title=settings.app_name, version="0.2.0") app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") @@ -26,6 +31,13 @@ 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") @@ -33,12 +45,16 @@ def index() -> FileResponse: @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()), } @@ -58,6 +74,49 @@ def reload_config() -> dict: 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() @@ -73,3 +132,21 @@ def api_chat(body: ChatRequest) -> dict: 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} diff --git a/engine/app/ragflow_client.py b/engine/app/ragflow_client.py new file mode 100644 index 0000000..53595eb --- /dev/null +++ b/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 []) diff --git a/engine/app/retrieval.py b/engine/app/retrieval.py index 95214f9..287d3b8 100644 --- a/engine/app/retrieval.py +++ b/engine/app/retrieval.py @@ -2,93 +2,79 @@ from __future__ import annotations from typing import Any -import httpx - 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]]): + 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 "mock").lower() - if backend == "ragflow": + backend = (settings.retrieval_backend or "ragflow").lower() + if backend == "mock": + return _retrieve_mock(slots, bundle) + try: return _retrieve_ragflow(slots, bundle) - return _retrieve_mock(slots, bundle) + except RagflowError as exc: + return RetrievalResult([], "ragflow", warning=str(exc)) 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 c in chunks: - if c.get("product") in bundle_set: - picked.append(c) - # tag overlap with slot values + for chunk in chunks: + if chunk.get("product") in bundle_set: + picked.append(chunk) blob = " ".join(slots.values()) - for c in chunks: - if c in picked: + for chunk in chunks: + if chunk in picked: continue - tags = c.get("tags") or [] - if any(str(t) in blob for t in tags): - picked.append(c) + 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]) + return RetrievalResult(picked[:4], "mock") def _retrieve_ragflow(slots: dict[str, str], bundle: list[str]) -> RetrievalResult: - """Minimal Ragflow retrieval hook. Falls back to mock if not configured/unavailable.""" - if not settings.ragflow_api_key: - return _retrieve_mock(slots, bundle) - + client = RagflowClient() query = " ".join( [ slots.get("tech_category", ""), slots.get("scenario", ""), slots.get("key_metric", ""), + slots.get("constraints", ""), " ".join(bundle), + "产品组合 适用场景 注意事项", ] - ).strip() or "产品方案" - dataset_ids = [x for x in settings.ragflow_dataset_ids.split(",") if x.strip()] - url = settings.ragflow_base_url.rstrip("/") + "/api/v1/retrieval" - headers = {"Authorization": f"Bearer {settings.ragflow_api_key}"} - payload: dict[str, Any] = {"question": query, "top_k": 4} - if dataset_ids: - payload["dataset_ids"] = dataset_ids - - try: - with httpx.Client(timeout=20.0) as client: - resp = client.post(url, headers=headers, json=payload) - resp.raise_for_status() - data = resp.json() - except Exception: - return _retrieve_mock(slots, bundle) - - # Ragflow response shapes vary by version; normalize best-effort - records = [] - if isinstance(data, dict): - records = data.get("data") or data.get("chunks") or data.get("records") or [] - if isinstance(records, dict): - records = records.get("chunks") or records.get("records") or [] + ).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 i, r in enumerate(records[:4]): - if not isinstance(r, dict): + for index, item in enumerate(raw[:4]): + if not isinstance(item, dict): continue normalized.append( { - "id": r.get("id") or f"rf-{i}", - "title": r.get("document_keyword") or r.get("title") or "Ragflow片段", - "product": r.get("document_id") or "", - "content": r.get("content") or r.get("text") or "", + "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"), } ) - if not normalized: - return _retrieve_mock(slots, bundle) - return RetrievalResult(normalized) + warning = "" if normalized else "知识库没有召回到可用片段，方案仅供草稿，建议转人工。" + return RetrievalResult(normalized, "ragflow", warning=warning) diff --git a/engine/app/scheme_engine.py b/engine/app/scheme_engine.py index edf6666..1880842 100644 --- a/engine/app/scheme_engine.py +++ b/engine/app/scheme_engine.py @@ -59,8 +59,15 @@ def handle_message(session_id: str, message: str) -> dict[str, Any]: "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")} + { + "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), @@ -75,18 +82,24 @@ def handle_message(session_id: str, message: str) -> dict[str, Any]: 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 c in scheme["citations"][:3]: - lines.append(f"- {c['title']}：{c['content']}") - if need_handoff: - lines.append("") - lines.append("系统建议转人工确认后再对外发送方案。回复「转人工」可进入人工流程。") + lines.append("知识依据：") + for citation in scheme["citations"][:3]: + lines.append(f"- {citation['title']}：{citation['content']}") else: lines.append("") - 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) diff --git a/engine/app/settings.py b/engine/app/settings.py index fe97c68..2a7c91d 100644 --- a/engine/app/settings.py +++ b/engine/app/settings.py @@ -8,15 +8,17 @@ KNOWLEDGE_DIR = ROOT / "knowledge" class Settings(BaseSettings): - model_config = SettingsConfigDict(env_file=".env", extra="ignore") + model_config = SettingsConfigDict(env_file=str(ROOT / ".env"), extra="ignore") - app_name: str = "鲜境方案引擎原型" + app_name: str = "鲜境智能客服" knowledge_dir: Path = KNOWLEDGE_DIR - retrieval_backend: str = "mock" # mock | ragflow + retrieval_backend: str = "ragflow" ragflow_base_url: str = "http://127.0.0.1:9380" ragflow_api_key: str = "" ragflow_dataset_ids: str = "" - llm_mode: str = "template" # template（无 Key 可跑）| openai_compatible（预留） + ragflow_web_url: str = "http://127.0.0.1:18080" + ragflow_email: str = "admin@xianjing.local" + ragflow_password: str = "XianjingDemo2026" settings = Settings() diff --git a/engine/app/static/app.js b/engine/app/static/app.js index 0442152..a7d1b87 100644 --- a/engine/app/static/app.js +++ b/engine/app/static/app.js @@ -6,7 +6,7 @@ const schemeEl = document.getElementById("scheme"); const metaEl = document.getElementById("meta"); const form = document.getElementById("form"); const input = document.getElementById("input"); -const resetBtn = document.getElementById("reset"); +const statusEl = document.getElementById("status"); function addBubble(role, text) { const div = document.createElement("div"); @@ -21,7 +21,7 @@ async function ensureSession() { const res = await fetch("/api/sessions", { method: "POST" }); const data = await res.json(); sessionId = data.session_id; - metaEl.textContent = `session: ${sessionId.slice(0, 8)}…`; + metaEl.textContent = `会话 ${sessionId.slice(0, 8)}`; return sessionId; } @@ -34,20 +34,18 @@ async function sendMessage(message) { body: JSON.stringify({ session_id: sessionId, message }), }); if (!res.ok) { - addBubble("sys", `错误：${res.status}`); + 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 = `session: ${sessionId.slice(0, 8)}… · phase: ${data.phase}`; + schemeEl.textContent = data.scheme ? JSON.stringify(data.scheme, null, 2) : "还没到出方案这一步"; + metaEl.textContent = `会话 ${sessionId.slice(0, 8)} · ${data.phase}`; } -form.addEventListener("submit", async (e) => { - e.preventDefault(); +form.addEventListener("submit", async (event) => { + event.preventDefault(); const text = input.value.trim(); if (!text) return; input.value = ""; @@ -60,14 +58,87 @@ form.addEventListener("submit", async (e) => { } }); -resetBtn.addEventListener("click", () => { +document.getElementById("reset").addEventListener("click", () => { sessionId = null; logEl.innerHTML = ""; slotsEl.textContent = "{}"; - schemeEl.textContent = "尚未生成"; + schemeEl.textContent = "还没到出方案这一步"; metaEl.textContent = ""; - addBubble("bot", "新会话已开始。可以直接说场景，例如：「产线检测，偏向 B 类」。"); + addBubble("bot", "新会话。可以直接说：「产线检测，偏向 B 类」。"); input.focus(); }); -addBubble("bot", "你好，我是鲜境方案引擎原型。先收集几个关键信息，再给出产品组合草稿。\n试试：「我们在户外做监测，指标还不确定」。"); +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(); diff --git a/engine/app/static/index.html b/engine/app/static/index.html index e9d2d4c..e464350 100644 --- a/engine/app/static/index.html +++ b/engine/app/static/index.html @@ -3,40 +3,76 @@ - 鲜境方案引擎原型 + 鲜境智能客服 - + -
-
-

鲜境物联 · 方案引擎原型

-

先跑通「追问 → 判定 → 出方案」

-

槽位与判定树在 knowledge/*.yaml，需求变了改配置即可。当前检索为 Mock，可后再接 Ragflow。

+
+
+
+

鲜境物联

+

方案型智能客服

+
+
正在检查知识库…
-
-
-
-
- - -
-
- - -
-
+ - -
+
+
+
+
+
+ + +
+
+ + +
+
+ +
+
+ +
+
+

Ragflow 知识库

+

检索走自托管 Ragflow。控制台可改文档、看召回；这里只看接入状态。示例资料待甲方替换。

+

打开 Ragflow 控制台

+ 
加载中
+

本地语料（灌库来源）

+
    +
    +
    + +
    +
    +

    已留资

    +

    演示库存在本机 SQLite，不是正式 CRM。

    +
    暂无
    +
    +
    diff --git a/engine/app/static/styles.css b/engine/app/static/styles.css index 0e986ca..c02ca7f 100644 --- a/engine/app/static/styles.css +++ b/engine/app/static/styles.css @@ -1,144 +1,75 @@ :root { - --bg0: #0f1c17; - --bg1: #163528; - --panel: rgba(248, 250, 246, 0.94); + --bg0: #10211a; + --bg1: #1b3b2c; + --panel: #f6f8f4; --ink: #14231c; - --muted: #5c7268; - --accent: #1f7a4c; - --accent-2: #c4f06a; + --muted: #5d7368; + --accent: #1d7a4d; --line: rgba(20, 35, 28, 0.12); - --bot: #e7f2ea; - --user: #1f7a4c; + --bot: #e7f3eb; } * { box-sizing: border-box; } body { margin: 0; min-height: 100vh; - font-family: "Noto Sans SC", "DM Sans", sans-serif; color: var(--ink); + font-family: "Noto Sans SC", "DM Sans", sans-serif; background: - radial-gradient(1200px 600px at 10% -10%, #2f6b4f 0%, transparent 55%), - radial-gradient(900px 500px at 100% 0%, #8fd36a 0%, transparent 45%), - linear-gradient(160deg, var(--bg0), var(--bg1)); + 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; } -.page { max-width: 1100px; margin: 0 auto; padding: 32px 20px 48px; } -.hero { color: #f4faf6; margin-bottom: 22px; } -.brand { - margin: 0 0 8px; - font-family: "DM Sans", sans-serif; - letter-spacing: 0.04em; - font-weight: 700; - font-size: 14px; - color: var(--accent-2); -} -.hero h1 { - margin: 0 0 10px; - font-size: clamp(1.6rem, 3vw, 2.2rem); - line-height: 1.25; - font-weight: 700; -} -.sub { margin: 0; max-width: 52rem; color: rgba(244, 250, 246, 0.82); } -.sub code { - font-family: ui-monospace, SFMono-Regular, Menlo, monospace; - font-size: 0.9em; - background: rgba(0,0,0,0.25); - padding: 0.1em 0.35em; - border-radius: 4px; +.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); } -.layout { - display: grid; - grid-template-columns: 1.4fr 0.9fr; - gap: 16px; -} -@media (max-width: 860px) { - .layout { grid-template-columns: 1fr; } -} +.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; - border: 1px solid rgba(255,255,255,0.35); - box-shadow: 0 18px 50px rgba(0,0,0,0.22); -} - -.chat { display: flex; flex-direction: column; min-height: 560px; overflow: hidden; } -.log { - flex: 1; - padding: 18px; - overflow-y: auto; - 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; + 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(--user); color: #fff; } -.bubble.sys { align-self: center; background: transparent; color: var(--muted); font-size: 12px; } - -.composer { - display: flex; - gap: 8px; - padding: 12px 14px; - border-top: 1px solid var(--line); +.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 input { - flex: 1; - border: 1px solid var(--line); - border-radius: 10px; - padding: 12px 14px; - font: inherit; +.composer button, .stack button, .ghost { + border: 0; border-radius: 10px; padding: 11px 14px; font: inherit; font-weight: 600; cursor: pointer; } -.composer button, .ghost { - border: 0; - border-radius: 10px; - padding: 12px 16px; - font: inherit; - font-weight: 600; - cursor: pointer; - background: var(--accent); - color: #fff; +.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; } -.ghost { - background: transparent; - color: var(--accent); - border: 1px solid var(--line); - padding: 8px 12px; -} -.toolbar { - display: flex; - align-items: center; - justify-content: space-between; - gap: 10px; - padding: 0 14px 14px; -} -.meta { color: var(--muted); font-size: 12px; } - -.side { padding: 16px 18px 20px; } -.side h2 { - margin: 12px 0 8px; - font-size: 13px; - letter-spacing: 0.04em; - text-transform: uppercase; - color: var(--muted); -} -.side h2:first-child { margin-top: 0; } -.side pre { - margin: 0; - padding: 12px; - background: #eef5f0; - border-radius: 10px; - font-size: 12px; - line-height: 1.45; - overflow: auto; - max-height: 220px; +.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); } diff --git a/engine/requirements.txt b/engine/requirements.txt index c727224..e4239f1 100644 --- a/engine/requirements.txt +++ b/engine/requirements.txt @@ -5,3 +5,4 @@ pydantic-settings==2.6.1 pyyaml==6.0.2 httpx==0.28.1 python-multipart==0.0.20 +pycryptodome==3.21.0 diff --git a/infra/ragflow-public.pem b/infra/ragflow-public.pem new file mode 100644 index 0000000..3fbcfe1 --- /dev/null +++ b/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----- diff --git a/knowledge/corpus/faq.md b/knowledge/corpus/faq.md new file mode 100644 index 0000000..5ee840c --- /dev/null +++ b/knowledge/corpus/faq.md @@ -0,0 +1,17 @@ +# 鲜境示例资料 · 常见询盘 + +## 下班后没人回怎么办 + +线上咨询由本系统 7×24 接待。答不上、置信度低、或客户要求人工时，留下称呼和电话，销售在工作时间跟进。 + +## 你们是卖设备还是卖方案 + +鲜境是卖方案。先判断走传感监测还是控制联动，再给产品组合，而不是一问一答式报价单。 + +## 能不能先给个大概价格 + +演示知识库没有价格。价格、折扣、交期一律转人工，不得估算。 + +## 资料里没有的参数 + +回复「资料未覆盖，需要工程师确认」，不要用常识补全型号参数。 diff --git a/knowledge/corpus/policies.md b/knowledge/corpus/policies.md new file mode 100644 index 0000000..e6dcc66 --- /dev/null +++ b/knowledge/corpus/policies.md @@ -0,0 +1,13 @@ +# 鲜境示例资料 · 对外边界 + +## 敏感与承诺 + +- 不报未入库的精度、认证、交期、折扣、质保年限。 +- 客户留下的电话只用于本次跟进，回复中不得复述其他客户信息。 +- 价格一律转人工。 + +## 转人工触发 + +- 客户明确说转人工、找销售、要真人。 +- 判定置信度低于 0.55。 +- 知识库没有命中依据却被追问具体参数。 diff --git a/knowledge/corpus/products-a.md b/knowledge/corpus/products-a.md new file mode 100644 index 0000000..0e6f3ce --- /dev/null +++ b/knowledge/corpus/products-a.md @@ -0,0 +1,28 @@ +# 鲜境示例资料 · A 类传感监测 + +> 本文件是演示知识，型号与参数为占位。正式上线前替换为甲方产品手册。未写入本库的参数，客服不得对外承诺。 + +## 技术路线 + +A 类面向监测与数据采集：现场传感器采集、网关汇聚、远程查看。典型客户说的是「要看数据、要报警、要远程」，而不是「要联动产线设备」。 + +## 推荐组合 + +- 主机：XJ-A100 +- 网关：XJ-A-GW + +XJ-A100 适用于户外与环境监测，宽温工作，需配合 XJ-A-GW 做多节点汇聚与协议转换。不要单独报主机而不报网关。 + +## 适用场景 + +- 户外监测 +- 环境采集 +- 实验室标定（轻量） + +## 必问再确认 + +防护等级、供电方式、是否已有采集平台。这三项资料未齐时，只能给草稿，建议转销售确认。 + +## 不适合 + +产线设备联动、大量 IO 点位、本地逻辑控制。这些走 B 类控制路线。 diff --git a/knowledge/corpus/products-b.md b/knowledge/corpus/products-b.md new file mode 100644 index 0000000..b92dc5c --- /dev/null +++ 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 类传感路线。 diff --git a/knowledge/corpus/solutions.md b/knowledge/corpus/solutions.md new file mode 100644 index 0000000..248ce0e --- /dev/null +++ b/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. 以上都不像：不要硬推型号，转人工 + +## 对外话术边界 + +方案正文只能复述知识库里出现过的事实。客户追问具体认证、交期、折扣时，一律转销售，不编造。 diff --git a/scripts/bootstrap_ragflow.py b/scripts/bootstrap_ragflow.py new file mode 100644 index 0000000..114cf11 --- /dev/null +++ b/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() diff --git a/scripts/fix_docker_forward.sh b/scripts/fix_docker_forward.sh new file mode 100644 index 0000000..a1f408e --- /dev/null +++ b/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}"