基于LangChain开发PDF对比工具遇Pydantic验证错误求助

问题描述 #

参照LangChain官方文档实现了一个基于Streamlit的PDF对比应用，运行时触发以下错误：

pydantic.v1.error_wrappers.ValidationError: 1 validation error for Tool
args_schema
subclass of BaseModel expected (type=type_error.subclass; expected_class=BaseModel)

应用代码如下：

import streamlit as st
import os
import tempfile
from pathlib import Path
from pydantic import BaseModel, Field
import streamlit as st
from langchain.chat_models import ChatOpenAI
from langchain.agents import Tool
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores import FAISS
from langchain.document_loaders import PyPDFLoader
from langchain.chains import RetrievalQA
from langchain.agents import initialize_agent
import openai
os.environ["OPENAI_API_KEY"] = ""
os.environ['OPENAI_API_TYPE'] = 'azure'
os.environ['OPENAI_API_VERSION'] = '2023-03-15-preview'
os.environ['OPENAI_API_BASE'] = "https://summarization"

#API settings for embedding
openai.api_type = "azure"
openai.api_base = "https://summarization"
openai.api_version = '2023-03-15-'
openai.api_key = ""
    
    
class DocumentInput(BaseModel):
    question: str = Field()

# Create a temporary directory in the script's folder
script_dir = Path(__file__).resolve().parent
temp_dir = os.path.join(script_dir, "tempDir")


def main():
    st.title("PDF Document Comparison")

    # Create a form to upload PDF files and enter a question
    st.write("Upload the first PDF file:")
    pdf1 = st.file_uploader("Choose a PDF file", type=["pdf"], key="pdf1")

    st.write("Upload the second PDF file:")
    pdf2 = st.file_uploader("Choose a PDF file", type=["pdf"], key="pdf2")

    question = st.text_input("Enter your question")
    submit_button = st.button("Compare PDFs")

    if submit_button:
        if pdf1 and pdf2:
            if not os.path.exists(temp_dir):
                os.makedirs(temp_dir)
            else:
                # Clear the previous contents of the "tempDir" folder
                for file in os.listdir(temp_dir):
                    file_path = os.path.join(temp_dir, file)
                    try:
                        if os.path.isfile(file_path):
                            os.unlink(file_path)
                    except Exception as e:
                        print(f"Error deleting file: {e}")

            # Save the PDF files to the "tempDir" directory
            pdf1_path = os.path.join(temp_dir, pdf1.name)
            with open(pdf1_path, 'wb') as f:
                f.write(pdf1.getbuffer())

            pdf2_path = os.path.join(temp_dir, pdf2.name)
            with open(pdf2_path, 'wb') as f:
                f.write(pdf2.getbuffer())



            llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo",engine="gpt-35-turbo")

            tools = []
            files = [

                {
                    "name": pdf1.name,
                    "path": pdf1_path,
                },

                {
                    "name": pdf2.name,
                    "path": pdf2_path,
                },
            ]

            for file in files:
                loader = PyPDFLoader(file["path"])
                pages = loader.load_and_split()
                text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
                docs = text_splitter.split_documents(pages)
                embeddings = OpenAIEmbeddings()
                retriever = FAISS.from_documents(docs, embeddings).as_retriever()

                # Wrap retrievers in a Tool
                tools.append(
                    Tool(
                        args_schema=DocumentInput,
                        name=file["name"],
                        description=f"useful when you want to answer questions about {file['name']}",
                        func=RetrievalQA.from_chain_type(llm=llm, retriever=retriever),
                    )
                )
            agent = initialize_agent(
                tools=tools,
                llm=llm,
                verbose=True,
            )

            st.write(agent({"input": question}))
            # Now you have both PDFs saved in the "tempDir" folder
            # You can perform your PDF comparison here


if __name__ == "__main__":
    main()

错误原因 #

LangChain的Tool类依赖Pydantic v1做参数校验，直接从pydantic导入BaseModel时，若环境中同时存在Pydantic v2，会导致args_schema类型不匹配，触发校验错误。另外，RetrievalQA链的默认输出格式也不符合Tool的预期要求。

解决步骤 #

1. 统一Pydantic导入版本 #

