Question Answering：2025年智能问答技术的全面解析与实践指南

# 2025年基于大型预训练语言模型的问答系统示例实现
import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import AutoTokenizer, AutoModelForQuestionAnswering, pipeline
from datasets import load_dataset, load_metric
import numpy as np
from tqdm import tqdm
import os
import json

class AdvancedQuestionAnswering:
    def __init__(self, model_name="google/gemma-2-9b-it", device=None):
        # 加载预训练的语言模型和分词器
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        
        # 尝试加载专用的问答模型，如果失败则使用通用模型
        try:
            self.model = AutoModelForQuestionAnswering.from_pretrained(model_name)
        except Exception as e:
            print(f"无法加载专用的问答模型，使用通用模型: {e}")
            # 对于不支持问答的通用模型，我们需要进行适配
            from transformers import AutoModelForCausalLM
            self.model = AutoModelForCausalLM.from_pretrained(model_name)
            self.use_generic_model = True
        else:
            self.use_generic_model = False
        
        # 设置设备
        self.device = device if device is not None else ("cuda" if torch.cuda.is_available() else "cpu")
        self.model.to(self.device)
        
        # 设置模型为评估模式
        self.model.eval()
        
        # 获取模型信息
        self.model_name = model_name
        
        # 初始化pipeline用于快速推理（对于支持的模型）
        if not self.use_generic_model:
            try:
                self.pipeline = pipeline(
                    "question-answering",
                    model=self.model,
                    tokenizer=self.tokenizer,
                    device=self.device
                )
            except Exception:
                self.pipeline = None
        else:
            self.pipeline = None
    
    def answer_question(self, context, question, max_length=1024, num_return_sequences=1):
        # 回答关于给定上下文的问题
        
        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"],
                    "start": result["start"],
                    "end": result["end"],
                    "model": self.model_name,
                    "question": question,
                    "context": context
                }
            except Exception as e:
                print(f"使用pipeline失败，回退到通用方法: {e}")
                # 失败时回退到通用方法
        
        # 生成提示文本
        prompt = f"上下文: {context}\n\n根据上面的上下文，回答以下问题: {question}\n回答: "
        
        # 准备输入
        inputs = self.tokenizer(prompt, return_tensors="pt", max_length=max_length, truncation=True).to(self.device)
        
        # 使用模型生成回答
        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_length=max_length,
                num_return_sequences=num_return_sequences,
                temperature=0.7,
                top_p=0.95,
                do_sample=True
            )
        
        # 解码生成的回答
        answers = []
        for i in range(num_return_sequences):
            # 提取生成的回答（去除输入部分）
            answer = self.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)
        
        # 返回结果
        if num_return_sequences == 1:
            return {
                "answer": answers[0],
                "model": self.model_name,
                "question": question,
                "context": context,
                "prompt": prompt
            }
        else:
            return {
                "answers": answers,
                "model": self.model_name,
                "question": question,
                "context": context,
                "prompt": prompt
            }
    
    def batch_answer_questions(self, contexts, questions, batch_size=8, max_length=1024):
        # 批量回答关于给定上下文的问题
        # 确保contexts和questions是列表格式
        if isinstance(contexts, str):
            contexts = [contexts] * len(questions)
        
        if isinstance(questions, str):
            questions = [questions] * len(contexts)
        
        # 确保contexts和questions长度相同
        if len(contexts) != len(questions):
            raise ValueError(f"contexts和questions长度不匹配: {len(contexts)} vs {len(questions)}")
        
        # 生成所有提示
        prompts = []
        for context, question in zip(contexts, questions):
            prompt = f"上下文: {context}\n\n根据上面的上下文，回答以下问题: {question}\n回答: "
            prompts.append(prompt)
        
        # 批量处理
        results = []
        for i in tqdm(range(0, len(prompts), batch_size), desc="Batch Processing"):
            batch_prompts = prompts[i:i+batch_size]
            batch_questions = questions[i:i+batch_size]
            batch_contexts = contexts[i:i+batch_size]
            
            # 准备输入
            inputs = self.tokenizer(
                batch_prompts,
                return_tensors="pt",
                padding=True,
                truncation=True,
                max_length=max_length
            ).to(self.device)
            
            # 使用模型生成回答
            with torch.no_grad():
                outputs = self.model.generate(
                    **inputs,
                    max_length=max_length,
                    temperature=0.7,
                    top_p=0.95,
                    do_sample=True
                )
            
            # 解码生成的回答
            for j in range(len(batch_prompts)):
                # 提取生成的回答
                answer = self.tokenizer.decode(outputs[j], 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()
                
                results.append({
                    "answer": answer,
                    "model": self.model_name,
                    "question": batch_questions[j],
                    "context": batch_contexts[j],
                    "prompt": batch_prompts[j]
                })
        
        return results
    
    def evaluate_with_dataset(self, dataset_name="squad", split="validation", sample_size=100):
        # 使用数据集评估模型性能
        try:
            # 加载数据集
            dataset = load_dataset(dataset_name, split=split)
            
            # 采样数据
            if sample_size < len(dataset):
                dataset = dataset.shuffle(seed=42).select(range(sample_size))
            
            # 评估指标
            correct = 0
            total = len(dataset)
            
            # 评估模型
            print(f"评估模型在{sample_size}个样本上的性能...")
            for i, example in enumerate(tqdm(dataset, desc="Evaluating")):
                # 获取上下文、问题和答案
                context = example["context"]
                question = example["question"]
                reference_answer = example["answers"]["text"][0] if "answers" in example else ""
                
                # 使用模型回答问题
                try:
                    result = self.answer_question(context, question)
                    predicted_answer = result["answer"]
                    
                    # 简单的正确性检查（实际应用中应使用更复杂的评估指标）
                    # 这里使用简单的文本匹配
                    if reference_answer.lower() in predicted_answer.lower() or predicted_answer.lower() in reference_answer.lower():
                        correct += 1
                    
                    # 每10个样本打印一次进度
                    if (i+1) % 10 == 0:
                        print(f"样本 {i+1}/{total}, 当前准确率: {correct/(i+1):.4f}")
                        print(f"问题: {question}")
                        print(f"参考答案: {reference_answer}")
                        print(f"模型回答: {predicted_answer}")
                        print("---")
                except Exception as e:
                    print(f"处理样本 {i} 时出错: {e}")
                    continue
            
            # 计算准确率
            accuracy = correct / total if total > 0 else 0
            print(f"评估完成，准确率: {accuracy:.4f}")
            
            return {
                "accuracy": accuracy,
                "correct": correct,
                "total": total,
                "dataset": dataset_name,
                "split": split
            }
        except Exception as e:
            print(f"评估过程中出错: {e}")
            return {
                "error": str(e),
                "dataset": dataset_name,
                "split": split
            }
    
    def get_model_info(self):
        # 获取模型信息
        return {
            "model_name": self.model_name,
            "device": self.device,
            "use_generic_model": self.use_generic_model,
            "has_pipeline": self.pipeline is not None,
            "vocab_size": self.tokenizer.vocab_size,
            "max_length": self.tokenizer.model_max_length
        }
    
    def save_model(self, save_path):
        # 保存模型
        try:
            # 创建保存目录
            os.makedirs(save_path, exist_ok=True)
            
            # 保存模型和分词器
            self.model.save_pretrained(save_path)
            self.tokenizer.save_pretrained(save_path)
            
            # 保存配置信息
            config = {
                "model_name": self.model_name,
                "use_generic_model": self.use_generic_model
            }
            with open(os.path.join(save_path, "config.json"), "w") as f:
                json.dump(config, f, indent=2)
            
            print(f"模型已保存到: {save_path}")
            return True
        except Exception as e:
            print(f"保存模型时出错: {e}")
            return False
    
    def load_model(self, load_path):
        # 加载模型
        try:
            # 加载配置信息
            with open(os.path.join(load_path, "config.json"), "r") as f:
                config = json.load(f)
            
            # 加载模型和分词器
            self.tokenizer = AutoTokenizer.from_pretrained(load_path)
            
            if config.get("use_generic_model", False):
                from transformers import AutoModelForCausalLM
                self.model = AutoModelForCausalLM.from_pretrained(load_path)
                self.use_generic_model = True
                self.pipeline = None
            else:
                try:
                    self.model = AutoModelForQuestionAnswering.from_pretrained(load_path)
                    self.use_generic_model = False
                    # 尝试初始化pipeline
                    try:
                        self.pipeline = pipeline(
                            "question-answering",
                            model=self.model,
                            tokenizer=self.tokenizer,
                            device=self.device
                        )
                    except Exception:
                        self.pipeline = None
                except Exception:
                    from transformers import AutoModelForCausalLM
                    self.model = AutoModelForCausalLM.from_pretrained(load_path)
                    self.use_generic_model = True
                    self.pipeline = None
            
            # 将模型移至设备
            self.model.to(self.device)
            self.model.eval()
            
            # 更新模型名称
            self.model_name = load_path
            
            print(f"模型已从{load_path}加载")
            return True
        except Exception as e:
            print(f"加载模型时出错: {e}")
            return False
    
    def visualize_answer(self, context, question, answer_result):
        # 可视化回答结果
        # 注意：这是一个简化的可视化实现
        # 实际应用中可能需要更复杂的可视化方法
        print("===== 问答结果可视化 ======")
        print(f"问题: {question}")
        print("\n上下文预览:")
        
        # 打印上下文预览（限制长度）
        if len(context) > 200:
            context_preview = context[:200] + "..."
        else:
            context_preview = context
        print(context_preview)
        
        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}")
        
        print("=========================")
        
        return {
            "question": question,
            "answer": answer_result.get("answer", answer_result.get("answers", [])),
            "context_preview": context_preview
        }

