文档问答系统

本系统利用先进的语言模型和检索技术，为用户提供基于上传文件内容的问答服务。支持多种文件格式，包括 Word、PDF、CSV、SQL 和 TXT 文件。

功能介绍

文件上传

• 用户可以同时上传多个文件。
• 支持的文件类型包括：.doc, .docx, .pdf, .csv, .sql, .txt。
• 系统会自动处理上传的文件，并提取文本内容。

问答处理

• 用户输入问题后，系统会根据上传的文件内容，使用检索增强生成（RAG）技术生成答案。
• RAG 技术结合了信息检索和语言生成，提高了答案的准确性和相关性。

技术细节

• 语言模型：使用 Ollama 模型处理语言生成任务。
• 检索系统：通过 Chroma 向量存储和检索框架，对文本内容进行索引和检索。
• 文本处理：使用递归字符文本分割器处理大文本文件，确保信息的完整性。

使用指南

1. 通过界面上传文件。可以手动放到upload目录。
2. 在提问框中输入您的问题。
3. 调整 temperature 和 top_p 参数以控制答案生成的随机性和多样性。
4. 提交问题，系统将显示基于文件内容的答案。

界面展示

系统提供一个简洁明了的界面，用户可以轻松上传文件和输入问题。温度和概率顶部参数可通过滑块调整，使用户能够根据需要定制答案的生成。

结论

本文档问答系统是一个强大的工具，适用于需要从特定文件内容中快速获取信息的用户。无论是学术研究、业务报告还是技术文档分析，本系统都能提供有效的支持。

import gradio as gr
import os
import uuid
from docx import Document
import PyPDF2
import pandas as pd
from langchain_community.llms import Ollama
from langchain.callbacks.manager import CallbackManager
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import OllamaEmbeddings
from langchain.chains import RetrievalQA
from langchain_core.prompts import PromptTemplate

def init_ollama_llm(model, temperature, top_p):
    return Ollama(model=model,
                  temperature=temperature,
                  top_p=top_p,
                  callback_manager=CallbackManager([StreamingStdOutCallbackHandler()])
                  )

def content_web(url):
    loader = WebBaseLoader(
        web_paths=(url,),
    )
    docs = loader.load()
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
    splits = text_splitter.split_documents(docs)
    return splits

def chroma_retriever_store_content(splits):
    vectorstore = Chroma.from_documents(documents=splits,
                                        embedding=OllamaEmbeddings(model='nomic-embed-text'))
    return vectorstore.as_retriever()

def rag_prompt():
    return PromptTemplate(
        input_variables=['context', 'question'],
        template=
        '''您是一个问答任务的助手，且需使用中文回答。请根据以下检索到的上下文信息用中文回答问题。如果您不知道答案，请直接回答“我不知道”，无需任何解释。问题：{question} 上下文：{context} 答案：'''
    )

def ollama_rag_chroma_web_content(web_url, question, temperature, top_p):
    llm = init_ollama_llm('llama3', temperature, top_p)
    splits = content_web(web_url)
    retriever = chroma_retriever_store_content(splits)
    qa_chain = RetrievalQA.from_chain_type(llm, retriever=retriever, chain_type_kwargs={'prompt': rag_prompt()})
    return qa_chain.invoke({'query': question})['result']

def process_files_and_chat(files, question, temperature, top_p):
    # 首先处理上传的文件
    if files is not None:
        upload_result, _ = upload_file(files)
        if "失败" in upload_result:
            return upload_result
        
    uploads_dir = 'uploads'
    file_paths = [os.path.join(uploads_dir, f) for f in os.listdir(uploads_dir) if os.path.isfile(os.path.join(uploads_dir, f))]
    
    if not file_paths:
        return "请上传文件"
    
    all_texts = []
    
    # 处理所有上传的文件
    for file_path in file_paths:
        print(f"处理文件: {file_path}")
        _, file_extension = os.path.splitext(file_path)
        
        try:
            if file_extension.lower() == '.pdf':
                print(f"开始处理 PDF 文件: {file_path}")
                with open(file_path, 'rb') as f:
                    pdf_reader = PyPDF2.PdfReader(f)
                    text = ''
                    for page in pdf_reader.pages:
                        text += page.extract_text() + '\n'
                print(text)
                all_texts.append(text)
                
            elif file_extension.lower() == '.sql':
                print(f"开始处理 SQL 文件: {file_path}")
                with open(file_path, 'r', encoding='utf-8') as f:
                    text = f.read()
                text = f"SQL文件内容:\n{text}"
                all_texts.append(text)
                
            elif file_extension.lower() == '.csv':
                print(f"开始处理 CSV 文件: {file_path}")
                df = pd.read_csv(file_path)
                text = f"���名: {', '.join(df.columns)}\n"
                text += df.to_string(index=False)
                all_texts.append(text)
                
            elif file_extension.lower() == '.txt':
                print(f"开始处理文本文件: {file_path}")
                with open(file_path, 'r', encoding='utf-8') as f:
                    text = f.read()
                all_texts.append(text)
                
            elif file_extension.lower() in ['.doc', '.docx']:
                print(f"开始处理 Word 文件: {file_path}")
                doc = Document(file_path)
                text = '\n'.join([paragraph.text for paragraph in doc.paragraphs])
                for table in doc.tables:
                    for row in table.rows:
                        for cell in row.cells:
                            text += '\n' + cell.text
                all_texts.append(text)
                
            else:
                print(f"跳过不支持的文件类型: {file_extension}")
                continue
                
        except Exception as e:
            print(f"处理文件 {file_path} 时出错: {str(e)}")
            continue
    
    if not all_texts:
        return "没有可处理的文件内容"
    
    # 合并所有文本
    combined_text = "\n\n".join(all_texts)
    print(f"成功处理 {len(file_paths)} 个文件，总文本长度: {len(combined_text)} 字符")
    
    # 使用RAG处理问题
    try:
        print("初始化LLM模型...")
        llm = init_ollama_llm('llama3', temperature, top_p)
        
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50,
            length_function=len,
            is_separator_regex=False
        )
        splits = text_splitter.split_text(combined_text)
        
        if not splits:
            return "文档内容为空或无法正确分割"
            
        print(f"文本分割完成，共 {len(splits)} 个片段")
        
        print("初始化嵌入模型...")
        embeddings = OllamaEmbeddings(
            model='nomic-embed-text',
            base_url="http://localhost:11434"
        )
        
        print("创建向量存储...")
        vectorstore = Chroma.from_texts(
            texts=splits,
            embedding=embeddings,
            collection_name="doc_qa",
            persist_directory="./chroma_file_db"
        )
        
        retriever = vectorstore.as_retriever(
            search_kwargs={"k": 1}
        )
        
        qa_chain = RetrievalQA.from_chain_type(
            llm=llm,
            retriever=retriever,
            chain_type_kwargs={
                'prompt': rag_prompt(),
                'verbose': True
            }
        )
        
        print("处理问题...")
        result = qa_chain.invoke({'query': question})
        return result['result']
        
    except Exception as e:
        print(f"处理问题时出错: {str(e)}")
        return f"处理问题时出错: {str(e)}"

