ai-robot-core/spec/ai-service/design.md

---
feature_id: "AISVC"
title: "Python AI 中台（ai-service）技术设计"
status: "draft"
version: "0.7.0"
last_updated: "2026-02-27"
inputs:
  - "spec/ai-service/requirements.md"
  - "spec/ai-service/openapi.provider.yaml"
  - "spec/ai-service/openapi.admin.yaml"
  - "java/openapi.deps.yaml"
---

# Python AI 中台（ai-service）技术设计（AISVC）

## 1. 设计目标与约束

### 1.1 设计目标
- 落地 `POST /ai/chat` 的 **non-streaming JSON** 与 **SSE streaming** 两种返回模式，并确保与契约一致：
  - non-streaming 响应字段必须包含 `reply/confidence/shouldTransfer`。
  - streaming 通过 `Accept: text/event-stream` 输出 `message/final/error` 事件序列。
- 实现 AI 侧会话记忆：基于 `(tenantId, sessionId)` 持久化与加载。
- 实现 RAG（MVP：向量检索）并预留图谱检索（Neo4j）插件点。
- 多租户隔离：
  - Qdrant：一租户一 collection（或一租户一 collection 前缀）。
  - PostgreSQL：按 `tenant_id` 分区/索引，保证跨租户不可见。

### 1.2 硬约束（来自契约与需求）
- API 对齐：`/ai/chat`、`/ai/health` 的路径/方法/状态码与 Java 侧 deps 对齐。
- 多租户：请求必须携带 `X-Tenant-Id`（网关/拦截器易处理），所有数据访问必须以 `tenant_id` 过滤。
- SSE：事件类型固定为 `message/final/error`，并保证顺序与异常语义清晰。

---

## 2. 总体架构与模块分层

### 2.1 分层概览
本服务按“职责单一 + 可插拔”的原则分为五层：

1) **API 层（Transport / Controller）**
- 职责：
  - HTTP 请求解析、参数校验（含 `X-Tenant-Id`）、鉴权/限流（如后续需要）。
  - 根据 `Accept` 头选择 non-streaming 或 SSE streaming。
  - 统一错误映射为 `ErrorResponse`。
- 输入：`X-Tenant-Id` header + `ChatRequest` body。
- 输出：
  - JSON：`ChatResponse`
  - SSE：`message/final/error` 事件流。

2) **编排层（Orchestrator / Use Case）**
- 职责：
  - 整体流程编排：加载会话记忆 → 合并上下文 →（可选）RAG 检索 → 组装 prompt → 调用 LLM → 计算置信度与转人工建议 → 写回记忆。
  - 在 streaming 模式下，将 LLM 的增量输出转为 SSE `message` 事件，同时维护最终 `reply`。
- 输入：`tenantId, sessionId, currentMessage, channelType, history?, metadata?`
- 输出：
  - non-streaming：一次性 `ChatResponse`
  - streaming：增量 token（或片段）流 + 最终 `ChatResponse`。

3) **记忆层（Memory）**
- 职责：
  - 持久化会话消息与摘要/记忆（最小：消息列表）。
  - 提供按 `(tenantId, sessionId)` 查询的会话上下文读取 API。
- 存储：PostgreSQL。

4) **检索层（Retrieval）**
- 职责：
  - 提供统一 `Retriever` 抽象接口。
  - MVP 实现：向量检索（Qdrant）。
  - 插件点：图谱检索（Neo4j）实现可新增而不改动 Orchestrator。

5) **LLM 适配层（LLM Adapter）**
- 职责：
  - 屏蔽不同 LLM 提供方差异（请求格式、流式回调、重试策略）。
  - 提供：一次性生成接口 + 流式生成接口（yield token/delta）。

### 2.2 关键数据流（文字版）
- API 层接收请求 → 提取 `tenantId`（Header）与 body → 调用 Orchestrator。
- Orchestrator：
  1) Memory.load(tenantId, sessionId)
  2) merge_context(local_history, external_history)
  3) Retrieval.retrieve(query, tenantId, channelType, metadata)（MVP 向量检索）
  4) build_prompt(merged_history, retrieved_docs, currentMessage)
  5) LLM.generate(...)（non-streaming）或 LLM.stream_generate(...)（streaming）
  6) compute_confidence(…)
  7) Memory.append(tenantId, sessionId, user/assistant messages)
  8) 返回 `ChatResponse`（或通过 SSE 输出）。

---

## 3. API 与协议设计要点

### 3.1 tenantId 放置与处理
- **主入口**：`X-Tenant-Id` header（契约已声明 required）。
- Orchestrator 与所有下游组件调用均显式传入 `tenantId`。
- 禁止使用仅 `sessionId` 定位会话，必须 `(tenantId, sessionId)`。

### 3.2 streaming / non-streaming 模式判定
- 以 `Accept` 头作为唯一判定依据：
  - `Accept: text/event-stream` → SSE streaming。
  - 其他 → non-streaming JSON。

---

## 4. RAG 管道设计

### 4.1 MVP：向量检索（Qdrant）流程

#### 4.1.1 步骤
1) **Query 规范化**
- 输入：`currentMessage`（可结合 `channelType` 与 metadata）。
- 规则：去噪、截断（防止超长）、可选的 query rewrite（MVP 可不做）。

2) **Embedding**
- 由 `EmbeddingProvider` 生成向量（可复用 LLM 适配层或独立适配层）。
- ✅ 已确认：Token 计数统一使用 `tiktoken` 进行精确计算（用于 history 截断与证据预算）。

3) **向量检索**（Qdrant）
- 按租户隔离选择 collection（见 5.1）。
- 使用 topK + score threshold 过滤。

4) **上下文构建**
- 将检索结果转为 “证据片段列表”，限制总 token 与片段数。
- 生成 prompt 时区分：系统指令 / 对话历史 / 证据 / 当前问题。

5) **生成与引用策略**
- 生成回答必须优先依据证据。
- 若证据不足：触发兜底策略（见 4.3）。

#### 4.1.2 关键参数（MVP 默认，可配置）
- topK（例如 5~10）
- scoreThreshold（相似度阈值）
- minHits（最小命中文档数）
- maxEvidenceTokens（证据总 token 上限）

### 4.2 图谱检索插件点（Neo4j）

#### 4.2.1 Retriever 抽象接口（概念设计）
设计统一接口，使 Orchestrator 不关心向量/图谱差异：

- `Retriever.retrieve(ctx) -> RetrievalResult`
  - 输入 `ctx`：包含 `tenantId`, `query`, `sessionId`, `channelType`, `metadata` 等。
  - 输出 `RetrievalResult`：
    - `hits[]`：证据条目（统一为 text + score + source + metadata）
    - `diagnostics`：检索调试信息（可选）

MVP 提供 `VectorRetriever(Qdrant)`。

#### 4.2.2 Neo4j 接入方式（未来扩展）
新增实现类 `GraphRetriever(Neo4j)`，实现同一接口：
- tenant 隔离：Neo4j 可采用 database per tenant / label+tenantId 过滤 / subgraph per tenant（视规模与授权能力选择）。
- 输出同构 `RetrievalResult`，由 ContextBuilder 使用。

> 约束：新增 GraphRetriever 不应要求修改 API 层与 Orchestrator 的业务流程，只需配置切换（策略模式/依赖注入）。

### 4.3 检索不中兜底与置信度策略（对应 AC-AISVC-17/18/19）

定义“检索不足”的判定：
- `hits.size < minHits` 或 `max(score) < scoreThreshold` 或 evidence token 超限导致可用证据过少。

兜底动作：
1) 回复策略：
- 明确表达“未从知识库确认/建议咨询人工/提供可执行下一步”。
- 避免编造具体事实性结论。
2) 置信度：
- 以 `T_low` 为阈值（可配置），检索不足场景通常产生较低 `confidence`。
3) 转人工建议：
- `confidence < T_low` 时 `shouldTransfer=true`，可附 `transferReason`。
- ✅ 已确认：MVP 阶段 `confidence` 优先基于 RAG 检索分数（Score）计算（并结合检索不中兜底下调）。

---

## 5. 多租户隔离方案

### 5.1 Qdrant（向量库）隔离：一租户一 Collection

#### 5.1.1 命名规则
- collection 命名：`kb_{tenantId}`（或 `kb_{tenantId}_{kbName}` 为未来多知识库预留）。

#### 5.1.2 读写路径
- 所有 upsert/search 操作必须先基于 `tenantId` 解析目标 collection。
- 禁止在同一 collection 内通过 payload filter 做租户隔离作为默认方案（可作为兜底/迁移手段），原因：
  - 更容易出现误用导致跨租户泄露。
  - 运维与配额更难隔离（单租户删除、重建、统计）。

#### 5.1.3 租户生命周期
- tenant 创建：初始化 collection（含向量维度与 index 参数）。
  - ✅ 已确认：采用**提前预置**模式，不通过业务请求动态创建 collection。
