一、使用本地知识库构建问答应用
上篇文章基于LangChain的Prompts提示管理构建特定领域模型，如果看过应该可以感觉出来ChatGPT还是非常强大的，但是对于一些特有领域的内容让GPT回答的话还是有些吃力的，比如让ChatGPT介绍下什么是LangChain：
from langchain.llms import OpenAIimport 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 Chromafrom langchain.chains import RetrievalQA# 文本转换为向量嵌入引擎from langchain.embeddings.openai import OpenAIEmbeddingsimport 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 Chromafrom langchain.chains import RetrievalQA# 文本转换为向量的嵌入引擎from langchain.embeddings.openai import OpenAIEmbeddings# 流式回调from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandlerimport 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./simple
创建问答服务接口server.py：
from tornado.concurrent import run_on_executorfrom tornado.web import RequestHandlerimport tornado.genimport utils_responsefrom langchain import OpenAI# 向量数据库from langchain.vectorstores import Chromafrom langchain.chains import RetrievalQA# 文本转换为向量的嵌入引擎from langchain.embeddings.openai import OpenAIEmbeddingsimport 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.webimport tornado.ioloopimport tornado.httpserverimport osfrom concurrent.futures.thread import ThreadPoolExecutorfrom 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()
下面使用PostMan进行测试：
已经成功获取到答案。
