Document Question Answering：2025年文档智能问答的全面解析

# 基于BERT的文档问答模型示例实现
import torch
import torch.nn as nn
from transformers import BertModel, BertTokenizer

class DocumentQA(nn.Module):
    def __init__(self, model_name):
        super().__init__()
        # 加载预训练的BERT模型
        self.bert = BertModel.from_pretrained(model_name)
        # 获取隐藏层大小
        hidden_size = self.bert.config.hidden_size
        # 答案开始位置预测层
        self.start_fc = nn.Linear(hidden_size, 1)
        # 答案结束位置预测层
        self.end_fc = nn.Linear(hidden_size, 1)
        # Dropout层用于防止过拟合
        self.dropout = nn.Dropout(0.3)
    
    def forward(self, input_ids, attention_mask, token_type_ids):
        # BERT模型前向传播
        outputs = self.bert(
            input_ids=input_ids,
            attention_mask=attention_mask,
            token_type_ids=token_type_ids
        )
        # 获取最后一层隐藏状态
        last_hidden_state = outputs.last_hidden_state  # [batch_size, seq_len, hidden_size]
        # 应用dropout
        last_hidden_state = self.dropout(last_hidden_state)
        # 预测答案开始位置
        start_logits = self.start_fc(last_hidden_state).squeeze(-1)  # [batch_size, seq_len]
        # 预测答案结束位置
        end_logits = self.end_fc(last_hidden_state).squeeze(-1)  # [batch_size, seq_len]
        # 返回预测结果
        return start_logits, end_logits

# 示例使用代码
tokenizer = BertTokenizer.from_pretrained("bert-base-uncased")
model = DocumentQA("bert-base-uncased")

# 准备输入数据
context = "2025年，文档问答技术已经成为自然语言处理领域的重要应用方向。这项技术能够让AI系统理解文档内容并回答用户提出的问题，极大地提升了文档信息获取的效率。"
question = "文档问答技术的主要作用是什么？"

# 编码输入
inputs = tokenizer.encode_plus(
    question, context,
    add_special_tokens=True,
    max_length=512,
    padding="max_length",
    truncation=True,
    return_tensors="pt"
)

# 模型推理
with torch.no_grad():
    start_logits, end_logits = model(**inputs)

# 获取预测的答案位置
start_idx = torch.argmax(start_logits)
end_idx = torch.argmax(end_logits)

# 解码答案
tokens = tokenizer.convert_ids_to_tokens(inputs["input_ids"][0])
answer = tokenizer.convert_tokens_to_string(tokens[start_idx:end_idx+1])
print(f"Question: {question}")
print(f"Answer: {answer}")

# 2025年高级文档问答系统完整实现
import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import AutoProcessor, AutoModelForCausalLM, AutoModelForQuestionAnswering, pipeline
import numpy as np
from PIL import Image
import requests
import io
import fitz  # PyMuPDF for PDF processing
import docx  # For Word document processing
import pandas as pd
from bs4 import BeautifulSoup  # For HTML processing
from tqdm import tqdm
import time
import os
import json
import re

