LangChain 基于 ChatGPT 构建本地知识库问答应用

一、使用本地知识库构建问答应用

上篇文章基于 LangChain 的Prompts 提示管理构建特定领域模型，如果看过应该可以感觉出来 ChatGPT 还是非常强大的，但是对于一些特有领域的内容让 GPT 回答的话还是有些吃力的，比如让 ChatGPT 介绍下什么是 LangChain：

from langchain.llms import OpenAI
import osopenai_api_key=os.environ["OPENAI_API_KEY"]
llm = OpenAI(model_name="gpt-3.5-turbo", openai_api_key=openai_api_key)
my_text = "介绍下 langChain "
print(llm(my_text))

在这里插入图片描述

可以看出回答的貌似不是我们想要的内容，还有在一些特定的场景下的问答，比如学校学生问答系统中，学生提问周一课程是什么？，这种直接让 GPT 回答的话也是有些吃力，那这种情况下怎么解决呢？

现在有了 LangChain 那实现起来就非常简答了：

例如：现有知识内容放在了 data 目录下，有如下内容：

在这里插入图片描述
其中 txt 文件记录了 LangChain 的介绍：

在这里插入图片描述
pdf 介绍了下 langchain 中的 Prompts ：

在这里插入图片描述
csv 记录的学生课程信息：

在这里插入图片描述

下面首先对知识内容进行载入和向量化。

3.1 文本载入及 Embedding 向量持久化

import os
# 向量数据库
from langchain.vectorstores import Chroma
# 文档加载器
from langchain.document_loaders import TextLoader, CSVLoader, PyPDFLoader
# 文本转换为向量的嵌入引擎
from langchain.embeddings.openai import OpenAIEmbeddings
# 文本拆分
from langchain.text_splitter import RecursiveCharacterTextSplitteropenai_api_key = os.environ["OPENAI_API_KEY"]knowledge_base_dir = "./data"doc = []
for item in os.listdir(knowledge_base_dir):if item.endswith("txt"):loader = TextLoader(file_path=os.path.join(knowledge_base_dir, item), encoding="utf-8")doc.append(loader.load())elif item.endswith("csv"):loader = CSVLoader(file_path=os.path.join(knowledge_base_dir, item), encoding="utf-8")doc.append(loader.load())elif item.endswith("pdf"):loader = PyPDFLoader(file_path=os.path.join(knowledge_base_dir, item))doc.append(loader.load())print("提取文本量：", len(doc))
# 拆分
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=400)
docs = []
for d in doc:docs.append(text_splitter.split_documents(d))print("拆分文档数：", len(docs))
# 准备嵌入引擎
embeddings = OpenAIEmbeddings(openai_api_key=openai_api_key)
# 向量化
# 会对 OpenAI 进行 API 调用
vectordb = Chroma(embedding_function=embeddings, persist_directory="./cut")
for d in docs:vectordb.add_documents(d)
# 持久化
vectordb.persist()

运行后可以在 ./cut 看到持久化的向量内容：

在这里插入图片描述

3.2 构建问答

from langchain import OpenAI
# 向量数据库
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
# 文本转换为向量嵌入引擎
from langchain.embeddings.openai import OpenAIEmbeddings
import osopenai_api_key = os.environ["OPENAI_API_KEY"]
llm = OpenAI(temperature=0, openai_api_key=openai_api_key)# 准备好你的嵌入引擎
embeddings = OpenAIEmbeddings(openai_api_key=openai_api_key)vectordb = Chroma(embedding_function=embeddings, persist_directory="./cut")
# 创建您的检索引擎
qa = RetrievalQA.from_chain_type(llm=llm, chain_type="stuff", retriever=vectordb.as_retriever())query = "介绍下什么是 langchain？"
res = qa.run(query)
print('问题：', query, 'LLM回答：', res)query = "介绍下 langchain 中的 prompts ？"
res = qa.run(query)
print('问题：', query, 'LLM回答：', res)query = "周一需要上什么课？"
res = qa.run(query)
print('问题：', query, 'LLM回答：', res)query = "周三上午需要上什么课？"
res = qa.run(query)
print('问题：', query, 'LLM回答：', res)

在这里插入图片描述
可以看出已经精准的做出回答。

正常我们使用 ChatGPT 的时候都是以流的形式进行返回，同样这里我们也可以修改为流的形式：

from langchain import OpenAI
# 向量数据库
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
# 文本转换为向量的嵌入引擎
from langchain.embeddings.openai import OpenAIEmbeddings
# 流式回调
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
import osopenai_api_key = os.environ["OPENAI_API_KEY"]
llm = OpenAI(temperature=0, openai_api_key=openai_api_key, streaming=True, callbacks=[StreamingStdOutCallbackHandler()])# 嵌入引擎
embeddings = OpenAIEmbeddings(openai_api_key=openai_api_key)vectordb = Chroma(embedding_function=embeddings, persist_directory="./cut")
# 创建您的检索引擎
qa = RetrievalQA.from_chain_type(llm=llm, chain_type="stuff", retriever=vectordb.as_retriever())query = "介绍下什么是 langchain？"
qa.run(query)

在这里插入图片描述

下面配合 tornado 高性能异步非阻塞web框架，实现接口调用问答。

二、部署 WEB 服务

安装 tornado 框架：

pip install tornado -i https://pypi.tuna.tsinghua.edu.cn/simple

创建问答服务接口 server.py ：

from tornado.concurrent import run_on_executor
from tornado.web import RequestHandler
import tornado.gen
import utils_response
from langchain import OpenAI
# 向量数据库
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
# 文本转换为向量的嵌入引擎
from langchain.embeddings.openai import OpenAIEmbeddings
import osclass QA(RequestHandler):# 准备模型openai_api_key = os.environ["OPENAI_API_KEY"]llm = OpenAI(temperature=0, openai_api_key=openai_api_key)# 准备好你的嵌入引擎embeddings = OpenAIEmbeddings(openai_api_key=openai_api_key)vectordb = Chroma(embedding_function=embeddings, persist_directory="./cut")# 检索引擎qa = RetrievalQA.from_chain_type(llm=llm, chain_type="stuff", retriever=vectordb.as_retriever())print("检索引擎已加载！")def prepare(self):self.executor = self.application.pool@tornado.gen.coroutinedef get(self):questions = self.get_query_argument('questions')if not questions or questions == '':return utils_response.fail(message='问题为空')result = yield self.detection(questions)self.write(result)@run_on_executordef detection(self, questions):# 开始检测res = self.qa.run(questions)return utils_response.ok(res)

路由配置，并启动服务 app.py ：

import tornado.web
import tornado.ioloop
import tornado.httpserver
import os
from concurrent.futures.thread import ThreadPoolExecutor
from server import QA## 配置
class Config():port = 8081base_path = os.path.dirname(__file__)settings = {# "debug":True,# "autore load":True,"static_path": os.path.join(base_path, "resources/static"),"template_path": os.path.join(base_path, "resources/templates"),"autoescape": None}# 路由
class Application(tornado.web.Application):def __init__(self):handlers = [("/qa", QA),("/(.*)$", tornado.web.StaticFileHandler, {"path": os.path.join(Config.base_path, "resources/static"),"default_filename": "index.html"})]super(Application, self).__init__(handlers, **Config.settings)self.pool = ThreadPoolExecutor(10)if __name__ == '__main__':app = Application()httpserver = tornado.httpserver.HTTPServer(app)httpserver.listen(Config.port)print("start success", "prot = ", Config.port)print("http://localhost:" + str(Config.port) + "/")tornado.ioloop.IOLoop.current().start()