企业官网建设流程全解析

EduRAG 的问答流水线由 Greeting、TF-IDF FAQ、RAG、联网搜索与 Fallback 串联而成。本章实现流水线的前两段：基于关键词匹配的 Greeting 短路，以及 FAQ 的 MySQL 入库与 TF-IDF 余弦相似度检索，并将其封装为QAPipeline，通过POST /api/ask对外提供服务。该链路不依赖向量库与大模型即可拦截高频问题，是后续 RAG 之前的高性价比预处理环节。

5.1 Greeting

项目的准备工作已经做好，接下来就正式进入对话流程的处理环节。按照日常对话习惯，用户上来就会先打个招呼。但这种招呼类的消息，是不应该进入流程的，直接给用户一个固定话术的回复即可。

第一步：greeting.py

services/greeting.py：

importrandom# 命中任一关键词，即走问候短路GREETING_KEYWORDS=["你好","您好","hi","hello","你是谁","您是谁","你叫什么","你的名字","who are you","在吗","在不在","有人吗",]# 随机取一条，避免每次回复完全一样GREETING_RESPONSES=["你好！我是智能问答助理，有什么可以帮你？","智能问答助理已就位，有什么可以帮你？","你好，我是智能问答助理，请问有什么可以协助你的？",]defmatch_greeting(query:str):text=(queryor"").strip().lower()# 去掉常见标点，兼容「在吗？」「你好！」forchin"？?！!。.，,":text=text.replace(ch,"")text=text.strip()iftextin{k.lower()forkinGREETING_KEYWORDS}:returnrandom.choice(GREETING_RESPONSES)returnNone

要点：1）关键词整句匹配（去标点），避免「你好，1+1等于几？」误命中；2）命中后从回复列表随机取一条。

第二步：ask 命令

commands/test_ask.py（本步先只接问候；与demo/test-clients一样走自动发现，无需改cli.py）：

