#RAG 实战（五）：FAQ 检索链路（Greeting + TF-IDF + 接口封装）

EduRAG 的问答流水线由 Greeting、TF-IDF FAQ、RAG、联网搜索与 Fallback 串联而成。本章实现流水线的前两段基于关键词匹配的 Greeting 短路以及 FAQ 的 MySQL 入库与 TF-IDF 余弦相似度检索并将其封装为QAPipeline通过POST /api/ask对外提供服务。该链路不依赖向量库与大模型即可拦截高频问题是后续 RAG 之前的高性价比预处理环节。5.1 Greeting项目的准备工作已经做好接下来就正式进入对话流程的处理环节。按照日常对话习惯用户上来就会先打个招呼。但这种招呼类的消息是不应该进入流程的直接给用户一个固定话术的回复即可。第一步greeting.pyservices/greeting.pyimportrandom# 命中任一关键词即走问候短路GREETING_KEYWORDS[你好,您好,hi,hello,你是谁,您是谁,你叫什么,你的名字,who are you,在吗,在不在,有人吗,]# 随机取一条避免每次回复完全一样GREETING_RESPONSES[你好我是智能问答助理有什么可以帮你,智能问答助理已就位有什么可以帮你,你好我是智能问答助理请问有什么可以协助你的,]defmatch_greeting(query:str):text(queryor).strip().lower()# 去掉常见标点兼容「在吗」「你好」forchin?!。.,:texttext.replace(ch,)texttext.strip()iftextin{k.lower()forkinGREETING_KEYWORDS}:returnrandom.choice(GREETING_RESPONSES)returnNone要点1关键词整句匹配去标点避免「你好11等于几」误命中2命中后从回复列表随机取一条。第二步ask 命令commands/test_ask.py本步先只接问候与demo/test-clients一样走自动发现无需改cli.pyimporttyperfromloggerimportloggerfromservices.greetingimportmatch_greetingdefregister(app):app.command(test-ask,help对话流程测试脚本)deftest_ask(questiontyper.Option(None,help用户问题)):ifnotquestionornotstr(question).strip():raiseSystemExit(用法: python cli.py test-ask --question你的问题)questionstr(question).strip()answermatch_greeting(question)ifanswer:logger.info(fstagegreeting, answer{answer})returnlogger.info(未命中问候语后续接入 FAQ)要点提供register(app)在内用app.command挂命令参数用typer.Option(..., help...)。python cli.py test-ask--question你好python cli.py test-ask--question你是谁python cli.py test-ask--questionPython 是什么验证「你好 / 你是谁 / 在吗」命中问候多次试「你好」话术可能不同「你好11等于几」不命中带真实问题留给后续检索普通问题提示「未命中问候语」5.2 导入 FAQ处理完打招呼的消息之后按流程就要正式进入 FAQFrequently Asked Questions的环节了。但是 FAQ 的数据还在文件里面需要先导入到数据库中然后才能进行后续的问答处理。认识数据用到的原始数据文件放在storage/data/edu_rag/ └── storage/data/ ├── faq.csv └── cn_stopwords.txt这是一份示例数据。可自行替换为自己业务的 FAQ 数据文件。faq.csv约 66 条三列分类,问题,答案 通用优惠,示例会员首月套餐多少钱,示例会员首月套餐原价为198元活动期间可享限时优惠价28元。 通用优惠,示例会员在哪些地区可以使用,示例会员适用于全国所有地区。 产品咨询,示例高级版的标准售价是多少钱,示例高级版标准售价为1980元活动期间享受限时优惠。 产品咨询,示例高级版有没有优惠,有限时优惠100元。cn_stopwords.txt是中文停用词表分词后去掉「的、了、吗」这类对检索帮助不大的词减轻噪声。词表来源goto456/stopwords本项目使用其中的cn_stopwords.txt。将faq.csv、cn_stopwords.txt放到storage/data/。第一步分词定义分词函数去除停用词。导入 FAQ、后续 TF-IDF 检索都会用到。utils/preprocess.pyimportjiebafrompathlibimportPathdef_get_stopwords():stopwords_pathPath(__file__).parent/../storage/data/cn_stopwords.txtwithopen(stopwords_path,r,encodingutf-8)asf:returnset(f.read().splitlines())STOPWORDS_get_stopwords()defpreprocess_text(text:str):texttext.lower()return[wordforwordinjieba.lcut(text)ifwordnotinSTOPWORDSandword.strip()]if__name____main__:print(preprocess_text(你好我是小明很高兴认识你))第二步读取 CSV先建ingest-faq命令只读 CSV 并打印前几行确认字段无误再入库。commands/ingest_faq.pyfrompathlibimportPathimportpandasaspdimporttyperdefregister(app):app.command(ingest-faq,help导入 FAQ 数据到 MySQL)defingest_faq():csv_pathPath(__file__).parent/../storage/data/faq.csvdatapd.read_csv(csv_path)print(data.head(5))python cli.py ingest-faq能看到「分类 / 问题 / 答案」三列即可。第三步写入 MySQL加工数据并批量入库列名映射为 Model 字段对「问题」分词后json.dumpslist 不能直接塞进一列再用 Session 写入。先确保表已迁移含tokens列alembic upgradeheadcommands/ingest_faq.py完整版本importjsonfrompathlibimportPathimportpandasaspdimporttyperfromdb.modelsimportFaqfromdb.sessionimportSessionLocalfromutils.preprocessimportpreprocess_textdefwrite_to_mysql(data):withSessionLocal()assession:forrowindata.to_dict(orientrecords):questionrow[问题]ifnotquestionornotstr(question).strip():continuehassession.query(Faq).filter(Faq.questionquestion).first()ifhas:continuetokensjson.dumps(preprocess_text(row[问题]),ensure_asciiFalse)faqFaq(categoryrow[分类],questionquestion,answerrow[答案],tokenstokens,)session.add(faq)session.commit()defregister(app):app.command(ingest-faq,help导入 FAQ 数据到 MySQL)defingest_faq():csv_pathPath(__file__).parent/../storage/data/faq.csvdatapd.read_csv(csv_path)write_to_mysql(data)print(fOK: FAQ 导入完成共{len(data)}条)python cli.py ingest-faq验证MySQLfaq约 66 条且tokens列为 JSON 分词。重复执行会先清空再导入避免重复条数。5.3 TF-IDF 检索FAQ 已导入 MySQL含tokens。下面使用 TF-IDF 余弦相似度做关键词检索启动时加载分词建索引查询时算相似度高于阈值则直接返回答案。BM25 进阶见后续章节。第一步配置阈值余弦相似度本身落在 01可直接设阈值低于阈值视为未命中留给后续 RAG / 联网。可按题库表现再调。# .env TFIDF_THRESHOLD0.5config.pyself.TFIDF_THRESHOLDfloat(os.getenv(TFIDF_THRESHOLD,0.5))第二步TF-IDF 封装对sklearn的TfidfVectorizer 余弦相似度做一层封装。入参是已经分好词的文档列表与 MySQLtokens一致用空格拼成字符串后analyzerstr.split避免再分一次词。services/tfidf.pyfromsklearn.feature_extraction.textimportTfidfVectorizerfromsklearn.metrics.pairwiseimportcosine_similarityfromloggerimportloggerclassTfidfSearch:def__init__(self,tokenized_docs):# tokens 已是 jieba 结果用空格拼接后按空格切开即可corpus[ .join(doc)fordocintokenized_docs]self.vectorizerTfidfVectorizer(analyzerstr.split)self.doc_vectorsself.vectorizer.fit_transform(corpus)logger.info(fTfidfVectorizer 初始化完成共{len(tokenized_docs)}条)defget_scores(self,tokenized_query):返回查询与各文档的余弦相似度01。q .join(tokenized_query)q_vecself.vectorizer.transform([q])returncosine_similarity(q_vec,self.doc_vectors).ravel()第三步FaqService从 MySQL 读出问答与tokensJSON建 TF-IDF查询时对用户问题分词、算相似度、过阈值。后面再抽QAPipeline并接到页面。services/faq.pyimportjsonimportnumpyasnpfromconfigimportconfigfromdb.modelsimportFaqfromdb.sessionimportSessionLocalfromloggerimportloggerfromservices.tfidfimportTfidfSearchfromutils.preprocessimportpreprocess_textclassFaqService:def__init__(self):self.ids[]self.questions[]self.answers[]self.tfidfNoneself._reload()def_reload(self):从 MySQL 读取问答 tokens建 TF-IDF。withSessionLocal()assession:db_rowssession.query(Faq).all()self.ids[]self.questions[]self.answers[]tokenized[]forrowindb_rows:ifnotrow.tokens:continuetry:tokensjson.loads(row.tokens)exceptException:logger.warning(fFAQ id{row.id}tokens 解析失败已跳过)continueifnottokens:continueself.ids.append(row.id)self.questions.append(row.question)self.answers.append(row.answer)tokenized.append(tokens)ifnottokenized:self.tfidfNonelogger.warning(MySQL 无可用 FAQ 分词请先执行: python cli.py ingest-faq)returnself.tfidfTfidfSearch(tokenized)logger.info(fFAQ 索引加载完成{len(self.ids)}条MySQL 问答 tokens)defsearch(self,query,thresholdNone):命中返回 (answer, score)未命中返回 (None, score)。thresholdconfig.TFIDF_THRESHOLDifthresholdisNoneelsethresholdifnotqueryornotself.tfidf:returnNone,0.0scoresself.tfidf.get_scores(preprocess_text(query))ifscoresisNoneorlen(scores)0:returnNone,0.0best_idxint(np.argmax(scores))best_scorefloat(scores[best_idx])logger.info(fTF-IDF 最高分{best_score:.3f}(阈值{threshold}))ifbest_scorethreshold:returnNone,best_scorereturnself.answers[best_idx],best_score要点启动时一次性json.loads建索引查询只对用户问题分词余弦相似度已在 01直接用TFIDF_THRESHOLD判断即可。第四步接入 test-ask先在命令里把流程写直问候 → FAQ 检索 → miss。后面再抽 Pipeline 并接到页面。commands/test_ask.pyimporttyperfromloggerimportloggerfromservices.faqimportFaqServicefromservices.greetingimportmatch_greetingdefregister(app):app.command(test-ask,help对话流程测试脚本)deftest_ask(questiontyper.Option(None,help用户问题)):ifnotquestionornotstr(question).strip():raiseSystemExit(用法: python cli.py test-ask --question你的问题)questionstr(question).strip()answermatch_greeting(question)ifanswer:logger.info(fstagegreeting, answer{answer})returnfaq_serviceFaqService()answer,scorefaq_service.search(question)ifanswer:logger.info(fstagefaq score{score:.3f})logger.info(fanswer{answer})returnlogger.info(fstagemiss score{score:.3f})logger.info(暂未在 FAQ 中找到可靠答案后续会接入 RAG / 联网搜索。)python cli.py test-ask--question你好python cli.py test-ask--question示例会员首月套餐多少钱python cli.py test-ask--questionPython 是什么验证「你好」→stagegreeting「示例会员首月套餐多少钱」→stagefaq答案接近 FAQ「Python 是什么」→stagemiss不在 FAQ 里日志有「FAQ 索引加载完成… 条MySQL 问答 tokens」5.4 接口封装上一节的流程写在test-ask里已经能跑。这里先抽成QAPipeline再暴露为POST /api/ask前端改为真实请求。第一步抽取 Pipeline命令只负责入参和打印业务分流放到QAPipeline.query。pipeline.pyfromloggerimportloggerfromservices.faqimportFaqServicefromservices.greetingimportmatch_greetingclassQAPipeline:def__init__(self):self.faq_serviceFaqService()defquery(self,question):question(questionor).strip()logger.info(f查询:{question!r})answermatch_greeting(question)ifanswer:return{answer:answer,stage:greeting,score:1.0}answer,scoreself.faq_service.search(question)ifanswer:return{answer:answer,stage:faq,score:score}logger.info(f阶段miss score{score:.3f})return{answer:暂未在 FAQ 中找到可靠答案后续会接入 RAG / 联网搜索。,stage:miss,score:score,}commands/test_ask.py改为调用 PipelineimporttyperfromloggerimportloggerfrompipelineimportQAPipelinedefregister(app):app.command(test-ask,help对话流程测试脚本)deftest_ask(questiontyper.Option(None,help用户问题)):ifnotquestionornotstr(question).strip():raiseSystemExit(用法: python cli.py test-ask --question你的问题)pipelineQAPipeline()resultpipeline.query(question)print(result)python cli.py test-ask--question示例会员首月套餐多少钱行为应与上一节一致之后接 HTTP / RAG 都复用QAPipeline。第二步APIroutes.py只加下面内容原有/、/health不动顶部新增 importfrompydanticimportBaseModel,FieldfrompipelineimportQAPipeline# 定义请求参数classQueryRequest(BaseModel):question:strField(...,min_length1)# 定义接口函数router.post(/api/ask)defask(body:QueryRequest):pipelineQAPipeline()returnpipeline.query(body.question)注意参数类型必须是QueryRequest不要写成 FastAPI 的Request——后者没有.question会报AttributeError。第三步前端改 fetchtemplates/index.html把原来的demoReply换成async function askBackend(session, question) { sendBtn.disabled true; try { const resp await fetch(/api/ask, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ question }), }); const data await resp.json(); if (!resp.ok) { throw new Error(data.detail || 请求失败); } const stage data.stage || ; const score typeof data.score number ? · ${data.score.toFixed(2)} : ; session.messages.push({ role: bot, text: ${data.answer || }${stage ? \n\n[${stage}${score}] : }, }); } catch (err) { session.messages.push({ role: bot, text: 请求失败${err.message || err}, }); } renderMessages(); sendBtn.disabled false; input.focus(); }提交时调用askBackend(session, text)。第四步验证目录结构edu_rag/ ├── pipeline.py ├── routes.py ├── app.py └── templates/ └── index.htmlpython app.py浏览器 1. 输入「你好」→ 问候话术[greeting]2. 输入 CSV 里已有问题 → FAQ 答案[faq · 0.xx]3. 输入无关问题 → miss 提示curl-s-XPOST http://127.0.0.1:8000/api/ask\-HContent-Type: application/json\-d{question:你好}小结Greeting 短路通过关键词整句匹配去标点实现命中即返回随机话术避免闲聊进入检索流程。FAQ 数据经 jieba 分词、去停用词后写入 MySQL 的tokens列JSON 字符串导入时做去重。TF-IDF 索引在FaqService初始化时一次性从 MySQL 构建查询仅对用户问题分词余弦相似度直接以阈值 0.5 判定命中。业务分流统一收敛到QAPipeline.queryGreeting → FAQ → miss 三段式便于后续接入 RAG 与联网搜索。通过POST /api/ask请求体{question: str}对外暴露前端以fetch提交并展示答案与stage/score。