- tenant 删除：删除 collection。
- tenant 扩容：独立配置 HNSW 参数或分片（依赖 Qdrant 部署模式）。

### 5.2 PostgreSQL（会话库）分区与约束

#### 5.2.1 表设计（概念）
- `chat_sessions`
  - `tenant_id` (NOT NULL)
  - `session_id` (NOT NULL)
  - `created_at`, `updated_at`
  - 主键/唯一约束：`(tenant_id, session_id)`

- `chat_messages`
  - `tenant_id` (NOT NULL)
  - `session_id` (NOT NULL)
  - `message_id` (UUID 或 bigserial)
  - `role` (user/assistant)
  - `content` (text)
  - `created_at`

#### 5.2.2 分区策略
根据租户规模选择：

**方案 A（MVP 推荐）：逻辑分区 + 复合索引**
- 不做 PG 分区表。
- 建立索引：
  - `chat_messages(tenant_id, session_id, created_at)`
  - `chat_sessions(tenant_id, session_id)`
- 好处：实现与运维简单。

**方案 B（规模化）：按 tenant_id 做 LIST/HASH 分区**
- `chat_messages` 按 `tenant_id` 分区（LIST 或 HASH）。
- 适合租户数量有限且单租户数据量大，或需要更强隔离与清理效率。

#### 5.2.3 防串租约束
- 所有查询必须带 `tenant_id` 条件；在代码层面提供 `TenantScopedRepository` 强制注入。
- 可选：启用 Row Level Security（RLS）并通过 `SET app.tenant_id` 做隔离（实现复杂度较高，后续可选）。

---

## 6. SSE 状态机设计（顺序与异常保证）

### 6.1 状态机
定义连接级状态：
- `INIT`：已建立连接，尚未输出。
- `STREAMING`：持续输出 `message` 事件。
- `FINAL_SENT`：已输出 `final`，准备关闭。
- `ERROR_SENT`：已输出 `error`，准备关闭。
- `CLOSED`：连接关闭。

### 6.2 事件顺序保证
- 在一次请求生命周期内，事件序列必须满足：
  - `message*`（0 次或多次） → **且仅一次** `final` → close
  - 或 `message*`（0 次或多次） → **且仅一次** `error` → close
- 禁止 `final` 之后再发送 `message`。
- 禁止同时发送 `final` 与 `error`。

实现策略（概念）：
- Orchestrator 维护一个原子状态变量（或单线程事件循环保证），在发送 `final/error` 时 CAS 切换状态。
- 对 LLM 流式回调进行包装：
  - 每个 delta 输出前检查状态必须为 `STREAMING`。
  - 发生异常立即进入 `ERROR_SENT` 并输出 `error`。

### 6.3 异常处理
- 参数错误：在进入流式生成前即可判定，直接发送 `error`（或返回 400，取决于是否已经选择 SSE；建议 SSE 模式同样用 `event:error` 输出 ErrorResponse）。
- 下游依赖错误（LLM/Qdrant/PG）：
  - 若尚未开始输出：可直接返回 503/500 JSON（non-streaming）或发送 `event:error`（streaming）。
  - 若已输出部分 `message`：必须以 `event:error` 收尾。
- 客户端断开：
  - 立即停止 LLM 流（如果适配层支持 cancel），并避免继续写入 response。

- ✅ 已确认：必须实现 SSE 心跳（Keep-alive），以注释行形式定期发送 `: ping`（不改变事件模型），防止网关/中间件断开连接。
- ✅ 已确认：Python 内部设置 **20s 硬超时**（包含 LLM 调用与检索/存储等关键步骤的总体超时控制），防止资源泄露与请求堆积。

---

## 7. 上下文合并规则（Java history + 本地持久化 history）

### 7.1 合并输入
- `H_local`：Memory 层基于 `(tenantId, sessionId)` 读取到的历史（按时间排序）。
- `H_ext`：Java 请求中可选的 `history`（按传入顺序）。

### 7.2 去重规则（确定性）
为避免重复注入导致 prompt 膨胀，定义 message 指纹：
- `fingerprint = hash(role + "|" + normalized(content))`
- normalized：trim + 统一空白（MVP 简化：trim）。

去重策略：
1) 先以 `H_local` 构建 `seen` 集合。
2) 遍历 `H_ext`：若 fingerprint 未出现，则追加到 merged；否则跳过。

> 解释：优先信任本地持久化历史，外部 history 作为补充。

### 7.3 优先级与冲突处理
- 若 `H_ext` 与 `H_local` 在末尾存在重复但内容略有差异：
  - MVP 采取“以 local 为准”策略（保持服务端一致性）。
  - 将差异记录到 diagnostics（可选）供后续排查。

### 7.4 截断策略（控制 token）
合并后历史 `H_merged` 需受 token 预算约束：
- 预算 = `maxHistoryTokens`（可配置）。
- 截断策略：保留最近的 N 条（从尾部向前累加 token 直到阈值）。
- 可选增强（后续）：对更早历史做摘要并作为系统记忆注入。

---

## 8. 关键接口（内部）与可插拔点

### 8.1 Orchestrator 依赖接口（概念）
- `MemoryStore`
  - `load_history(tenantId, sessionId) -> messages[]`
  - `append_messages(tenantId, sessionId, messages[])`
- `Retriever`
  - `retrieve(tenantId, query, metadata) -> RetrievalResult`
- `LLMClient`
  - `generate(prompt, params) -> text`
  - `stream_generate(prompt, params) -> iterator[delta]`

### 8.2 插件点
- Retrieval：VectorRetriever / GraphRetriever / HybridRetriever
- LLM：OpenAICompatibleClient / LocalModelClient
- ConfidencePolicy：可替换策略（基于检索质量 + 模型信号）

---

## 9. 风险与后续工作
- SSE 的网关兼容性：需确认网关是否支持 `text/event-stream` 透传与超时策略。
- 租户级 collection 数量增长：若租户数量巨大，Qdrant collection 管理成本上升；可在规模化阶段切换为“单 collection + payload tenant filter”并加强隔离校验。
- 上下文膨胀：仅截断可能影响长会话体验；后续可引入摘要记忆与检索式记忆。
- 置信度定义：MVP 先以规则/阈值实现，后续引入离线评测与校准。

---

## 10. v0.6.0 智能客服增强 — 总体架构升级

### 10.1 升级后的 Orchestrator 数据流

原有 8 步 pipeline 升级为 12 步，新增步骤用 `[NEW]` 标记：

```
API 层接收请求 → 提取 tenantId + body → 调用 Orchestrator

Orchestrator:
  1)  Memory.load(tenantId, sessionId)
  2)  merge_context(local_history, external_history)
  3)  [NEW] InputGuardrail.scan(currentMessage)           → 前置禁词检测（仅记录，不阻断）
  4)  [NEW] FlowEngine.check_active_flow(sessionId)       → 检查是否有进行中的话术流程
      ├─ 有活跃流程 → FlowEngine.advance(user_input) → 返回话术内容 → 跳到步骤 11
      └─ 无活跃流程 → 继续步骤 5
  5)  [NEW] IntentRouter.match(currentMessage, tenantId)   → 意图识别（关键词+正则）
      ├─ fixed  → 返回固定回复 → 跳到步骤 11
      ├─ flow   → FlowEngine.start(flowId, sessionId) → 返回首步话术 → 跳到步骤 11
      ├─ transfer → shouldTransfer=true + 转人工话术 → 跳到步骤 11
      ├─ rag    → 设置 target_kb_ids → 继续步骤 6
      └─ 未命中 → target_kb_ids=按优先级全部 → 继续步骤 6
  6)  [NEW] QueryRewriter.rewrite(currentMessage, history) → Query 改写（LLM 调用，解析指代词）
  7)  Retrieval.retrieve(rewritten_query, tenantId, target_kb_ids)  → 多知识库定向检索
  8)  [NEW] ResultRanker.rank(hits, kb_priorities)         → 分层排序（按知识库类型优先级）
  9)  [NEW] PromptBuilder.build(template, evidence, history, message) → 从数据库模板构建 Prompt
  10) LLM.generate(messages) 或 LLM.stream_generate(messages)
  11) [NEW] OutputGuardrail.filter(reply)                  → 后置禁词过滤（mask/replace/block）
  12) compute_confidence(retrieval_result)
  13) Memory.append(tenantId, sessionId, user + assistant messages)
  14) 返回 ChatResponse
```

### 10.2 新增模块与现有模块的关系