# 示例使用代码
# # 初始化问答系统
# qa_system = AdvancedQuestionAnswering(
#     model_name="google/gemma-2-9b-it"
# )
# print("模型信息:", qa_system.get_model_info())
# 
# # 示例上下文
# context = """
# 人工智能（AI）是计算机科学的一个分支，旨在创建能够执行通常需要人类智能的任务的系统。这些任务包括理解自然语言、识别图像、解决问题和学习。AI可以分为弱AI（或窄AI）和强AI。弱AI被设计来执行特定任务，如语音识别或图像分类。强AI，也称为通用人工智能（AGI），是一种假设的AI形式，能够理解、学习和应用知识到任何问题上，类似于人类。
# 
# 机器学习（ML）是AI的一个子领域，它使计算机系统能够通过数据而不是明确的编程来学习和改进。深度学习是机器学习的一个子集，它使用人工神经网络（受人类大脑结构启发）来学习数据的表示。深度学习在过去十年中取得了显著进展，特别是在计算机视觉、自然语言处理和语音识别领域。
# 
# AI的应用非常广泛，包括医疗保健（诊断、药物发现）、金融（欺诈检测、算法交易）、交通（自动驾驶汽车）、教育（个性化学习）和娱乐（推荐系统、游戏AI）等。随着技术的不断发展，AI有望在更多领域创造价值，同时也带来了关于隐私、伦理和就业影响的重要问题。
# """
# 
# # 示例问题
# questions = [
#     "什么是人工智能？",
#     "弱AI和强AI的区别是什么？",
#     "机器学习和深度学习的关系是什么？",
#     "AI在医疗保健领域有哪些应用？",
#     "AI发展带来了哪些挑战？"
# ]
# 
# # 回答单个问题
# print("回答单个问题...")
# question = questions[0]
# result = qa_system.answer_question(context, question)
# print(f"问题: {question}")
# print(f"回答: {result['answer']}")
# 
# # 可视化回答结果
# print("\n可视化回答结果...")
# qa_system.visualize_answer(context, question, result)
# 
# # 回答多个问题
# print("\n回答多个问题...")
# for question in questions[1:]:
#     result = qa_system.answer_question(context, question)
#     print(f"问题: {question}")
#     print(f"回答: {result['answer']}")
#     print("---")
# 
# # 批量回答问题
# print("\n批量回答问题...")
# batch_results = qa_system.batch_answer_questions(context, questions, batch_size=2)
# for i, result in enumerate(batch_results):
#     print(f"问题 {i+1}: {result['question']}")
#     print(f"回答: {result['answer']}")
#     print("---")
# 
# # 评估模型性能（注意：这需要下载相应的数据集，可能需要一定时间和存储空间）
# # print("\n评估模型性能...")
# # eval_results = qa_system.evaluate_with_dataset(sample_size=10)
# # print("评估结果:", eval_results)
# 
# # 保存模型
# # print("\n保存模型...")
# # qa_system.save_model("qa_model")
# 
# # 加载模型
# # print("\n加载模型...")
# # qa_system.load_model("qa_model")

