构建用户查询到智能回复的全流程：基于向量数据库和 LLM 的实践探索

上一篇提到基于 Elasticsearch 和 LLM 的实践探索(https://zhuanlan.zhihu.com/p/12528541608) 讲到基于关系数据库检索，定位不到用户真正关心问题的地方，这篇尝试下向量数据库，基于文本向量相似度来提高检索的准确性本文构建一个调用流程：用户Query -> 向量数据库检索 -> Prompt -> LLM -> 回复的 # 准备数据参考上一篇涉及到的内容和代码 ## 本地搭建检索引擎向量数据库介绍在介绍向量数据库前清楚几个概念： 1. 向量数据库的意义是快速的检索； 2. 向量数据库本身不生成向量，向量是由 Embedding 模型产生的； 3. 向量数据库与传统的关系型数据库是互补的，不是替代关系，在实际应用中根据实际需求经常同时使用。 ## 主流向量数据库功能对比生产环境推荐Milvus，Weaviate ● Milvus: 开源向量数据库，同时有云服务 https://milvus.io/ ● Weaviate: 开源向量数据库，同时有云服务 https://weaviate.io/ ● FAISS: Meta 开源的向量检索引擎 https://github.com/facebookresearch/faiss ● Pinecone: 商用向量数据库，只有云服务 https://www.pinecone.io/ ● Qdrant: 开源向量数据库，同时有云服务 https://qdrant.tech/ ● PGVector: Postgres 的开源向量检索引擎 https://github.com/pgvector/pgvector ● RediSearch: Redis 的开源向量检索引擎 https://github.com/RediSearch/RediSearch ● ElasticSearch 也支持向量检索 https://www.elastic.co/enterprise-search/vector-search ## 安装chromadb 为了方便演示，我这用内存向量数据库chromadb 安装依赖包：pip install chromadb ## 代码实现关键词检索实现将pdf文档内融合灌入到db库中，注意client.embeddings.create文本向量化比较耗时，可以先测试小文本 ``` import chromadb from chromadb.config import Settings from numpy import dot from numpy.linalg import norm from openai import OpenAI from dotenv import load_dotenv, find_dotenv import load_pdf as cxt import tiktoken import concurrent.futures import re _ = load_dotenv(find_dotenv()) # 读取本地 .env 文件，里面定义了 OPENAI_API_KEY client = OpenAI() def count_tokens(text, model="text-embedding-ada-002"): encoding = tiktoken.encoding_for_model(model) return sum(len(encoding.encode(cxt)) for cxt in text) class MyVectorDBConnector: def __init__(self, collection_name, embedding_fn): chroma_client = chromadb.Client(Settings(allow_reset=True)) # 为了演示，实际不需要每次 reset() chroma_client.reset() # 创建一个 collection self.collection = chroma_client.get_or_create_collection( name=collection_name) self.embedding_fn = embedding_fn def add_documents(self, documents): '''向 collection 中添加文档与向量''' embeddings = self.embedding_fn(documents) if len(embeddings) != len(documents): print(f"Mismatch: {len(embeddings)} embeddings for {len(documents)} documents") return self.collection.add( embeddings=embeddings, documents=documents, ids=[f"id{i}" for i in range(len(documents))] ) def add_documents_pool(self, documents, batch_size=10): '''Add documents and vectors to the collection in batches''' with concurrent.futures.ThreadPoolExecutor() as executor: futures = [] for i in range(0, len(documents), batch_size): batch_documents = documents[i:i + batch_size] futures.append(executor.submit(self.embedding_fn, batch_documents)) for i, future in enumerate(concurrent.futures.as_completed(futures)): try: embeddings = future.result() if embeddings is not None: self.collection.add( embeddings=embeddings, # 每个文档的向量 documents=batch_documents, # 当前批次的文档 ids=[f"id{j}" for j in range(i * batch_size, i * batch_size + len(batch_documents))] # 当前批次的文档 id ) except Exception as e: print(f"An error occurred while processing batch {i}: {e}") def search(self, query, top_n): '''检索向量数据库''' results = self.collection.query( query_embeddings=self.embedding_fn([query]), n_results=top_n ) return results def check_texts_type(texts): if isinstance(texts, list): print("texts is a list") elif isinstance(texts, str): print("texts is a string") else: print(f"texts is of type {type(texts)}") def get_embeddings(texts, model="text-embedding-ada-002"): try: if not isinstance(texts, (list, tuple)): print("错误：'texts' 不是可迭代的类型。请确保它是列表或元组。") response = client.embeddings.create(input=texts, model=model) if response.data is None: print("Response data is None") return [] return [x.embedding for x in response.data] except Exception as e: print(f"An error occurred: {e}") return [] # 使用示例 # asyncio.run(get_embeddings_async(paragraphs, model="text-embedding-ada-002")) def sent_tokenize(input_string): """按标点断句""" # 按标点切分 sentences = re.split(r'(?<=[。！？；?!])', input_string) # 去掉空字符串 return [sentence for sentence in sentences if sentence.strip()] def split_text(paragraphs, chunk_size=300, overlap_size=100): '''按指定 chunk_size 和 overlap_size 交叠割文本''' sentences = [s.strip() for p in paragraphs for s in sent_tokenize(p)] chunks = [] i = 0 while i < len(sentences): chunk = sentences[i] overlap = '' prev_len = 0 prev = i - 1 # 向前计算重叠部分 while prev >= 0 and len(sentences[prev])+len(overlap) <= overlap_size: overlap = sentences[prev] + ' ' + overlap prev -= 1 chunk = overlap+chunk next = i + 1 # 向后计算当前chunk while next < len(sentences) and len(sentences[next])+len(chunk) <= chunk_size: chunk = chunk + ' ' + sentences[next] next += 1 chunks.append(chunk) i = next return chunks def main(): paragraphs = ["文本1", "文本2", "文本3"] # 示例文本 vector_db = MyVectorDBConnector("demo", get_embeddings) vector_db.add_documents(paragraphs) # 运行异步主函数 #asyncio.run(main()) if __name__ == "__main__": #paragraphs = cxt.extract_text_from_pdf("/Users/liuqiang/code/ai/lq/RAG/llama2.pdf", min_line_length=10) paragraphs = cxt.extract_text_from_pdf("/Users/liuqiang/code/ai/lq/RAG/haikangweishi.pdf", min_line_length=10) # 创建一个向量数据库对象 vector_db = MyVectorDBConnector("demo", get_embeddings) #paragraphs = ["文本1", "文本2", "文本3"] # 示例文本 chunks = split_text(paragraphs, 300, 100) # 向向量数据库中添加文档 vector_db.add_documents(chunks) # user_query = "Llama 2有多少参数" user_query = "海康威视2023年营收是多少?" #user_query = "Does Llama 2 have a conversational variant" results = vector_db.search(user_query, 2) for para in results['documents'][0]: print(para+"\n") ``` 从运行结果看: ![](https://upload-images.jianshu.io/upload_images/20056691-ddc1ee7783e6582f.png) **向量数据库比之前的ES检索要准确**，给出了具体营收情况，下面结合LLM给出拟人化的回答 # 调用LLM 接口代码如下： ``` from openai import OpenAI from dotenv import load_dotenv, find_dotenv from myVertor import MyVectorDBConnector, get_embeddings,client import load_pdf as cxt import myVertor as mv prompt_template = """ 你是一个问答机器人。你的任务是根据下述给定的已知信息回答用户问题。已知信息: {context} 用户问： {query} 如果已知信息不包含用户问题的答案，或者已知信息不足以回答用户的问题，请直接回复"我无法回答您的问题"。请不要输出已知信息中不包含的信息或答案。请用中文回答用户问题。 """ def get_completion(prompt, model='gpt-4o'): '''封装 openai 接口''' messages = [{"role": "user", "content": prompt}] response = client.chat.completions.create( model=model, messages=messages, temperature=0, # 模型输出的随机性，0 表示随机性最小 ) return response.choices[0].message.content def build_prompt(prompt_template, **kwargs): '''将 Prompt 模板赋值''' inputs = {} for k, v in kwargs.items(): if isinstance(v, list) and all(isinstance(elem, str) for elem in v): val = '\n\n'.join(v) else: val = v inputs[k] = val return prompt_template.format(**inputs) class RAG_Robot: def __init__(self, vector_db, llm_api, n_results=2): self.vector_db = vector_db self.llm_api = llm_api self.n_results = n_results def chat(self, user_query): # 1. 检索 search_results = self.vector_db.search(user_query, self.n_results) # 2. 构建 Prompt prompt = build_prompt( prompt_template, context=search_results['documents'][0], query=user_query) print("===Prompt===") print(prompt) # 3. 调用 LLM response = self.llm_api(prompt) return response def loadDb(vector_db): paragraphs = cxt.extract_text_from_pdf("/Users/liuqiang/code/ai/lq/RAG/haikangweishi.pdf", min_line_length=10) # 创建一个向量数据库对象 #paragraphs = ["文本1", "文本2", "文本3"] # 示例文本 chunks = mv.split_text(paragraphs, 300, 100) # 向向量数据库中添加文档 vector_db.add_documents(chunks) if __name__ == '__main__': vector_db = MyVectorDBConnector("demo", get_embeddings) loadDb(vector_db) print("===Prompt===") bot = RAG_Robot( vector_db, llm_api=get_completion ) user_query = "海康威视2023年营收是多少?" response = bot.chat(user_query) print(response) ``` 从代码中的prompt组成，我们的问题加上向量数据库检索到的补充信息，经LLM加工处理后返回答案，如下图 ![](https://upload-images.jianshu.io/upload_images/20056691-c952decdb166f4d5.png) 和财报中的数据一致 ![](https://upload-images.jianshu.io/upload_images/20056691-cade91b37124ec0d.png) 本文由[mdnice](https://mdnice.com/?platform=6)多平台发布

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 219,928评论 6赞 509
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 93,748评论 3赞 396
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 166,282评论 0赞 357
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 59,065评论 1赞 295
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 68,101评论 6赞 395
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 51,855评论 1赞 308
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 40,521评论 3赞 420
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 39,414评论 0赞 276
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 45,931评论 1赞 319
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 38,053评论 3赞 340
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 40,191评论 1赞 352
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 35,873评论 5赞 347
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 41,529评论 3赞 331
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 32,074评论 0赞 23
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 33,188评论 1赞 272
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 48,491评论 3赞 375
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 45,173评论 2赞 357

构建用户查询到智能回复的全流程：基于向量数据库和 LLM 的实践探索

推荐阅读更多精彩内容