1. 为什么这一系列会写到第六篇:先聊聊我现在在做什么
坦白说,接到这个系列第六篇的题目时,我自己也愣了一下——按照原计划,前面五篇已经把GPT-4 API的基本调用、Prompt工程、函数调用(Function Calling)、流式输出这些基础路线都跑通了,该讲的坑也讲得差不多了,还能写点什么?
但真正动手准备这一篇的内容时,我发现事情远没有结束。上一篇最后给大家留了一个悬念:我们调通了GPT-4的对话能力,但一个真正的AI驱动应用,不能只依靠大模型的文本输出。你还需要音频输入、需要向量检索来扩展模型的知识边界,需要把这些能力像搭积木一样组合在一起。这篇要做的,就是把这三块拼图完整地装进同一个应用里:用Whisper把用户语音转成文字,用Weaviate做向量数据库来管理私有知识库,再让GPT-4在检索结果的基础上生成最终回复。
先说清楚这篇文章适合谁。如果你已经把OpenAI的API调通,能跑通基本的Chat Completions请求,但对Whisper的本地部署和API调用还停留在"听说过"的阶段;如果你正在纠结"到底要不要上向量数据库""上哪个向量数据库",面对一堆技术选型文档不知道从哪下手;又或者你已经决定用Weaviate,但第一次接触这个工具,想知道它的数据模型和查询逻辑到底怎么设计——那这篇文章就是为你写的。如果你是纯新手,连pip install openai都还没跑过,建议先把系列前几篇翻出来补齐基础,不然中间有些概念衔接会比较吃力。
整体思路是这样:我先用一个真实的项目需求(一个能听懂人话、能翻企业知识库的语音问答助手)串起整条链路,然后逐个拆解Whisper的接入方式、Weaviate的数据建模和检索逻辑,最后把它们和GPT-4组装成完整应用。每一部分我都会给出可以直接复制的代码,同时讲清楚为什么这么做——这是我最在意的事:授人以鱼,也授人以渔。
注意:这篇涉及Whisper的本地部署,涉及如何用API调用各项能力。整体内容完全基于OpenAI官方API的公开用法和个人实践,不涉及任何非官方访问渠道和工具。如果你所在网络环境对OpenAI API有访问限制,请先自行查阅官方文档和所在地区合规的网络访问方案,这篇文章只讨论代码和架构层面的实现思路。
2. 项目背景:一个"能听会说"的企业知识库语音助手
2.1 真实场景是什么样的
我前阵子帮朋友做了个小项目,需求特别朴素:公司内部的规章制度、产品手册散落在几十份文档里,员工想问个"年假怎么休""报销流程是什么",得自己去翻文件夹,偶尔还要问HR、问行政,效率很低。他们想做一个内部用的问答机器人,挂在企业微信或者Web页面上,员工直接发语音提问,机器人给出答案。
这个需求听起来简单,但真拆解起来,涉及的问题一个比一个现实:
- 语音输入怎么处理:员工在手机上说话,发过来是一段音频(也可能是视频里的音轨),得先转成文字。
- 回答要基于公司内部文档:GPT-4训练数据里不可能有这家公司的内部制度,直接问大模型,它只能给一个"员工手册通常包含以下内容"式的泛泛回答。必须先把文档灌进某个检索系统,让模型基于检索到的内容作答。
- 回答速度要够快:员工等不了十秒才出结果,语音转文字、向量检索、GPT生成这三段链路,每一步都要做性能取舍。
- 容错率要够高:语音识别错了怎么办?员工问得含糊怎么办?如果直接甩一个"我没听懂",这个产品基本就废了。
这个项目的技术栈正是题目里的三件套:Whisper负责语音转文字,Weaviate负责文档的向量检索,GPT-4负责基于检索结果的答案生成。整个系统架构就是典型的语音问答RAG(Retrieval-Augmented Generation)流水线。
2.2 技术选型:为什么是Whisper + Weaviate + GPT-4
技术选型是所有人盯着的第一步,我直接说结论,然后解释理由。
| 环节 | 选项 | 为什么这么选 |
|---|---|---|
| 语音转文字 | OpenAI Whisper(API / 本地部署) | 识别准确率高,中英文混说场景表现远好于传统ASR方案 |
| 向量数据库 | Weaviate | 原生支持混合检索(BM25 + 向量),自带模块化架构,对Python开发者友好 |
| 回答生成 | GPT-4 + Function Calling | 基于检索到的文档做引用式回答,并允许用工具调用做后续动作 |
先说Whisper。做语音转文字的方案其实不少,讯飞、阿里云、百度都有现成的API,为什么选Whisper?最核心的原因是大模型领域的工程化经验可以复用:Whisper的API和GPT系列在同一个平台管理,账号体系、鉴权方式、SDK调用逻辑完全一致,少学一套东西。另一个关键点是Whisper对中英文混合语音的处理能力。很多国内项目的语音里,"报销流程""ERP系统""OKR"这种中英混杂是常态,传统ASR在这种场景下往往会把英文单词识别成奇怪的中文谐音。Whisper在这类场景的鲁棒性,我实测下来是好一个档次的。
再说Weaviate。向量数据库这个赛道的选择实在太多,Pinecone、Milvus、Qdrant、Chroma……我自己在两三个项目里做过对比,选Weaviate的核心理由有三条:
第一,自带混合检索能力。Weaviate从1.x版本开始就内置了BM25稀疏检索和向量稠密检索的混合(Hybrid Search),这意味着同一个查询既可以用关键词精确匹配(比如员工问"报销流程",文档里频繁出现"报销"两个字),又可以用语义相似度匹配(员工问"我出差花的钱怎么报",文档里说的是"差旅费用报销",传统BM25会因为缺少字面匹配而漏掉)。做知识库问答,这种"既要字面匹配、又要语义扩展"的需求是刚需,Weaviate是开箱即用做得最顺的。
第二,模块化架构。Weaviate的vectorizer(向量化模块)可以配置成调用OpenAI的Embedding接口,也可以换成别的模型,甚至用本地模型。这意味着你在开发阶段可以快速用官方模块跑通,生产环境再换成自己部署的向量化模型,迁移成本很低。
第三,Python客户端成熟。Weaviate官方Python客户端的API设计很贴近开发直觉,collection.query.hybrid()这种写法读起来就是"在这个集合上做混合查询",没有繁琐的底层细节,非常适合做原型验证。
GPT-4这个环节就不多说了,前几篇已经覆盖了基础调用。这一篇的重点在于:如何把检索到的文档内容组织成Prompt,让GPT-4输出一个带引用来源的、可信的回答。这是RAG应用能不能落地到企业内部的关键——员工需要的不是一段"看起来很有道理"的话,而是"这段话是从制度文档第几节来的"。这部分后面用专门一节讲。
2.3 整条链路的运行流程
把三块拼图串起来,整个应用的运行时数据流是这样的:
- 员工在页面按住说话,松开后上传音频文件(或直接传音频URL)。
- 后端拿到音频,调用Whisper转成文本。
- 文本交给Weaviate的混合检索模块,从知识库中召回最相关的若干条文档片段。
- 把文档片段、用户原始问题、系统提示词一起拼成Prompt,发给GPT-4。
- GPT-4生成带引用的回答,返回给前端展示。
我在本地的开发机上是这样模拟这个流程的:用一个Flask应用接收音频上传,跑通全链路。后面第三部分讲Whisper,第四部分讲Weaviate,第五部分把GPT-4组装进来,第六部分讲我在实际开发中踩到的一堆坑。每一步我都会把代码和当时的思考过程写清楚。
3. Whisper接入:从API调用到本地部署的完整方案
3.1 先用OpenAI API快速验证
如果只是想验证"Whisper能不能满足我的识别需求",最快的办法是直接调API。OpenAI的音频转文字接口很简单,Python SDK里一个方法就搞定了。先装依赖:
pip install openai python-dotenv然后写一个最小可用的转写函数:
import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) def transcribe_audio(file_path: str, language: str = "zh") -> str: with open(file_path, "rb") as audio_file: transcript = client.audio.transcriptions.create( model="whisper-1", file=audio_file, language=language, response_format="text" ) return transcript if __name__ == "__main__": result = transcribe_audio("assets/demo_question.m4a") print(result)这个接口的参数有几个值得注意的地方:
model="whisper-1":目前API上可用的Whisper模型只有一个标识,底层实现是不断迭代的,不需要自己去指定具体的大小版本(base、large之类)。language="zh":建议显式指定。如果你不确定用户说什么语言,可以不传,让Whisper自动检测。但对于内部知识库场景,业务语言基本是确定的(比如中文为主的团队),指定语言能减少识别歧义,速度也会稍快一些。response_format="text":返回纯文本。如果想拿到带时间戳的分段结果(比如要逐字校对),可以换成"verbose_json",返回的JSON里会有segments数组,每个segment有起止时间和置信度。
API方式的识别效果在大多数场景下非常能打,我拿一段20秒的中文语音测试,全程没有一个字识别错,包括"报销""差旅费"这种容易出错的词。但API方案有两个硬伤:
第一,按分钟计费,量大了成本可观。企业内部高频使用的话,一天几千条语音,费用会是个问题。 第二,延迟不可控。虽然通常也就一两秒,但遇到高峰期排队,体验就会变差。
所以在项目从原型走向生产时,我必然会考虑:把Whisper本地化部署。
3.2 本地部署Whisper服务的实操细节
本地部署Whisper,看起来只是pip install openai-whisper的事,实际做起来坑比想象的多。我先给出一套完整的、可用的搭建流程,然后逐个讲哪些地方容易翻车。
第一步,装基础依赖。Whisper底层的推理依赖PyTorch,所以环境安装顺序是:Python环境 → PyTorch → openai-whisper。建议用conda管理环境,避免系统Python被玩坏:
# 创建独立环境 conda create -n whisper-serve python=3.10 -y conda activate whisper-serve # 安装PyTorch(CPU版) pip install torch torchvision torchaudio # 安装Whisper pip install -U openai-whisper如果是NVIDIA GPU机器,PyTorch安装命令换成官方CUDA版即可,推理速度会有数量级提升。CPU版能不能用?能用,但速度很感人。一个20秒的音频,CPU上跑base模型大概需要2到5秒,跑large模型可能要30秒以上,生产环境基本没法用。如果对实时性有要求,最好上GPU,哪怕是最低端的CUDA显卡,速度也能提升十倍以上。
第二步,写一个Flask服务暴露转写接口。这里有个重要的工程决策:不要每次请求都重新加载模型。Whisper模型加载到内存要花好几秒,如果每次请求都load一遍,整个服务的QPS会惨不忍睹。正确做法是服务启动时加载一次,后续请求复用。
import os import tempfile import whisper from flask import Flask, request, jsonify app = Flask(__name__) # 只加载一次 MODEL_SIZE = os.getenv("WHISPER_MODEL_SIZE", "base") model = whisper.load_model(MODEL_SIZE) @app.route("/transcribe", methods=["POST"]) def transcribe(): if "file" not in request.files: return jsonify({"error": "no file uploaded"}), 400 audio_file = request.files["file"] with tempfile.NamedTemporaryFile(suffix=".m4a", delete=False) as tmp: audio_file.save(tmp.name) tmp_path = tmp.name try: result = model.transcribe(tmp_path, language="zh", fp16=False) return jsonify({"text": result["text"].strip()}) except Exception as e: return jsonify({"error": str(e)}), 500 finally: os.unlink(tmp_path) if __name__ == "__main__": app.run(host="0.0.0.0", port=9000, debug=False)这里我故意标了一个fp16=False,这是CPU部署最容易踩的坑之一。PyTorch在部分CPU平台上跑fp16会直接编译失败或者推理出错,显式关掉可以保证兼容性。如果是GPU部署,fp16=True能明显加速,可以留着。
第三步,性能评估。我把本地base模型和API的识别效果做了对比:纯中文短句,两者差异不大;但中英文夹杂的句子,API的Whisper-1明显更强。本地部署适合预算敏感、数据敏感(音频不出内网)的场景,API则适合追求顶配效果、开发验证阶段。这个取舍后面我会再细说。
实操提示:本地部署Whisper不意味着一定要用
openai-whisper这个包。现在还有一些更工程化的方案,比如用faster-whisper(基于CTranslate2的推理加速),在CPU上就能跑到接近实时的速度,且显存占用更低。如果你的团队对延迟特别敏感,建议直接调研faster-whisper。我在这篇里用官方包是因为它最通用、最容易复现。
3.3 为什么我不建议在生产环境直接调Whisper API
生产环境的选择,很多时候不是技术问题,而是成本和组织问题。我帮朋友做这个项目时,一开始确实想直接用API,毕竟省事,一个HTTP请求就完事了。但算了一笔账后,决定改成本地部署。
算账的方式很简单:假设每天有300条语音提问,每条平均15秒。Whisper的计费是按分钟算的,15秒音频算0.25分钟,300条就是75分钟。API价格是每分钟0.006美元,一天就是0.45美元,一个月13.5美元。看着不多,但这只是转写费用。如果后续要做得更精细,比如每段音频做说话人分离、做关键词高亮,费用还会涨。
本地部署的一次性成本是:一台带GPU的开发机(或者云上租一台带GPU的实例),加上模型文件(base模型约150MB,large约3GB),加上日常运维。如果团队已经有GPU机器在跑其他模型,边际成本几乎为零。
另外还有一个被很多人忽略的因素:音频数据的内网合规性。企业内部员工的语音提问,多多少少涉及业务内容甚至个人信息。如果音频文件全部传到第三方API,很多公司的信息安全部门会直接一票否决。本地部署把音频处理留在内网,这个在合规审查时是很有分量的加分项。
当然,如果只是做一个Demo、一个MVP,我强烈建议先用API跑通逻辑。不要在一开始就陷入本地部署的基建泥潭——先把产品逻辑验证对了,再决定要不要为规模化做本地化改造。这是我从好几次"一上来就像搭生产环境结果被琐事拖垮"的经历里总结出的教训。
4. Weaviate知识库搭建:数据建模与混合检索
4.1 Weaviate的基础概念和部署方式
Whisper把语音变成文本后,下一步的文本要进Weaviate做检索。先花点篇幅讲清楚Weaviate的几个基本概念,因为后面代码里会涉及,而且网上很多教程把这几个概念写得云里雾里。
Weaviate是开源的向量数据库,核心抽象有两层:
- Collection(集合),可以类比关系数据库里的表。
- Object(对象),集合里的一条数据记录,类比表的行。每个对象可以有自己的向量(vector)和属性(properties)。
特别要说明的一点是:Weaviate的向量不一定要自己算好再传进去。它支持"模块化"机制,也就是在创建Collection时指定一个向量化模块(如text2vec-openai),插入文档时Weaviate会自动调用OpenAI的Embedding接口把文本转成向量存储,查询时也会自动把你的query转成向量。这种设计大大降低了上手门槛——你不用关心向量是怎么来的,只需要关心业务对象长什么样。
部署Weaviate最简单的方式是用Docker。下面这段docker-compose.yml是官方文档的标准配置,我加了一条注释说明环境变量的作用:
version: '3.4' services: weaviate: image: semitechnologies/weaviate:1.24.0 command: - --host - 0.0.0.0 - --port - '8080' - --scheme - http environment: QUERY_DEFAULTS_LIMIT: 25 AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED: 'true' PERSISTENCE_DATA_PATH: '/var/lib/weaviate' DEFAULT_VECTORIZER_MODULE: 'text2vec-openai' ENABLE_MODULES: 'text2vec-openai,generative-openai' OPENAI_APIKEY: 'sk-你的key' CLUSTER_HOSTNAME: 'node1' ports: - 8080:8080 volumes: - weaviate_data:/var/lib/weaviate volumes: weaviate_data:几个要点:
ENABLE_MODULES里我开了两个模块:text2vec-openai(文本向量化)和generative-openai(生成回复)。后者在Weaviate里可以直接调用GPT做生成,但我在实际项目中更倾向自己控制Prompt,所以这个模块开了但主要用它做Embedding。OPENAI_APIKEY是让Weaviate调用OpenAI接口时用的。这里有个安全提示:生产环境不要把这个key硬编码在compose文件里,应该用环境变量注入或密钥管理服务。- 端口映射用
8080:8080,So后续Python客户端连接地址是http://localhost:8080。
启动命令:
docker compose up -d启动完可以用Python确认服务状态:
import weaviate client = weaviate.connect_to_local() print(client.is_ready()) # True说明服务已就绪 weaviate.close # 注意这个写法在最新版SDK中已改为 client.close()4.2 数据建模:把文档切成"合适大小"的块
知识库文档不能整篇塞进一个对象里,那样检索效果会非常差。正确做法是把文档拆分成多个"chunk"(文本块),每块作为一个独立对象存入Weaviate。
为什么不能整篇存?道理很简单:向量检索的本质是"找到一个和你问题语义最接近的文本块"。如果一个文本块是三千字的完整制度文档,员工问"年假怎么休",这个文档的语义被大量其他内容稀释,和问题的向量相似度就会被拉低。反之,如果把文档切成一两百字的小块,"年假申请流程:工作满一年可享受5天年假"这样的独立文本块,和问题的语义距离就非常近,检索命中率会更高。
分块策略我踩过不少坑,总结出一个相对可靠的默认配置:
- 每块长度:200到400个token左右。太短则语义不完整,太长则检索不精准。中文按字符算,我一般用250到600个字符。
- 块与块之间要有重叠:比如每块300字,滑动步长250字,前块尾部50字会出现在下一块开头。重叠避免切分时把完整句子切断,导致答案语义不全。
- 按语义边界切分:优先在段落结束、句子结束的地方切,避免在句子中间硬切。简单做法是按
\n\n切第一轮,再按句号切第二轮。
我写了一个简化版的分块器(基于文本自然边界),方便理解思路:
import re def split_document(text: str, chunk_size: int = 300, overlap: int = 50) -> list[str]: # 先把文章按段落拆开,尽量保证块边界落在自然段落 paragraphs = re.split(r'\n\s*\n', text.strip()) chunks = [] current_chunk = "" for para in paragraphs: if len(current_chunk) + len(para) <= chunk_size: current_chunk += "\n" + para else: if current_chunk: chunks.append(current_chunk.strip()) # 这里做overlap:保留当前块最后一段的尾部,作为下一块的起始 tail = current_chunk[-overlap:] if current_chunk and overlap else "" current_chunk = tail + "\n" + para if current_chunk.strip(): chunks.append(current_chunk.strip()) return chunks实际生产中,分块器和关键词提取、标题结构分析往往是联动的。比如公司制度文档里有明确的一级标题和二级标题,我会把标题信息跟着chunk一起存进对象属性里,以便生成回答时能给引用出处。这个属性设计在下一节说。
4.3 创建Collection、灌入文档数据的完整代码
在Weaviate中新建Collection的Python代码,核心是定义properties。我把每个chunk设计成四个字段:
| 属性名 | 类型 | 说明 |
|---|---|---|
document_id | string | 原始文档唯一标识 |
title | string | 文档标题或章节标题 |
chunk_text | string | 分块后的文本内容 |
page_no | int | 页码(可选) |
创建Collection的代码:
import weaviate import weaviate.classes as wvc client = weaviate.connect_to_local() # 如果集合已经存在,先删除(开发阶段用) if client.collections.exists("KnowledgeChunk"): client.collections.delete("KnowledgeChunk") collection = client.collections.create( name="KnowledgeChunk", properties=[ wvc.config.Property(name="document_id", data_type=wvc.config.DataType.TEXT), wvc.config.Property(name="title", data_type=wvc.config.DataType.TEXT), wvc.config.Property(name="chunk_text", data_type=wvc.config.DataType.TEXT), wvc.config.Property(name="page_no", data_type=wvc.config.DataType.INT), ], vectorizer_config=wvc.config.Configure.Vectorizer( text2vec_openai=wvc.config.Configure.Text2VecOpenAI( model="text-embedding-3-small" ) ), ) print(f"Collection created: {collection.name}") client.close()向量化模型我选了text-embedding-3-small而不是更大的text-embedding-3-large。原因有两方面:一是embedding的维度直接影响存储成本和检索速度,small版本只有1536维,足够处理企业内部制度文档这种语义复杂度不高的文本;二是small模型在API上的价格远低于large,在这个项目里没必要为不太可能出现的极端语义场景多花钱。
灌入数据的代码:
from datasets import load_dataset # 这里以上传本地文档为例 import os client = weaviate.connect_to_local() collection = client.collections.get("KnowledgeChunk") # 假设你已经有了一个 documents 列表,每个元素是 {"doc_id": ..., "title": ..., "text": ...} documents = [ {"doc_id": "policy-001", "title": "年假管理办法", "text": "第一条 员工入职满一年可享受5天带薪年假……"}, # 更多文档... ] with collection.batch.fixed_size(batch_size=100) as batch: for doc in documents: chunks = split_document(doc["text"]) for idx, chunk in enumerate(chunks): batch.add_object( properties={ "document_id": doc["doc_id"], "title": doc["title"], "chunk_text": chunk, "page_no": idx + 1, } ) print("Data ingestion completed") client.close()灌数据这段我在开发时踩过一个特别隐蔽的坑:with collection.batch.fixed_size(...)这个上下文管理器内部的错误处理。如果你在批量插入时某个对象因为格式问题报错(比如文本字段超过了对象大小上限),默认配置会让整个batch异常退出,而不是跳过出错的那条。开发阶段不明显,数据量大了之后会让人很头痛。后来我在batch里加了callback来逐个捕获错误,才把问题定位清楚。这个细节我放到第六节"实战坑点"再展开。
4.4 混合检索:BM25 + 向量的实战用法
检索是知识库问答的灵魂。Weaviate支持三种检索模式:纯向量检索(near_text)、关键词BM25检索、以及两者的混合(hybrid)。我的核心推荐是默认用hybrid,理由前面已经铺垫过了。
下面这段代码演示了hybrid检索的完整用法,并带着解释每个参数的含义:
import weaviate client = weaviate.connect_to_local() collection = client.collections.get("KnowledgeChunk") query_text = "员工出差产生的打车费怎么报销" response = collection.query.hybrid( query=query_text, limit=5, alpha=0.5, query_properties=["chunk_text"], ) for obj in response.objects: print(f"标题: {obj.properties['title']}") print(f"内容: {obj.properties['chunk_text'][:100]}...") print(f"分数: {obj.metadata.score}") print("---") client.close()参数alpha是hybrid检索最核心的旋钮,取值范围0到1。alpha=0时,该查询变成纯BM25关键词检索;alpha=1时,变成纯向量检索;alpha=0.5是官方推荐的默认值,表示两种检索权重各半。实际调参时我建议从0.5起步,然后根据检索效果微调:
- 如果发现检索结果经常漏掉"换了个说法但语义相同"的内容,适当调高
alpha偏向向量检索。 - 如果发现检索结果经常返回一些字面相关但语义跑偏的内容,适当降低
alpha偏向关键词。
limit=5控制返回的候选片段数量。这个值不是越大越好,因为最终这些片段会全部塞进Prompt里,多了不仅浪费token,还会让GPT-4的回答不聚焦。我测试下来,知识库文档质量中等时,5到8条是性价比最高的区间。
另外多说一句:query_properties指定检索的字段范围。我们这里只检索chunk_text,不检索title。如果你希望"员工问问题时部分靠标题就能命中"(比如员工问"年假政策"),可以把title也加进检索字段,但这样可能会引入一些标题相关但内容不相关的噪声。我个人的习惯是主检索只打在文本内容字段上,标题的作用放到生成答案阶段作为引用来源,检索靠内容本身说话。
跑通检索之后,我强烈建议做一个简单的评估脚本:准备20个典型问题,逐个跑hybrid检索,看返回的前5条里有多少条真正和答案相关。不需要多复杂的评估指标,人工看一眼就行。这个"20条问题集"在后续调整分块大小、调整alpha参数时非常有用,每次改完参数跑一遍,好坏一目了然。
5. 组装GPT-4:构建带引用来源的Answer API
5.1 构造检索增强Prompt的核心方法论
检索拿到了相关文档片段,但把片段一股脑扔给GPT还不够。很多人做RAG翻车,就翻在Prompt构造这一步——要么不给模型明确指令让它"只基于文档回答",要么给了指令却没给引用来源的格式规范。
我的Prompt模板是这样的,逻辑很清晰:角色设定 → 指令边界 → 检索内容 → 用户问题 → 输出格式规范。
SYSTEM_PROMPT = """ 你是一个企业知识库问答助手。你的任务是根据提供的【检索文档片段】回答用户问题。 严格遵守以下规则: 1. 只根据检索文档片段回答,不得使用你自己的常识或猜测补充事实性内容。 2. 如果检索片段中没有相关信息,直接回答"知识库中未找到相关信息",不要编造。 3. 回答必须包含引用来源,引用格式为【来源:{title}】。 4. 如果检索片段之间有内容冲突,如实说明冲突,并引用不同来源。 5. 回答语言与用户提问语言保持一致,简洁通顺,总字数尽量不超过200字。 """关键在第3条:必须包含引用来源。这是企业内部知识库应用能不能被信任的分水岭。员工看到答案旁边跟着"来源:年假管理办法",他才会相信这不是模型瞎编的。我在开发时也验证过:不加引用约束时,GPT-4偶尔会用自己的通用知识补全内容,而这些通用知识放在企业特定制度下很可能是错的。加了引用约束后,这种行为被明显抑制。
构造Prompt的代码,注意检索出来的片段要带上相关性和完整上下文:
def build_prompt(query_text: str, search_results: list) -> str: context_text = "\n\n".join([ f"[片段{i+1}|来源:{obj.properties['title']}]\n{obj.properties['chunk_text']}" for i, obj in enumerate(search_results) ]) user_content = f"""检索到以下文档片段: ===开始=== {context_text} ===结束=== 用户问题:{query_text} 请根据片段回答问题,务必包含引用来源。""" return user_content这里有个细节:每个片段前加了[片段i|来源:xxx]标记。这样做有两个好处——一是让GPT明确看到每个片段的边界,二是当它引用第几个片段时,我们能根据最终的引用标签回溯到原始文档,做审计和校对。
5.2 调用GPT-4并解析引用输出的完整代码
调GPT-4的代码和系列前面几篇大同小异,关键是在输出层面预留扩展点。我不建议直接把GPT返回的纯文本甩给前端,而是先用一个函数把引用来源从文本里识别出来,结构化成前端友好的数据。
先看基础调用:
import os from openai import OpenAI from dotenv import load_dotenv import weaviate load_dotenv() client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) weaviate_client = weaviate.connect_to_local() collection = weaviate_client.collections.get("KnowledgeChunk") def answer_question(query_text: str) -> dict: # 1. 检索 response = collection.query.hybrid( query=query_text, limit=5, alpha=0.5, query_properties=["chunk_text"], ) search_results = response.objects if not search_results: return {"answer": "知识库中未找到相关信息", "sources": []} # 2. 构造Prompt user_content = build_prompt(query_text, search_results) # 3. 调用GPT-4 completion = client.chat.completions.create( model="gpt-4", messages=[ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_content}, ], temperature=0.2, max_tokens=1024, ) answer_text = completion.choices[0].message.content # 4. 从答案里提取【来源:xxx】引用标签 import re sources = re.findall(r"【来源:([^】]+)】", answer_text) sources = list(set(sources)) return {"answer": answer_text, "sources": sources}temperature=0.2是我在知识库问答场景偏好的设置。知识库问答核心诉求是"准确、忠实",不是"有创造力",所以温度要压低,让模型更少自由发挥。
max_tokens=1024对大部分企业内部问答足够。如果某些制度问题需要长篇回答,可以适当调高,但200字限制的Prompt已经约束了输出长度,不需要无脑给大值。
提取引用标签用正则【来源:xxx】,简单可靠。如果人格外喜欢用别的引用格式,就调整正则,这里不展开。
5.3 应用组装:Flask档案接口把全链路串起来
有了上面两个核心函数,接下来把它包成一个Flask接口。这个接口接收两部分数据:一段音频文件,以及可选的文本问题。如果是音频就先用Whisper转文字,再走问答链路。
import os import tempfile from flask import Flask, request, jsonify import whisper app = Flask(__name__) # 加载Whisper模型(全局只加载一次) asr_model = whisper.load_model("base") def speech_to_text(audio_path: str) -> str: result = asr_model.transcribe(audio_path, language="zh", fp16=False) return result["text"].strip() @app.route("/api/ask", methods=["POST"]) def ask(): # 支持两种输入:audio文件 或 query文本 file = request.files.get("audio") text = request.form.get("query", "") if file is not None: with tempfile.NamedTemporaryFile(suffix=".m4a", delete=False) as tmp: file.save(tmp.name) tmp_path = tmp.name try: text = speech_to_text(tmp_path) finally: os.unlink(tmp_path) if not text: return jsonify({"error": "empty query"}), 400 result_data = answer_question(text) return jsonify(result_data) if __name__ == "__main__": app.run(host="0.0.0.0", port=5000, debug=False)到这一步,完整链路就跑通了。用curl测试一下:
curl -X POST http://localhost:5000/api/ask \ -F "audio=@demo_question.m4a" \ -F "query="返回示例:
{ "answer": "根据《差旅费用管理制度》,员工出差产生的市内交通费可以凭发票报销,原则上凭票实报实销。【来源:差旅费用管理制度】", "sources": ["差旅费用管理制度"] }看起来已经是一个可用的MVP了。但真正把它用在生产环境之前,有几个结构性问题和坑我务必要单独写一节,因为这些是我实际做完项目后觉得最有价值的东西。
6. 实测中踩过的坑与调优方案
6.1 批量导入数据时的静默失败:一个定位了很久的问题
接前文埋的伏笔。运行灌数据脚本时,几千个对象导入看似成功,日志也没报错,但检索时明显发现某些文档的内容永远搜不到。我一度怀疑是分块逻辑的问题,排查了一圈,最后定位到Weaviate的batch写入机制。
问题出在这里:collection.batch.fixed_size()的上下文管理器,默认情况下如果batch中某一个对象的写入失败了,它不会把错误抛出来打断整个流程,而是可能静默丢弃,或者等到上下文退出时才统一报错。数据量小的时候你注意不到,几千条里偶尔丢一两条,检索结果就缺了一块。
解决办法是在batch创建时传入一个callback,显式捕获错误并记录:
def batch_callback(result): if result.has_errors: for err in result.errors: print(f"写入失败: {err.object_id if err.object_id else 'unknown'} - {err.message}") with collection.batch.fixed_size(batch_size=100, callback=batch_callback) as batch: # ... 添加对象 ...加了callback之后,丢数据的问题立即现形。最常见的原因是对象属性值包含空字符串或格式异常,其次是page_no类型不对(比如传了字符串"1"而不是整数1)。Weaviate对类型敏感度比关系数据库高不少,字段类型不匹配时会直接拒绝写入。这是个很疼、但定位后很好改的坑。
6.2 语音识别的两类高频错误和处理策略
Whisper虽然强大,但远非万能。我实测中遇到两类特别典型的问题:
第一类是专业名词被"音译"。比如员工语音说"ERP系统",Whisper可能识别成"伊尔皮系统"或者"ERP 系统"(是的,它有时候能识别对),GPT-4拿着这个文本去做向量检索,和文档里的"ERP"匹配度大打折扣。更夸张的是"OA审批"被识别成"欧艾审批",这种错误在混合检索阶段会造成灾难性的检索失败。
处理这类问题我试过几种方案,最后比较有效的是:
- 在分块和检索时都保留中英文别名:比如在文档中把"ERP系统"改成"ERP系统(企业资源计划系统)",让向量检索有更多语义锚点。
- 做语音文本的后处理纠错:维护一个常见术语表(term list),Whisper转写完成后做一轮替换。比如正则替换
伊尔皮→ERP,欧艾→OA。术语表的维护成本很低,但收益非常直接,尤其是内部有特定缩写体系的企业。
TERM_PATCHES = [ (r"伊尔皮\s*系统", "ERP系统"), (r"欧艾\s*审批", "OA审批"), # 按需扩充 ] def patch_transcript(text: str) -> str: for pattern, replacement in TERM_PATCHES: text = re.sub(pattern, replacement, text, flags=re.IGNORECASE) return text第二类是语音分割不当导致的问题。用户可能一口气说了两句话,中间有停顿,Whisper默认会把它当作一整段处理,转写出的文本可能把句号漏掉。这在后续向量检索时不是大问题,但Prompt里给GPT-4的文本会显得不自然。我建议在转写后做一个简单的句子切分(按句号、问号分句),对后续的引用处理也有帮助。
6.3 引用来源的稳定性:让GPT-4输出结构化引用的两个技巧
【来源:xxx】这种正则提取方式首次测试就很灵,但用了一段时间后我发现一个问题:如果检索到5条结果,其中两条标题非常相似(比如"差旅费用管理制度"和"差旅费用管理制度(修订版)"),GPT-4在回答时可能自己改写标题,比如把来源写成"差旅管理制度",正则匹配后sources里这条来源就无法和原始检索结果对应上。
针对这个问题,我在Prompt里加了一条约束:引用标题必须完全使用片段标记中提供的原样标题,不要改写、不要缩写。同时把提取和校验逻辑改为:提取引用标签后,与原始搜索结果中的title列表做匹配,如果某条引用在原始列表中不存在,就归为"未匹配来源",至少前端还能展示它,不会凭空丢失。
另一种更稳的做法是放弃让GPT生成引用格式,改为让它输出引用编号,比如"[1]""[3]"对应检索片段的编号。编号比标题字符串更不容易被改写,也不占用额外的token。后端把编号和标题映射表返回给前端。这个方案我后面的项目里用得越来越多,只是这篇为了贴近"最简可复现"的初衷,演示了字符串引用的做法,升级方向说清楚即可。
6.4 延迟优化:从全链路耗时到逐环节消耗分析
企业内部问答对延迟的敏感度比想象的高。用户语音发出去,转圈超过3秒,很多人就退了。我把全链路大致拆成了三部分,逐一测耗时:
| 环节 | 耗时(本地CPU部署Whisper base + 云上GPT-4 + 本地Weaviate) | 优化优先级 |
|---|---|---|
| Whisper语音转文字(20秒音频) | 约1.5~3秒 | 高 |
| Weaviate混合检索(5条结果) | 约0.1~0.3秒 | 低 |
| GPT-4生成回答 | 约2~5秒 | 高 |
Whisper和GPT-4加起来占了90%以上的耗时。优化思路有这么几条:
- Whisper换faster-whisper或上GPU。前面说过,这是最直接的手段。CPU上base模型转20秒音频需要两三秒,换faster-whisper在同样CPU上能压到一秒以内,换成GPU更是百毫秒级。
- GPT-4用流式输出。这部分我在系列前几篇详细讲过,核心是不要让用户干等完整回答,而是让回答像打字机一样一个字一个字蹦出来。对语音问答这种场景,用户看文字逐步出现,心理等待时间会大幅缩短。
- 把Whisper和GPT-4两个串行环节改成流水线:语音还在转写时,如果用户的问题是纯文本输入(比如在Web页面上除了语音框还有个文本输入框),文本问题可以直接先走检索,等语音转写完成后合并处理。但如果是纯语音输入,这两个环节天然串行,只能靠单点加速。
我最终给朋友的方案是:CPU服务器上部署faster-whisper + 云上GPT-4流式输出,全链路实测感知延迟在2秒左右,用户能接受。这也是我这篇文章的落地结论之一——技术选型不是非黑即白,API和本地部署可以混用,关键是每一环都有它的性能账要算清楚。
7. 完整代码仓库目录结构与一键启动说明
为了让大家能直接跑起来,我把整个应用的代码结构整理成了下面这样,方便你对照着搭建:
ai-voice-assistant/ ├── app/ │ ├── __init__.py │ ├── asr.py # Whisper转写封装 │ ├── vector_store.py # Weaviate检索封装 │ ├── llm.py # GPT-4问答封装 │ └── main.py # Flask应用入口 ├── data/ │ └── raw_docs/ # 原始PDF/Word文档,可自行替换 ├── scripts/ │ ├── ingest_docs.py # 文档解析 + 分块 + 灌入Weaviate │ └── eval_retrieval.py # 20条问题集检索效果评估 ├── assets/ │ └── demo_question.m4a # 测试音频文件 ├── docker-compose.yml └── requirements.txtrequirements.txt内容如下:
openai>=1.0.0 weaviate-client>=4.0.0 flask>=3.0.0 openai-whisper>=20231117 python-dotenv>=1.0.0启动顺序是:先把Weaviate容器拉起来(docker compose up -d),然后运行灌数据脚本(python scripts/ingest_docs.py),最后启动Flask服务(python app/main.py)。第一次跑通全链路,我建议先用assets里的测试音频验证,不要一上来就换自己的文档和音频——先把管线跑通,后面再怎么替换数据都不怕。
如果要换成自己的知识库,核心要做两件事:一是把data/raw_docs/里的文档换成真实制度文件,二是在ingest_docs.py里把PDF/Word的解析逻辑补上(简单方案是用pypdf或docx库提取纯文本)。分块参数、检索参数都可以用scripts/eval_retrieval.py评估后微调。
8. 回到开头:关于"艺术与科学"的一点后记
这个系列的标题叫"构建AI驱动应用的艺术与科学",前五篇更多在讲"科学"——API怎么调、参数怎么设、代码怎么写。这一篇把语音、检索、生成三块串成一个完整应用后,我反而更强烈地感受到"艺术"的部分在哪里:真正让一个AI应用可用的,往往是那些在文档里找不到的取舍。
比如Whisper用API还是本地部署,本质上不是技术问题,是成本和合规的权衡;检索用向量还是混合,不是越"先进"越好,而是要和你的文档类型、用户问法匹配;引用格式用标题字符串还是编号,直接决定了这个回答将来能不能被追溯审计。这些决策都没有标准答案,只能靠对自己业务的理解去判断。
如果让我给一条最核心的建议,就是:先做一个能跑的粗糙版本,然后用真实数据去评估,再做优化。不要一开始就把每个环节都调到极致,那是永远调不完的。我见过太多人卡在"向量数据库选型对比了三个月"——你只需要选一个顺手能跑的,跑起来再说。
这篇文章写完,我后续打算单独写一篇关于流式输出+语音问答的长链接组合优化,以及一篇生产环境多租户隔离的文章。如果你看完这篇把自己那条链路跑通了,遇到任何问题,欢迎在评论区把具体报错或者检索效果描述出来,我们评论区见。