langchain pdf链检索，提问式表单（实体命名识别）

目录

PDF检索

提问式表单

PDF检索

stuff 链，重排链，RetrievalQA 链

from PyPDF2 import PdfReader
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores import FAISS
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.chains.question_answering import load_qa_chain
from langchain.chains import RetrievalQA
import os
from dotenv import load_dotenv
from langchain_community.llms import Tongyi
 
 
load_dotenv('key.env')  # 指定加载 env 文件
key = os.getenv('DASHSCOPE_API_KEY')  # 获得指定环境变量
DASHSCOPE_API_KEY = os.environ["DASHSCOPE_API_KEY"]  # 获得指定环境变量
model = Tongyi(temperature=1)
 
# 打开 pdf
pdf_data = 'langchain入门/full_course_langchain-main/code/user_cases/impromptu-rh.pdf'
doc_reader = PdfReader(pdf_data)
 
# 把文本读取进来
raw_text = ''
for i, page in enumerate(doc_reader.pages):
    text = page.extract_text()
    if text:
        raw_text += text
 
# 文本拆分
text_splitter = CharacterTextSplitter(
    separator="\n",
    chunk_size=1000,
    chunk_overlap=200, #striding over the text
    length_function=len,
)
texts = text_splitter.split_text(raw_text)
 
# 加载 embedding
embeddings = HuggingFaceEmbeddings(model_name='bge-small-zh-v1.5', model_kwargs={'device': 'cpu'})
 
# 向量存储
docsearch = FAISS.from_texts(texts, embeddings)
 
 
 
# # 创建问答 stuff 链，意思是吧全部检索结果穿给大模型
# chain = load_qa_chain(model, chain_type="stuff")
# # 检索
# query = "这本书是哪些人创作的？请用中文回答"
# docs = docsearch.similarity_search(query, k=6)
# res = chain.run(input_documents=docs, question=query)
# print(res)
 
 
 
# # 重排链，对检索到的文档大模型打分，得到分数最高的座位结果
# chain = load_qa_chain(model,
#                       chain_type="map_rerank",
#                       return_intermediate_steps=True  # 可以看看到 map_rerank 是如何对检索到的文档打分的
#                       )
# query = "OpenAI 的创始人是谁?"
# docs = docsearch.similarity_search(query,k=10)
# results = chain({"input_documents": docs, "question": query}, return_only_outputs=True)
# print(results['output_text'])
# print(chain.llm_chain.prompt.template)  # 打印提示词模版
 
 
# RetrievalQA 链是 Langchain 已经封装好的索引查询问答链。实例化之后，我们可以直接把问题扔给它，而不需要 chain.run()。简化了很多步骤，获得了比较稳定的查询结果
docsearch = FAISS.from_texts(texts, embeddings) # 检索器是向量库数据
retriever = docsearch.as_retriever(search_type="similarity", search_kwargs={"k":4})  # as_retriever
rqa = RetrievalQA.from_chain_type(llm=model,
                                  chain_type="stuff",
                                  retriever=retriever,
                                  return_source_documents=True)
# 如果我们不需要中间步骤和源文档，只需要最终答案，那么我们可以直接请求返回结果。将代码:return_source_documents=True  改为 return_source_documents=False
query = "OpenAI 是什么?"
res = rqa(query)['result']
print(res)

提问式表单

通过用户输入的内容，识别需要填写的字段，有点实体命名识别的感觉，当需要从程序中识别特定实体时可以参考

#!/usr/bin/env python
# coding: utf-8
 
# In[ ]:
 
 
get_ipython().system('pip -q install  openai tiktoken')
get_ipython().run_line_magic('pip', 'install git+https://github.com/hwchase17/langchain')
 
 
# In[2]:
 
 
import os
 
os.environ["OPENAI_API_KEY"] = ""
 
 
# In[3]:
 
 
get_ipython().system('pip show langchain')
 
 
# ## Classification / Tagging
# 
 
# In[44]:
 
 
from langchain.chat_models import ChatOpenAI
from langchain.chains import LLMChain
from langchain.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from enum import Enum
from langchain.chains.openai_functions import (
    create_tagging_chain,
    create_tagging_chain_pydantic,
)
 
 
# In[11]:
 
 
class PersonalDetails(BaseModel):
    # 定义数据的类型
    name: str = Field(
        ...,
        description = "这是用户输入的名字"
    )
    city: str = Field(
        ...,
        description = "这是用户输入的居住城市"
    )
    email: str = Field(
        ...,
        description = "这是用户输入的邮箱地址"
    )
 
 
# In[7]:
 
 
llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo-0613")
 
 
# In[22]:
 
 
chain = create_tagging_chain_pydantic(PersonalDetails,llm)
 
 
# In[184]:
 
 
test_str1 = "你好，我是美丽，我住在上海浦东，我的邮箱是： liteli1987@gmail.com"
test_res1 = chain.run(test_str1)
test_res1
 
 
# 第二个测试，我只告诉她我的邮箱。机器人只记录了邮箱。
 