```
app/
├── services/
│   ├── orchestrator.py          # [修改] 升级为 12 步 pipeline
│   ├── prompt/                  # [新增] Prompt 模板服务
│   │   ├── template_service.py  # 模板 CRUD + 版本管理 + 缓存
│   │   └── variable_resolver.py # 变量替换引擎
│   ├── intent/                  # [新增] 意图识别与路由
│   │   ├── router.py            # IntentRouter：规则匹配引擎
│   │   └── rule_service.py      # 规则 CRUD
│   ├── flow/                    # [新增] 话术流程引擎
│   │   ├── engine.py            # FlowEngine：状态机执行
│   │   └── flow_service.py      # 流程 CRUD
│   ├── guardrail/               # [新增] 输出护栏
│   │   ├── input_scanner.py     # 输入前置检测
│   │   ├── output_filter.py     # 输出后置过滤
│   │   └── word_service.py      # 禁词/行为规则 CRUD
│   ├── retrieval/
│   │   ├── optimized_retriever.py  # [修改] 支持 target_kb_ids 参数
│   │   ├── query_rewriter.py       # [新增] Query 改写
│   │   └── result_ranker.py        # [新增] 分层排序
│   ├── kb.py                    # [修改] 支持多知识库 CRUD
│   └── ...（现有模块不变）
├── api/
│   └── admin/
│       ├── prompt_templates.py  # [新增] Prompt 模板管理 API
│       ├── intent_rules.py      # [新增] 意图规则管理 API
│       ├── script_flows.py      # [新增] 话术流程管理 API
│       ├── guardrails.py        # [新增] 护栏管理 API（禁词+行为规则）
│       ├── kb.py                # [修改] 新增知识库 CRUD 端点
│       └── ...（现有 API 不变）
├── models/
│   └── entities.py              # [修改] 新增实体定义
└── core/
    └── prompts.py               # [修改] 改为从数据库加载，保留硬编码作为 fallback
```

---

## 11. Prompt 模板系统设计

### 11.1 数据模型

```
prompt_templates 表
├── id: UUID (PK)
├── tenant_id: VARCHAR (NOT NULL, FK)
├── name: VARCHAR (模板名称，如"默认客服人设")
├── scene: VARCHAR (场景标签：chat/rag_qa/greeting/farewell)
├── description: TEXT (模板描述)
├── is_default: BOOLEAN (是否为该场景的默认模板)
├── created_at: TIMESTAMP
├── updated_at: TIMESTAMP
└── INDEX: (tenant_id, scene)

prompt_template_versions 表
├── id: UUID (PK)
├── template_id: UUID (FK → prompt_templates.id)
├── version: INTEGER (自增版本号)
├── status: VARCHAR (draft/published/archived)
├── system_instruction: TEXT (系统指令内容，支持 {{variable}} 占位符)
├── variables: JSONB (变量定义列表，如 [{"name":"persona_name","default":"小N","description":"人设名称"}])
├── created_at: TIMESTAMP
└── INDEX: (template_id, status)
└── UNIQUE: 同一 template_id 下仅一个 status=published
```

### 11.2 变量替换引擎

内置变量（系统自动注入，无需用户定义）：

| 变量 | 说明 | 示例值 |
|------|------|--------|
| `{{persona_name}}` | 人设名称 | 小N |
| `{{current_time}}` | 当前时间 | 2026-02-27 14:30 |
| `{{channel_type}}` | 渠道类型 | wechat |
| `{{tenant_name}}` | 租户名称 | 某某公司 |
| `{{session_id}}` | 会话ID | kf_001_wx123 |

自定义变量：由模板定义，管理员在模板中声明变量名和默认值。

替换流程：
1. 加载已发布版本的 `system_instruction`
2. 合并内置变量 + 自定义变量默认值
3. 执行 `{{variable}}` 模式替换
4. 注入行为规则（从 guardrails 加载，追加到系统指令末尾）
5. 输出最终 system message

### 11.3 缓存策略

- 使用内存缓存（dict），key = `(tenant_id, scene)`，value = 已发布版本的完整模板
- 发布/回滚操作时主动失效缓存
- 缓存 TTL = 300s（兜底过期，防止分布式场景下缓存不一致）
- fallback：缓存未命中且数据库无模板时，使用现有硬编码的 `SYSTEM_PROMPT` 作为兜底

---

## 12. 多知识库设计

### 12.1 数据模型

```
knowledge_bases 表（扩展现有 KnowledgeBase 实体）
├── id: VARCHAR (PK, 如 "kb_product_001")
├── tenant_id: VARCHAR (NOT NULL)
├── name: VARCHAR (知识库名称)
├── kb_type: VARCHAR (product/faq/script/policy/general)
├── description: TEXT
├── priority: INTEGER (优先级权重，数值越大越优先，默认 0)
├── is_enabled: BOOLEAN (默认 true)
├── doc_count: INTEGER (文档数量，冗余统计)
├── created_at: TIMESTAMP
├── updated_at: TIMESTAMP
└── INDEX: (tenant_id, kb_type)
```

### 12.2 Qdrant Collection 命名升级

现有：`kb_{tenant_id}`（单 collection）

升级为：`kb_{tenant_id}_{kb_id}`（每个知识库独立 collection）

兼容策略：
- 新创建的知识库使用新命名
- 现有 `kb_{tenant_id}` collection 映射为 `kb_default` 知识库（自动迁移）
- 检索时如果 target_kb_ids 包含 `kb_default`，同时搜索新旧两种命名的 collection

### 12.3 多知识库检索流程

```
target_kb_ids（来自意图路由或默认全部）
  → 按 kb_type 优先级排序：script > faq > product > policy > general
  → 并行检索各 collection（使用现有 OptimizedRetriever）
  → 合并结果，按 (kb_type_priority, score) 双维度排序
  → 截断到 maxEvidenceTokens
  → 输出 ranked_hits
```

---

## 13. 意图识别与规则引擎设计

### 13.1 数据模型

```
intent_rules 表
├── id: UUID (PK)
├── tenant_id: VARCHAR (NOT NULL)
├── name: VARCHAR (意图名称，如"退货意图")
├── keywords: JSONB (关键词列表，如 ["退货","退款","不想要了"])
├── patterns: JSONB (正则模式列表，如 ["退.*货","怎么退"])
├── priority: INTEGER (优先级，数值越大越先匹配)
├── response_type: VARCHAR (flow/rag/fixed/transfer)
├── target_kb_ids: JSONB (rag 类型时关联的知识库 ID 列表)
├── flow_id: UUID (flow 类型时关联的流程 ID)
├── fixed_reply: TEXT (fixed 类型时的固定回复内容)
├── transfer_message: TEXT (transfer 类型时的转人工话术)
├── is_enabled: BOOLEAN (默认 true)
├── hit_count: BIGINT (命中统计，默认 0)
├── created_at: TIMESTAMP
├── updated_at: TIMESTAMP
└── INDEX: (tenant_id, is_enabled, priority DESC)
```

### 13.2 匹配算法

```python
class IntentRouter:
    def match(self, message: str, tenant_id: str) -> Optional[IntentMatchResult]:
        rules = self._load_rules(tenant_id)  # 按 priority DESC 排序，已缓存
        for rule in rules:
            if not rule.is_enabled:
                continue
            # 1. 关键词匹配（任一命中即匹配）
            for keyword in rule.keywords:
                if keyword in message:
                    return IntentMatchResult(rule=rule, match_type="keyword", matched=keyword)
            # 2. 正则匹配（任一命中即匹配）
            for pattern in rule.patterns:
                if re.search(pattern, message):
                    return IntentMatchResult(rule=rule, match_type="regex", matched=pattern)
        return None  # 未命中，走默认 RAG
```

### 13.3 缓存策略

- 规则列表按 `tenant_id` 缓存在内存中
- 规则 CRUD 操作时主动失效缓存
- 缓存 TTL = 60s

---

## 14. 话术流程引擎设计

### 14.1 数据模型

```
script_flows 表
├── id: UUID (PK)
├── tenant_id: VARCHAR (NOT NULL)
├── name: VARCHAR (流程名称)
├── description: TEXT
├── steps: JSONB (步骤列表，见下方结构)
├── is_enabled: BOOLEAN (默认 true)
├── created_at: TIMESTAMP
├── updated_at: TIMESTAMP
└── INDEX: (tenant_id)

steps JSONB 结构:
[
  {
    "step_no": 1,
    "content": "您好，请问您的订单号是多少？",
    "wait_input": true,
    "timeout_seconds": 120,
    "timeout_action": "repeat",  // repeat/skip/transfer
    "next_conditions": [
      {"keywords": ["不知道","忘了"], "goto_step": 3},
      {"pattern": "\\d{10,}", "goto_step": 2}
    ],
    "default_next": 2  // 无条件匹配时的下一步
  },
  ...
]

flow_instances 表（运行时状态）
├── id: UUID (PK)
├── tenant_id: VARCHAR (NOT NULL)
├── session_id: VARCHAR (NOT NULL)
├── flow_id: UUID (FK → script_flows.id)
├── current_step: INTEGER (当前步骤序号)
├── status: VARCHAR (active/completed/timeout/cancelled)
├── context: JSONB (流程执行上下文，存储用户输入等)
├── started_at: TIMESTAMP
├── updated_at: TIMESTAMP
├── completed_at: TIMESTAMP (nullable)
└── UNIQUE: (tenant_id, session_id, status='active')  -- 同一会话同时只有一个活跃流程
```

### 14.2 状态机

