---
id: scenario.customer-service-kb
title: 客服知识库问答：从文档接入到带引用的流式回答、权限、评测和转人工
type: scenario
domain: cross
stack: [rag, vector-db, elasticsearch, fastapi, vue, redis, data-pipeline, agent-ops]
tags: [RAG, 混合检索, 权限, 流式输出, 评测, 转人工, 引用]
level: intermediate
prerequisites: [ai.rag.index, python.fastapi.index]
est_hours: 3
related: [ai.rag.scenario.permission-aware-kb, backend.elasticsearch.scenario.kb-search-sync, frontend.vue.scenario.ai-chat-stream, ai.agent-ops.scenario.agent-quality-gate]
week: [5, 6, 13, 15]
---

# 客服知识库问答：从文档接入到带引用的流式回答、权限、评测和转人工

> 跨技术栈的综合场景，串联前端、后端、AI 和运维。BOSS直聘 2026-10-23 的调研中，Agent（22 个岗位）、RAG（17 个）是出现最多的要求，这是最常见的落地形态，也适合作为作品集项目。

## 业务背景与问题

一家 SaaS 公司的客服每天处理大量重复问题（退货、发票、账号、错误码）。要做一个 AI 客服：

1. 回答必须基于公司文档，不能编造，要给出引用来源；
2. 不同客户（租户）、不同部门的文档权限不同；
3. 文档经常更新，更新后很快生效；
4. 回答不了或用户不满意时转人工，并把上下文交给坐席；
5. 需要证明效果：上线前和每次改动后都要有数据。

## 约束

- 首 token 延迟 2 秒以内，回答流式输出。
- 模型调用有成本预算。
- 数据不能出公司（或只能使用指定的模型服务）。
- 团队：1 名前端、1 名后端、你（全栈 + AI）。

## 方案设计

| 层 | 方案 | 专题 |
| --- | --- | --- |
| 文档接入 | 连接器增量拉取 → 解析（PDF / Word / 网页，表格单独处理）→ 按标题层级分块 → 嵌入 → 写入；outbox + 队列 + 幂等消费者，每日对账 | [数据管道](../topics/backend/data-pipeline/index.md)、[RAG 切分](../topics/ai/rag/01-pipeline-chunking.md) |
| 检索 | ES BM25 + 向量 kNN 两路召回，RRF 融合，重排取前 5；租户和部门权限作为 filter | [ES 混合检索](../topics/backend/elasticsearch/scenarios/kb-search-sync.md)、[权限感知知识库](../topics/ai/rag/scenarios/permission-aware-kb.md) |
| 生成 | 只依据资料回答、资料不足时拒答，输出带引用编号；结构化输出（答案、引用、置信度、是否建议转人工） | [提示词与结构化输出](../topics/ai/prompt-engineering/index.md) |
| 接口 | FastAPI SSE 流式接口，事件：`citations`、`token`、`done`、`error`；断开时取消生成 | [FastAPI 流式接口](../topics/python/fastapi/scenarios/streaming-chat-api.md) |
| 前端 | Vue 对话组件，增量渲染 Markdown（安全清洗），引用可点击看原文，“没解决”按钮转人工 | [Vue AI 对话流](../topics/frontend/vue/scenarios/ai-chat-stream.md)、[AI 应用安全](../topics/cs/web-security/scenarios/ai-app-security.md) |
| 缓存与限流 | 通用问题语义缓存（按租户隔离、文档更新时失效）；按用户限流 | [Redis](../topics/backend/redis/index.md) |
| 转人工 | 置信度低 / 用户点击 / 反复追问时创建工单，附对话摘要和检索资料 | [工单系统](ticket-system.md) |
| 评测与监控 | 标注集（问题 → 应引用文档 → 参考答案），CI 中跑召回率@K、引用正确率、拒答正确率；线上记录 Bad Case、成本、延迟 | [Agent 质量门禁](../topics/ai/agent-ops/scenarios/agent-quality-gate.md) |

## 关键代码

接口骨架（示意；完整实现见第 5~6 周项目和各专题示例）：

```python
@app.post("/chat")
async def chat(req: ChatRequest, user: User = Depends(current_user)):
    async def events():
        hits = await retrieve(req.question, acl=user.acl, k=20)          # 两路召回 + 权限过滤
        top = await rerank(req.question, hits, k=5)
        if not top or top[0].score < MIN_SCORE:
            await create_ticket(user, req, reason="no_context")
            yield sse("done", {"answer": "抱歉，资料中没有找到相关内容，已为你转人工。", "handoff": True})
            return
        yield sse("citations", [c.public() for c in top])               # 先发引用，前端先展示来源
        async for token in llm.stream(build_prompt(req.question, top, history=req.history)):
            yield sse("token", {"text": token})
        yield sse("done", {"handoff": False})
    return StreamingResponse(events(), media_type="text/event-stream")
```

评测门禁（CI 中运行，比基线差则失败）：

```python
report = evaluate(dataset="eval/kb_v3.jsonl", retriever=retrieve, generator=answer)
assert report.recall_at_5 >= baseline.recall_at_5
assert report.citation_accuracy >= 0.9
assert report.refusal_accuracy >= baseline.refusal_accuracy      # 该拒答的要拒答
```

## 取舍

| 决策 | 选择 | 放弃的 | 原因 |
| --- | --- | --- | --- |
| 检索 | 混合检索 + 重排 | 只用向量 | 错误码、型号等精确词召回差 |
| 框架 | 主流程手写，部分组件用 LangChain | 全部用框架 | 便于调试和讲清原理 |
| 低代码 | Dify 做 POC，正式版写代码 | 直接上线 Dify | 权限、评测、定制 UI 受限，见 [POC 到生产](../topics/ai/lowcode/scenarios/poc-to-production.md) |
| 缓存 | 只缓存无个人信息的通用问题 | 所有问题都缓存 | 防止串数据 |

## 易错点

- 没有评测集，靠感觉调参。
- 权限只在检索后过滤，导致结果不足或泄露。
- 文档删除、权限变更没有同步。
- 回答渲染成 HTML 时没有清洗。
- 转人工时没有带上下文，用户要从头再说一遍。

## 面试怎么讲

> “这是我做的客服知识库问答。文档通过增量同步进 ES，同时存文本和向量；查询时 BM25 和向量两路召回、RRF 融合、重排取前 5，租户和部门权限作为 filter。生成只依据资料、带引用编号，资料不足就拒答并转人工。接口是 FastAPI 的 SSE，先发引用再流式发答案。我维护了一份评测集，CI 里跑召回率、引用正确率和拒答正确率，任何改动比基线差就不合并。”（数字请用你自己项目的实测结果。）