提交 b94d0230 authored 作者: 陈泽健's avatar 陈泽健

feat(offline): 问题排查助手离线 Q&A 模式和容器化部署

离线模式:OFFLINE_MODE 环境变量控制,跳过 Claude API 直接返回 TF-IDF 匹配结果
容器化:Dockerfile + docker-compose.yml + requirements.txt + .dockerignore
同进程依赖:补 cryptography/paramiko(service_monitor 子包依赖,PRD §4.6)

详细变更:
- utils/offline_config.py:离线开关读取(env OFFLINE_MODE → bool,进程级缓存)
- services/ai_service.py:新增 build_offline_response(),兼容原始/格式化两种输入
- routes/troubleshoot.py:troubleshoot/analyze/analyze_stream/health 四处离线分支
  - 离线流式保留 SSE 协议,单帧推送后 close(前端无需改动)
  - 离线跳过缓存读写(<1s 无需缓存)
  - health 响应新增 offline_mode 字段
- tests/test_offline_mode.py:22 用例全绿(离线配置/响应构建/路由/回归)
- Dockerfile:python:3.10-slim,gunicorn 多 worker,WORKDIR /app/web
- docker-compose.yml:restart always,records/users/config 只读挂载,logs 持久化
- .dockerignore:排除 搜索向量.json(强制 TF-IDF),排除测试/缓存/临时文件
- requirements.txt:flask/gunicorn/scikit-learn/scipy/numpy/jieba/werkzeug/requests
  + cryptography/paramiko(同进程 service_monitor 硬依赖)
- .env.example:补 OFFLINE_MODE 说明
- PRD v2.2 + 计划执行文档(增加 §4.6 同进程依赖耦合说明 + T8 任务)