```
                    ┌─────────────┐
                    │   IDLE      │ (无活跃流程)
                    └──────┬──────┘
                           │ 意图命中 flow 规则
                           ▼
                    ┌─────────────┐
              ┌────►│   ACTIVE    │◄────┐
              │     └──────┬──────┘     │
              │            │            │
              │     用户输入匹配条件     │ 用户输入不匹配
              │            │            │ → 重复当前步骤
              │            ▼            │
              │     推进到下一步 ────────┘
              │            │
              │     到达最后一步
              │            │
              │            ▼
              │     ┌─────────────┐
              │     │  COMPLETED  │
              │     └─────────────┘
              │
              │     超时 / 用户触发退出
              │            │
              │            ▼
              │     ┌─────────────┐
              └─────│  TIMEOUT /  │
                    │  CANCELLED  │
                    └─────────────┘
```

### 14.3 FlowEngine 核心逻辑

```python
class FlowEngine:
    async def check_active_flow(self, tenant_id: str, session_id: str) -> Optional[FlowInstance]:
        """检查会话是否有活跃流程"""
        return await self.repo.get_active_instance(tenant_id, session_id)

    async def start(self, flow_id: str, tenant_id: str, session_id: str) -> str:
        """启动流程，返回第一步话术"""
        flow = await self.repo.get_flow(flow_id)
        instance = FlowInstance(flow_id=flow_id, session_id=session_id, current_step=1, status="active")
        await self.repo.save_instance(instance)
        return flow.steps[0]["content"]

    async def advance(self, instance: FlowInstance, user_input: str) -> FlowAdvanceResult:
        """根据用户输入推进流程"""
        flow = await self.repo.get_flow(instance.flow_id)
        current = flow.steps[instance.current_step - 1]

        # 匹配下一步条件
        next_step = self._match_next(current, user_input)

        if next_step > len(flow.steps):
            # 流程结束
            instance.status = "completed"
            await self.repo.save_instance(instance)
            return FlowAdvanceResult(completed=True, message=None)

        instance.current_step = next_step
        await self.repo.save_instance(instance)
        return FlowAdvanceResult(completed=False, message=flow.steps[next_step - 1]["content"])
```

---

## 15. 输出护栏设计

### 15.1 数据模型

```
forbidden_words 表
├── id: UUID (PK)
├── tenant_id: VARCHAR (NOT NULL)
├── word: VARCHAR (禁词)
├── category: VARCHAR (competitor/sensitive/political/custom)
├── strategy: VARCHAR (mask/replace/block)
├── replacement: TEXT (replace 策略时的替换文本)
├── fallback_reply: TEXT (block 策略时的兜底话术)
├── is_enabled: BOOLEAN (默认 true)
├── hit_count: BIGINT (命中统计，默认 0)
├── created_at: TIMESTAMP
├── updated_at: TIMESTAMP
└── INDEX: (tenant_id, is_enabled)

behavior_rules 表
├── id: UUID (PK)
├── tenant_id: VARCHAR (NOT NULL)
├── rule_text: TEXT (行为约束描述，如"不允许承诺具体赔偿金额")
├── category: VARCHAR (compliance/tone/boundary/custom)
├── is_enabled: BOOLEAN (默认 true)
├── created_at: TIMESTAMP
├── updated_at: TIMESTAMP
└── INDEX: (tenant_id, is_enabled)
```

### 15.2 输出过滤流程

```python
class OutputGuardrail:
    async def filter(self, reply: str, tenant_id: str) -> GuardrailResult:
        words = self._load_words(tenant_id)  # 已缓存
        triggered = []
        filtered_reply = reply

        for word in words:
            if not word.is_enabled:
                continue
            if word.word in filtered_reply:
                triggered.append(word)
                if word.strategy == "block":
                    # 整条拦截，返回兜底话术
                    return GuardrailResult(
                        reply=word.fallback_reply or "抱歉，让我换个方式回答您",
                        blocked=True,
                        triggered_words=[w.word for w in triggered]
                    )
                elif word.strategy == "mask":
                    filtered_reply = filtered_reply.replace(word.word, "*" * len(word.word))
                elif word.strategy == "replace":
                    filtered_reply = filtered_reply.replace(word.word, word.replacement)

        return GuardrailResult(
            reply=filtered_reply,
            blocked=False,
            triggered_words=[w.word for w in triggered]
        )
```

### 15.3 Streaming 模式下的护栏处理

SSE 流式输出时，禁词过滤需要特殊处理：

- 维护一个滑动窗口缓冲区（buffer），大小 = 最长禁词长度
- 每次收到 LLM delta 时追加到 buffer
- 当 buffer 长度超过窗口大小时，对已确认安全的前缀执行输出
- 在 `final` 事件前对剩余 buffer 做最终检查
- `block` 策略在流式模式下：检测到禁词后立即停止输出，发送 `error` 事件并附带兜底话术

### 15.4 行为规则注入

行为规则不做运行时检测，而是注入到 Prompt 中作为 LLM 的行为约束：

```
[系统指令]
{模板内容}

[行为约束 - 以下规则必须严格遵守]
1. 不允许承诺具体赔偿金额
2. 不允许透露内部流程
3. 不允许评价竞品
...
```

---

## 16. 智能 RAG 增强设计

### 16.1 Query 改写

```python
class QueryRewriter:
    REWRITE_PROMPT = """根据对话历史，改写用户的最新问题，使其语义完整、适合知识库检索。
规则：
- 解析指代词（"它"、"这个"等），替换为具体实体
- 补全省略的主语或宾语
- 保持原意，不添加额外信息
- 如果问题已经足够清晰，直接返回原文

对话历史：
{history}

用户最新问题：{query}

改写后的检索查询："""

    async def rewrite(self, query: str, history: list, llm_client: LLMClient) -> str:
        if not history or len(history) < 2:
            return query  # 无历史或历史太短，不改写
        messages = [{"role": "user", "content": self.REWRITE_PROMPT.format(
            history=self._format_history(history[-6:]),  # 最近 3 轮
            query=query
        )}]
        result = await llm_client.generate(messages, max_tokens=100, temperature=0)
        return result.content.strip() or query
```

### 16.2 分层排序

```python
KB_TYPE_PRIORITY = {
    "script": 50,   # 话术模板最高
    "faq": 40,      # FAQ 次之
    "product": 30,  # 产品知识
    "policy": 20,   # 政策规范
    "general": 10,  # 通用最低
}

class ResultRanker:
    def rank(self, hits: list[RetrievalHit], kb_map: dict[str, KnowledgeBase]) -> list[RetrievalHit]:
        """按 (kb_type_priority DESC, score DESC) 双维度排序"""
        def sort_key(hit):
            kb = kb_map.get(hit.kb_id)
            type_priority = KB_TYPE_PRIORITY.get(kb.kb_type, 0) if kb else 0
            custom_priority = kb.priority if kb else 0
            return (-(type_priority + custom_priority), -hit.score)
        return sorted(hits, key=sort_key)
```

---

## 17. 新增数据库实体汇总

v0.6.0 新增以下 SQLModel 实体（均包含 `tenant_id` 字段，遵循现有多租户隔离模式）：

| 实体 | 表名 | 用途 |
|------|------|------|
| PromptTemplate | prompt_templates | Prompt 模板主表 |
| PromptTemplateVersion | prompt_template_versions | 模板版本表 |
| KnowledgeBase（扩展） | knowledge_bases | 知识库主表（新增 kb_type/priority/is_enabled） |
| IntentRule | intent_rules | 意图规则表 |
| ScriptFlow | script_flows | 话术流程定义表 |
| FlowInstance | flow_instances | 流程运行实例表 |
| ForbiddenWord | forbidden_words | 禁词表 |
| BehaviorRule | behavior_rules | 行为规则表 |

---

## 18. v0.6.0 风险与待澄清

- Query 改写的 LLM 调用会增加约 0.5-1s 延迟和额外 token 消耗；可通过配置开关控制是否启用。
- 流式模式下的禁词滑动窗口可能导致输出延迟（等待 buffer 填满）；需要在实时性和安全性之间权衡窗口大小。
- 多知识库并行检索会增加 Qdrant 负载；需要评估并发 collection 搜索的性能影响。
- 话术流程的超时检测依赖调用方（Java 侧）触发；需要与 Java 侧约定超时回调机制。
- 现有 `kb_default` 到多知识库的数据迁移需要平滑过渡，不能中断现有服务。

---

## 19. v0.7.0 测试与监控系统设计

### 19.1 设计目标与范围

#### 19.1.1 核心目标
- **可测试性**：为 v0.6.0 新增的四大功能（Prompt 模板、意图规则、话术流程、输出护栏）提供独立测试能力。
- **可观测性**：提供细粒度的运行时监控数据，支持规则命中率、流程执行状态、护栏拦截统计等。
- **可追溯性**：完整记录对话流程的 12 步执行细节，支持问题排查与效果评估。
- **可导出性**：支持对话数据导出，便于离线分析与模型优化。