class AdvancedDocumentQuestionAnswering:
    def __init__(self, model_name="google/gemma-7b-it", device=None):
        # 设置设备
        self.device = device if device is not None else ("cuda" if torch.cuda.is_available() else "cpu")
        
        # 加载预训练的语言模型和处理器
        print(f"加载文档问答模型: {model_name}")
        
        try:
            # 尝试加载专用的问答模型
            self.model = AutoModelForQuestionAnswering.from_pretrained(
                model_name,
                device_map="auto" if torch.cuda.is_available() else None,
                trust_remote_code=True
            )
            self.processor = AutoProcessor.from_pretrained(
                model_name,
                trust_remote_code=True
            )
            self.use_generic_model = False
        except Exception as e:
            print(f"无法加载专用的问答模型，使用通用语言模型: {e}")
            # 使用通用语言模型
            try:
                self.model = AutoModelForCausalLM.from_pretrained(
                    model_name,
                    device_map="auto" if torch.cuda.is_available() else None,
                    trust_remote_code=True
                )
                self.processor = AutoProcessor.from_pretrained(
                    model_name,
                    trust_remote_code=True
                )
                self.use_generic_model = True
            except Exception as e2:
                raise RuntimeError(f"无法加载模型: {e2}")
        
        # 设置模型为评估模式
        self.model.eval()
        
        # 获取模型信息
        self.model_name = model_name
        
        # 初始化pipeline用于快速推理（对于支持的模型）
        self.pipeline = None
        if not self.use_generic_model:
            try:
                self.pipeline = pipeline(
                    "question-answering",
                    model=self.model,
                    tokenizer=self.processor.tokenizer,
                    device=self.device
                )
            except Exception:
                pass
        
        # 初始化文档解析器
        self.doc_parsers = {
            "pdf": self.parse_pdf_document,
            "docx": self.parse_docx_document,
            "txt": self.parse_text_document,
            "html": self.parse_html_document,
            "csv": self.parse_csv_document
        }
    
    def load_document(self, document_path_or_url):
        # 加载文档
        if isinstance(document_path_or_url, str):
            if document_path_or_url.startswith((\'http://\', \'https://\')):
                # 从URL加载文档
                try:
                    response = requests.get(document_path_or_url, stream=True)
                    response.raise_for_status()  # 检查请求是否成功
                    
                    # 根据URL扩展名确定文档类型
                    file_extension = document_path_or_url.split(\'.\')[-1].lower()
                    if file_extension not in self.doc_parsers:
                        # 默认当作文本处理
                        text = response.text
                    else:
                        # 使用相应的解析器
                        parser = self.doc_parsers[file_extension]
                        text = parser(response.content)
                    
                    return text
                except Exception as e:
                    raise ValueError(f"无法从URL加载文档: {e}")
            else:
                # 从本地路径加载文档
                try:
                    # 获取文件扩展名
                    file_extension = document_path_or_url.split(\'.\')[-1].lower()
                    
                    if file_extension not in self.doc_parsers:
                        # 默认当作文本处理
                        with open(document_path_or_url, \'r\', encoding=\'utf-8\', errors=\'replace\') as f:
                            text = f.read()
                    else:
                        # 使用相应的解析器
                        parser = self.doc_parsers[file_extension]
                        with open(document_path_or_url, \'rb\') as f:
                            content = f.read()
                        text = parser(content)
                    
                    return text
                except Exception as e:
                    raise ValueError(f"无法从本地路径加载文档: {e}")
        elif isinstance(document_path_or_url, bytes):
            # 如果输入是字节流，尝试解析
            # 注意：这里我们无法确定文档类型，可能需要用户提供更多信息
            try:
                # 尝试当作文本处理
                return document_path_or_url.decode(\'utf-8\', errors=\'replace\')
            except UnicodeDecodeError:
                # 如果不是文本，返回错误信息
                return "无法识别的二进制文档格式"
        elif isinstance(document_path_or_url, str):
            # 如果输入已经是文本，直接返回
            return document_path_or_url
        else:
            raise TypeError("文档必须是路径字符串、URL字符串或字节流")
    
    def parse_pdf_document(self, content):
        # 解析PDF文档
        try:
            text = ""
            with fitz.open(stream=content, filetype="pdf") as doc:
                for page in doc:
                    text += page.get_text()
            return text
        except Exception as e:
            raise ValueError(f"解析PDF文档失败: {e}")
    
    def parse_docx_document(self, content):
        # 解析Word文档
        try:
            text = ""
            with io.BytesIO(content) as f:
                doc = docx.Document(f)
                for paragraph in doc.paragraphs:
                    text += paragraph.text + \'\n\'
            return text
        except Exception as e:
            raise ValueError(f"解析Word文档失败: {e}")
    
    def parse_text_document(self, content):
        # 解析文本文档
        try:
            return content.decode(\'utf-8\', errors=\'replace\')
        except Exception as e:
            raise ValueError(f"解析文本文档失败: {e}")
    
    def parse_html_document(self, content):
        # 解析HTML文档
        try:
            soup = BeautifulSoup(content, \'html.parser\')
            # 提取文本内容
            text = soup.get_text(separator=\'\n\', strip=True)
            # 清理多余的空行
            text = re.sub(r\'\n\s*\n\', \'\n\', text)
            return text
        except Exception as e:
            raise ValueError(f"解析HTML文档失败: {e}")
    
    def parse_csv_document(self, content):
        # 解析CSV文档
        try:
            with io.StringIO(content.decode(\'utf-8\', errors=\'replace\')) as f:
                df = pd.read_csv(f)
            # 将DataFrame转换为文本表示
            text = "CSV表格数据:\n"
            text += df.to_string(index=False)
            return text
        except Exception as e:
            raise ValueError(f"解析CSV文档失败: {e}")
    
    def chunk_document(self, document, max_chunk_size=1000, overlap=100):
        # 将文档分块处理，处理长文档
        words = document.split()
        chunks = []
        current_chunk = []
        
        for i, word in enumerate(words):
            current_chunk.append(word)
            
            # 检查当前块的大小
            if len(\' \'.join(current_chunk)) >= max_chunk_size:
                # 添加当前块
                chunks.append(\' \'.join(current_chunk))
                # 重置当前块，包含重叠部分
                current_chunk = words[max(0, i - overlap + 1):i + 1]
        
        # 添加最后一个块
        if current_chunk:
            chunks.append(\' \'.join(current_chunk))
        
        return chunks
    
    def answer_question(self, document, question, max_chunk_size=1000, overlap=100, max_length=2048, num_return_sequences=1, temperature=0.7):
        # 回答关于给定文档的问题
        
        # 确保文档是文本格式
        if not isinstance(document, str):
            document = self.load_document(document)
        
        # 分块处理文档（如果文档太长）
        chunks = self.chunk_document(document, max_chunk_size, overlap)
        
        # 准备提示
        if self.use_generic_model:
            # 对于通用语言模型，需要构建适当的提示
            # 如果文档很小，直接使用
            if len(chunks) == 1:
                prompt = f"基于以下文档，回答问题:\n\n文档: {chunks[0]}\n\n问题: {question}\n\n回答: "
            else:
                # 如果文档很大，使用RAG（检索增强生成）的思路
                # 1. 初步检索相关块
                relevant_chunks = self.retrieve_relevant_chunks(chunks, question, top_k=3)
                # 2. 构建提示
                retrieved_docs = "\n\n".join([f"文档片段{i+1}: {chunk}" for i, chunk in enumerate(relevant_chunks)])
                prompt = f"基于以下检索到的文档片段，回答问题:\n\n{retrieved_docs}\n\n问题: {question}\n\n回答: "
        else:
            # 对于专用问答模型，直接使用问题和文档
            # 如果文档很小，直接使用
            if len(chunks) == 1:
                context = chunks[0]
            else:
                # 如果文档很大，使用RAG的思路
                relevant_chunks = self.retrieve_relevant_chunks(chunks, question, top_k=3)
                context = "\n\n".join(relevant_chunks)
        
        # 使用模型生成回答
        if not self.use_generic_model and self.pipeline is not None:
            # 如果有专用的问答pipeline，使用它
            try:
                result = self.pipeline(question=question, context=context)
                # 格式化结果
                return {
                    "answer": result["answer"],
                    "confidence": result["score"],
                    "model": self.model_name,
                    "question": question,
                    "context": context[:200] + "..." if len(context) > 200 else context
                }
            except Exception as e:
                print(f"使用pipeline失败，回退到通用方法: {e}")
                # 失败时回退到通用方法
        
        # 对于通用语言模型或pipeline失败的情况
        if self.use_generic_model:
            # 准备输入
            inputs = self.processor(
                text=prompt,
                return_tensors="pt",
                max_length=max_length,
                truncation=True
            )
            
            # 移至设备
            if not hasattr(self.model, "hf_device_map") or self.model.hf_device_map is None:
                inputs = {k: v.to(self.device) for k, v in inputs.items()}
            
            # 使用模型生成回答
            with torch.no_grad():
                outputs = self.model.generate(
                    **inputs,
                    max_length=max_length,
                    num_return_sequences=num_return_sequences,
                    temperature=temperature,
                    top_p=0.95,
                    do_sample=True
                )
                
                # 解码生成的回答
                answers = []
                for i in range(num_return_sequences):
                    # 提取生成的回答（去除输入部分）
                    answer = self.processor.tokenizer.decode(outputs[i], skip_special_tokens=True)
                    
                    # 尝试从生成文本中提取回答部分
                    if "回答:" in answer:
                        answer = answer.split("回答:")[-1].strip()
                    elif "答案是" in answer:
                        answer = answer.split("答案是")[-1].strip()
                    elif "基于文档" in answer:
                        answer = answer.split("基于文档")[-1].strip()
                    
                    answers.append(answer)
                
                answer = answers[0] if num_return_sequences == 1 else answers
        else:
            # 对于专用问答模型，但没有pipeline的情况
            # 这里使用简化的方法，实际应用中需要根据模型具体情况调整
            answer = "无法使用专用模型回答，建议使用通用语言模型。"
        
        # 返回结果
        if num_return_sequences == 1:
            return {
                "answer": answer,
                "model": self.model_name,
                "question": question,
                "prompt": prompt[:200] + "..." if self.use_generic_model and len(prompt) > 200 else (prompt if self.use_generic_model else None),
                "num_chunks": len(chunks)
            }
        else:
            return {
                "answers": answers,
                "model": self.model_name,
                "question": question,
                "prompt": prompt[:200] + "..." if self.use_generic_model and len(prompt) > 200 else (prompt if self.use_generic_model else None),
                "num_chunks": len(chunks)
            }
    
    def retrieve_relevant_chunks(self, chunks, question, top_k=3):
        # 检索与问题最相关的文档块
        # 这里使用简化的TF-IDF方法进行相关性排序
        # 实际应用中可以使用更高级的检索方法，如密集检索
        
        # 提取问题中的关键词
        question_keywords = set(re.findall(r\'\w+\', question.lower()))
        
        # 计算每个块的相关度分数
        scores = []
        for chunk in chunks:
            # 提取块中的关键词
            chunk_keywords = set(re.findall(r\'\w+\', chunk.lower()))
            # 计算关键词交集的大小作为相关度分数
            score = len(question_keywords.intersection(chunk_keywords))
            # 为了处理没有重叠关键词的情况，可以添加其他特征
            if score == 0:
                # 例如，检查问题中的词是否在块中出现
                for word in question_keywords:
                    if word in chunk.lower():
                        score += 0.5
            # 添加块长度的归一化因子
            score = score / (1 + np.log1p(len(chunk) / 100))
            scores.append(score)
        
        # 获取相关性最高的前top_k个块
        top_indices = np.argsort(scores)[-top_k:][::-1]
        relevant_chunks = [chunks[i] for i in top_indices]
        
        return relevant_chunks
    
    def batch_answer_questions(self, document, questions, batch_size=4, max_chunk_size=1000, overlap=100, max_length=2048):
        # 批量回答关于给定文档的多个问题
        
        # 确保文档是文本格式
        if not isinstance(document, str):
            document = self.load_document(document)
        
        # 确保questions是列表格式
        if isinstance(questions, str):
            questions = [questions]
        
        # 分块处理文档（如果文档太长）
        chunks = self.chunk_document(document, max_chunk_size, overlap)
        
        # 批量处理问题
        results = []
        for i in tqdm(range(0, len(questions), batch_size), desc="Batch Processing Questions"):
            batch_questions = questions[i:i+batch_size]
            
            # 对每个问题进行处理
            for question in batch_questions:
                # 回答问题
                result = self.answer_question(
                    document=document,  # 这里重新传入完整文档，以便函数内部处理
                    question=question,
                    max_chunk_size=max_chunk_size,
                    overlap=overlap,
                    max_length=max_length
                )
                results.append(result)
        
        return results
    
    def summarize_document(self, document, max_length=512, temperature=0.7):
        # 生成文档摘要
        
        # 确保文档是文本格式
        if not isinstance(document, str):
            document = self.load_document(document)
        
        # 如果文档很长，先分块处理
        chunks = self.chunk_document(document, max_chunk_size=2000, overlap=100)
        
        # 构建提示
        if len(chunks) == 1:
            prompt = f"请为以下文档生成简明扼要的摘要（约100-200字）:\n\n文档: {document}\n\n摘要: "
        else:
            # 对于长文档，先为每个块生成摘要，然后整合
            chunk_summaries = []
            for i, chunk in enumerate(tqdm(chunks, desc="Summarizing Chunks")):
                chunk_prompt = f"请为以下文档片段生成简明的摘要:\n\n文档片段{i+1}: {chunk}\n\n摘要: "
                
                # 使用模型生成块摘要
                inputs = self.processor(
                    text=chunk_prompt,
                    return_tensors="pt",
                    max_length=max_length,
                    truncation=True
                )
                
                # 移至设备
                if not hasattr(self.model, "hf_device_map") or self.model.hf_device_map is None:
                    inputs = {k: v.to(self.device) for k, v in inputs.items()}
                
                # 使用模型生成回答
                with torch.no_grad():
                    outputs = self.model.generate(
                        **inputs,
                        max_length=max_length // 2,  # 块摘要可以短一些
                        temperature=temperature,
                        top_p=0.95,
                        do_sample=True
                    )
                    
                    # 解码生成的摘要
                    chunk_summary = self.processor.tokenizer.decode(outputs[0], skip_special_tokens=True)
                    
                    # 尝试从生成文本中提取摘要部分
                    if "摘要:" in chunk_summary:
                        chunk_summary = chunk_summary.split("摘要:")[-1].strip()
                    
                    chunk_summaries.append(chunk_summary)
            
            # 整合块摘要
            combined_summaries = "\n\n".join([f"文档片段{i+1}摘要: {summary}" for i, summary in enumerate(chunk_summaries)])
            prompt = f"请根据以下各个文档片段的摘要，生成整个文档的综合摘要（约200-300字）:\n\n{combined_summaries}\n\n综合摘要: "
        
        # 准备输入
        inputs = self.processor(
            text=prompt,
            return_tensors="pt",
            max_length=max_length,
            truncation=True
        )
        
        # 移至设备
        if not hasattr(self.model, "hf_device_map") or self.model.hf_device_map is None:
            inputs = {k: v.to(self.device) for k, v in inputs.items()}
        
        # 使用模型生成摘要
        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_length=max_length,
                temperature=temperature,
                top_p=0.95,
                do_sample=True
            )
            
            # 解码生成的摘要
            summary = self.processor.tokenizer.decode(outputs[0], skip_special_tokens=True)
            
            # 尝试从生成文本中提取摘要部分
            if "摘要:" in summary:
                summary = summary.split("摘要:")[-1].strip()
            elif "综合摘要:" in summary:
                summary = summary.split("综合摘要:")[-1].strip()
        
        return {
            "summary": summary,
            "model": self.model_name,
            "num_chunks": len(chunks)
        }
    
    def get_model_info(self):
        # 获取模型信息
        # 估算模型大小
        def get_model_size(model):
            param_size = 0
            for param in model.parameters():
                param_size += param.nelement() * param.element_size()
            buffer_size = 0
            for buffer in model.buffers():
                buffer_size += buffer.nelement() * buffer.element_size()
            return param_size + buffer_size
        
        model_size = get_model_size(self.model)
        
        return {
            "model_name": self.model_name,
            "device": self.device,
            "use_generic_model": self.use_generic_model,
            "has_pipeline": self.pipeline is not None,
            "estimated_size_mb": model_size / 1024 / 1024,
            "supports_batching": True,
            "supported_formats": list(self.doc_parsers.keys())
        }
    
    def visualize_answer(self, document, question, answer_result):
        # 可视化回答结果
        print("===== 文档问答结果可视化 ======")
        print(f"问题: {question}")
        print(f"\n文档信息: 长度约{len(document)}字符, {answer_result.get('num_chunks', 1)}个处理块")
        
        print("\n回答:")
        if "answer" in answer_result:
            print(answer_result["answer"])
            if "confidence" in answer_result:
                print(f"置信度: {answer_result['confidence']:.4f}")
        elif "answers" in answer_result:
            for i, ans in enumerate(answer_result["answers"]):
                print(f"答案 {i+1}: {ans}")
        
        if "context" in answer_result:
            print("\n使用的上下文:")
            print(answer_result["context"])
        
        print("=============================")

# 使用示例
if __name__ == "__main__":
    # 初始化文档问答系统
    qa_system = AdvancedDocumentQuestionAnswering(model_name="google/gemma-7b-it")
    
    # 加载文档
    document = "2025年，文档问答技术已经成为自然语言处理领域的重要应用方向。这项技术能够让AI系统理解文档内容并回答用户提出的问题，极大地提升了文档信息获取的效率。从企业知识管理到教育科研，从医疗文献检索到法律文件分析，文档问答技术正在各个领域展现出强大的应用潜力。"
    
    # 提问
    question = "文档问答技术的主要应用领域有哪些？"
    
    # 获取答案
    result = qa_system.answer_question(document=document, question=question)
    
    # 可视化结果
    qa_system.visualize_answer(document=document, question=question, answer_result=result)
    
    # 生成文档摘要
    summary = qa_system.summarize_document(document=document)
    print(f"\n文档摘要:\n{summary['summary']}")