def detect_file_type(file_content, original_name=''):
    """根据文件内容和名称检测文件类型"""
    # 首先尝试从文件名获取扩展名
    if original_name:
        ext = os.path.splitext(original_name)[1].lower()
        if ext in ['.doc', '.docx', '.pdf', '.txt', '.csv', '.sql']:
            return ext
    
    # 如果没有文件名或扩展名不支持，尝试通过内容判断
    try:
        # 检查文件头部特征
        if file_content.startswith(b'%PDF'):
            return '.pdf'
        # Word文档的特征（.docx 文件是 ZIP 格式，以 PK 开头）
        elif file_content.startswith(b'PK'):
            return '.docx'
        
        # 尝试解码为文本
        try:
            content_start = file_content[:1000].decode('utf-8')
            # SQL文件特征检测
            sql_keywords = ['SELECT', 'INSERT', 'UPDATE', 'DELETE', 'CREATE', 'DROP', 'ALTER']
            if any(keyword in content_start.upper() for keyword in sql_keywords):
                return '.sql'
            # CSV文件特征检测
            if ',' in content_start and '\n' in content_start:
                return '.csv'
            # 默认作为文本文件
            return '.txt'
        except UnicodeDecodeError:
            # 如果无法解码为文本，检查是否是 DOC 文件
            if b'\xD0\xCF\x11\xE0\xA1\xB1\x1A\xE1' in file_content[:10]:
                return '.doc'
            return None
    except Exception as e:
        print(f"文件类型检测失败: {str(e)}")
        return None

def upload_file(files):
    try:
        if not os.path.exists("uploads"):
            os.makedirs("uploads")
            
        file_paths = []
        for file in files:
            try:
                # 读取文件内容
                if isinstance(file, bytes):
                    file_content = file
                elif hasattr(file, 'read'):
                    file_content = file.read()
                    if hasattr(file, 'seek'):
                        file.seek(0)  # 重置文件指针
                else:
                    print(f"警告：无法读取文件内容")
                    continue
                
                # 获取原始文件名（如果有）
                original_name = ''
                if isinstance(file, dict) and 'name' in file:
                    original_name = file['name']
                elif hasattr(file, 'name'):
                    original_name = file.name
                elif hasattr(file, 'filename'):
                    original_name = file.filename
                
                # 检测文件类型
                ext = detect_file_type(file_content, original_name)
                if not ext:
                    print(f"警告：无法识别文件类型")
                    continue
                
                # 生成唯一文件名
                unique_filename = str(uuid.uuid4()) + ext
                save_path = os.path.join("uploads", unique_filename)
                
                print(f"保存文件: {original_name or '未命名文件'} -> {save_path}")
                
                # 保存文件
                with open(save_path, "wb") as f:
                    if isinstance(file_content, str):
                        f.write(file_content.encode())
                    else:
                        f.write(file_content)
                    
                file_paths.append(save_path)
                
            except Exception as e:
                print(f"处理文件出错: {str(e)}")
                continue
                
        if not file_paths:
            return "没有成功上传任何文件", None
            
        return f"成功上传 {len(file_paths)} 个文件", file_paths
        
    except Exception as e:
        print(f"上传过程出错: {str(e)}")
        return f"文件上传失败: {str(e)}", None






# 更新界面
ol = gr.Interface(
    fn=process_files_and_chat,
    inputs=[
        gr.File(label='上传文件', type='binary', file_count='multiple'),  # 移除 optional 参数
        gr.Textbox(label='问题', placeholder='请输入您的问题'),
        gr.Slider(label='temperature', minimum=0, maximum=1, step=0.1, value=0.1),
        gr.Slider(label='top_p', minimum=0, maximum=1, step=0.1, value=0.4)
    ],
    outputs='text',
    title='文档问答系统',
    description='支持同时上传多个文件（Word、PDF、CSV、SQL、TXT等），输入问题获取基于文档内容的答案'
)

ol.launch()