#### 19.1.2 设计约束
- **性能优先**：监控数据采集不能显著影响对话生成性能（目标：<5% 延迟增加）。
- **存储可控**：完整流程测试的详细日志仅保留 7 天，避免存储膨胀。
- **租户隔离**：所有测试与监控数据必须按 `tenant_id` 隔离。
- **向后兼容**：新增监控不影响现有 `/ai/chat` 接口的行为与性能。

---

### 19.2 总体架构

#### 19.2.1 监控数据流

```
┌─────────────────────────────────────────────────────────────┐
│                    Admin API Layer                          │
│  /admin/test/*  /admin/monitoring/*  /admin/dashboard/*    │
└────────────┬────────────────────────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────────────────────────┐
│                  Monitoring Service Layer                    │
│  ├─ FlowTestService      (完整流程测试)                      │
│  ├─ IntentMonitor        (意图规则监控)                      │
│  ├─ PromptMonitor        (Prompt 模板监控)                   │
│  ├─ FlowMonitor          (话术流程监控)                      │
│  ├─ GuardrailMonitor     (护栏监控)                          │
│  └─ ConversationTracker  (对话追踪)                          │
└────────────┬────────────────────────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────────────────────────┐
│                    Orchestrator (增强)                       │
│  12-step pipeline + 监控埋点 (可选开关)                      │
└────────────┬────────────────────────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────────────────────────┐
│                  Data Storage Layer                          │
│  ├─ PostgreSQL (统计数据、对话记录)                          │
│  └─ Redis (缓存、实时统计)                                   │
└─────────────────────────────────────────────────────────────┘
```

#### 19.2.2 监控模式

**生产模式（默认）**：
- 仅记录关键指标（命中次数、错误率、平均延迟）
- 不记录详细的步骤执行日志
- 性能影响 <2%

**测试模式（显式开启）**：
- 记录完整的 12 步执行细节
- 包含每步的输入/输出/耗时/错误
- 仅用于 `/admin/test/*` 端点
- 数据保留 7 天

---

### 19.3 Dashboard 统计增强设计

#### 19.3.1 新增统计指标（对应 AC-AISVC-91, AC-AISVC-92）

在现有 `GET /admin/dashboard/stats` 响应中新增以下字段：

```python
# 意图规则统计
intentRuleHitRate: float          # 命中率 = 命中次数 / 总对话次数
intentRuleHitCount: int           # 总命中次数
intentRuleTopHits: list[dict]     # Top 5 命中规则 [{"ruleId", "name", "hitCount"}]

# Prompt 模板统计
promptTemplateUsageCount: int     # 模板使用总次数
promptTemplateActiveCount: int    # 已发布模板数量
promptTemplateTopUsed: list[dict] # Top 5 使用模板 [{"templateId", "name", "usageCount"}]

# 话术流程统计
scriptFlowActivationCount: int    # 流程激活总次数
scriptFlowCompletionRate: float   # 完成率 = 完成次数 / 激活次数
scriptFlowTopActive: list[dict]   # Top 5 活跃流程 [{"flowId", "name", "activationCount"}]

# 护栏统计
guardrailBlockCount: int          # 拦截总次数
guardrailBlockRate: float         # 拦截率 = 拦截次数 / 总对话次数
guardrailTopWords: list[dict]     # Top 5 触发禁词 [{"word", "category", "hitCount"}]
```

#### 19.3.2 数据来源与计算

**实时统计（Redis）**：
- 使用 Redis Hash 存储租户级计数器
- Key 格式：`stats:{tenant_id}:{metric}:{date}`
- 每次对话结束时异步更新（不阻塞响应）
- TTL = 90 天

**聚合统计（PostgreSQL）**：
- 从现有表的 `hit_count` 字段聚合（intent_rules, forbidden_words）
- 从 `flow_instances` 表统计流程激活与完成
- 从 `chat_messages` 表关联 `prompt_template_id`（需新增字段）

#### 19.3.3 性能优化

- Dashboard 统计结果缓存 60 秒（Redis）
- Top N 排行榜每 5 分钟预计算一次（后台任务）
- 避免实时聚合大表，使用增量计数器

---

### 19.4 完整流程测试台设计

#### 19.4.1 测试接口（对应 AC-AISVC-93 ~ AC-AISVC-96）

**端点**：`POST /admin/test/flow-execution`

**请求体**：
```json
{
  "message": "我想退货",
  "sessionId": "test_session_001",
  "channelType": "wechat",
  "history": [...],  // 可选
  "metadata": {...}, // 可选
  "enableDetailedLog": true  // 是否记录详细日志
}
```

**响应体**：
```json
{
  "executionId": "exec_uuid",
  "steps": [
    {
      "step": 1,
      "name": "Memory.load",
      "status": "success",
      "durationMs": 12,
      "input": {"sessionId": "test_session_001"},
      "output": {"messageCount": 5},
      "error": null,
      "metadata": {}
    },
    {
      "step": 3,
      "name": "InputGuardrail.scan",
      "status": "success",
      "durationMs": 8,
      "input": {"message": "我想退货"},
      "output": {"triggered": false, "words": []},
      "error": null,
      "metadata": {}
    },
    {
      "step": 5,
      "name": "IntentRouter.match",
      "status": "success",
      "durationMs": 15,
      "input": {"message": "我想退货"},
      "output": {
        "matched": true,
        "ruleId": "rule_001",
        "ruleName": "退货意图",
        "responseType": "flow",
        "flowId": "flow_return_001"
      },
      "error": null,
      "metadata": {"priority": 100, "matchType": "keyword"}
    },
    // ... 其他步骤
  ],
  "finalResponse": {
    "reply": "您好，请问您的订单号是多少？",
    "confidence": 0.95,
    "shouldTransfer": false
  },
  "totalDurationMs": 1250,
  "summary": {
    "successSteps": 11,
    "failedSteps": 0,
    "skippedSteps": 1
  }
}
```

#### 19.4.2 实现策略

**Orchestrator 增强**：
```python
class Orchestrator:
    async def generate_with_monitoring(
        self,
        request: ChatRequest,
        tenant_id: str,
        enable_detailed_log: bool = False
    ) -> tuple[ChatResponse, Optional[list[StepLog]]]:
        """
        增强版生成方法，支持可选的详细日志记录
        """
        step_logs = [] if enable_detailed_log else None

        # Step 1: Memory.load
        step_start = time.time()
        try:
            history = await self.memory.load_history(tenant_id, request.sessionId)
            if step_logs is not None:
                step_logs.append(StepLog(
                    step=1,
                    name="Memory.load",
                    status="success",
                    durationMs=int((time.time() - step_start) * 1000),
                    input={"sessionId": request.sessionId},
                    output={"messageCount": len(history)}
                ))
        except Exception as e:
            if step_logs is not None:
                step_logs.append(StepLog(
                    step=1,
                    name="Memory.load",
                    status="failed",
                    durationMs=int((time.time() - step_start) * 1000),
                    error=str(e)
                ))
            raise

        # ... 其他步骤类似

        return response, step_logs
```

**测试端点实现**：
```python
@router.post("/admin/test/flow-execution")
async def test_flow_execution(
    request: FlowTestRequest,
    tenant_id: str = Depends(get_current_tenant_id),
    session: AsyncSession = Depends(get_session)
):
    # 调用增强版 Orchestrator
    response, step_logs = await orchestrator.generate_with_monitoring(
        ChatRequest(
            message=request.message,
            sessionId=request.sessionId,
            channelType=request.channelType,
            history=request.history,
            metadata=request.metadata
        ),
        tenant_id=tenant_id,
        enable_detailed_log=request.enableDetailedLog
    )

    # 保存测试记录（可选，用于历史查询）
    if request.enableDetailedLog:
        test_record = FlowTestRecord(
            tenant_id=tenant_id,
            session_id=request.sessionId,
            steps=step_logs,
            final_response=response,
            created_at=datetime.utcnow()
        )
        await save_test_record(session, test_record)

    return FlowExecutionResult(
        executionId=str(uuid.uuid4()),
        steps=step_logs,
        finalResponse=response,
        totalDurationMs=sum(s.durationMs for s in step_logs),
        summary={
            "successSteps": sum(1 for s in step_logs if s.status == "success"),
            "failedSteps": sum(1 for s in step_logs if s.status == "failed"),
            "skippedSteps": sum(1 for s in step_logs if s.status == "skipped")
        }
    )
```

---

### 19.5 意图规则测试与监控设计

#### 19.5.1 独立测试接口（对应 AC-AISVC-97 ~ AC-AISVC-99）

**端点**：`POST /admin/intent-rules/{ruleId}/test`

**请求体**：
```json
{
  "testMessages": [
    "我想退货",
    "能退款吗",
    "这个产品怎么样"
  ]
}
```