# In[174]:
 
 
test_str2 = "我的邮箱是： liteli1987@gmail.com"
test_res2 = chain.run(test_str2)
test_res2
 
 
# 我们可以再来第三个测试，不告诉机器人我的名字，但是告诉他邮箱，而且还故意告诉他我弟弟的邮箱。
 
# In[177]:
 
 
test_str3 = "我叫美丽，我弟弟的邮箱是：1106968391@qq.com"
test_res3 = chain.run(test_str3)
test_res3
 
 
# 我们可以看到结果是'', 如果没有匹配我们定义的PersonalDetails对象里的数据类型，机器人并不会瞎编乱造。
 
# In[35]:
 
 
user_007_personal_details = PersonalDetails(name="",city="",email="")
 
 
# In[36]:
 
 
user_007_personal_details
 
 
# 定义一个函数，用于检查数据是否填写完整。
 
# In[37]:
 
 
def check_what_is_empty(user_personal_details):
    ask_for = []
    # 检查项目是否为空
    for field,value in user_personal_details.dict().items():
        if value in [None, "", 0]: 
            print(f"Field '{field}' 为空" )
            ask_for.append(f'{field}')
    return ask_for        
 
 
# 我们来测试一下用户007是否填写完整了。
 
# In[38]:
 
 
ask_for = check_what_is_empty(user_007_personal_details)
ask_for
 
 
# 我们再来定义一个函数，用于获取用户的输入信息并且更新用户的信息。
 
# In[180]:
 
 
def add_non_empty_details(current_details:PersonalDetails, new_details:PersonalDetails):
    # 这是已经填好的用户信息
    non_empty_details = {k:v for k,v in new_details.dict().items() if v not in [None, "", 0]}
    update_details = current_details.copy(update=non_empty_details)
    return update_details
 
 
# In[181]:
 
 
res = chain.run("我的名字007") 
user_007_personal_details = add_non_empty_details(user_007_personal_details,res)
user_007_personal_details
 
 
# In[149]:
 
 
res = chain.run("我住在南京") 
user_007_personal_details = add_non_empty_details(user_007_personal_details,res)
user_007_personal_details
 
 
# In[150]:
 
 
res = chain.run("我的邮箱是XX@qq.com") 
user_007_personal_details = add_non_empty_details(user_007_personal_details,res)
user_007_personal_details
 
 
# 测试一下哪一项没有填写
 
# In[182]:
 
 
ask_for = check_what_is_empty(user_007_personal_details)
ask_for
 
 
# In[152]:
 
 
if not ask_for:
    print("谢谢您的回答！我没有问题了")
 
 
# 使用自定义提示模板，实现机器人发起提问。
 
# In[61]:
 
 
def ask_for_info(ask_for=["name","city","email"]):
    # 定义一个提示模板
    first_prompt = ChatPromptTemplate.from_template(
        """
        假设你现在是一名前台，你现在需要对用户进行询问他个人的具体信息。
        不要跟用户打招呼！你可以解释你需要什么信息。不要说“你好！”！
        接下来你和用户之间的对话都是你来提问，凡是你说的都是问句。
        你每次随机选择{ask_for}列表中的一个项目，向用户提问。
        比如["name","city"]列表，你可以随机选择一个"name", 你的问题就是“请问你的名字是？”
        """
    )
    info_gathering_chain = LLMChain(llm=llm, prompt=first_prompt)
    chat_chain = info_gathering_chain.run(ask_for=ask_for)
    return chat_chain
    
 
 
# In[183]:
 
 
ask_for_info(ask_for=["name","city","email"])
 
 
# 定义一个处理模型返回信息的函数。
 
# In[104]:
 
 
def filter_response(text_input, user_details):
    # 我们使用到openAI提供的打标签链, 用户的输入可以被这个链解析和映射。
    chain = create_tagging_chain_pydantic(PersonalDetails,llm)
    res = chain.run(text_input)
    # 更新用户信息
    user_details = add_non_empty_details(user_details,res)
    ask_for = check_what_is_empty(user_details)
    return user_details, ask_for
 
 
# 测试一下效果，我们要获取的就是最新的用户信息，筛选出来哪些项目还没有填写。
# 
 
# In[155]:
 
 
def decide_ask(ask_for=["name","city","email"]):
    if ask_for:
        ai_res = ask_for_info(ask_for=ask_for)
        print(ai_res)
    else:
        print("全部填写完整")
decide_ask(ask_for)        
 
 
# In[167]:
 
 
user_999_personal_details = PersonalDetails(name="",city="",email="")
user_999_personal_details
 
 
# In[168]:
 
 
decide_ask(ask_for)
 
 
# In[169]:
 
 
str999 = "我的名字是999,我住在北京"
 
 
# In[170]:
 
 
user_999_personal_details, ask_for_999 = filter_response(str999,user_999_personal_details)
decide_ask(ask_for_999)
 
 
# In[173]:
 
 
str999 = "XX@XX.com"
user_999_personal_details, ask_for_999 = filter_response(str999,user_999_personal_details)
decide_ask(ask_for_999)