# 问答模型量化技术优化示例
import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import AutoTokenizer, AutoModelForQuestionAnswering, AutoModelForCausalLM, BitsAndBytesConfig
import numpy as np
from tqdm import tqdm
import time
import os
import json
from torch.utils.data import DataLoader, Dataset

# 2025年基于大型预训练语言模型的表问答示例实现
import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import AutoTokenizer, AutoModelForTableQuestionAnswering, pipeline
from datasets import load_dataset, load_metric
import numpy as np
import pandas as pd
from tqdm import tqdm
import os
import json

class AdvancedTableQA:
    def __init__(self, model_name="google/gemma-2-9b-it", device=None):
        # 加载预训练的语言模型和分词器
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        
        # 尝试加载专用的表问答模型，如果失败则使用通用模型
        try:
            self.model = AutoModelForTableQuestionAnswering.from_pretrained(model_name)
        except Exception as e:
            print(f"无法加载专用的表问答模型，使用通用模型: {e}")
            # 对于不支持表问答的通用模型，我们需要进行适配
            from transformers import AutoModelForCausalLM
            self.model = AutoModelForCausalLM.from_pretrained(model_name)
            self.use_generic_model = True
        else:
            self.use_generic_model = False
        
        # 设置设备
        self.device = device if device is not None else ("cuda" if torch.cuda.is_available() else "cpu")
        self.model.to(self.device)
        
        # 设置模型为评估模式
        self.model.eval()
        
        # 获取模型信息
        self.model_name = model_name
        
        # 初始化pipeline用于快速推理（对于支持的模型）
        if not self.use_generic_model:
            try:
                self.pipeline = pipeline(
                    "table-question-answering",
                    model=self.model,
                    tokenizer=self.tokenizer,
                    device=self.device
                )
            except Exception:
                self.pipeline = None
        else:
            self.pipeline = None
    
    def process_table(self, table):
        # 处理表格数据
        # 支持多种格式的表格输入
        if isinstance(table, pd.DataFrame):
            # 如果是pandas DataFrame，转换为字典格式
            table_dict = {
                "header": table.columns.tolist(),
                "rows": table.values.tolist()
            }
        elif isinstance(table, dict) and "header" in table and "rows" in table:
            # 如果已经是标准的字典格式，直接使用
            table_dict = table
        else:
            # 其他格式，尝试转换为DataFrame再处理
            try:
                df = pd.DataFrame(table)
                table_dict = {
                    "header": df.columns.tolist(),
                    "rows": df.values.tolist()
                }
            except Exception as e:
                raise ValueError(f"不支持的表格格式: {e}")
        
        return table_dict
    
    def generate_prompt(self, table_dict, question):
        # 生成提示文本
        # 将表格转换为文本描述
        table_text = "表格:\n"
        # 添加表头
        table_text += " | ".join(table_dict["header"]) + "\n"
        # 添加分隔线
        table_text += " | ".join(["---"] * len(table_dict["header"])) + "\n"
        # 添加数据行
        for row in table_dict["rows"]:
            # 将每个元素转换为字符串
            row_str = [str(cell) if cell is not None else "" for cell in row]
            table_text += " | ".join(row_str) + "\n"
        
        # 添加问题
        prompt = f"{table_text}\n\n根据上面的表格，回答以下问题: {question}"
        
        return prompt
    
    def answer_question(self, table, question, max_length=1024, num_return_sequences=1):
        # 回答关于表格的问题
        # 处理表格
        table_dict = self.process_table(table)
        
        if not self.use_generic_model and self.pipeline is not None:
            # 如果有专用的表问答pipeline，使用它
            try:
                result = self.pipeline(table=table_dict, query=question)
                return {
                    "answer": result["answer"],
                    "confidence": result.get("score", None),
                    "model": self.model_name,
                    "question": question
                }
            except Exception as e:
                print(f"使用pipeline失败，回退到通用方法: {e}")
                # 失败时回退到通用方法
        
        # 生成提示文本
        prompt = self.generate_prompt(table_dict, question)
        
        # 准备输入
        inputs = self.tokenizer(prompt, return_tensors="pt", max_length=max_length, truncation=True).to(self.device)
        
        # 使用模型生成回答
        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_length=max_length,
                num_return_sequences=num_return_sequences,
                temperature=0.7,
                top_p=0.95,
                do_sample=True
            )
        
        # 解码生成的回答
        answers = []
        for i in range(num_return_sequences):
            # 提取生成的回答（去除输入部分）
            answer = self.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)
        
        # 返回结果
        if num_return_sequences == 1:
            return {
                "answer": answers[0],
                "model": self.model_name,
                "question": question,
                "prompt": prompt
            }
        else:
            return {
                "answers": answers,
                "model": self.model_name,
                "question": question,
                "prompt": prompt
            }
    
    def batch_answer_questions(self, tables, questions, batch_size=8, max_length=1024):
        # 批量回答关于表格的问题
        # 确保tables和questions是列表格式
        if isinstance(tables, (pd.DataFrame, dict)):
            tables = [tables] * len(questions)
        
        if isinstance(questions, str):
            questions = [questions] * len(tables)
        
        # 确保tables和questions长度相同
        if len(tables) != len(questions):
            raise ValueError(f"tables和questions长度不匹配: {len(tables)} vs {len(questions)}")
        
        # 处理所有表格
        processed_tables = []
        for table in tables:
            processed_tables.append(self.process_table(table))
        
        # 生成所有提示
        prompts = []
        for table_dict, question in zip(processed_tables, questions):
            prompts.append(self.generate_prompt(table_dict, question))
        
        # 批量处理
        results = []
        for i in tqdm(range(0, len(prompts), batch_size), desc="Batch Processing"):
            batch_prompts = prompts[i:i+batch_size]
            batch_questions = questions[i:i+batch_size]
            
            # 准备输入
            inputs = self.tokenizer(
                batch_prompts,
                return_tensors="pt",
                padding=True,
                truncation=True,
                max_length=max_length
            ).to(self.device)
            
            # 使用模型生成回答
            with torch.no_grad():
                outputs = self.model.generate(
                    **inputs,
                    max_length=max_length,
                    temperature=0.7,
                    top_p=0.95,
                    do_sample=True
                )
            
            # 解码生成的回答
            for j in range(len(batch_prompts)):
                # 提取生成的回答
                answer = self.tokenizer.decode(outputs[j], 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()
                
                results.append({
                    "answer": answer,
                    "model": self.model_name,
                    "question": batch_questions[j],
                    "prompt": batch_prompts[j]
                })
        
        return results
    
    def evaluate_with_dataset(self, dataset_name="wikitablequestions", split="validation", sample_size=100):
        # 使用数据集评估模型性能
        try:
            # 加载数据集
            dataset = load_dataset(dataset_name, split=split)
            
            # 采样数据
            if sample_size < len(dataset):
                dataset = dataset.shuffle(seed=42).select(range(sample_size))
            
            # 评估指标
            correct = 0
            total = len(dataset)
            
            # 评估模型
            print(f"评估模型在{sample_size}个样本上的性能...")
            for i, example in enumerate(tqdm(dataset, desc="Evaluating")):
                # 获取表格、问题和答案
                table = example["table"]
                question = example["question"]
                reference_answer = example["answer"]["text"][0] if "answer" in example else ""
                
                # 使用模型回答问题
                try:
                    result = self.answer_question(table, question)
                    predicted_answer = result["answer"]
                    
                    # 简单的正确性检查（实际应用中应使用更复杂的评估指标）
                    # 这里使用简单的文本匹配
                    if reference_answer.lower() in predicted_answer.lower() or predicted_answer.lower() in reference_answer.lower():
                        correct += 1
                    
                    # 每10个样本打印一次进度
                    if (i+1) % 10 == 0:
                        print(f"样本 {i+1}/{total}, 当前准确率: {correct/(i+1):.4f}")
                        print(f"问题: {question}")
                        print(f"参考答案: {reference_answer}")
                        print(f"模型回答: {predicted_answer}")
                        print("---")
                except Exception as e:
                    print(f"处理样本 {i} 时出错: {e}")
                    continue
            
            # 计算准确率
            accuracy = correct / total if total > 0 else 0
            print(f"评估完成，准确率: {accuracy:.4f}")
            
            return {
                "accuracy": accuracy,
                "correct": correct,
                "total": total,
                "dataset": dataset_name,
                "split": split
            }
        except Exception as e:
            print(f"评估过程中出错: {e}")
            return {
                "error": str(e),
                "dataset": dataset_name,
                "split": split
            }
    
    def get_model_info(self):
        # 获取模型信息
        return {
            "model_name": self.model_name,
            "device": self.device,
            "use_generic_model": self.use_generic_model,
            "has_pipeline": self.pipeline is not None,
            "vocab_size": self.tokenizer.vocab_size,
            "max_length": self.tokenizer.model_max_length
        }
    
    def save_model(self, save_path):
        # 保存模型
        try:
            # 创建保存目录
            os.makedirs(save_path, exist_ok=True)
            
            # 保存模型和分词器
            self.model.save_pretrained(save_path)
            self.tokenizer.save_pretrained(save_path)
            
            # 保存配置信息
            config = {
                "model_name": self.model_name,
                "use_generic_model": self.use_generic_model
            }
            with open(os.path.join(save_path, "config.json"), "w") as f:
                json.dump(config, f, indent=2)
            
            print(f"模型已保存到: {save_path}")
            return True
        except Exception as e:
            print(f"保存模型时出错: {e}")
            return False
    
    def load_model(self, load_path):
        # 加载模型
        try:
            # 加载配置信息
            with open(os.path.join(load_path, "config.json"), "r") as f:
                config = json.load(f)
            
            # 加载模型和分词器
            self.tokenizer = AutoTokenizer.from_pretrained(load_path)
            
            if config.get("use_generic_model", False):
                from transformers import AutoModelForCausalLM
                self.model = AutoModelForCausalLM.from_pretrained(load_path)
                self.use_generic_model = True
                self.pipeline = None
            else:
                try:
                    self.model = AutoModelForTableQuestionAnswering.from_pretrained(load_path)
                    self.use_generic_model = False
                    # 尝试初始化pipeline
                    try:
                        self.pipeline = pipeline(
                            "table-question-answering",
                            model=self.model,
                            tokenizer=self.tokenizer,
                            device=self.device
                        )
                    except Exception:
                        self.pipeline = None
                except Exception:
                    from transformers import AutoModelForCausalLM
                    self.model = AutoModelForCausalLM.from_pretrained(load_path)
                    self.use_generic_model = True
                    self.pipeline = None
            
            # 将模型移至设备
            self.model.to(self.device)
            self.model.eval()
            
            # 更新模型名称
            self.model_name = load_path
            
            print(f"模型已从{load_path}加载")
            return True
        except Exception as e:
            print(f"加载模型时出错: {e}")
            return False
    
    def visualize_answer(self, table, question, answer_result):
        # 可视化回答结果
        # 注意：这是一个简化的可视化实现
        # 实际应用中可能需要更复杂的可视化方法
        print("===== 表问答结果可视化 ======")
        print(f"问题: {question}")
        print("\n表格:")
        
        # 处理表格
        table_dict = self.process_table(table)
        
        # 打印表头
        header = " | ".join(table_dict["header"])
        print(header)
        print(" | ".join(["---"] * len(table_dict["header"])))
        
        # 打印前5行数据
        max_rows = 5
        for i, row in enumerate(table_dict["rows"]):
            row_str = [str(cell) if cell is not None else "" for cell in row]
            print(" | ".join(row_str))
            if i >= max_rows - 1 and len(table_dict["rows"]) > max_rows:
                print("...")
                break
        
        print("\n回答:")
        if "answer" in answer_result:
            print(answer_result["answer"])
        elif "answers" in answer_result:
            for i, ans in enumerate(answer_result["answers"]):
                print(f"答案 {i+1}: {ans}")
        
        print("=========================")
        
        return {
            "question": question,
            "answer": answer_result.get("answer", answer_result.get("answers", [])),
            "table_preview": table_dict
        }

# 示例使用代码
# # 初始化表问答系统
# table_qa_system = AdvancedTableQA(
#     model_name="google/gemma-2-9b-it"
# )
# print("模型信息:", table_qa_system.get_model_info())
# 
# # 创建一个示例表格
# data = {
#     "产品名称": ["笔记本电脑A", "笔记本电脑B", "平板电脑C", "智能手机D", "智能手表E"],
#     "价格": [5999, 7999, 3299, 4599, 1299],
#     "销量": [1200, 800, 2500, 3200, 1800],
#     "评分": [4.5, 4.8, 4.2, 4.7, 4.3]
# }
# df = pd.DataFrame(data)
# 
# # 示例问题
# questions = [
#     "价格最高的产品是什么？",
#     "销量最高的产品是什么？",
#     "评分超过4.5的产品有哪些？",
#     "价格在4000到6000之间的产品有哪些？",
#     "平均销量是多少？"
# ]
# 
# # 回答单个问题
# print("回答单个问题...")
# question = questions[0]
# result = table_qa_system.answer_question(df, question)
# print(f"问题: {question}")
# print(f"回答: {result['answer']}")
# 
# # 可视化回答结果
# print("\n可视化回答结果...")
# table_qa_system.visualize_answer(df, question, result)
# 
# # 回答多个问题
# print("\n回答多个问题...")
# for question in questions[1:]:
#     result = table_qa_system.answer_question(df, question)
#     print(f"问题: {question}")
#     print(f"回答: {result['answer']}")
#     print("---")
# 
# # 批量回答问题
# print("\n批量回答问题...")
# batch_results = table_qa_system.batch_answer_questions(df, questions, batch_size=2)
# for i, result in enumerate(batch_results):
#     print(f"问题 {i+1}: {result['question']}")
#     print(f"回答: {result['answer']}")
#     print("---")
# 
# # 评估模型性能（注意：这需要下载相应的数据集，可能需要一定时间和存储空间）
# # print("\n评估模型性能...")
# # eval_results = table_qa_system.evaluate_with_dataset(sample_size=10)
# # print("评估结果:", eval_results)
# 
# # 保存模型
# # print("\n保存模型...")
# # table_qa_system.save_model("table_qa_model")
# 
# # 加载模型
# # print("\n加载模型...")
# # table_qa_system.load_model("table_qa_model")