**响应体**：
```json
{
  "ruleId": "rule_001",
  "ruleName": "退货意图",
  "results": [
    {
      "message": "我想退货",
      "matched": true,
      "matchedKeywords": ["退货"],
      "matchedPatterns": [],
      "matchType": "keyword",
      "priority": 100,
      "conflictRules": []
    },
    {
      "message": "能退款吗",
      "matched": true,
      "matchedKeywords": ["退款"],
      "matchedPatterns": [],
      "matchType": "keyword",
      "priority": 100,
      "conflictRules": []
    },
    {
      "message": "这个产品怎么样",
      "matched": false,
      "matchedKeywords": [],
      "matchedPatterns": [],
      "matchType": null,
      "priority": 100,
      "conflictRules": [
        {
          "ruleId": "rule_002",
          "ruleName": "产品咨询",
          "priority": 80,
          "reason": "可能匹配产品咨询规则"
        }
      ]
    }
  ],
  "summary": {
    "totalTests": 3,
    "matchedCount": 2,
    "matchRate": 0.67
  }
}
```

#### 19.5.2 冲突检测算法

```python
class IntentRuleTester:
    async def test_rule(
        self,
        rule: IntentRule,
        test_messages: list[str],
        tenant_id: str
    ) -> IntentRuleTestResult:
        """测试意图规则并检测冲突"""
        all_rules = await self.rule_service.get_rules(tenant_id)
        results = []

        for message in test_messages:
            # 测试当前规则
            matched = self._match_rule(rule, message)

            # 检测冲突：查找其他也能匹配的规则
            conflict_rules = []
            for other_rule in all_rules:
                if other_rule.id == rule.id:
                    continue
                if self._match_rule(other_rule, message):
                    conflict_rules.append({
                        "ruleId": other_rule.id,
                        "ruleName": other_rule.name,
                        "priority": other_rule.priority,
                        "reason": f"同时匹配（优先级：{other_rule.priority}）"
                    })

            results.append(IntentRuleTestCase(
                message=message,
                matched=matched,
                conflictRules=conflict_rules
            ))

        return IntentRuleTestResult(
            ruleId=rule.id,
            ruleName=rule.name,
            results=results,
            summary={
                "totalTests": len(test_messages),
                "matchedCount": sum(1 for r in results if r.matched),
                "matchRate": sum(1 for r in results if r.matched) / len(test_messages)
            }
        )
```

#### 19.5.3 监控统计接口（对应 AC-AISVC-100）

**端点**：`GET /admin/monitoring/intent-rules`

**查询参数**：
- `startDate`: 开始日期（ISO 8601）
- `endDate`: 结束日期（ISO 8601）
- `limit`: 返回数量（默认 10）

**响应体**：
```json
{
  "totalHits": 1250,
  "totalConversations": 5000,
  "hitRate": 0.25,
  "rules": [
    {
      "ruleId": "rule_001",
      "ruleName": "退货意图",
      "hitCount": 450,
      "hitRate": 0.09,
      "responseType": "flow",
      "avgResponseTime": 1200,
      "lastHitAt": "2026-02-27T14:30:00Z"
    }
  ],
  "timeSeriesData": [
    {
      "date": "2026-02-27",
      "totalHits": 120,
      "ruleBreakdown": {
        "rule_001": 45,
        "rule_002": 30,
        "rule_003": 45
      }
    }
  ]
}
```

---

### 19.6 Prompt 模板测试与监控设计

#### 19.6.1 模板预览接口（对应 AC-AISVC-101）

**端点**：`POST /admin/prompt-templates/{templateId}/preview`

**请求体**：
```json
{
  "variables": {
    "persona_name": "小助手",
    "custom_var": "测试值"
  },
  "sampleHistory": [
    {"role": "user", "content": "你好"},
    {"role": "assistant", "content": "您好，有什么可以帮您？"}
  ],
  "sampleMessage": "我想了解产品信息"
}
```

**响应体**：
```json
{
  "templateId": "tpl_001",
  "templateName": "默认客服人设",
  "version": 3,
  "renderedSystemPrompt": "你是小助手，一个专业的客服助手...\n\n[行为约束]\n1. 不允许承诺具体赔偿金额\n...",
  "finalMessages": [
    {
      "role": "system",
      "content": "你是小助手，一个专业的客服助手..."
    },
    {
      "role": "user",
      "content": "你好"
    },
    {
      "role": "assistant",
      "content": "您好，有什么可以帮您？"
    },
    {
      "role": "user",
      "content": "我想了解产品信息"
    }
  ],
  "tokenCount": {
    "systemPrompt": 450,
    "history": 120,
    "currentMessage": 30,
    "total": 600
  },
  "warnings": []
}
```

#### 19.6.2 模板使用统计（对应 AC-AISVC-102）

**端点**：`GET /admin/monitoring/prompt-templates`

**响应体**：
```json
{
  "totalUsage": 5000,
  "templates": [
    {
      "templateId": "tpl_001",
      "templateName": "默认客服人设",
      "scene": "chat",
      "usageCount": 3500,
      "usageRate": 0.70,
      "currentVersion": 3,
      "avgTokenCount": 450,
      "lastUsedAt": "2026-02-27T14:30:00Z"
    }
  ],
  "sceneBreakdown": {
    "chat": 3500,
    "rag_qa": 1200,
    "greeting": 300
  }
}
```

#### 19.6.3 实现策略

**Token 计数**：
```python
import tiktoken

class PromptTemplateMonitor:
    def __init__(self):
        self.tokenizer = tiktoken.get_encoding("cl100k_base")

    async def preview_template(
        self,
        template: PromptTemplate,
        variables: dict,
        sample_history: list[dict],
        sample_message: str
    ) -> PromptPreviewResult:
        """预览模板渲染结果并计算 token"""
        # 1. 渲染系统指令
        version = await self.get_published_version(template.id)
        system_prompt = self.variable_resolver.resolve(
            version.system_instruction,
            variables
        )

        # 2. 注入行为规则
        behavior_rules = await self.get_behavior_rules(template.tenant_id)
        if behavior_rules:
            system_prompt += "\n\n[行为约束]\n" + "\n".join(
                f"{i+1}. {rule.rule_text}"
                for i, rule in enumerate(behavior_rules)
            )

        # 3. 构建完整消息列表
        messages = [{"role": "system", "content": system_prompt}]
        messages.extend(sample_history)
        messages.append({"role": "user", "content": sample_message})

        # 4. 计算 token
        token_counts = {
            "systemPrompt": len(self.tokenizer.encode(system_prompt)),
            "history": sum(
                len(self.tokenizer.encode(msg["content"]))
                for msg in sample_history
            ),
            "currentMessage": len(self.tokenizer.encode(sample_message)),
        }
        token_counts["total"] = sum(token_counts.values())

        # 5. 检查警告
        warnings = []
        if token_counts["total"] > 4000:
            warnings.append("总 token 数超过 4000，可能影响性能")
        if token_counts["systemPrompt"] > 2000:
            warnings.append("系统指令过长，建议精简")

        return PromptPreviewResult(
            templateId=template.id,
            renderedSystemPrompt=system_prompt,
            finalMessages=messages,
            tokenCount=token_counts,
            warnings=warnings
        )
```

---

### 19.7 话术流程测试与监控设计

#### 19.7.1 流程模拟测试接口（对应 AC-AISVC-103 ~ AC-AISVC-105）

**端点**：`POST /admin/script-flows/{flowId}/simulate`

**请求体**：
```json
{
  "userInputs": [
    "12345678901234",
    "质量问题",
    "是的"
  ]
}
```

**响应体**：
```json
{
  "flowId": "flow_001",
  "flowName": "退货流程",
  "simulation": [
    {
      "stepNo": 1,
      "botMessage": "您好，请问您的订单号是多少？",
      "userInput": "12345678901234",
      "matchedCondition": {
        "type": "pattern",
        "pattern": "\\d{10,}",
        "gotoStep": 2
      },
      "nextStep": 2,
      "durationMs": 50
    },
    {
      "stepNo": 2,
      "botMessage": "请问退货原因是什么？",
      "userInput": "质量问题",
      "matchedCondition": {
        "type": "default",
        "gotoStep": 3
      },
      "nextStep": 3,
      "durationMs": 45
    },
    {
      "stepNo": 3,
      "botMessage": "已为您登记退货申请，是否需要上门取件？",
      "userInput": "是的",
      "matchedCondition": {
        "type": "keyword",
        "keywords": ["是", "需要"],
        "gotoStep": 4
      },
      "nextStep": 4,
      "durationMs": 40
    }
  ],
  "result": {
    "completed": true,
    "totalSteps": 3,
    "totalDurationMs": 135,
    "finalMessage": "好的，我们会在 24 小时内安排快递上门取件。"
  },
  "coverage": {
    "totalSteps": 5,
    "coveredSteps": 3,
    "coverageRate": 0.60,
    "uncoveredSteps": [4, 5]
  },
  "issues": [
    "流程覆盖率低于 80%，建议增加测试用例"
  ]
}
```

#### 19.7.2 流程覆盖率分析