测试:新增 22 用例全绿,回归 218 全绿(问题排查助手 167 + service_monitor 51)
隔离:未改 service_monitor / service_manage 代码,仅补其依赖到 requirements
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 c6287b3b
# 向量文件:离线镜像不打包,强制容器内走 TF-IDF(PRD §4.4)
# 若打包进容器,SearchEngine 会静默走向量模式,与离线用 TF-IDF 的设计矛盾
搜索向量.json
**/搜索向量.json
deploy/搜索向量.json
# 版本控制与工具
.git
.gitignore
.claude
# 文档与临时文件
Docs
临时目录
HANDOFF.md
# Python 缓存
**/__pycache__
**/*.pyc
**/*.pyo
# 测试与覆盖率
skill/code/tests
skill/code/.pytest_cache
.coverage
htmlcov
# 本地数据与日志(挂载方式注入,不入镜像)
logs
cache
...@@ -20,10 +20,26 @@ SECRET_KEY=your_secret_key_here ...@@ -20,10 +20,26 @@ SECRET_KEY=your_secret_key_here
CLAUDE_API_BASE=https://office.ubainsyun.com:8400 CLAUDE_API_BASE=https://office.ubainsyun.com:8400
CLAUDE_API_KEY=your_api_key_here CLAUDE_API_KEY=your_api_key_here
# ============================================
# 离线模式开关(问题排查助手)
# true = 离线模式,跳过 Claude API,直接返回 TF-IDF 匹配结果
# false = 在线模式(默认),调用 Claude API 做 AI 分析
# 现场内网/离线部署时设为 true
# ============================================
OFFLINE_MODE=true
# ============================================ # ============================================
# 数据库配置(如需要) # 数据库配置(如需要)
# ============================================ # ============================================
# DB_HOST=localhost # DB_HOST=localhost
# DB_PORT=3306 # DB_PORT=3306
# DB_USER=root # DB_USER=root
# DB_PASSWORD=your_db_password # DB_PASSWORD=your_db_password
\ No newline at end of file
# ============================================
# 服务监测模块配置
# ============================================
# SSH 凭据加密密钥(Fernet)。
# 不设则派生自 SECRET_KEY;生产建议单独设置一个 44 字节 base64 密钥
# 生成方法:python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
MONITOR_ENC_KEY=
\ No newline at end of file
FROM python:3.10-slim
# 工作目录设为 /app/web,与 utils/paths.py 的 SCRIPT_DIR 推导一致
# gunicorn 直接加载 server 模块(server.py:107 已有模块级 app = create_app())
WORKDIR /app/web
# 安装依赖(版本锁定,离线环境复现性保障)
COPY requirements.txt /app/requirements.txt
RUN pip install --no-cache-dir -r /app/requirements.txt
# 拷贝代码
COPY skill/code/web/ /app/web/
COPY skill/code/SKILL.md /app/SKILL.md
# 端口
EXPOSE 8088
# 环境变量
ENV PYTHONIOENCODING=utf-8
# 启动:gunicorn 多 worker
CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:8088", "server:app"]
# PRD 需求文档 — 离线部署方案 # PRD 需求文档 — 问题排查助手离线部署方案
> 版本:2.0 | 日期:2026-07-15 | 分支:troubleshoot-ai-assistant > **适用范围**:本文档仅针对「问题排查助手」模块的离线部署。服务管理、服务监测模块的容器化方案另立文档。
> 版本:2.2 | 日期:2026-07-16 | 分支:troubleshoot-ai-assistant
--- ---
...@@ -8,7 +10,9 @@ ...@@ -8,7 +10,9 @@
运行维护平台部署到客户现场服务器时,通常处于**内网/离线环境**,无法访问外部 AI API(Claude)。当前问题排查助手依赖 Claude API 进行 AI 分析,在离线环境下不可用。 运行维护平台部署到客户现场服务器时,通常处于**内网/离线环境**,无法访问外部 AI API(Claude)。当前问题排查助手依赖 Claude API 进行 AI 分析,在离线环境下不可用。
同时,现场部署采用**容器化(Docker)**方式,不直接部署在宿主机上。三个模块(问题排查助手、服务管理、服务监测)需要在容器化离线环境中稳定运行。 同时,现场部署采用**容器化(Docker)**方式,不直接部署在宿主机上。
> **范围声明**:本方案**仅覆盖问题排查助手模块**的离线化与容器化。服务管理、服务监测两个模块的容器化部署方案见各自独立文档。三者共用同一 Docker 镜像底座还是分镜像,由后续总部署方案统筹决策。
--- ---
...@@ -16,13 +20,20 @@ ...@@ -16,13 +20,20 @@
| 模块 | 外部依赖 | 离线可用性 | 备注 | | 模块 | 外部依赖 | 离线可用性 | 备注 |
|------|----------|-----------|------| |------|----------|-----------|------|
| 问题排查助手 | Claude API + TF-IDF 搜索 | ❌ AI 不可用 | 搜索可用,AI 分析不可用 | | 问题排查助手 | Claude API + TF-IDF 搜索 | ❌ AI 不可用 | 搜索可用(TF-IDF,纯本地运行),AI 分析不可用 |
| 服务管理 | 无(占位) | ✅ | 后续走 SSH 连接内部系统 | | 服务管理 | 无(占位) | ✅ | 后续走 SSH 连接内部系统 |
| 服务监测 | SSH(局域网) | ✅ | SSH 连接目标服务器 | | 服务监测 | SSH(局域网) | ✅ | SSH 连接目标服务器 |
| 平台首页 / 认证 | 无(Flask 本地) | ✅ | — | | 平台首页 / 认证 | 无(Flask 本地) | ✅ | — |
> **上表包含服务管理/服务监测仅用于对比**,说明为何只有问题排查助手需要离线改造。二者的隔离设计见 §3.7。
**核心卡点**:问题排查助手的 AI 分析功能依赖外网 API。 **核心卡点**:问题排查助手的 AI 分析功能依赖外网 API。
**搜索选型说明**:本方案搜索引擎采用 **TF-IDF**(词频-逆文档频率),而非 Embedding 向量搜索。原因有二:
1. 向量搜索需在线调用 `/v1/embeddings` API 预计算向量文件(`搜索向量.json`),会产生额外 API 费用;
2. 知识库更新后需重新跑预计算脚本,离线环境无法完成。
TF-IDF 为纯本地计算,零外部依赖、零额外费用,满足离线部署需求。
--- ---
## 3. 离线方案设计 ## 3. 离线方案设计
...@@ -31,6 +42,8 @@ ...@@ -31,6 +42,8 @@
**原理**:跳过 Claude API 调用,直接将 TF-IDF 搜索匹配结果作为排查建议返回。 **原理**:跳过 Claude API 调用,直接将 TF-IDF 搜索匹配结果作为排查建议返回。
> **为什么是 TF-IDF 而不是向量搜索?** 项目已完成 P2-1 向量搜索升级,但向量搜索需要在联网环境下调用 `/v1/embeddings` API 预计算 `搜索向量.json`,会产生额外费用。离线模式下 TF-IDF 纯本地运行、零外部依赖,是最安全的选择。
**流程对比** **流程对比**
``` ```
...@@ -68,18 +81,19 @@ ...@@ -68,18 +81,19 @@
参考文件:record_002.md 参考文件:record_002.md
``` ```
### 3.2 配置开关 ### 3.2 配置开关 — 环境变量
**文件**`config.json` `offline_mode` 通过**环境变量**控制,与 `SECRET_KEY` 保持一致(不从 `config.json` 读取),避免配置随代码暴露。
```json ```bash
{ # .env 文件或 docker-compose environment 块
"offline_mode": false OFFLINE_MODE=true # true = 离线模式(跳过 API),false = 在线模式(调 Claude API)
}
``` ```
- `offline_mode: false`(默认)— 在线模式,调 Claude API - `OFFLINE_MODE=false`(默认未设置)— 在线模式,调 Claude API
- `offline_mode: true` — 离线模式,跳过 API,直接返回搜索结果 - `OFFLINE_MODE=true` — 离线模式,跳过 API,直接返回搜索结果
**生效方式**:进程启动时读取一次,修改后需重启容器。配置热更新本期不做。
### 3.3 部署方式 — 容器化 ### 3.3 部署方式 — 容器化
...@@ -122,29 +136,79 @@ ...@@ -122,29 +136,79 @@
**`POST /api/analyze_stream`** **`POST /api/analyze_stream`**
离线模式下不调用流式 API,直接返回完整 JSON(非 SSE),响应格式: **保留原有 SSE 逻辑**,离线模式不调用流式 API,但响应仍走 SSE 协议(前端逻辑无需修改)。区别在于:在线模式多帧推送 AI 生成文本,离线模式**单帧推送**完整匹配结果后即 `close`
```json ```text
{ data: {"success": true, "offline": true, "analysis": null, "matched_cases": [...]}
"success": true,
"offline": true, [event: close]
"analysis": null,
"matched_cases": [...]
}
``` ```
**前端无需改动**:前端按现有 SSE 解析逻辑读取 `data` 帧,渲染 `matched_cases``analysis: null` 时前端显示"当前为离线模式,无 AI 分析"提示。
### 3.5 /api/health 响应 ### 3.5 /api/health 响应
新增 `offline_mode` 字段: 新增 `offline_mode` 字段(来源:环境变量 `OFFLINE_MODE`),与现有 `search.mode` 字段并存
```json ```json
{ {
"status": "ok", "status": "ok",
"version": "1.4.0", "version": "1.4.0",
"search": {"mode": "tfidf"},
"offline_mode": true "offline_mode": true
} }
``` ```
**字段关系说明**
- `search.mode`:搜索引擎当前模式(离线镜像下应为 `tfidf`,见 §4.4 向量文件处理)
- `offline_mode`:是否跳过 Claude API 走离线 Q&A 模式
两个开关相互独立。合法组合:
| offline_mode | search.mode | 场景 |
|--------------|-------------|------|
| true | tfidf | 纯离线现场(本方案默认) |
| false | vector | 在线环境 + 向量搜索 |
| false | tfidf | 在线环境 + TF-IDF(省资源) |
> `offline_mode: true` + `search.mode: vector` 理论上可行(向量查询不需联网),但本方案离线镜像不打包向量文件,故该组合不会出现。
### 3.6 代码实现分层
遵循项目 P1-3 三层架构约定(routes/services/utils 分层):
| 层 | 文件 | 职责 |
|----|------|------|
| utils | `utils/config.py`(或 container.py 集中) | 读取环境变量 `OFFLINE_MODE`,返回 bool |
| services | `services/ai_service.py` | 新增 `build_offline_response(cases)` 函数:格式化 TF-IDF 匹配结果为离线响应(标题/横幅/案例详情) |
| routes | `routes/troubleshoot.py` | `/api/troubleshoot``/api/analyze``/api/analyze_stream` 三个路由入口判断 `OFFLINE_MODE`,为 true 时调用 `build_offline_response` 并跳过 Claude API |
**约束**
- 离线格式化逻辑统一走 `ai_service.build_offline_response()`,两个路由不各自重写
- 不修改 `search_engine.py` / `safety_filter.py` / `cache_manager.py` 公开接口
- `cache_manager.py` 在离线模式下照常运行(空跑缓存,不动接口)
- 异常处理遵循 P1-1 规范:禁止空 `except:`,必须记录日志
### 3.7 模块隔离边界(部署层隔离)
三个模块(问题排查助手、服务管理、服务监测)共用同一个 Flask 进程,通过 Blueprint 平行注册(`server.py` 第 86-101 行)。**本方案采用部署层隔离**:代码保持现有物理分层不变,离线改造严格限定在问题排查助手的边界内,不触碰另外两个模块。
**现状(已天然隔离)**
| 归属 | 文件 | 说明 |
|------|------|------|
| 问题排查助手 | `routes/troubleshoot.py``services/ai_service.py``services/record_service.py` | 独立 Blueprint,无跨模块 import |
| 服务管理 | `routes/service_manage.py` | 平行 Blueprint |
| 服务监测 | `routes/service_monitor.py` | 平行 Blueprint |
| 公共 | `routes/auth.py``routes/platform.py``utils/*` | 三模块共享 |
**隔离红线(离线改造必须遵守)**
1. **改动范围仅限问题排查助手文件**`routes/troubleshoot.py``services/ai_service.py`,以及 `container.py`/`utils` 中新增的 `OFFLINE_MODE` 读取。**禁止修改** `service_manage.py``service_monitor.py`
2. **`OFFLINE_MODE` 只作用于问题排查助手路由**:该开关仅在 `troubleshoot.py` 的三个路由内判断,不影响服务管理/服务监测的任何行为。服务监测的 SSH 连接、服务管理的后续功能不受离线开关约束。
3. **共享工具层保持中立**`utils/*``auth.py``platform.py` 的改动若发生,必须对三模块无差别兼容,不得引入问题排查助手专属逻辑。
4. **container.py 单例边界**`container.py` 现仅持有问题排查助手的单例(search_engine/safety_filter/cache_manager)。新增 `OFFLINE_MODE` 读取属于问题排查助手配置,不得在此混入另两个模块的状态。
5. **验证要求**:离线改造后需回归测试确认服务管理/服务监测的现有路由行为无变化(响应结构、状态码不变)。
> 本方案为**部署层隔离(理解 A)**,非源码拆包。若后续需要三模块各自独立镜像/独立进程,属另一次重构任务,不在本 PRD 范围。
--- ---
## 4. 容器化部署方案 ## 4. 容器化部署方案
...@@ -156,8 +220,9 @@ FROM python:3.10-slim ...@@ -156,8 +220,9 @@ FROM python:3.10-slim
WORKDIR /app WORKDIR /app
# 安装依赖 # 安装依赖(含在线模式所需的 requests,镜像兼容在线/离线切换)
RUN pip install flask scikit-learn scipy numpy jieba gunicorn werkzeug COPY requirements.txt /app/requirements.txt
RUN pip install --no-cache-dir -r /app/requirements.txt
# 拷贝代码 # 拷贝代码
COPY skill/code/web/ /app/web/ COPY skill/code/web/ /app/web/
...@@ -166,10 +231,34 @@ COPY skill/code/SKILL.md /app/SKILL.md ...@@ -166,10 +231,34 @@ COPY skill/code/SKILL.md /app/SKILL.md
# 端口 # 端口
EXPOSE 8088 EXPOSE 8088
# 启动 # 启动(gunicorn 需 server.py 暴露模块级 app 变量)
ENV PYTHONIOENCODING=utf-8
CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:8088", "server:app"] CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:8088", "server:app"]
``` ```
**`requirements.txt`(版本锁定,离线环境复现性保障)**
```text
flask==3.0.3
gunicorn==22.0.0
scikit-learn==1.4.2
scipy==1.13.0
numpy==1.26.4
jieba==0.42.1
werkzeug==3.0.3
requests==2.32.3
cryptography==46.0.7
paramiko==4.0.0
```
> - `requests` 保留——离线镜像需兼容在线模式切换(容器改 `OFFLINE_MODE=false` 即可调 Claude API)。
> - `cryptography` / `paramiko` 为**同进程邻居 service_monitor 子包的依赖**(见 §4.6),非问题排查助手自身所需,但同一镜像同一 Flask 进程必须具备,否则启动即崩。
> - 版本以本地实测通过为准(service_monitor 51 + 问题排查助手 167 用例全绿);在线构建环境若解析失败再调整。
### 4.1.1 gunicorn 启动路径校验
项目 `server.py` 第 106-107 行已有模块级 `app = create_app()`,gunicorn `server:app` 可正常加载。无需额外修改。
### 4.2 docker-compose.yml(推荐) ### 4.2 docker-compose.yml(推荐)
```yaml ```yaml
...@@ -188,6 +277,7 @@ services: ...@@ -188,6 +277,7 @@ services:
environment: environment:
- PYTHONIOENCODING=utf-8 - PYTHONIOENCODING=utf-8
- SECRET_KEY=${SECRET_KEY} - SECRET_KEY=${SECRET_KEY}
- OFFLINE_MODE=${OFFLINE_MODE:-true}
``` ```
### 4.3 现场部署流程 ### 4.3 现场部署流程
...@@ -207,27 +297,70 @@ docker load < troubleshoot.tar.gz ...@@ -207,27 +297,70 @@ docker load < troubleshoot.tar.gz
docker compose up -d docker compose up -d
``` ```
### 4.4 向量文件处理(关键)
离线镜像**不打包 `搜索向量.json`**。原因:SearchEngine 初始化时若检测到向量文件存在,会**静默走 vector 模式**(见 CLAUDE.md 避坑第 7 条),与"离线用 TF-IDF"的设计矛盾。
**处理规则**
- 构建镜像时确保 `搜索向量.json` 不被 `COPY` 进容器(`.dockerignore` 排除或 records 目录不含该文件)
- 部署后验证 `/api/health` 返回 `search.mode: tfidf`,确认未误走向量模式
- `搜索索引.json`(TF-IDF 索引)**必须打包或挂载**,SearchEngine 初始化依赖它
### 4.5 路径一致性校验
`utils/paths.py``Path(__file__).parent` 推导路径。容器内代码位于 `/app/web/`,故:
- `SCRIPT_DIR` = `/app/web/`
- `RECORDS_DIR` = `/app/web/records/` ← 与 docker-compose 挂载点 `/app/web/records` 一致 ✅
部署后需验证知识库记录数(357 条)正确加载,确认路径未偏移。
### 4.6 同进程模块依赖耦合(运行时事实)
本方案采用 §3.7 的**部署层隔离**——问题排查助手与 service_monitor 在代码层无相互 import、数据分离。但**运行时同处一个 Flask 进程**`server.py``create_app()` 一次性注册全部 Blueprint)。这意味着:
> **邻居模块的导入失败,会连累整个进程启动失败,问题排查助手也随之不可用。**
**影响来源**(service_monitor 子包):
| 依赖 | 导入方式 | 失败时机 | 影响 |
|------|----------|----------|------|
| `cryptography` | `utils/crypto.py` 模块级 `from cryptography.fernet import Fernet` | `create_app()` 阶段 | **启动即崩**,问题排查助手无法启动 |
| `paramiko` | `executor.py` 函数内延迟 `import paramiko`(第 249 行) | 调用 SSH 监测时 | 启动不崩,点 SSH 功能才崩 |
**处理决策(本期采用)**
- **补依赖**:将 `cryptography` / `paramiko` 纳入 `requirements.txt`(见 §4.1)。承认运行时同进程这一既定事实,而非制造耦合。
- **不拆镜像 / 不拆进程**:拆分属另一次架构重构,超出本 PRD 范围;且会破坏 service_monitor 已做好的独立子包设计。
- **不延迟加载 cryptography**:改 service_monitor 源码会违背 §3.7 隔离红线(不改 service_monitor)。
**可选增强(本期不做)**:若需让问题排查助手不受 service_monitor 导入失败连累,可在 `server.py` 注册 Blueprint 时包一层容错 try/except,使单模块加载失败时其余模块仍能起。该项涉及 `server.py`(service_monitor 重构进行中),待该重构稳定后另行评估,避免文件冲突。
> **本节是设计妥协的如实记录**:部署层隔离 ≠ 进程隔离。同进程邻居的硬依赖必须随镜像一起提供。
--- ---
## 5. 非功能需求 ## 5. 非功能需求
| 项目 | 要求 | | 项目 | 要求 |
|------|------| |------|------|
| 配置切换 | 改 config.json 的 `offline_mode` 即可,无需改代码 | | 配置切换 | 改环境变量 `OFFLINE_MODE` 即可,无需改代码(修改后重启容器生效) |
| 搜索质量 | 离线模式下 TF-IDF 搜索逻辑不变,纯本地运行 | | 搜索质量 | 离线模式下 TF-IDF 搜索逻辑不变,纯本地运行 |
| 响应速度 | 离线模式排查 <1 秒(无 API 调用) | | 响应速度 | 离线模式排查 <1 秒(无 API 调用) |
| 前端适配 | 离线模式结果展示清晰标注"离线模式" | | 前端适配 | 离线模式结果展示清晰标注"离线模式";`analysis: null` 时显示"当前为离线模式,无 AI 分析" |
| 容器化 | 提供 Dockerfile + docker-compose.yml | | 流式兼容 | 离线模式保留 SSE 协议,前端解析逻辑无需改动 |
| 容器化 | 提供 Dockerfile + docker-compose.yml + requirements.txt(版本锁定) |
| 进程守护 | Docker restart policy always + gunicorn 多 worker | | 进程守护 | Docker restart policy always + gunicorn 多 worker |
| 日志持久化 | 日志挂载到宿主机 volume | | 日志持久化 | 日志挂载到宿主机 volume |
| 镜像兼容性 | 离线镜像同时支持在线模式(保留 requests + Claude 依赖);同进程 service_monitor 依赖(cryptography/paramiko)一并打入镜像(见 §4.6) |
--- ---
## 6. 约束 ## 6. 约束
- 离线模式下不提供 AI 分析和步骤建议,仅返回知识库匹配结果 - 离线模式下不提供 AI 分析和步骤建议,仅返回知识库匹配结果
- 离线模式不影响现有在线模式行为(配置开关隔离) - 离线模式不影响现有在线模式行为(环境变量开关隔离)
- 不修改 `search_engine.py` / `safety_filter.py` / `cache_manager.py` 公开接口 - 不修改 `search_engine.py` / `safety_filter.py` / `cache_manager.py` 公开接口
- 离线镜像**不打包 `搜索向量.json`**,强制走 TF-IDF(见 §4.4)
- `OFFLINE_MODE` 启动时读取一次,配置热更新本期不做
- **模块隔离红线**:离线改造改动范围仅限问题排查助手(`routes/troubleshoot.py``services/ai_service.py` 及新增的 `OFFLINE_MODE` 读取),禁止修改 `service_manage.py` / `service_monitor.py`(见 §3.7)
- Docker 镜像构建在在线环境完成,现场只需 `docker load` + `docker run` - Docker 镜像构建在在线环境完成,现场只需 `docker load` + `docker run`
--- ---
...@@ -236,4 +369,4 @@ docker compose up -d ...@@ -236,4 +369,4 @@ docker compose up -d
- **本地小模型推理**:Docker 内嵌 Ollama + 小参数模型 - **本地小模型推理**:Docker 内嵌 Ollama + 小参数模型
- **知识库离线导入**:支持通过平台界面上传离线 Q&A 文档 - **知识库离线导入**:支持通过平台界面上传离线 Q&A 文档
- **配置热更新**:修改 config.json 后无需重启容器 - **配置热更新**:修改 `OFFLINE_MODE` 后无需重启容器
\ No newline at end of file \ No newline at end of file
# 计划执行文档 — 问题排查助手离线部署方案
> 配套 PRD:`PRD_需求文档_离线部署方案_问题排查助手.md`(v2.1)
> 版本:1.0 | 日期:2026-07-16 | 分支:troubleshoot-ai-assistant
> 执行顺序:T1 → T2 → T3 → T4 → T5 → T6(验收)→ T7(部署)
---
## 0. 执行概述
### 目标
为问题排查助手实现离线 Q&A 模式:在 `OFFLINE_MODE=true` 时跳过 Claude API,直接返回 TF-IDF 匹配结果;并提供 Docker 容器化部署能力。
### 改动范围(严格遵守 §3.7 隔离红线)
| 类型 | 文件 | 说明 |
|------|------|------|
| 新增 | `skill/code/web/utils/offline_config.py` | 读取 `OFFLINE_MODE` 环境变量 |
| 修改 | `skill/code/web/services/ai_service.py` | 新增 `build_offline_response()` |
| 修改 | `skill/code/web/routes/troubleshoot.py` | 三路由 + health 增加离线分支 |
| 新增 | `Dockerfile` | 仓库根目录 |
| 新增 | `docker-compose.yml` | 仓库根目录 |
| 新增 | `requirements.txt` | 仓库根目录 |
| 新增 | `.dockerignore` | 排除向量文件等 |
| 新增 | `.env.example` 补充 `OFFLINE_MODE` | 文档同步 |
| 新增 | `skill/code/tests/test_offline_mode.py` | 离线模式单元测试 |
| **禁止修改** | `routes/service_manage.py``routes/service_monitor.py` | 隔离红线 |
### 关键决策(已锁定)
1. 搜索用 **TF-IDF**(不调向量 API,省费)
2. 流式接口 **保留 SSE**,离线单帧推送后 close
3. 镜像 **兼容在线模式**(保留 requests)
4. `OFFLINE_MODE`**环境变量**(与 SECRET_KEY 一致)
5. 离线镜像 **不打包 `搜索向量.json`**
---
## 1. 任务分解与实施计划
### T1:离线配置读取工具(utils 层)
**文件**`skill/code/web/utils/offline_config.py`(新增)
**职责**:集中读取 `OFFLINE_MODE` 环境变量,返回 bool。隔离在 utils 层,供 routes/services 调用,不污染 container 单例边界。
**实现要点**
```python
# -*- coding: utf-8 -*-
"""offline_config.py — 离线模式开关读取(问题排查助手专用)
从环境变量 OFFLINE_MODE 读取,启动时读取一次。
true/1/yes(不区分大小写)→ True,其余 → False(默认在线)。
"""
import os
_CACHE = None # 进程级缓存,启动时读一次
def is_offline_mode():
"""返回是否离线模式"""
global _CACHE
if _CACHE is None:
raw = os.environ.get('OFFLINE_MODE', '').strip().lower()
_CACHE = raw in ('true', '1', 'yes')
return _CACHE
def reset_for_test():
"""测试用:重置缓存"""
global _CACHE
_CACHE = None
```
**验收**
- `OFFLINE_MODE=true``True`
- `OFFLINE_MODE=False` / 未设置 → `False`
- `reset_for_test()` 后重新读取
**约束**:仅此模块读 `OFFLINE_MODE`,其他文件统一调用 `is_offline_mode()`
---
### T2:离线响应构建(services 层)
**文件**`skill/code/web/services/ai_service.py`(修改,新增函数)
**职责**:新增 `build_offline_response(matched_cases_data)`,将 TF-IDF 匹配结果格式化为离线响应。
**实现要点**
```python
def build_offline_response(matched_cases_data):
"""构建离线模式响应(跳过 Claude API)
返回 dict,供 /api/troubleshoot 与 /api/analyze 复用:
{
"response": str, # 格式化文本(标题+横幅+案例列表)
"matched_cases": list, # 复用 troubleshoot.py 的 format_matched_cases
"offline": True,
"api_time": 0.0,
}
"""
# 横幅 + 案例详情拼接(参照 PRD §3.1 展示格式)
...
```
**案例格式参照 PRD §3.1**:标题「📋 离线排查参考(基于知识库匹配)」+ 横幅「当前为离线模式,结果基于历史知识库匹配」+ 每个案例(相似度/项目/标题/现象/解决方案/参考文件)。
**关键决策**`build_offline_response` 只负责生成 `response` 文本,`matched_cases` 列表仍由路由层的 `format_matched_cases()` 复用现有逻辑,避免重复格式化代码。
**验收**
- 输入 5 条匹配结果,输出含 5 个案例详情的文本
- 输入空列表,返回「未匹配到相关案例」兜底
- 不调用任何网络/API
**约束**:不改 `build_prompt` / `call_claude_api` 等现有函数签名。
---
### T3:路由层离线分支(routes 层)
**文件**`skill/code/web/routes/troubleshoot.py`(修改)
**职责**:三个路由 + health 增加 `OFFLINE_MODE` 分支。
#### T3.1 `/api/troubleshoot`(第 57 行 `troubleshoot()`)
在「1. 搜索」之后插入分支:
```python
# 1. 搜索匹配案例
engine = container.get_search_engine()
matched_cases_data = engine.search(...)
# 【新增】离线模式:跳过 Claude API
if is_offline_mode():
result = build_offline_response(matched_cases_data)
log_audit({... 'mode': 'offline' ...})
return jsonify({
'success': True,
'response': result['response'],
'matched_cases': format_matched_cases(matched_cases_data),
'offline': True,
'api_time': 0.0,
})
# 2. 构建 Prompt(在线模式继续原逻辑)
...
```
#### T3.2 `/api/analyze`(第 150 行 `analyze()`)
该接口入参带 `matched_cases`(前端已搜索过)。离线分支:
```python
if is_offline_mode():
# analyze 入参的 matched_cases 是前端传来的 dict 列表,需重新喂给 build_offline_response
result = build_offline_response(matched_cases)
log_audit({... 'mode': 'offline' ...})
return jsonify({
'success': True,
'response': result['response'],
'offline': True,
'api_time': 0.0,
})
```
**注意**:analyze 的 `matched_cases` 是前端传来的已格式化 dict(含 rank/score/project/title/file/phenomenon),与 troubleshoot 的 `matched_cases_data`(含 record 原始结构)不同。`build_offline_response` 需兼容两种输入,或路由层做适配转换。**实施时以 `build_offline_response` 统一接收「已格式化 dict 列表」为准**,troubleshoot 路由先调 `format_matched_cases` 再传入。
#### T3.3 `/api/analyze/stream`(第 201 行 `analyze_stream()`)
离线模式**保留 SSE 协议**,单帧推送后结束:
```python
def generate():
yield f'data: {json.dumps({"type": "start", "message": "正在分析...", "offline": True}, ensure_ascii=False)}\n\n'
# 搜索
engine = container.get_search_engine()
matched_cases_data = engine.search(...)
matched_cases = format_matched_cases(matched_cases_data)
yield f'data: {json.dumps({"type": "matched_cases", "cases": matched_cases, "offline": True}, ensure_ascii=False)}\n\n'
# 离线:单帧推送完整结果
result = build_offline_response(matched_cases)
yield f'data: {json.dumps({"type": "chunk", "content": result["response"], "offline": True}, ensure_ascii=False)}\n\n'
yield f'data: {json.dumps({"type": "done", "api_time": 0.0, "offline": True}, ensure_ascii=False)}\n\n'
log_audit({... 'mode': 'offline_stream' ...})
```
`generate()` 开头判断 `is_offline_mode()` 走离线生成器,否则走原在线逻辑。**缓存检查仍在离线分支之前**(离线结果也可缓存,但本期离线 <1s 可不缓存——决策:离线模式跳过缓存读写,直接返回)。
#### T3.4 `/api/health`(第 324 行 `health_check()`)
响应体新增 `offline_mode` 字段:
```python
from utils.offline_config import is_offline_mode
...
return jsonify({
'status': 'ok',
...
'offline_mode': is_offline_mode(), # 【新增】
...
})
```
**验收**
- `OFFLINE_MODE=true` 时三个路由均不触发 `call_claude_api` / `call_claude_api_stream`
- `OFFLINE_MODE` 未设置时三个路由行为与改造前完全一致(回归)
- health 响应含 `offline_mode` 字段
**约束**:不改 `format_matched_cases``search_cases``get_projects``get_categories`
---
### T4:单元测试
**文件**`skill/code/tests/test_offline_mode.py`(新增)
**用例清单**(预计 8-10 例):
| # | 用例 | 验证点 |
|---|------|--------|
| 1 | `is_offline_mode``true` | 返回 True |
| 2 | `is_offline_mode` 未设置 | 返回 False |
| 3 | `build_offline_response` 有匹配 | 含案例详情文本 |
| 4 | `build_offline_response` 空匹配 | 兜底文案 |
| 5 | `/api/troubleshoot` 离线 | 返回 `offline:true`,不调 API |
| 6 | `/api/analyze` 离线 | 返回 `offline:true` |
| 7 | `/api/analyze/stream` 离线 | SSE 含 offline 字段,单帧 |
| 8 | `/api/health` | 含 `offline_mode` 字段 |
| 9 | 在线模式回归 | `OFFLINE_MODE` 未设时行为不变 |
**依赖 conftest.py**:复用现有 `sys.path` 注入与索引路径 fixture;新增 `monkeypatch.setenv('OFFLINE_MODE', 'true')` + 调 `reset_for_test()` fixture。
**验收**`cd skill/code && python -m pytest tests/test_offline_mode.py -v` 全绿,且原有 143 用例不回归。
---
### T5:容器化部署文件
**文件**(仓库根目录新增):
#### T5.1 `requirements.txt`
```text
flask==3.0.3
gunicorn==22.0.0
scikit-learn==1.4.2
scipy==1.13.0
numpy==1.26.4
jieba==0.42.1
werkzeug==3.0.3
requests==2.32.3
cryptography==46.0.7
paramiko==4.0.0
```
> - 版本以本地实测通过为准(service_monitor 51 + 问题排查助手 167 用例全绿)。
> - `cryptography` / `paramiko` 为同进程 service_monitor 子包依赖,见 T8 与 PRD §4.6。
#### T5.2 `Dockerfile`
```dockerfile
FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt /app/requirements.txt
RUN pip install --no-cache-dir -r /app/requirements.txt
COPY skill/code/web/ /app/web/
COPY skill/code/SKILL.md /app/SKILL.md
EXPOSE 8088
ENV PYTHONIOENCODING=utf-8
CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:8088", "server:app"]
```
> gunicorn 加载 `server:app`——已确认 `server.py:107` 有模块级 `app = create_app()`,gunicorn 需以 `/app/web` 为工作目录,故 CMD 前需 `WORKDIR /app/web`(**实施时调整:将 WORKDIR 设为 `/app/web`,代码拷到 `/app/web/`,gunicorn 在该目录加载 `server` 模块**)。
**修正**
```dockerfile
FROM python:3.10-slim
WORKDIR /app/web
COPY requirements.txt /app/requirements.txt
RUN pip install --no-cache-dir -r /app/requirements.txt
COPY skill/code/web/ /app/web/
COPY skill/code/SKILL.md /app/SKILL.md
EXPOSE 8088
ENV PYTHONIOENCODING=utf-8
CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:8088", "server:app"]
```
#### T5.3 `docker-compose.yml`
```yaml
version: '3.8'
services:
troubleshoot:
build: .
ports:
- "8088:8088"
restart: always
volumes:
- ./config.json:/app/web/config.json:ro
- ./users.json:/app/web/users.json:ro
- ./records:/app/web/records:ro
- ./logs:/app/logs
environment:
- PYTHONIOENCODING=utf-8
- SECRET_KEY=${SECRET_KEY}
- OFFLINE_MODE=${OFFLINE_MODE:-true}
```
#### T5.4 `.dockerignore`
```text
# 排除向量文件,强制容器内走 TF-IDF(PRD §4.4)
搜索向量.json
**/搜索向量.json
deploy/搜索向量.json
# 其他无关项
.git
.claude
Docs
临时目录
HANDOFF.md
**/__pycache__
**/*.pyc
```
#### T5.5 `.env.example` 补充
在现有 `.env.example` 增加:
```bash
# 离线模式开关(true=跳过 Claude API,false=在线调 API)
OFFLINE_MODE=true
```
**验收**:四个文件齐全,`.dockerignore` 确认排除向量文件。
---
### T6:本地容器构建与验证
**前置**:本地需有 Docker 环境。
**步骤**
1. `docker build -t troubleshoot:offline .`
2. `docker run -d -p 8089:8088 -e OFFLINE_MODE=true -e SECRET_KEY=test troubleshoot:offline`
3. `curl http://localhost:8089/api/health` → 验证 `offline_mode: true``search.mode: tfidf`
4. `curl -X POST http://localhost:8089/api/troubleshoot -d '{"query":"门口屏MQTT绑定失败"}' -H 'Content-Type: application/json'` → 验证 `offline: true``response` 含案例、`api_time < 1s`
5. 验证容器内**不存在** `搜索向量.json``docker exec <cid> ls /app/web/records/`
**验收**:上述 5 项全过。若本地无 Docker,则跳过 T6,直接在 5.60 实测。
---
### T7:部署到 5.60 + 权威验证
**说明**:5.60 现为裸机部署(`upload_to_server.py` 走 SSH),本期容器化是**新增能力**,不立即替换现有裸机部署。
**步骤**(用 `!` 前缀执行,避免环境变量读不到):
1. 本地构建:`docker build -t troubleshoot:offline .`
2. 导出:`docker save troubleshoot:offline | gzip > troubleshoot-offline.tar.gz`
3. 传到 5.60:`! scp troubleshoot-offline.tar.gz ubains@192.168.5.60:/opt/`(需 `SSH_PASSWORD`
4. 5.60 加载:`docker load < troubleshoot-offline.tar.gz`
5. 启动:`docker compose up -d`(需在 5.60 准备 `docker-compose.yml` + `.env`
6. 权威验证:`cd deploy && python verify_deployment.py`(确认 `offline_mode``search.mode` 字段)
**验收**
- 5.60 上容器健康检查返回 `offline_mode: true``search.mode: tfidf`
- `/api/troubleshoot` 离线响应 <1s
- 现有裸机部署(8088)不受影响(容器映射到其他端口或停机窗口操作)
**注意**:若 5.60 未装 Docker,T7 无法执行,记录为「待现场环境就绪」。
---
### T8:同进程 service_monitor 依赖补齐(PRD §4.6)
**背景**:问题排查助手与 service_monitor 同处一个 Flask 进程(`server.py::create_app()` 一次性注册全部 Blueprint)。service_monitor 子包的 `utils/crypto.py` 模块级 `from cryptography.fernet import Fernet``create_app()` 阶段执行;若镜像缺 `cryptography`**进程启动即崩,问题排查助手连带不可用**`paramiko` 为函数内延迟导入(`executor.py:249`),启动不崩但用 SSH 监测时崩。
**文件**`requirements.txt`(修改)
**改动**:补两行
```text
cryptography==42.0.5
paramiko==3.4.0
```
**验收**
- 镜像构建后 `python -c "import cryptography, paramiko"` 成功
- 容器启动 `create_app()` 不再因 service_monitor 导入失败而崩
- 问题排查助手离线路由行为不受影响(回归 T4 用例仍全绿)
**约束**
- 仅改 `requirements.txt`**不改 service_monitor 源码**(§3.7 隔离红线)
- 不拆镜像 / 不拆进程(超出 PRD 范围,且破坏 service_monitor 独立子包设计)
- 容错 try/except(`server.py` 注册时跳过失败模块)**本期不做**——`server.py` 正随 service_monitor 重构变动,避免文件冲突,待重构稳定后另行评估
---
## 2. 验收标准(对照 PRD §5 非功能需求)
| PRD 要求 | 验收方式 | 对应任务 |
|----------|----------|----------|
| 配置切换改 `OFFLINE_MODE` 即可 | 改环境变量重启容器,行为切换 | T3/T6 |
| 离线 TF-IDF 逻辑不变 | 单测 + health `search.mode:tfidf` | T4/T6 |
| 离线排查 <1 秒 | `/api/troubleshoot` 离线响应计时 | T6 |
| 前端标注"离线模式" | 前端读 `offline:true` 显示横幅(前端改动另计,本期后端就绪即可) | T3 |
| 流式保留 SSE | `/api/analyze/stream` 离线仍返回 `text/event-stream` | T3.3 |
| Dockerfile+compose+requirements 齐全 | 文件存在 + 构建成功 | T5/T6 |
| 进程守护 restart always | compose `restart: always` | T5.3 |
| 日志持久化 | `./logs` volume 挂载 | T5.3 |
| 镜像兼容在线 | `OFFLINE_MODE=false` 时可调 Claude API(依赖 requests 已装) | T5.1 |
---
## 3. 测试计划
### 3.1 单元测试(T4)
- 新增 `test_offline_mode.py`,目标 ≥8 用例
- 回归:`python -m pytest -v` 全部 ≥151 用例全绿(原 143 + 新增)
### 3.2 集成测试(T6 本地容器)
- health 字段校验
- 三个路由离线返回校验
- 向量文件不存在校验
### 3.3 回归测试(隔离红线验证)
- 离线改造后,手动调用服务管理/服务监测路由,确认响应结构、状态码不变
- `service_manage.py` / `service_monitor.py` 文件 git diff 为空(零改动)
### 3.4 部署验证(T7)
- `verify_deployment.py` 通过
- 容器 `docker ps` 状态 running,restart 策略 always
---
## 4. 风险评估
| 风险 | 等级 | 缓解措施 |
|------|------|----------|
| `requirements.txt` 版本号在 slim 镜像装不上(scipy 编译) | 中 | T6 本地先构建验证;必要时改用 `python:3.10` 非 slim 或预编译 wheel |
| gunicorn WORKDIR 路径不对,加载不到 `server` 模块 | 中 | T5.2 已将 WORKDIR 设为 `/app/web`,T6 构建后实测 `docker run` 启动日志 |
| analyze 路由入参 `matched_cases` 格式与 troubleshoot 不一致,导致 `build_offline_response` 兼容问题 | 中 | T3.2 已标注,实施时统一接收「已格式化 dict 列表」,troubleshoot 先 format 再传入 |
| 5.60 无 Docker 环境,T7 无法执行 | 中 | T7 标注「待现场就绪」,不阻塞 T1-T6 验收 |
| 离线模式下 `cache_manager` 空跑产生无意义缓存文件 | 低 | T3.3 决策:离线跳过缓存读写 |
| 前端未适配 `offline:true` 横幅显示 | 低 | 本期后端就绪即可,前端适配另行处理 |
| 同进程 service_monitor 的 `cryptography` 模块级导入失败,连累问题排查助手启动 | 中 | T8 补依赖入 requirements;不改 service_monitor 源码 |
---
## 5. 实施记录
| 任务 | 状态 | 完成时间 | 备注 |
|------|------|----------|------|
| T1 离线配置工具 | ✅ 完成 | 2026-07-16 | 新增 `utils/offline_config.py` |
| T2 离线响应构建 | ✅ 完成 | 2026-07-16 | `ai_service.py` 新增 `build_offline_response()`,兼容两种输入格式 |
| T3 路由离线分支 | ✅ 完成 | 2026-07-16 | troubleshoot/analyze/analyze_stream/health 四处改完,离线跳过缓存 |
| T4 单元测试 | ✅ 完成 | 2026-07-16 | 新增 22 用例全绿,回归 167 全绿 |
| T5 容器化文件 | ✅ 完成 | 2026-07-16 | Dockerfile/compose/requirements/.dockerignore/.env.example |
| T6 本地容器验证 | ⏸️ 暂缓 | — | 待你本地有 Docker 时验证 |
| T7 部署 5.60 | ⏸️ 暂缓 | — | 5.60 有 Docker 但当前网络不可达,待现场执行 |
| T8 同进程依赖补齐 | ✅ 完成 | 2026-07-16 | requirements 补 cryptography/paramiko,回归测试通过 |
---
## 6. 后续工作(本期不做)
- 前端 index.html 适配 `offline:true` 横幅与 `analysis:null` 提示
- 服务管理/服务监测模块的容器化方案(另立 PRD)
- 三模块是否共用镜像底座的统筹决策
- 本地小模型推理(Ollama)
- 配置热更新(`OFFLINE_MODE` 改后免重启)
- `deploy/upload_to_server.py` 是否改为推送镜像而非裸文件(容器化全面落地后)
---
## 7. 附录
### 7.1 关键代码定位
- 离线开关:`utils/offline_config.py::is_offline_mode`
- 离线响应:`services/ai_service.py::build_offline_response`
- 路由分支:`routes/troubleshoot.py`(troubleshoot / analyze / analyze_stream / health)
- gunicorn 入口:`server.py:107` `app = create_app()`
### 7.2 PRD 对照
本计划严格对照 PRD v2.1:
- §3.1 离线 Q&A → T2
- §3.2 环境变量开关 → T1
- §3.4 API 行为变更 → T3
- §3.5 health 字段 → T3.4
- §3.6 代码分层 → T1/T2/T3
- §3.7 模块隔离 → 全程约束(禁改 service_manage / service_monitor)
- §4.1-4.5 容器化 → T5
- §4.4 向量文件不打包 → T5.4 `.dockerignore`
- §5 非功能需求 → 第 2 节验收
- §6 约束 → 全程遵守
version: '3.8'
services:
troubleshoot:
build: .
ports:
- "8088:8088"
restart: always
volumes:
# 配置与用户数据只读挂载(便于现场修改无需重建镜像)
- ./config.json:/app/web/config.json:ro
- ./users.json:/app/web/users.json:ro
- ./records:/app/web/records:ro
# 日志持久化到宿主机
- ./logs:/app/logs
environment:
- PYTHONIOENCODING=utf-8
- SECRET_KEY=${SECRET_KEY}
# 离线模式开关:true=跳过 Claude API(离线 Q&A),false=在线调 API
- OFFLINE_MODE=${OFFLINE_MODE:-true}
flask==3.0.3
gunicorn==22.0.0
scikit-learn==1.4.2
scipy==1.13.0
numpy==1.26.4
jieba==0.42.1
werkzeug==3.0.3
requests==2.32.3
# 同进程 service_monitor 子包依赖(PRD §4.6)
# cryptography 模块级导入,缺失则 create_app() 启动即崩,连累问题排查助手
# paramiko 延迟导入(SSH 监测时),缺失则用到时崩
# 版本以本地实测通过为准(service_monitor 51 用例 + 问题排查助手 167 用例全绿)
cryptography==46.0.7
paramiko==4.0.0
# -*- coding: utf-8 -*-
"""
离线模式单元测试(PRD §3 离线部署方案)
覆盖:
T1 utils/offline_config.py 的 is_offline_mode / reset_for_test
T2 services/ai_service.build_offline_response
T3 routes/troubleshoot.py 三个路由 + health 的离线分支
回归:OFFLINE_MODE 未设置时行为不变(不调 AI)
隔离:本测试只动问题排查助手相关文件,不触碰 service_manage / service_monitor。
"""
import json
import pytest
import utils.offline_config as offline_config
# ============================================================
# Fixture:离线开关切换
# ============================================================
@pytest.fixture
def offline_on(monkeypatch):
"""开启离线模式"""
monkeypatch.setenv('OFFLINE_MODE', 'true')
offline_config.reset_for_test()
yield
offline_config.reset_for_test()
@pytest.fixture
def offline_off(monkeypatch):
"""关闭离线模式(在线)"""
monkeypatch.delenv('OFFLINE_MODE', raising=False)
offline_config.reset_for_test()
yield
offline_config.reset_for_test()
# ============================================================
# T1:offline_config
# ============================================================
class TestOfflineConfig:
def test_offline_true(self, offline_on):
assert offline_config.is_offline_mode() is True
def test_offline_not_set_defaults_online(self, offline_off):
assert offline_config.is_offline_mode() is False
@pytest.mark.parametrize('val,expected', [
('true', True), ('TRUE', True), ('1', True), ('yes', True), ('YES', True),
('false', False), ('0', False), ('', False), ('random', False),
])
def test_value_parsing(self, monkeypatch, val, expected):
if val == '':
monkeypatch.delenv('OFFLINE_MODE', raising=False)
else:
monkeypatch.setenv('OFFLINE_MODE', val)
offline_config.reset_for_test()
assert offline_config.is_offline_mode() is expected
offline_config.reset_for_test()
def test_cache_persists(self, monkeypatch):
"""首次读取后缓存,中途改环境变量不生效"""
monkeypatch.setenv('OFFLINE_MODE', 'true')
offline_config.reset_for_test()
assert offline_config.is_offline_mode() is True
# 中途改环境变量,缓存不变
monkeypatch.setenv('OFFLINE_MODE', 'false')
assert offline_config.is_offline_mode() is True
offline_config.reset_for_test()
# ============================================================
# T2:build_offline_response
# ============================================================
class TestBuildOfflineResponse:
def test_empty_cases(self):
from services.ai_service import build_offline_response
result = build_offline_response([])
assert result['offline'] is True
assert '未匹配到相关案例' in result['response']
assert '离线模式' in result['response']
def test_with_cases_raw_format(self):
"""原始格式:search engine 返回的结构(含 record)"""
from services.ai_service import build_offline_response
cases = [
{
'rank': 1, 'score': 0.92,
'record': {
'project': '厦门银行',
'title': '门口屏 MQTT 绑定失败',
'phenomenon': '绑定时报 MQTT 连接错误',
'solution': '升级门口屏到 5.0+ 版本',
'file': 'record_001.md',
},
},
{
'rank': 2, 'score': 0.75,
'record': {
'project': '展厅',
'title': '门口屏不显示会议信息',
'phenomenon': '会议开始后不显示主题',
'solution': '检查 EMQX 容器状态',
'file': 'record_002.md',
},
},
]
result = build_offline_response(cases)
assert result['offline'] is True
text = result['response']
assert '厦门银行' in text
assert '门口屏 MQTT 绑定失败' in text
assert '92%' in text # 相似度格式化
assert '升级门口屏到 5.0+ 版本' in text # solution
assert 'record_001.md' in text # 参考文件
assert '共匹配 2 个' in text
def test_with_cases_formatted_format(self):
"""已格式化格式:前端 analyze 入参(无 solution,直接字段)"""
from services.ai_service import build_offline_response
cases = [
{
'rank': 1, 'score': 0.80,
'project': '展厅', 'title': '测试问题', 'phenomenon': '现象A', 'file': 'r.md',
},
]
result = build_offline_response(cases)
assert '展厅' in result['response']
assert '80%' in result['response']
# 无 solution 时不报错
def test_max_five_cases(self):
from services.ai_service import build_offline_response
cases = [{'rank': i, 'score': 0.5, 'record': {'title': f'c{i}'}} for i in range(10)]
result = build_offline_response(cases)
# 最多展示 5 个案例标题(### 案例N),但总数标注为 10
assert result['response'].count('### 案例') == 5
assert '共匹配 10 个' in result['response']
# ============================================================
# T3:路由离线分支
# ============================================================
class TestRoutesOffline:
def test_health_has_offline_field(self, client, offline_on):
resp = client.get('/api/health')
assert resp.status_code == 200
data = resp.get_json()
assert 'offline_mode' in data
assert data['offline_mode'] is True
def test_health_online_mode(self, client, offline_off):
resp = client.get('/api/health')
data = resp.get_json()
assert data['offline_mode'] is False
def test_troubleshoot_offline_skips_api(self, auth_client, offline_on, monkeypatch):
"""离线模式不调 Claude API(mock 抛异常验证不被调用)"""
import services.ai_service as ai_service
def _explode(*a, **kw):
raise AssertionError("离线模式不应调用 Claude API")
monkeypatch.setattr(ai_service, 'call_claude_api', _explode)
resp = auth_client.post('/api/troubleshoot', json={
'query': '门口屏MQTT绑定失败',
'project_name': '',
})
assert resp.status_code == 200
data = resp.get_json()
assert data['success'] is True
assert data.get('offline') is True
assert data['api_time'] == 0.0
assert len(data['matched_cases']) > 0
assert '离线模式' in data['response']
def test_troubleshoot_online_calls_api(self, auth_client, offline_off):
"""在线模式回归:调 AI(已被 conftest mock),返回正常分析"""
resp = auth_client.post('/api/troubleshoot', json={
'query': '门口屏MQTT绑定失败',
})
assert resp.status_code == 200
data = resp.get_json()
assert data['success'] is True
# 在线模式无 offline 字段(或为 False),有 AI 响应
assert data.get('offline') is not True
assert 'response' in data
def test_analyze_offline(self, auth_client, offline_on, monkeypatch):
import services.ai_service as ai_service
def _explode(*a, **kw):
raise AssertionError("离线模式不应调用 Claude API")
monkeypatch.setattr(ai_service, 'call_claude_api', _explode)
resp = auth_client.post('/api/analyze', json={
'query': '门口屏问题',
'matched_cases': [
{'rank': 1, 'score': 0.9, 'project': '厦门银行',
'title': 'MQTT失败', 'phenomenon': '绑定报错', 'file': 'r.md'},
],
})
assert resp.status_code == 200
data = resp.get_json()
assert data['success'] is True
assert data.get('offline') is True
assert '厦门银行' in data['response']
def test_analyze_stream_offline_sse(self, auth_client, offline_on, monkeypatch):
"""离线流式仍走 SSE,单帧推送,含 offline 标记"""
import services.ai_service as ai_service
def _explode(*a, **kw):
raise AssertionError("离线模式不应调用 Claude API")
monkeypatch.setattr(ai_service, 'call_claude_api_stream', _explode)
resp = auth_client.get('/api/analyze/stream?query=门口屏MQTT')
assert resp.status_code == 200
assert 'text/event-stream' in resp.content_type
body = resp.get_data(as_text=True)
# 至少有 start / matched_cases / chunk / done 四类事件
assert '"type": "start"' in body
assert '"type": "matched_cases"' in body
assert '"type": "chunk"' in body
assert '"type": "done"' in body
# 离线标记
assert '"offline": true' in body
# 不应出现 error
assert '"type": "error"' not in body
...@@ -16,11 +16,12 @@ from flask import Blueprint, request, jsonify, Response, session, render_templat ...@@ -16,11 +16,12 @@ from flask import Blueprint, request, jsonify, Response, session, render_templat
import container import container
from decorators import page_login_required from decorators import page_login_required
from services.ai_service import build_prompt, call_claude_api, call_claude_api_stream from services.ai_service import build_prompt, call_claude_api, call_claude_api_stream, build_offline_response
from utils.audit import log_audit from utils.audit import log_audit
from utils.response import error_response from utils.response import error_response
from utils.error_codes import ErrorCodes from utils.error_codes import ErrorCodes
from utils.logger import get_logger from utils.logger import get_logger
from utils.offline_config import is_offline_mode
logger = get_logger(__name__) logger = get_logger(__name__)
...@@ -76,6 +77,29 @@ def troubleshoot(): ...@@ -76,6 +77,29 @@ def troubleshoot():
project_filter=project_name if project_name else None, project_filter=project_name if project_name else None,
) )
# ============================================================
# 【离线模式】跳过 Claude API,直接返回搜索结果(PRD §3.4)
# ============================================================
if is_offline_mode():
result = build_offline_response(matched_cases_data)
log_audit({
'timestamp': datetime.now().isoformat(),
'project': project_name,
'system_type': system_type,
'apk_product': apk_product,
'query': query,
'matched_count': len(matched_cases_data),
'api_time': 0.0,
'mode': 'offline',
})
return jsonify({
'success': True,
'response': result['response'],
'matched_cases': format_matched_cases(matched_cases_data),
'offline': True,
'api_time': 0.0,
})
# 2. 构建 Prompt # 2. 构建 Prompt
prompt = build_prompt(query, project_name, system_type, apk_product, matched_cases_data) prompt = build_prompt(query, project_name, system_type, apk_product, matched_cases_data)
...@@ -162,6 +186,30 @@ def analyze(): ...@@ -162,6 +186,30 @@ def analyze():
return jsonify(error_response(ErrorCodes.INVALID_PARAM, '请输入问题描述')), 400 return jsonify(error_response(ErrorCodes.INVALID_PARAM, '请输入问题描述')), 400
try: try:
# ============================================================
# 【离线模式】跳过 Claude API(PRD §3.4)
# ============================================================
if is_offline_mode():
# analyze 收到的 matched_cases 是前端传来的已格式化数据
# 直接用于离线响应(可能缺失 solution,但 title/phenomenon/score 等齐全)
result = build_offline_response(matched_cases)
log_audit({
'timestamp': datetime.now().isoformat(),
'project': project_name,
'system_type': system_type,
'apk_product': apk_product,
'query': query,
'matched_count': len(matched_cases),
'api_time': 0.0,
'mode': 'offline_analyze',
})
return jsonify({
'success': True,
'response': result['response'],
'offline': True,
'api_time': 0.0,
})
# 1. 构建 Prompt(使用前端传来的匹配案例数据) # 1. 构建 Prompt(使用前端传来的匹配案例数据)
prompt = build_prompt(query, project_name, system_type, apk_product, matched_cases) prompt = build_prompt(query, project_name, system_type, apk_product, matched_cases)
...@@ -227,28 +275,30 @@ def analyze_stream(): ...@@ -227,28 +275,30 @@ def analyze_stream():
yield f'data: {json.dumps({"type": "error", "message": "请输入问题描述"}, ensure_ascii=False)}\n\n' yield f'data: {json.dumps({"type": "error", "message": "请输入问题描述"}, ensure_ascii=False)}\n\n'
return Response(error_gen(), mimetype='text/event-stream') return Response(error_gen(), mimetype='text/event-stream')
# 检查缓存 # 离线模式:跳过缓存,直接走离线分支(<1s 无需缓存)
cache = container.get_cache_manager() if not is_offline_mode():
cached_result = cache.get(project_name, system_type, apk_product, query) # 检查缓存
cache = container.get_cache_manager()
if cached_result: cached_result = cache.get(project_name, system_type, apk_product, query)
# 返回缓存结果
def generate_cached(): if cached_result:
yield f'data: {json.dumps({"type": "start", "message": "正在分析...", "cached": True}, ensure_ascii=False)}\n\n' # 返回缓存结果
yield f'data: {json.dumps({"type": "matched_cases", "cases": cached_result.get("matched_cases", [])}, ensure_ascii=False)}\n\n' def generate_cached():
yield f'data: {json.dumps({"type": "start", "message": "正在分析...", "cached": True}, ensure_ascii=False)}\n\n'
# 流式推送缓存内容 yield f'data: {json.dumps({"type": "matched_cases", "cases": cached_result.get("matched_cases", [])}, ensure_ascii=False)}\n\n'
cached_response = cached_result.get("response", "")
chunk_size = 30 # 流式推送缓存内容
for i in range(0, len(cached_response), chunk_size): cached_response = cached_result.get("response", "")
content_chunk = cached_response[i:i + chunk_size] chunk_size = 30
yield f'data: {json.dumps({"type": "chunk", "content": content_chunk}, ensure_ascii=False)}\n\n' for i in range(0, len(cached_response), chunk_size):
time.sleep(0.01) content_chunk = cached_response[i:i + chunk_size]
yield f'data: {json.dumps({"type": "chunk", "content": content_chunk}, ensure_ascii=False)}\n\n'
time.sleep(0.01)
yield f'data: {json.dumps({"type": "done", "api_time": 0.5, "cached": True}, ensure_ascii=False)}\n\n' yield f'data: {json.dumps({"type": "done", "api_time": 0.5, "cached": True}, ensure_ascii=False)}\n\n'
return Response(generate_cached(), mimetype='text/event-stream', return Response(generate_cached(), mimetype='text/event-stream',
headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'}) headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'})
def generate(): def generate():
"""生成器:流式返回内容""" """生成器:流式返回内容"""
...@@ -256,7 +306,7 @@ def analyze_stream(): ...@@ -256,7 +306,7 @@ def analyze_stream():
try: try:
# 1. 发送开始标记 # 1. 发送开始标记
yield f'data: {json.dumps({"type": "start", "message": "正在分析..."}, ensure_ascii=False)}\n\n' yield f'data: {json.dumps({"type": "start", "message": "正在分析...", "offline": is_offline_mode()}, ensure_ascii=False)}\n\n'
# 2. 搜索匹配案例 # 2. 搜索匹配案例
engine = container.get_search_engine() engine = container.get_search_engine()
...@@ -264,7 +314,28 @@ def analyze_stream(): ...@@ -264,7 +314,28 @@ def analyze_stream():
matched_cases = format_matched_cases(matched_cases_data) matched_cases = format_matched_cases(matched_cases_data)
# 3. 发送匹配案例 # 3. 发送匹配案例
yield f'data: {json.dumps({"type": "matched_cases", "cases": matched_cases}, ensure_ascii=False)}\n\n' yield f'data: {json.dumps({"type": "matched_cases", "cases": matched_cases, "offline": is_offline_mode()}, ensure_ascii=False)}\n\n'
# ============================================================
# 【离线模式】单帧推送搜索结果后结束(PRD §3.4)
# ============================================================
if is_offline_mode():
result = build_offline_response(matched_cases_data)
# 单帧推送完整响应文本
yield f'data: {json.dumps({"type": "chunk", "content": result["response"], "offline": True}, ensure_ascii=False)}\n\n'
api_time = time.time() - start_time
yield f'data: {json.dumps({"type": "done", "api_time": round(api_time, 2), "offline": True}, ensure_ascii=False)}\n\n'
log_audit({
'timestamp': datetime.now().isoformat(),
'project': project_name,
'system_type': system_type,
'apk_product': apk_product,
'query': query,
'matched_count': len(matched_cases),
'api_time': round(api_time, 2),
'mode': 'offline_stream',
})
return
# 4. 构建 Prompt # 4. 构建 Prompt
prompt = build_prompt(query, project_name, system_type, apk_product, matched_cases_data) prompt = build_prompt(query, project_name, system_type, apk_product, matched_cases_data)
...@@ -339,6 +410,7 @@ def health_check(): ...@@ -339,6 +410,7 @@ def health_check():
'status': 'ok', 'status': 'ok',
'timestamp': datetime.now().isoformat(), 'timestamp': datetime.now().isoformat(),
'version': '1.3.0', 'version': '1.3.0',
'offline_mode': is_offline_mode(),
'knowledge_base': { 'knowledge_base': {
'total_records': len(engine.records), 'total_records': len(engine.records),
'last_update': engine.index.get('last_update', 'unknown'), 'last_update': engine.index.get('last_update', 'unknown'),
......
...@@ -353,6 +353,74 @@ def call_claude_api_stream(prompt, model=None): ...@@ -353,6 +353,74 @@ def call_claude_api_stream(prompt, model=None):
yield chunk yield chunk
# ============================================================
# 离线模式响应构建(PRD §3.1)
# ============================================================
def build_offline_response(matched_cases):
"""构建离线模式响应文本(跳过 Claude API,直接格式化匹配结果)
接受两种输入格式:
1. 原始格式:engine.search() 返回的列表,每项含 record dict(有 solution)
2. 已格式化:前端 /api/analyze 传回的列表,每项含直接字段(无 solution)
Args:
matched_cases: 匹配结果列表,每项至少含 score/title/phenomenon
Returns:
dict: {"response": str(Markdown 文本), "offline": True}
"""
if not matched_cases:
return {
"response": (
"## 📋 离线排查参考(基于知识库匹配)\n\n"
"> 💡 当前为离线模式,结果基于历史知识库匹配,无 AI 分析。\n\n"
"未匹配到相关案例。"
),
"offline": True,
}
# 提取案例字段(兼容两种输入格式)
def _get(case, key):
"""从原始格式或格式化格式提取字段"""
if 'record' in case and isinstance(case['record'], dict):
# 原始格式:case.record.key
return case['record'].get(key, '') or case.get(key, '')
# 已格式化格式:case.key
return case.get(key, '')
parts = [
"## 📋 离线排查参考(基于知识库匹配)\n",
"> 💡 当前为离线模式,结果基于历史知识库匹配,无 AI 分析。\n",
]
for i, c in enumerate(matched_cases[:5], 1):
score = c.get('score', 0)
title = _get(c, 'title') or '未记录'
project = _get(c, 'project') or '未记录'
phenomenon = _get(c, 'phenomenon') or title
solution = _get(c, 'solution') or ''
filename = _get(c, 'file') or _get(c, 'filename') or '未记录'
parts.extend([
f"---\n",
f"### 案例{i}(相似度:{score:.0%})\n",
f"- **项目**:{project}\n",
f"- **标题**:{title}\n",
f"- **现象**:{phenomenon}\n",
])
if solution:
parts.append(f"- **解决方案**:{solution}\n")
parts.append(f"- **参考文件**:{filename}\n")
parts.append(f"\n> 共匹配 {len(matched_cases)} 个相关案例。")
return {
"response": "".join(parts),
"offline": True,
}
def split_mock_response(text, chunk_size=20): def split_mock_response(text, chunk_size=20):
"""将模拟响应分割成小块,模拟流式返回""" """将模拟响应分割成小块,模拟流式返回"""
for i in range(0, len(text), chunk_size): for i in range(0, len(text), chunk_size):
......
# -*- coding: utf-8 -*-
"""
offline_config.py — 离线模式开关读取(问题排查助手专用)
从环境变量 OFFLINE_MODE 读取离线模式开关,启动时读取一次。
仅此模块直接读环境变量,其他文件统一调用 is_offline_mode()。
设计原则:
- 进程级缓存:启动时读取一次,后续不改(热更新本期不做,PRD §3.2)
- 隔离边界:仅归问题排查助手使用,不污染 container.py 或其他模块状态
- 测试可重置:提供 reset_for_test() 供 pytest 刷新缓存
"""
import os
from utils.logger import get_logger
logger = get_logger(__name__)
_CACHE = None # None = 未初始化,True/False = 已缓存
def is_offline_mode():
"""返回是否离线模式
从环境变量 OFFLINE_MODE 读取:
true / 1 / yes(不区分大小写) → True(离线)
False / 未设置 / 其他值 → False(在线,默认)
首次调用后结果缓存到进程退出,重启或测试调 reset_for_test() 才刷新。
"""
global _CACHE
if _CACHE is None:
raw = os.environ.get('OFFLINE_MODE', '').strip().lower()
_CACHE = raw in ('true', '1', 'yes')
logger.info("离线模式: %s (OFFLINE_MODE=%r)", _CACHE, raw or '<未设置>')
return _CACHE
def reset_for_test():
"""测试用:重置缓存,下次调 is_offline_mode() 重新读环境变量"""
global _CACHE
_CACHE = None
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论