核心内容摘要
在公有云上试用英伟达账单超乎预期，期待昇腾950今年能够在公有云上出现
基于RAGFlow的智能客服问答系统从架构设计到性能优化实战背景痛点传统客服的“三慢”顽疾做ToB SaaS客服平台三年最怕听到客户吐槽“你们机器人答非所问”。
传统FAQ-bot的通病可以
总结成“三慢”知识更新慢运营同学改一次Excel研发再导库上线至少两天热点政策早就变了。
长尾响应慢越冷门的问题越依赖关键字匹配一旦没命中就转人工排队10分钟起步。
语义理解慢同义词、口语化表达全靠穷举维护成本指数级上升。
去年618大促我们峰值QPS
2k人工坐席溢出率飙到47%老板直接拍桌子两周内必须上语义升级方案。
于是把视线投向了RAGRetrieval-Augmented Generation。
技术选型为什么不是BERTFAQ、也不是Fine-tune LLM对比实验在内部沙箱跑了7天结论一句话BERTFAQ召回率92%但更新需重训分类器依旧“两天上线”。
Fine-tune LLM生成效果惊艳可10w条领域数据才收敛GPU账单5w/月。
RAGFlow检索与生成解耦知识库10分钟级增量更新成本≈LLM的1/5。
RAGFlow还把Milvus、Faiss、Elasticsearch等组件做了DSL封装对我们这种“运维人力紧张”的小团队极度友好于是直接All-in。
架构设计一张图看懂数据流先上图再拆解网关层NginxLua做灰度分流按uid哈希到不同版本RAG链。
检索模块语义编码bge-small-zh-v
5768维延迟35ms。
向量库Milvus
3分区按“业务线版本”做软隔离方便回滚。
精排用向量相似度计算score1再叠加BM25的score2加权融合。
生成模块提示模板“系统人设检索结果用户历史3轮”token控制在2k以内。
大模型Qwen-14B-Chat-int4单卡A10G可跑TPS≈8。
缓存层Redis缓存“同一知识版本问题指纹”的生成结果TTL300s命中率38%。
反馈闭环用户点“解决/未解决”落Kafka离线负采样做困难例挖掘次日晨跑批处理更新索引。
核心实现代码直接能跑以下片段来自生产仓库已脱敏可直接粘进IDE跑通。
知识库构建含chunk、embedding、写入# kb_builder.py import os, json, time, logging from ragflow import DocumentSet, EmbeddingEncoder from concurrent.futures import ThreadPoolExecutor, as_completed logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) def chunk_text(text: str, max_len384, overlap
: 按标点长度双重切片避免截断语义 import re split_re re.compile(r[。
]) sentences split_re.split(text) buf, chunks , [] for sent in sentences: if len(buf sent) max_len: buf sent else: if buf : buf.strip(): chunks.append(buf) buf sent if buf: chunks.append(buf) return chunks def build_kb(file_path: str, ds: DocumentSet, encoder: EmbeddingEncoder, batch
: 多线程批量写入IO与CPU解耦 tasks, cnt [], 0 with open(file_path, encodingutf-
as f: for line in f: data json.loads(line) chunks chunk_text(data[content]) for chk in chunks: tasks.append((chk, {source: data[id]})) if len(tasks) batch: _flush(ds, encoder, tasks) tasks.clear() if tasks: _flush(ds, encoder, tasks) logging.info(kb build done, total chunks%s, ds.count()) def _flush(ds, encoder, tasks): texts [t[0] for t in tasks] embeddings encoder.encode(texts) # shape[batch,768] ds.upsert(records[{text: t[0], vec: e, meta: t[1]} for t, e in zip(tasks, embeddings)]) if __name__ __main__: encoder EmbeddingEncoder(bge-small-zh) ds DocumentSet(cs_kb_v
# 对应Milvus collection build_kb(qa_corpus.jsonl, ds, encoder)
在线问答接口含异常、日志、缓存# rag_service.py from flask import Flask, request, jsonify from ragflow import Retriever, Generator from redis import Redis import hashlib, time, logging app Flask(__name__) redis_cli Redis(decode_responsesTrue) ret Retriever(cs_kb_v
# 绑定同一张量库 gen Generator(qwen-14b-chat) # 本地vLLM推理 def make_key(q, kv_versionv
: 生成问题指纹 return frag:{kv_version}: hashlib.md5(q.encode()).hexdigest() app.route(/ask, methods[POST]) def ask(): st time.time() question request.json.get(question, ).strip() history request.json.get(history, []) if not question: return jsonify({code: 400, msg: empty question}), 400 try: #
缓存命中 key make_key(question) if ans : redis_cli.get(key): logging.info(cache hit, key%s, key) return jsonify({answer: ans, latency: time.time()-st, source: cache}) #
检索top5 docs ret.retrieve(question, topk5, score_threshold
0.
contexts [d[text] for d in docs] #
生成 prompt format_prompt(question, contexts, history) answer gen.generate(prompt, max_tokens512, temperature
0.
#
写缓存 日志 redis_cli.setex(key, 300, answer) logging.info(rag ok, q%s, latency%.2f, docs%s, question, time.time()-st, len(docs)) return jsonify({answer: answer, latency: time.time()-st, source: rag}) except Exception as e: logging.exception(rag error) return jsonify({code: 500, msg: internal error}), 500 def format_prompt(q, ctxs, hist): 极简模板tokenhistctxsq 2k hist_str \n.join([fUser:{h[q]}\nAssistant:{h[a]} for h in hist[-3:]]) ctx_str \n.join([f[{i1}] {c} for i, c in enumerate(ctxs)]) return f你是客服机器人请依据以下资料回答问题\n{ctx_str}\n历史对话\n{hist_str}\n用户{q}\n助理性能优化把延迟压到400ms以内上线第一版平均延迟900ms老板一句“不如人工快”直接打回。
我们做了三轮压测最终P99400ms关键动作如下索引并行化Milvus的index_file_size1024MB单次build CPU打满把segment_row_limit降到512k并给create_index开nprobe32并行度8核机器索引时间从45min缩到7min。
模型量化连续批处理vLLM支持--quantization int4显存占用减半同时打开--max-num-seqs 256把单卡吞吐从
6提到
2 req/s延迟反而降了15%。
预检索过滤业务线彼此独立先在元数据里加product_id字段利用Milvus的partition_key做剪枝把候选池从200w降到5w向量相似度计算耗时从120ms降到18ms。
缓存分层除了Redis再加一层本地Caffeine基于GuavaTTL30s命中率又多12%对超高频“密码怎么改”类问题极有效。
避坑指南生产级血泪
总结切片粒度过细→召回冗余早期按128字符切结果top5里同一段出现3次用户直呼啰嗦。
把overlap调为max_len*1/6并加“句子边界”正则后冗余率从34%降到7%。
时间字段没同步→答案过期政策库带生效时间第一次没把effective_date写进meta检索到去年答案被投诉“误导”。
后在Retriever层加filtereffective_datetoday日更脚本同步T0。
高并发下Milvus OOM默认cache_insert_datatrue写入高峰把内存吃满。
关闭该参数并调queryNode.memory60%同时把retriever的batch_search拆成4次并发内存稳在70%以下。
生成“幻觉”甩锅给检索用户问“能否退款”模型答“可以全额退”实际需满足7天无理由。
解决方式在提示模板里加“若资料未提及请回答‘暂无相关信息’”并给contexts标号让模型引用编号减少自由发挥。
日志没脱敏→泄露手机号开发期把请求全文打印被安全扫描揪出。
统一用logging.Filter把1\d{10}替换为1*********并关闭debug级别合规通过。
安全考量数据隐私与访问控制向量库隔离敏感企业数据单独建databaseMilvus支持RBAC检索服务启动时只读授权账号。
传输加密Nginx层强制TLS
3内网服务间用mTLS双向校验杜绝明文嗅探。
prompt攻击防护在Generator层加正则黑名单含“忽略前面”“转为中文”等指令直接拦截。
审计留痕把user_idquestionanswer_hash写进ES保留30天方便合规部门抽查。
开放性问题目前我们仅支持3轮历史如果要做多轮对话状态跟踪DST你会如何把“用户意图槽位”与RAG的检索结果融合欢迎一起探讨。