```python
class ScriptFlowTester:
    async def simulate_flow(
        self,
        flow: ScriptFlow,
        user_inputs: list[str]
    ) -> FlowSimulationResult:
        """模拟流程执行并分析覆盖率"""
        simulation = []
        current_step = 1
        visited_steps = set()

        for user_input in user_inputs:
            if current_step > len(flow.steps):
                break

            step_def = flow.steps[current_step - 1]
            visited_steps.add(current_step)

            # 匹配下一步条件
            matched_condition, next_step = self._match_next_step(
                step_def,
                user_input
            )

            simulation.append(FlowSimulationStep(
                stepNo=current_step,
                botMessage=step_def["content"],
                userInput=user_input,
                matchedCondition=matched_condition,
                nextStep=next_step
            ))

            current_step = next_step

        # 分析覆盖率
        total_steps = len(flow.steps)
        covered_steps = len(visited_steps)
        coverage_rate = covered_steps / total_steps

        # 检测问题
        issues = []
        if coverage_rate < 0.8:
            issues.append("流程覆盖率低于 80%，建议增加测试用例")

        # 检测死循环
        if len(simulation) > total_steps * 2:
            issues.append("检测到可能的死循环")

        # 检测未覆盖的分支
        uncovered_steps = set(range(1, total_steps + 1)) - visited_steps
        if uncovered_steps:
            issues.append(f"未覆盖步骤：{sorted(uncovered_steps)}")

        return FlowSimulationResult(
            flowId=flow.id,
            simulation=simulation,
            coverage={
                "totalSteps": total_steps,
                "coveredSteps": covered_steps,
                "coverageRate": coverage_rate,
                "uncoveredSteps": list(uncovered_steps)
            },
            issues=issues
        )
```

#### 19.7.3 流程监控统计（对应 AC-AISVC-106）

**端点**：`GET /admin/monitoring/script-flows`

**响应体**：
```json
{
  "totalActivations": 850,
  "totalCompletions": 680,
  "completionRate": 0.80,
  "flows": [
    {
      "flowId": "flow_001",
      "flowName": "退货流程",
      "activationCount": 450,
      "completionCount": 380,
      "completionRate": 0.84,
      "avgDuration": 180,
      "avgStepsCompleted": 4.2,
      "dropOffPoints": [
        {
          "stepNo": 2,
          "dropOffCount": 50,
          "dropOffRate": 0.11
        }
      ],
      "lastActivatedAt": "2026-02-27T14:30:00Z"
    }
  ]
}
```

---

### 19.8 输出护栏测试与监控设计

#### 19.8.1 禁词测试接口（对应 AC-AISVC-107）

**端点**：`POST /admin/guardrails/forbidden-words/test`

**请求体**：
```json
{
  "testTexts": [
    "我们的产品比竞品 A 更好",
    "可以给您赔偿 1000 元",
    "这是正常的回复"
  ]
}
```

**响应体**：
```json
{
  "results": [
    {
      "originalText": "我们的产品比竞品 A 更好",
      "triggered": true,
      "triggeredWords": [
        {
          "word": "竞品 A",
          "category": "competitor",
          "strategy": "replace",
          "replacement": "其他品牌"
        }
      ],
      "filteredText": "我们的产品比其他品牌更好",
      "blocked": false
    },
    {
      "originalText": "可以给您赔偿 1000 元",
      "triggered": true,
      "triggeredWords": [
        {
          "word": "赔偿",
          "category": "sensitive",
          "strategy": "block",
          "fallbackReply": "关于补偿问题，请联系人工客服处理"
        }
      ],
      "filteredText": "关于补偿问题，请联系人工客服处理",
      "blocked": true
    },
    {
      "originalText": "这是正常的回复",
      "triggered": false,
      "triggeredWords": [],
      "filteredText": "这是正常的回复",
      "blocked": false
    }
  ],
  "summary": {
    "totalTests": 3,
    "triggeredCount": 2,
    "blockedCount": 1,
    "triggerRate": 0.67
  }
}
```

#### 19.8.2 护栏监控统计（对应 AC-AISVC-108）

**端点**：`GET /admin/monitoring/guardrails`

**响应体**：
```json
{
  "totalBlocks": 120,
  "totalTriggers": 450,
  "blockRate": 0.024,
  "words": [
    {
      "wordId": "word_001",
      "word": "竞品 A",
      "category": "competitor",
      "strategy": "replace",
      "hitCount": 85,
      "blockCount": 0,
      "lastHitAt": "2026-02-27T14:30:00Z"
    },
    {
      "wordId": "word_002",
      "word": "赔偿",
      "category": "sensitive",
      "strategy": "block",
      "hitCount": 45,
      "blockCount": 45,
      "lastHitAt": "2026-02-27T14:25:00Z"
    }
  ],
  "categoryBreakdown": {
    "competitor": 85,
    "sensitive": 45,
    "political": 0,
    "custom": 20
  }
}
```

---

### 19.9 对话追踪与导出设计

#### 19.9.1 对话追踪接口（对应 AC-AISVC-109）

**端点**：`GET /admin/monitoring/conversations`

**查询参数**：
- `startDate`: 开始日期（ISO 8601）
- `endDate`: 结束日期（ISO 8601）
- `sessionId`: 会话 ID（可选）
- `channelType`: 渠道类型（可选）
- `hasError`: 是否包含错误（可选）
- `limit`: 返回数量（默认 20）
- `offset`: 偏移量（默认 0）

**响应体**：
```json
{
  "total": 1250,
  "conversations": [
    {
      "sessionId": "kf_001_wx123",
      "channelType": "wechat",
      "messageCount": 12,
      "startTime": "2026-02-27T14:00:00Z",
      "lastMessageTime": "2026-02-27T14:15:00Z",
      "duration": 900,
      "intentRulesHit": [
        {"ruleId": "rule_001", "ruleName": "退货意图", "hitCount": 2}
      ],
      "flowsActivated": [
        {"flowId": "flow_001", "flowName": "退货流程", "status": "completed"}
      ],
      "guardrailTriggered": true,
      "errorCount": 0,
      "avgConfidence": 0.85,
      "transferRequested": false
    }
  ]
}
```

#### 19.9.2 对话详情接口

**端点**：`GET /admin/monitoring/conversations/{sessionId}`

**响应体**：
```json
{
  "sessionId": "kf_001_wx123",
  "channelType": "wechat",
  "startTime": "2026-02-27T14:00:00Z",
  "messages": [
    {
      "messageId": "msg_001",
      "role": "user",
      "content": "我想退货",
      "timestamp": "2026-02-27T14:00:00Z"
    },
    {
      "messageId": "msg_002",
      "role": "assistant",
      "content": "您好，请问您的订单号是多少？",
      "timestamp": "2026-02-27T14:00:02Z",
      "confidence": 0.95,
      "intentMatched": {
        "ruleId": "rule_001",
        "ruleName": "退货意图",
        "responseType": "flow"
      },
      "flowActivated": {
        "flowId": "flow_001",
        "flowName": "退货流程",
        "currentStep": 1
      },
      "guardrailResult": {
        "triggered": false,
        "words": []
      },
      "latencyMs": 1200,
      "totalTokens": 450,
      "promptTokens": 380,
      "completionTokens": 70
    }
  ],
  "summary": {
    "totalMessages": 12,
    "userMessages": 6,
    "assistantMessages": 6,
    "avgConfidence": 0.85,
    "avgLatency": 1150,
    "totalTokens": 5400,
    "intentRulesHit": 2,
    "flowsActivated": 1,
    "guardrailTriggered": false,
    "errorOccurred": false
  }
}
```

#### 19.9.3 对话导出接口（对应 AC-AISVC-110）

**端点**：`POST /admin/monitoring/conversations/export`

**请求体**：
```json
{
  "startDate": "2026-02-20T00:00:00Z",
  "endDate": "2026-02-27T23:59:59Z",
  "format": "csv",
  "filters": {
    "channelType": "wechat",
    "hasError": false,
    "minConfidence": 0.7
  },
  "fields": [
    "sessionId",
    "channelType",
    "messageCount",
    "avgConfidence",
    "intentRulesHit",
    "flowsActivated"
  ]
}
```

**响应体**：
```json
{
  "exportId": "export_uuid",
  "status": "processing",
  "estimatedRows": 1250,
  "downloadUrl": null,
  "expiresAt": null
}
```

**导出状态查询**：`GET /admin/monitoring/conversations/export/{exportId}`

**响应体**：
```json
{
  "exportId": "export_uuid",
  "status": "completed",
  "totalRows": 1250,
  "downloadUrl": "/admin/monitoring/conversations/export/export_uuid/download",
  "expiresAt": "2026-02-28T14:30:00Z",
  "createdAt": "2026-02-27T14:25:00Z"
}
```

#### 19.9.4 实现策略