将原代码中的Pydantic导入替换为LangChain封装的v1版本，避免版本冲突：

from langchain.pydantic_v1 import BaseModel, Field

2. 调整Tool的func输出格式 #

RetrievalQA.from_chain_type返回的链默认输出是字典，需要提取其中的结果字段作为字符串返回给Tool，用lambda包装实现：

tools.append(
    Tool(
        args_schema=DocumentInput,
        name=file["name"],
        description=f"useful when you want to answer questions about {file['name']}",
        func=lambda x: RetrievalQA.from_chain_type(llm=llm, retriever=retriever).run(x["question"]),
    )
)

3. 可选优化：自动管理临时文件 #

使用tempfile.TemporaryDirectory自动处理临时文件的创建和清理，替代手动管理逻辑，避免残留文件：

with tempfile.TemporaryDirectory() as temp_dir:
    # 保存PDF文件到临时目录
    pdf1_path = os.path.join(temp_dir, pdf1.name)
    with open(pdf1_path, 'wb') as f:
        f.write(pdf1.getbuffer())
    
    pdf2_path = os.path.join(temp_dir, pdf2.name)
    with open(pdf2_path, 'wb') as f:
        f.write(pdf2.getbuffer())
    
    # 后续LLM、Tool、Agent逻辑放在此处

修改后的完整代码 #

import streamlit as st
import os
import tempfile
from pathlib import Path
from langchain.pydantic_v1 import BaseModel, Field
from langchain.chat_models import ChatOpenAI
from langchain.agents import Tool
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores import FAISS
from langchain.document_loaders import PyPDFLoader
from langchain.chains import RetrievalQA
from langchain.agents import initialize_agent
import openai

os.environ["OPENAI_API_KEY"] = ""
os.environ['OPENAI_API_TYPE'] = 'azure'
os.environ['OPENAI_API_VERSION'] = '2023-03-15-preview'
os.environ['OPENAI_API_BASE'] = "https://summarization"

# API settings for embedding
openai.api_type = "azure"
openai.api_base = "https://summarization"
openai.api_version = '2023-03-15-preview'
openai.api_key = ""
    
class DocumentInput(BaseModel):
    question: str = Field()

def main():
    st.title("PDF Document Comparison")

    pdf1 = st.file_uploader("Upload the first PDF file:", type=["pdf"], key="pdf1")
    pdf2 = st.file_uploader("Upload the second PDF file:", type=["pdf"], key="pdf2")
    question = st.text_input("Enter your question")
    submit_button = st.button("Compare PDFs")

    if submit_button:
        if pdf1 and pdf2:
            with tempfile.TemporaryDirectory() as temp_dir:
                # Save PDF files to temporary directory
                pdf1_path = os.path.join(temp_dir, pdf1.name)
                with open(pdf1_path, 'wb') as f:
                    f.write(pdf1.getbuffer())

                pdf2_path = os.path.join(temp_dir, pdf2.name)
                with open(pdf2_path, 'wb') as f:
                    f.write(pdf2.getbuffer())

                llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo", engine="gpt-35-turbo")

                tools = []
                files = [
                    {"name": pdf1.name, "path": pdf1_path},
                    {"name": pdf2.name, "path": pdf2_path}
                ]

                for file in files:
                    loader = PyPDFLoader(file["path"])
                    pages = loader.load_and_split()
                    text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
                    docs = text_splitter.split_documents(pages)
                    embeddings = OpenAIEmbeddings()
                    retriever = FAISS.from_documents(docs, embeddings).as_retriever()

                    # Wrap retrievers in Tool with corrected func and schema
                    tools.append(
                        Tool(
                            args_schema=DocumentInput,
                            name=file["name"],
                            description=f"useful when you want to answer questions about {file['name']}",
                            func=lambda x: RetrievalQA.from_chain_type(llm=llm, retriever=retriever).run(x["question"]),
                        )
                    )

                agent = initialize_agent(
                    tools=tools,
                    llm=llm,
                    verbose=True,
                    agent="zero-shot-react-description"
                )

                result = agent({"input": question})
                st.subheader("Comparison Result")
                st.write(result["output"])

if __name__ == "__main__":
    main()

内容的提问来源于stack exchange，提问作者Stetco Oana