zhuangang/scenarios/customer-service-kb.md

id, title, type, domain, stack, tags, level, prerequisites, est_hours, related, week

id	title	type	domain	stack	tags	level	prerequisites	est_hours	related	week
scenario.customer-service-kb	客服知识库问答：从文档接入到带引用的流式回答、权限、评测和转人工	scenario	cross	rag vector-db elasticsearch fastapi vue redis data-pipeline agent-ops	RAG 混合检索 权限 流式输出 评测 转人工 引用	intermediate	ai.rag.index python.fastapi.index	3	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	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 + 队列 + 幂等消费者，每日对账	数据管道、RAG 切分
检索	ES BM25 + 向量 kNN 两路召回，RRF 融合，重排取前 5；租户和部门权限作为 filter	ES 混合检索、权限感知知识库
生成	只依据资料回答、资料不足时拒答，输出带引用编号；结构化输出（答案、引用、置信度、是否建议转人工）	提示词与结构化输出
接口	FastAPI SSE 流式接口，事件：citations、token、done、error；断开时取消生成	FastAPI 流式接口
前端	Vue 对话组件，增量渲染 Markdown（安全清洗），引用可点击看原文，“没解决”按钮转人工	Vue AI 对话流、AI 应用安全
缓存与限流	通用问题语义缓存（按租户隔离、文档更新时失效）；按用户限流	Redis
转人工	置信度低 / 用户点击 / 反复追问时创建工单，附对话摘要和检索资料	工单系统
评测与监控	标注集（问题 → 应引用文档 → 参考答案），CI 中跑召回率@K、引用正确率、拒答正确率；线上记录 Bad Case、成本、延迟	Agent 质量门禁

关键代码

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

@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 中运行，比基线差则失败）：

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 到生产
缓存	只缓存无个人信息的通用问题	所有问题都缓存	防止串数据

易错点

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

面试怎么讲

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