最近在极客时间学习《AI 大模型应用开发实战营》，老师留了一个项目作业，因为最近一直忙，居然拖到最后一天才写（梦回学生时代寒暑假TOT）。这个项目是利用大语言模型对非结构化的pdf进行翻译，我重新从零来复现一下这个项目，在这里简单记录下开发过程和心得体会，供有兴趣的同学参考。

把大象装冰箱需要三步，这个作业也需要三步：

1. 读取pdf

2. 把pdf喂给大模型，得到翻译结果

3. 保存翻译结果

把大象塞冰箱里说起来容易，做起来全是细节。我把在复现过程中的心得体会写在了程序的注释中，复现的项目有不足之处，还望共同学习。

由于这个项目复现的仓促，一些类似Argparse这样锦上添花的功能没有实现。

整个项目的模块都平铺在根目录，方便读者阅读，我尽量保证整个项目是低耦合的。

0. logger.py

# 首先写一个日志记录器，这个东西在项目开发中十分有用，开发人员永远不可能盯着终端打印看，另外终端打印也没有持久化的方式，所以一个可以往任何地方输出的logger就十分的重要
# Python自带的log包的使用微微有些复杂，这里借用老师的loguru包，这个包使用起来就很友好了。

from loguru import logger

LOG_FILE = "translation.log"
ROTATION_TIME = "02:00"

class Logger:

    def __init__(self, name="translation", log_dir="logs", level="DEBUG"):
        if not os.path.exists(log_dir):
            os.makedirs(log_dir)
        log_file_path = os.path.join(log_dir, LOG_FILE)

        logger.remove()    # loguru的logger默认自带一个将信息重定向到标准错误输出的handler，这个用处不大，直接移除

        logger.add(sys.stdout, level=level)    # 往终端上打印
        logger.add(log_file_path, rotation=ROTATION_TIME, level="DEBUG")    # 往日志文件里写，每天经过ROTATION_TIME后，日志被清空

        self.logger = logger

LOG = Logger().logger

1. main.py

import os, sys
sys.path.append(os.path.dirname(os.path.abspath(__file__)))    # 让Python解释器知道你这个项目应该把什么位置作为根目录来导包，不理解也没事儿。

# 有的同学想在项目的子模块中进行测试，而不是启动整个项目进行测试。如果将子模块作为入口点，那么python解释器会将当前子模块作为导包根目录，可能让程序无法正确找到包的位置。
# 这时候上面这个语句就十分关键，dirname中的参数可以设置为你项目的根目录，它可以让程序在非入口处正确引导python解释器从正确的根目录导包。
# 这种入侵式的测试代码修改其实是不推荐的，有一种可以将项目根目录位置写入本地环境的.pth文件的方法可以改进这种入侵式修改，网上一大堆，请自行学习。
# 对于没看懂我在说什么的同学，无所谓，忽略它。

from models import LLMModel
from logger import LOG
from pdf_reader import PDFParser, PDFWriter    # 一个负责pdf解析，一个负责pdf保存



pdf_file_path = '/pdf/The_Old_Man_of_the_Sea.pdf'
pdf_paraser = PDFParser(pdf_file_path, page_num=2)
pages_text = pdf_paraser.parse_pdf()

model = LLMModel(model_name='gpt-turbo')

res_text = []

for page_text in pages_text:
    temp_list = []
    for text in page_text:
        response, flag = model.translate(text)
        temp_list.append(response)
    res_text.append(temp_list)


pdf_writer = PDFWriter(save_path='/pdf/translators.pdf')
pdf_writer.save_pdf(res_text)

2. pdf_reader.py

import pdfplumber
from logger import LOG
import pandas as pd
from reportlab.lib import colors, pagesizes, units
from reportlab.platypus import (
    SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, PageBreak
)
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle


class PDFParser:
    def __init__(self, path: str, page_num: int) -> None:
        self.path = path
        self.page_num = page_num

    def parse_pdf(self):
        path = self.path
        with pdfplumber.open(path) as f:
            pages = f.pages[:self.page_num]
            pages_txt = []

            for idx, page in enumerate(pages):
                raw_text = page.extract_text() # 提取所有文本
                tables = page.extract_tables() # 提取所有表格

                table_list = []
                table_df_list = []

                for table_data in tables:
                    table_text = ""        
                    for row in table_data:
                        for cell in row:
                            table_text += cell
                    table_list.append(table_text) # 表格中的文字
                    table_df_list.append(pd.DataFrame(table_data)) # 表格中的数据
                assert len(table_list) == len(table_df_list)
            i = 0    
            flag = True
            len_table_list = len(table_list)

            if raw_text:
                raw_text_lines = raw_text.splitlines()
                cleaned_raw_text_list = []
                for line in raw_text_lines:
                    if line.strip():
                        if i < len_table_list and line in table_list[i] and not flag: # 遇到了表格
                            i += 1
                            flag = False
                            cleaned_raw_text_list.append(table_df_list[i])    
                        else:
                            flag = True
                        cleaned_raw_text_list.append(line.strip())
                        # 这个位置不要直接拼接，尽量保持短句的格式，需要拼接的时候再拼接
                        # 有时候一页的文字特别的多，大模型不能一次性吃那么多的文字（原因自行百度），所以可以一句一句翻译
                        # 给模型扩充上下文窗口，让模型可以吃下更多的文字，也是当下的研究热点之一
                    pages_txt.append(cleaned_raw_text_list)
                    cleaned_raw_text = "\n".join(cleaned_raw_text_list)
                    LOG.debug(f"[raw_text{idx}]\n {cleaned_raw_text}")    

                return pages_txt


class PDFWriter:
    def __init__(self, save_path:str) -> None:
        self.save_path = save_path
        self.doc = SimpleDocTemplate(save_path, pagesize=pagesizes.letter)    

    def save_pdf(self, pages_list: list):
        story = []
        simsun_style = ParagraphStyle('SimSun', fontName='SimSun', fontSize=12, leading=14)
        for page_list in pages_list:
            for page in page_list:
                para = Paragraph(page, simsun_style)
                story.append(para)
        self.doc.build(story) # 保存pdf文件

3. model.py

import os, openai
import requests, simplejson, time
from logger import LOG

class LLMModel():
    def __init__(self, model_name='gpt-turbo') -> None:
        openai.api_type = os.getenv("OPENAI_API_TYPE")
        openai.api_base = os.getenv("OPENAI_API_BASE")
        openai.api_version = os.getenv("OPENAI_API_VERSION")
        openai.api_key = os.getenv("OPENAI_API_KEY")
        self.model_name = model_name

    def get_deployed_models(self) -> None:
        deployment_list = openai.Deployment.list()['data']
        for deployment in deployment_list:
            print(f"model:{deployment['model']}")
            print(f"deployment:{deployment['id']}")

    def translate(self, prompt):
        attempts = 0
        prompt = f'翻译成中文：{prompt}'
        while attempts < 3:
        try:
            translation = ""
            response = openai.ChatCompletion.create(
                            engine=self.model_name,
                            messages=[
                                {"role": "user", "content": prompt}
                            ]
                        )
            translation = response.choices[0].message.get('content',None).strip()
            return translation, True
        except openai.error.RateLimitError:
            attempts += 1
            if attempts < 3:
                LOG.warning("Rate limit reached. Waiting for 60 seconds before retrying.")
                time.sleep(60)
            else:
                raise Exception("Rate limit reached. Maximum attempts exceeded.")
        except requests.exceptions.RequestException as e:
            raise Exception(f"请求异常：{e}")
        except requests.exceptions.Timeout as e:
            raise Exception(f"请求超时：{e}")
        except simplejson.errors.JSONDecodeError as e:
            raise Exception("Error: response is not valid JSON format.")
        except Exception as e:
            raise Exception(f"发生了未知错误：{e}")
    return "", False