**异步导出处理**：
```python
import asyncio
import csv
from io import StringIO

class ConversationExporter:
    async def export_conversations(
        self,
        tenant_id: str,
        filters: dict,
        fields: list[str],
        format: str = "csv"
    ) -> str:
        """异步导出对话数据"""
        export_id = str(uuid.uuid4())

        # 创建导出任务记录
        export_task = ExportTask(
            id=export_id,
            tenant_id=tenant_id,
            status="processing",
            created_at=datetime.utcnow()
        )
        await self.save_export_task(export_task)

        # 异步执行导出
        asyncio.create_task(self._process_export(
            export_id,
            tenant_id,
            filters,
            fields,
            format
        ))

        return export_id

    async def _process_export(
        self,
        export_id: str,
        tenant_id: str,
        filters: dict,
        fields: list[str],
        format: str
    ):
        """后台处理导出任务"""
        try:
            # 1. 查询对话数据（分批处理，避免内存溢出）
            batch_size = 1000
            offset = 0
            output = StringIO()
            writer = csv.DictWriter(output, fieldnames=fields)
            writer.writeheader()

            while True:
                conversations = await self.query_conversations(
                    tenant_id,
                    filters,
                    limit=batch_size,
                    offset=offset
                )

                if not conversations:
                    break

                for conv in conversations:
                    row = {field: conv.get(field) for field in fields}
                    writer.writerow(row)

                offset += batch_size

            # 2. 保存到临时文件
            file_path = f"/tmp/exports/{export_id}.csv"
            with open(file_path, "w", encoding="utf-8") as f:
                f.write(output.getvalue())

            # 3. 更新导出任务状态
            await self.update_export_task(
                export_id,
                status="completed",
                file_path=file_path,
                total_rows=offset,
                expires_at=datetime.utcnow() + timedelta(hours=24)
            )

        except Exception as e:
            logger.error(f"Export failed: {e}")
            await self.update_export_task(
                export_id,
                status="failed",
                error=str(e)
            )
```

---

### 19.10 新增数据库实体汇总

v0.7.0 新增以下 SQLModel 实体（均包含 `tenant_id` 字段，遵循现有多租户隔离模式）：

| 实体 | 表名 | 用途 | 关键字段 |
|------|------|------|----------|
| FlowTestRecord | flow_test_records | 完整流程测试记录 | session_id, steps (JSONB), final_response (JSONB) |
| ExportTask | export_tasks | 对话导出任务 | status, file_path, total_rows, expires_at |

**扩展现有实体**：

| 实体 | 表名 | 新增字段 | 用途 |
|------|------|----------|------|
| ChatMessage | chat_messages | prompt_template_id (UUID, nullable) | 关联使用的 Prompt 模板 |
| ChatMessage | chat_messages | intent_rule_id (UUID, nullable) | 关联命中的意图规则 |
| ChatMessage | chat_messages | flow_instance_id (UUID, nullable) | 关联的话术流程实例 |
| ChatMessage | chat_messages | guardrail_triggered (BOOLEAN) | 是否触发护栏 |
| ChatMessage | chat_messages | guardrail_words (JSONB, nullable) | 触发的禁词列表 |

**索引优化**：
```sql
-- 监控查询优化
CREATE INDEX idx_chat_messages_monitoring
ON chat_messages(tenant_id, created_at DESC, role);

-- 意图规则统计优化
CREATE INDEX idx_chat_messages_intent
ON chat_messages(tenant_id, intent_rule_id)
WHERE intent_rule_id IS NOT NULL;

-- 流程统计优化
CREATE INDEX idx_flow_instances_monitoring
ON flow_instances(tenant_id, status, started_at DESC);

-- 护栏统计优化
CREATE INDEX idx_chat_messages_guardrail
ON chat_messages(tenant_id, guardrail_triggered, created_at DESC)
WHERE guardrail_triggered = true;
```

---

### 19.11 性能优化策略

#### 19.11.1 缓存策略

**Redis 缓存层次**：
```
Level 1: Dashboard 统计（TTL 60s）
  - Key: stats:{tenant_id}:dashboard
  - 内容：聚合统计数据

Level 2: Top N 排行榜（TTL 300s）
  - Key: stats:{tenant_id}:top:intent_rules
  - Key: stats:{tenant_id}:top:prompt_templates
  - Key: stats:{tenant_id}:top:script_flows
  - Key: stats:{tenant_id}:top:forbidden_words

Level 3: 实时计数器（TTL 90天）
  - Key: stats:{tenant_id}:counter:{metric}:{date}
  - 内容：增量计数器
```

**缓存更新策略**：
- Dashboard 统计：每次对话结束后异步更新计数器
- Top N 排行榜：后台任务每 5 分钟重新计算
- 实时计数器：使用 Redis INCR 原子操作

#### 19.11.2 数据库优化

**分区策略**：
```sql
-- flow_test_records 按日期分区（保留 7 天）
CREATE TABLE flow_test_records (
    id UUID PRIMARY KEY,
    tenant_id VARCHAR NOT NULL,
    created_at TIMESTAMP NOT NULL,
    ...
) PARTITION BY RANGE (created_at);

CREATE TABLE flow_test_records_2026_02_27
PARTITION OF flow_test_records
FOR VALUES FROM ('2026-02-27') TO ('2026-02-28');

-- 自动清理过期分区（定时任务）
DROP TABLE IF EXISTS flow_test_records_2026_02_20;
```

**查询优化**：
- 使用覆盖索引减少回表查询
- 对大表使用 LIMIT + 游标分页
- 避免 SELECT *，只查询需要的字段
- 使用 EXPLAIN ANALYZE 分析慢查询

#### 19.11.3 监控埋点优化

**最小化性能影响**：
```python
class MonitoringMiddleware:
    async def __call__(self, request, call_next):
        # 仅在测试模式或采样时记录详细日志
        enable_detailed_log = (
            request.url.path.startswith("/admin/test/") or
            self._should_sample()  # 1% 采样率
        )

        if enable_detailed_log:
            # 记录详细步骤日志
            request.state.monitoring_enabled = True

        response = await call_next(request)

        # 异步更新统计（不阻塞响应）
        if hasattr(request.state, "monitoring_data"):
            asyncio.create_task(
                self._update_stats_async(request.state.monitoring_data)
            )

        return response

    def _should_sample(self) -> bool:
        """1% 采样率"""
        return random.random() < 0.01
```

---

### 19.12 v0.7.0 风险与待澄清

#### 19.12.1 性能风险

- **完整流程测试**：记录 12 步详细日志会增加 10-15% 的延迟，仅用于测试环境。
- **对话导出**：大批量导出（>10000 条）可能导致内存压力，需要流式处理。
- **实时统计**：高并发场景下 Redis 计数器可能成为瓶颈，考虑使用 Redis Cluster。

#### 19.12.2 存储风险

- **测试日志膨胀**：完整流程测试日志每条约 5KB，需严格执行 7 天清理策略。
- **导出文件管理**：导出文件需要定期清理（24 小时过期），避免磁盘占用。
- **索引膨胀**：新增多个索引会增加写入开销，需监控索引使用率。

#### 19.12.3 功能待澄清

- **对话导出格式**：是否需要支持 JSON/Excel 格式？当前仅实现 CSV。
- **实时监控推送**：是否需要 WebSocket 实时推送监控数据？当前仅支持轮询。
- **历史数据迁移**：现有对话数据是否需要回填 `prompt_template_id` 等新字段？
- **权限控制**：测试与监控接口是否需要更细粒度的权限控制（如只读/读写）？

#### 19.12.4 兼容性风险

- **数据库迁移**：新增字段和索引需要在生产环境谨慎执行，建议分批迁移。
- **API 版本**：新增监控接口不影响现有 `/ai/chat` 接口，向后兼容。
- **前端适配**：Dashboard 新增统计字段需要前端同步更新，否则显示为空。

---

## 20. 总结

v0.7.0 测试与监控系统为 AI 中台提供了完整的可观测性与可测试性能力：

**核心价值**：
- **独立测试**：为四大功能（Prompt 模板、意图规则、话术流程、输出护栏）提供独立测试能力
- **完整追踪**：12 步流程的详细执行日志，支持问题排查与效果评估
- **实时监控**：细粒度的运行时统计，支持规则命中率、流程完成率、护栏拦截率等
- **数据导出**：支持对话数据导出，便于离线分析与模型优化

**技术亮点**：
- **性能优先**：生产模式性能影响 <2%，测试模式仅在显式开启时生效
- **存储可控**：测试日志 7 天自动清理，导出文件 24 小时过期
- **租户隔离**：所有监控数据按 `tenant_id` 隔离，保证多租户安全
- **向后兼容**：新增监控不影响现有接口行为与性能

**实施建议**：
1. 优先实现 Dashboard 统计增强（AC-AISVC-91, AC-AISVC-92）
2. 其次实现完整流程测试台（AC-AISVC-93 ~ AC-AISVC-96）
3. 再实现各功能的独立测试接口（AC-AISVC-97 ~ AC-AISVC-108）
4. 最后实现对话追踪与导出（AC-AISVC-109, AC-AISVC-110）