day14 chatDocument 文件分析器(二)：混合資料格式解析!完全版chatDocument接受多檔案多來源的文件分析器

16th鐵人賽 langchain rag llm (python) streamlit

jay0810

團隊不時以註解遮羞的實習同學

2024-09-13 04:24:04

777 瀏覽

分享至

前言

昨天我們利用streamlit快速建構出csv檔案的智能文件分析器，今天我們將文件分析器利用自定義的Loader擴充py、text、pdf等資料格式，甚至我們還能上傳多個不同格式的檔案，使得LLM接受的來源更加多元!

正文

首先我們會進行一些修改，但大部分的內容都是重複的!

匯入模型

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
import os
# 1. Create model
os.environ["OPENAI_API_KEY"] = "你的OpenAI API key"

model = ChatOpenAI(model="gpt-4o")

embeddingmodel = OpenAIEmbeddings(model="text-embedding-ada-002")

製作Loader

製作Loader時，我們會在lazy_load方法當中，查看尾端的副檔名，並且導入到不同的Loader之中

from langchain_core.document_loaders import BaseLoader
from typing import Iterator
from langchain_core.documents import Document
from langchain_community.document_loaders.csv_loader import CSVLoader
from langchain_community.document_loaders import PyPDFLoader
# 2. 製作loader
class CustomDocumentLoader(BaseLoader):
    def __init__(self, file_path: str) -> None:
        self.file_path = file_path

    def lazy_load(self) -> Iterator[Document]:
        if self.file_path.endswith('.csv'):
            loader = CSVLoader(self.file_path)
            for doc in loader.lazy_load():
                yield doc
        elif self.file_path.endswith('.pdf'):
            loader = PyPDFLoader(self.file_path)
            for doc in loader.lazy_load():
                yield doc
        else:
            with open(self.file_path, encoding="utf-8") as f:
                line_number = 0
                for line in f:
                    yield Document(
                        page_content=line,
                        metadata={"line_number": line_number,
                                  "source": self.file_path},
                    )
                    line_number += 1

process_documents

這裡是作為streamlit的資料處理，也就是接受檔案以及使用者輸入的處理流程
主要的更改在前面的files輸入，我們在streamlit的修改會變成多檔案的接收，因此在使用for 並且用一個all_docs的list 來載入多檔案資料源

from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
def process_documents(files, user_input):
    all_docs = []

    for uploaded_file in files:
        # 暫存上檔案
        temp_file_path = f'''temp_uploaded_file_{
            uploaded_file.name.split('.')[-1]}'''
        with open(temp_file_path, "wb") as temp_file:
            temp_file.write(uploaded_file.getvalue())

        # 1. 載入檔案
        loader = CustomDocumentLoader(temp_file_path)
        docs = list(loader.lazy_load())
        all_docs.extend(docs)

        # 刪除暫存檔案
        os.remove(temp_file_path)

    # 2. 切分成數個chunk，自定切割大小，自訂overlap
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=400, chunk_overlap=100)
    splits = text_splitter.split_documents(docs)

    # 3. 轉換成embedding，儲存進Chroma向量資料庫
    vectorstore = Chroma.from_documents(
        documents=splits, embedding=embeddingmodel)
    retriever = vectorstore.as_retriever()

    # 4. 做成chain.
    template = """
    你是一個人工智慧輔助系統，可以根據文件的內容進行簡單的總結，並且根據使用者需求來進行分析
    以下為文件的內容:
    {context}
    使用者提供的額外資訊：
    {user_input}
    """
    custom_rag_prompt = PromptTemplate.from_template(template)

    def format_docs(docs):
        return "\n\n".join(doc.page_content for doc in docs)

    rag_chain = (
        {"context": retriever | format_docs, "user_input": lambda x: x}
        | custom_rag_prompt
        | model
        | StrOutputParser()
    )

    # 5. 呼叫
    result = rag_chain.invoke(user_input)
    return result

run_streamlit

主要的變更是file_uploader 多了個accept_multiple_files=True的參數，可以在介面上傳多個檔案
另外type的部分改成 type=["txt", "py", "csv", "pdf"]，接收更多元的資料