importtyperfromloggerimportloggerfromservices.greetingimportmatch_greetingdefregister(app):@app.command("test-ask",help="对话流程测试脚本")deftest_ask(question=typer.Option(None,help="用户问题")):ifnotquestionornotstr(question).strip():raiseSystemExit('用法: python cli.py test-ask --question="你的问题"')question=str(question).strip()answer=match_greeting(question)ifanswer:logger.info(f"stage=greeting, 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--question="Python 是什么"

验证

• 「你好 / 你是谁 / 在吗」命中问候（多次试「你好」，话术可能不同）
• 「你好，1+1等于几？」不命中（带真实问题，留给后续检索）
• 普通问题提示「未命中问候语」

5.2 导入 FAQ

处理完打招呼的消息之后，按流程，就要正式进入 FAQ（Frequently 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.py：

importjiebafrompathlibimportPathdef_get_stopwords():stopwords_path=Path(__file__).parent/"../storage/data/cn_stopwords.txt"withopen(stopwords_path,"r",encoding="utf-8")asf:returnset(f.read().splitlines())STOPWORDS=_get_stopwords()defpreprocess_text(text:str):text=text.lower()return[wordforwordinjieba.lcut(text)ifwordnotinSTOPWORDSandword.strip()]if__name__=="__main__":print(preprocess_text("你好，我是小明，很高兴认识你"))

第二步：读取 CSV

先建ingest-faq命令，只读 CSV 并打印前几行，确认字段无误再入库。

commands/ingest_faq.py：

frompathlibimportPathimportpandasaspdimporttyperdefregister(app):@app.command("ingest-faq",help="导入 FAQ 数据到 MySQL")defingest_faq():csv_path=Path(__file__).parent/"../storage/data/faq.csv"data=pd.read_csv(csv_path)print(data.head(5))

python cli.py ingest-faq

能看到「分类 / 问题 / 答案」三列即可。

第三步：写入 MySQL

加工数据并批量入库：列名映射为 Model 字段；对「问题」分词后json.dumps（list 不能直接塞进一列），再用 Session 写入。

先确保表已迁移（含tokens列）：

alembic upgradehead

commands/ingest_faq.py（完整版本）：

importjsonfrompathlibimportPathimportpandasaspdimporttyperfromdb.modelsimportFaqfromdb.sessionimportSessionLocalfromutils.preprocessimportpreprocess_textdefwrite_to_mysql(data):withSessionLocal()assession:forrowindata.to_dict(orient="records"):question=row["问题"]ifnotquestionornotstr(question).strip():continuehas=session.query(Faq).filter(Faq.question==question).first()ifhas:continuetokens=json.dumps(preprocess_text(row["问题"]),ensure_ascii=False)faq=Faq(category=row["分类"],question=question,answer=row["答案"],tokens=tokens,)session.add(faq)session.commit()defregister(app):@app.command("ingest-faq",help="导入 FAQ 数据到 MySQL")defingest_faq():csv_path=Path(__file__).parent/"../storage/data/faq.csv"data=pd.read_csv(csv_path)write_to_mysql(data)print(f"OK: FAQ 导入完成，共{len(data)}条")

python cli.py ingest-faq

验证：MySQLfaq约 66 条，且tokens列为 JSON 分词。重复执行会先清空再导入，避免重复条数。

5.3 TF-IDF 检索

FAQ 已导入 MySQL（含tokens）。下面使用 TF-IDF + 余弦相似度做关键词检索：启动时加载分词建索引，查询时算相似度，高于阈值则直接返回答案。BM25 进阶见后续章节。

第一步：配置阈值

余弦相似度本身落在 0～1，可直接设阈值：低于阈值视为未命中，留给后续 RAG / 联网。可按题库表现再调。

# .env TFIDF_THRESHOLD=0.5

config.py：

self.TFIDF_THRESHOLD=float(os.getenv("TFIDF_THRESHOLD","0.5"))

第二步：TF-IDF 封装

对sklearn的TfidfVectorizer+ 余弦相似度做一层封装。入参是已经分好词的文档列表（与 MySQLtokens一致），用空格拼成字符串后analyzer=str.split，避免再分一次词。

services/tfidf.py：

fromsklearn.feature_extraction.textimportTfidfVectorizerfromsklearn.metrics.pairwiseimportcosine_similarityfromloggerimportloggerclassTfidfSearch:def__init__(self,tokenized_docs):# tokens 已是 jieba 结果，用空格拼接后按空格切开即可corpus=[" ".join(doc)fordocintokenized_docs]self.vectorizer=TfidfVectorizer(analyzer=str.split)self.doc_vectors=self.vectorizer.fit_transform(corpus)logger.info(f"TfidfVectorizer 初始化完成，共{len(tokenized_docs)}条")defget_scores(self,tokenized_query):"""返回查询与各文档的余弦相似度（0～1）。"""q=" ".join(tokenized_query)q_vec=self.vectorizer.transform([q])returncosine_similarity(q_vec,self.doc_vectors).ravel()

第三步：FaqService

从 MySQL 读出问答与tokens（JSON），建 TF-IDF；查询时对用户问题分词、算相似度、过阈值。后面再抽QAPipeline并接到页面。

services/faq.py：

importjsonimportnumpyasnpfromconfigimportconfigfromdb.modelsimportFaqfromdb.sessionimportSessionLocalfromloggerimportloggerfromservices.tfidfimportTfidfSearchfromutils.preprocessimportpreprocess_textclassFaqService:def__init__(self):self.ids=[]self.questions=[]self.answers=[]self.tfidf=Noneself._reload()def_reload(self):"""从 MySQL 读取问答 + tokens，建 TF-IDF。"""withSessionLocal()assession:db_rows=session.query(Faq).all()self.ids=[]self.questions=[]self.answers=[]tokenized=[]forrowindb_rows:ifnotrow.tokens:continuetry:tokens=json.loads(row.tokens)exceptException:logger.warning(f"FAQ 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.tfidf=Nonelogger.warning("MySQL 无可用 FAQ 分词，请先执行: python cli.py ingest-faq")returnself.tfidf=TfidfSearch(tokenized)logger.info(f"FAQ 索引加载完成：{len(self.ids)}条（MySQL 问答 + tokens）")defsearch(self,query,threshold=None):"""命中返回 (answer, score)，未命中返回 (None, score)。"""threshold=config.TFIDF_THRESHOLDifthresholdisNoneelsethresholdifnotqueryornotself.tfidf:returnNone,0.0scores=self.tfidf.get_scores(preprocess_text(query))ifscoresisNoneorlen(scores)==0:returnNone,0.0best_idx=int(np.argmax(scores))best_score=float(scores[best_idx])logger.info(f"TF-IDF 最高分={best_score:.3f}(阈值={threshold})")ifbest_score<threshold:returnNone,best_scorereturnself.answers[best_idx],best_score

要点：启动时一次性json.loads建索引；查询只对用户问题分词；余弦相似度已在 0～1，直接用TFIDF_THRESHOLD判断即可。

第四步：接入 test-ask

先在命令里把流程写直：问候 → FAQ 检索 → miss。后面再抽 Pipeline 并接到页面。

commands/test_ask.py：

importtyperfromloggerimportloggerfromservices.faqimportFaqServicefromservices.greetingimportmatch_greetingdefregister(app):@app.command("test-ask",help="对话流程测试脚本")deftest_ask(question=typer.Option(None,help="用户问题")):ifnotquestionornotstr(question).strip():raiseSystemExit('用法: python cli.py test-ask --question="你的问题"')question=str(question).strip()answer=match_greeting(question)ifanswer:logger.info(f"stage=greeting, answer={answer}")returnfaq_service=FaqService()answer,score=faq_service.search(question)ifanswer:logger.info(f"stage=faq score={score:.3f}")logger.info(f"answer={answer}")returnlogger.info(f"stage=miss score={score:.3f}")logger.info("暂未在 FAQ 中找到可靠答案，后续会接入 RAG / 联网搜索。")

python cli.py test-ask--question="你好"python cli.py test-ask--question="示例会员首月套餐多少钱？"python cli.py test-ask--question="Python 是什么"

验证

• 「你好」→stage=greeting
• 「示例会员首月套餐多少钱？」→stage=faq，答案接近 FAQ
• 「Python 是什么」→stage=miss（不在 FAQ 里）
• 日志有「FAQ 索引加载完成：… 条（MySQL 问答 + tokens）」

5.4 接口封装

上一节的流程写在test-ask里已经能跑。这里先抽成QAPipeline，再暴露为POST /api/ask，前端改为真实请求。

第一步：抽取 Pipeline

命令只负责入参和打印；业务分流放到QAPipeline.query。

pipeline.py：

fromloggerimportloggerfromservices.faqimportFaqServicefromservices.greetingimportmatch_greetingclassQAPipeline:def__init__(self):self.faq_service=FaqService()defquery(self,question):question=(questionor"").strip()logger.info(f"查询:{question!r}")answer=match_greeting(question)ifanswer:return{"answer":answer,"stage":"greeting","score":1.0}answer,score=self.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（改为调用 Pipeline）：

importtyperfromloggerimportloggerfrompipelineimportQAPipelinedefregister(app):@app.command("test-ask",help="对话流程测试脚本")deftest_ask(question=typer.Option(None,help="用户问题")):ifnotquestionornotstr(question).strip():raiseSystemExit('用法: python cli.py test-ask --question="你的问题"')pipeline=QAPipeline()result=pipeline.query(question)print(result)

python cli.py test-ask--question="示例会员首月套餐多少钱？"

行为应与上一节一致；之后接 HTTP / RAG 都复用QAPipeline。

第二步：API

routes.py（只加下面内容，原有/、/health不动）：

顶部新增 import：

frompydanticimportBaseModel,FieldfrompipelineimportQAPipeline# 定义请求参数classQueryRequest(BaseModel):question:str=Field(...,min_length=1)# 定义接口函数@router.post("/api/ask")defask(body:QueryRequest):pipeline=QAPipeline()returnpipeline.query(body.question)

注意：参数类型必须是QueryRequest，不要写成 FastAPI 的Request——后者没有.question，会报AttributeError。

第三步：前端改 fetch

templates/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.html

python app.py

浏览器： 1. 输入「你好」→ 问候话术，[greeting]2. 输入 CSV 里已有问题 → FAQ 答案，[faq · 0.xx]3. 输入无关问题 → miss 提示

curl-s-XPOST http://127.0.0.1:8000/api/ask\-H'Content-Type: application/json'\-d'{"question":"你好"}'

小结

• Greeting 短路通过关键词整句匹配（去标点）实现，命中即返回随机话术，避免闲聊进入检索流程。
• FAQ 数据经 jieba 分词、去停用词后写入 MySQL 的tokens列（JSON 字符串），导入时做去重。
• TF-IDF 索引在FaqService初始化时一次性从 MySQL 构建，查询仅对用户问题分词，余弦相似度直接以阈值 0.5 判定命中。
• 业务分流统一收敛到QAPipeline.query，Greeting → FAQ → miss 三段式便于后续接入 RAG 与联网搜索。
• 通过POST /api/ask（请求体{question: str}）对外暴露，前端以fetch提交并展示答案与stage/score。