import streamlit as st
def run_streamlit():
    st.set_page_config(
        page_title="文件分析與總結器", page_icon="📄")
    st.title("文件分析與總結器")
    st.write("")
    st.write("### Step 1: 上傳文件")
    uploaded_files = st.file_uploader(
        "Choose a file", type=["txt", "py", "csv"], label_visibility="visible", accept_multiple_files=True)
    st.write("")
    st.write("### Step 2: 輸入額外資訊，讓人工智慧處理")
    user_input = st.text_input(
        "Enter Additional Information", "")
    st.write("")
    st.write("### Step 3: 提交")
    submit_button = st.button("Submit", use_container_width=True)
    if submit_button:
        if uploaded_files is not None:
            file_names = ", ".join([file.name for file in uploaded_files])
            st.write(f"**檔案名稱:** {file_names}")
            st.write(f"**額外資訊:** {user_input}")

            # Process the document
            summary = process_documents(uploaded_files, user_input)

            st.write("")
            st.write("### Summary:")
            st.write(summary)
        else:
            st.error("請先上傳檔案")


if __name__ == "__main__":
    run_streamlit()

完整程式碼

# streamlit
import streamlit as st
# langChain
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
import os
from langchain_core.document_loaders import BaseLoader
from typing import Iterator
from langchain_core.documents import Document
from langchain_community.document_loaders.csv_loader import CSVLoader
from langchain_community.document_loaders import PyPDFLoader
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate

# 啟動命令 streamlit run filename.py

# 1. Create model
os.environ["OPENAI_API_KEY"] = "你的OpenAI API key"

model = ChatOpenAI(model="gpt-4o")

embeddingmodel = OpenAIEmbeddings(model="text-embedding-ada-002")


# 2. 製作loader
class CustomDocumentLoader(BaseLoader):
    def __init__(self, file_path: str) -> None:
        self.file_path = file_path

    def lazy_load(self) -> Iterator[Document]:
        if self.file_path.endswith('.csv'):
            loader = CSVLoader(self.file_path)
            for doc in loader.lazy_load():
                yield doc
        elif self.file_path.endswith('.pdf'):
            loader = PyPDFLoader(self.file_path)
            for doc in loader.lazy_load():
                yield doc
        else:
            with open(self.file_path, encoding="utf-8") as f:
                line_number = 0
                for line in f:
                    yield Document(
                        page_content=line,
                        metadata={"line_number": line_number,
                                  "source": self.file_path},
                    )
                    line_number += 1


def process_documents(files, user_input):
    all_docs = []

    for uploaded_file in files:
        # 暫存上檔案
        temp_file_path = f'''temp_uploaded_file_{
            uploaded_file.name.split('.')[-1]}'''
        with open(temp_file_path, "wb") as temp_file:
            temp_file.write(uploaded_file.getvalue())

        # 1. 載入檔案
        loader = CustomDocumentLoader(temp_file_path)
        docs = list(loader.lazy_load())
        all_docs.extend(docs)

        # 刪除暫存檔案
        os.remove(temp_file_path)

    # 2. 切分成數個chunk，自定切割大小，自訂overlap
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=400, chunk_overlap=100)
    splits = text_splitter.split_documents(docs)

    # 3. 轉換成embedding，儲存進Chroma向量資料庫
    vectorstore = Chroma.from_documents(
        documents=splits, embedding=embeddingmodel)
    retriever = vectorstore.as_retriever()

    # 4. 做成chain.
    template = """
    你是一個人工智慧輔助系統，可以根據文件的內容進行簡單的總結，並且根據使用者需求來進行分析
    以下為文件的內容:
    {context}
    使用者提供的額外資訊：
    {user_input}
    """
    custom_rag_prompt = PromptTemplate.from_template(template)

    def format_docs(docs):
        return "\n\n".join(doc.page_content for doc in docs)

    rag_chain = (
        {"context": retriever | format_docs, "user_input": lambda x: x}
        | custom_rag_prompt
        | model
        | StrOutputParser()
    )

    # 5. 呼叫
    result = rag_chain.invoke(user_input)
    return result


def run_streamlit():
    st.set_page_config(
        page_title="文件分析與總結器", page_icon="📄")
    st.title("文件分析與總結器")
    st.write("")
    st.write("### Step 1: 上傳文件")
    uploaded_files = st.file_uploader(
        "Choose a file", type=["txt", "py", "csv", "pdf"], label_visibility="visible", accept_multiple_files=True)
    st.write("")
    st.write("### Step 2: 輸入額外資訊，讓人工智慧處理")
    user_input = st.text_input(
        "Enter Additional Information", "")
    st.write("")
    st.write("### Step 3: 提交")
    submit_button = st.button("Submit", use_container_width=True)
    if submit_button:
        if uploaded_files is not None:
            file_names = ", ".join([file.name for file in uploaded_files])
            st.write(f"**檔案名稱:** {file_names}")
            st.write(f"**額外資訊:** {user_input}")

            # Process the document
            summary = process_documents(uploaded_files, user_input)

            st.write("")
            st.write("### Summary:")
            st.write(summary)
        else:
            st.error("請先上傳檔案")


if __name__ == "__main__":
    run_streamlit()