基于大模型 Prompt + LangChain 架构的 AI 翻译系统全流程解析

from loguru import loggerimport osimport sys
# 定义日志文件名和轮转时间LOG_FILE = "exp.log"ROTATION_TIME = "01:00"
class Logger:"""日志类，用于配置和管理日志记录器。参数:- name: 日志记录器的名称，默认为"translation"。- log_dir: 存放日志文件的目录，默认为"logs"。- debug: 是否开启调试模式，默认为False。若开启，日志级别为DEBUG，否则为INFO。返回值:- 无"""def __init__(self, name="translation", log_dir="logs", debug=False):# 创建日志目录，如果不存在的话if not os.path.exists(log_dir):os.makedirs(log_dir)log_file_path = os.path.join(log_dir, LOG_FILE)
# 移除loguru的默认处理器logger.remove()
# 添加控制台处理器，带有特定的日志级别level = "DEBUG" if debug else "INFO"logger.add(sys.stdout, level=level)# 添加文件处理器，带有特定日志级别和定时轮转logger.add(log_file_path, rotation=ROTATION_TIME, level="DEBUG")self.logger = logger
# 配置全局日志记录器，开启调试模式LOG = Logger(debug=True).logger
if __name__ == "__main__":# 在主程序中创建和使用日志记录器log = Logger().logger
# 示例日志消息log.debug("This is a debug message.")log.info("This is an info message.")log.warning("This is a warning message.")log.error("This is an error message.")

from .page import Page
class Book:"""代表一本书的类，可以包含多个页面。参数:- pdf_file_path (str): PDF文件的路径。属性:- pdf_file_path (str): 存储PDF文件路径。- pages (list): 存储书中的页面对象。"""def __init__(self, pdf_file_path):"""初始化Book实例。参数:- pdf_file_path (str): PDF文件的路径。"""self.pdf_file_path = pdf_file_pathself.pages = []
def add_page(self, page: Page):"""向书中添加一个页面。参数:- page (Page): 要添加的页面对象。"""self.pages.append(page)

from .content import Content
class Page:"""Page类用于创建和管理页面内容。属性:contents (list): 存储Content对象的列表。"""def __init__(self):"""初始化Page对象，创建一个空的内容列表。"""self.contents = []# 初始化一个空列表，用于存储页面内容
def add_content(self, content: Content):"""向页面中添加内容。参数:content (Content): 待添加到页面的内容对象。"""self.contents.append(content)# 将新内容添加到内容列表中

import pandas as pd
from enum import Enum, autofrom PIL import Image as PILImagefrom io import StringIOfrom utils import LOG
# 定义内容类型枚举class ContentType(Enum):TEXT = auto()# 文本类型TABLE = auto()# 表格类型IMAGE = auto()# 图像类型
# 定义内容类，支持文本、表格、图像内容的存储和翻译class Content:def __init__(self, content_type, original, translation=None):"""初始化内容对象。:param content_type: 内容类型（ContentType枚举）。:param original: 原始内容。:param translation: 翻译后的内容（默认为None）。"""self.content_type = content_typeself.original = originalself.translation = translationself.status = False# 翻译状态标志
def set_translation(self, translation, status):"""设置翻译后的内容并更新状态。:param translation: 翻译后的内容。:param status: 翻译状态（True或False）。:raises ValueError: 当翻译类型与期望类型不匹配时抛出。"""if not self.check_translation_type(translation):raise ValueError(f"Invalid translation type. Expected {self.content_type}, but got {type(translation)}")self.translation = translationself.status = status
def check_translation_type(self, translation):"""检查翻译内容的类型是否匹配。:param translation: 待检查的翻译内容。:return: 布尔值，类型匹配返回True，否则返回False。"""if self.content_type == ContentType.TEXT and isinstance(translation, str):return Trueelif self.content_type == ContentType.TABLE and isinstance(translation, list):return Trueelif self.content_type == ContentType.IMAGE and isinstance(translation, PILImage.Image):return Truereturn False
def __str__(self):return self.original# 返回原始内容的字符串表示
# 表格内容类，继承自Content类，提供特定于表格内容的操作class TableContent(Content):def __init__(self, data, translation=None):"""初始化表格内容对象。:param data: 表格数据，二维列表形式。:param translation: 翻译后的表格数据（默认为None）。:raises ValueError: 当数据与创建的DataFrame对象的行数或列数不匹配时抛出。"""df = pd.DataFrame(data)
# 验证数据和DataFrame对象的行数、列数是否匹配if len(data) != len(df) or len(data[0]) != len(df.columns):raise ValueError("The number of rows and columns in the extracted table data and DataFrame object do not match.")super().__init__(ContentType.TABLE, df)
def set_translation(self, translation, status):"""设置翻译后的表格内容并更新状态。:param translation: 翻译后的表格内容，字符串形式。:param status: 翻译状态（True或False）。:raises ValueError: 当翻译格式不正确或类型不匹配时抛出。"""try:if not isinstance(translation, str):raise ValueError(f"Invalid translation type. Expected str, but got {type(translation)}")
LOG.debug(f"[translation]\n{translation}")# 从字符串解析表格头和数据header = translation.split(']')[0][1:].split(', ')data_rows = translation.split('] ')[1:]data_rows = [row[1:-1].split(', ') for row in data_rows]translated_df = pd.DataFrame(data_rows, columns=header)LOG.debug(f"[translated_df]\n{translated_df}")self.translation = translated_dfself.status = statusexcept Exception as e:LOG.error(f"An error occurred during table translation: {e}")self.translation = Noneself.status = False
def __str__(self):return self.original.to_string(header=False, index=False)# 返回表格的字符串表示，不包含表头和索引
def iter_items(self, translated=False):"""遍历表格项。:param translated: 是否遍历翻译后的表格（默认为False，遍历原始表格）。:return: 生成器，每次返回一行的索引和值。"""target_df = self.translation if translated else self.originalfor row_idx, row in target_df.iterrows():for col_idx, item in enumerate(row):yield (row_idx, col_idx, item)
def update_item(self, row_idx, col_idx, new_value, translated=False):"""更新表格项的值。:param row_idx: 行索引。:param col_idx: 列索引。:param new_value: 新值。:param translated: 是否更新翻译后的表格项（默认为False，更新原始表格项）。"""target_df = self.translation if translated else self.originaltarget_df.at[row_idx, col_idx] = new_value
def get_original_as_str(self):"""获取原始表格的字符串表示。:return: 原始表格的字符串表示，不包含表头和索引。"""return self.original.to_string(header=False, index=False)

import yaml
class TranslationConfig:_instance = Nonedef __new__(cls):"""实现单例模式的构造方法。返回：TranslationConfig的单例实例。"""if cls._instance is None:cls._instance = super(TranslationConfig, cls).__new__(cls)cls._instance._config = Nonereturn cls._instancedef initialize(self, args):"""初始化配置，读取配置文件并允许通过命令行参数覆盖配置。参数：args: 包含配置文件路径的命名空间（argparse的返回值）。"""with open(args.config_file, "r") as f:config = yaml.safe_load(f)
# 使用命令行参数覆盖配置文件中的值overridden_values = {key: value for key, value in vars(args).items() if key in config and value is not None}config.update(overridden_values)# 存储原始配置字典self._instance._config = config
def __getattr__(self, name):"""重写getattr方法，从配置字典中获取属性值。参数：name: 尝试获取的属性名。返回：如果属性存在于配置字典中，则返回其值；否则抛出AttributeError。"""# 尝试从_config中获取属性if self._instance._config and name in self._instance._config:return self._instance._config[name]raise AttributeError(f"'TranslationConfig' object has no attribute '{name}'")

from typing import Optionalfrom translator.pdf_parser import PDFParserfrom translator.writer import Writerfrom translator.translation_chain import TranslationChainfrom utils import LOG
class PDFTranslator:"""PDFTranslator类用于将PDF文档从一种语言翻译成另一种语言。参数:- model_name: str，翻译模型的名称。"""def __init__(self, model_name: str):"""初始化PDFTranslator实例。参数:- model_name: str，翻译模型的名称。"""self.translate_chain = TranslationChain(model_name)# 创建翻译链实例self.pdf_parser = PDFParser()# 创建PDF解析器实例self.writer = Writer()# 创建写入器实例
def translate_pdf(self,input_file: str,output_file_format: str = 'markdown',source_language: str = "English",target_language: str = 'Chinese',pages: Optional[int] = None):"""翻译PDF文档并将其保存为指定格式的文件。参数:- input_file: str，输入的PDF文件路径。- output_file_format: str，输出文件的格式，默认为'markdown'。- source_language: str，源语言，默认为'English'。- target_language: str，目标语言，默认为'Chinese'。- pages: Optional[int]，要翻译的PDF页面范围，可以是单个页面或页面范围，None表示所有页面。返回:- str，翻译后文件的保存路径。"""self.book = self.pdf_parser.parse_pdf(input_file, pages)# 解析PDF文档
# 遍历并翻译每一页的内容for page_idx, page in enumerate(self.book.pages):for content_idx, content in enumerate(page.contents):# 对内容进行翻译translation, status = self.translate_chain.run(content, source_language, target_language)# 将翻译结果直接更新到页面内容中self.book.pages[page_idx].contents[content_idx].set_translation(translation, status)return self.writer.save_translated_book(self.book, output_file_format)# 保存翻译后的文档

import pdfplumberfrom typing import Optionalfrom book import Book, Page, Content, ContentType, TableContentfrom translator.exceptions import PageOutOfRangeExceptionfrom utils import LOG

class PDFParser:"""PDF解析器类，用于解析PDF文件并提取文本和表格内容。"""
def __init__(self):"""初始化PDF解析器。"""pass
def parse_pdf(self, pdf_file_path: str, pages: Optional[int] = None) -> Book:"""解析PDF文件，提取每页的文本和表格内容。
参数:- pdf_file_path: str，PDF文件的路径。- pages: Optional[int]，要解析的页面数，若为None则解析所有页面。
返回:- Book，包含解析得到的文本和表格内容的书对象。"""book = Book(pdf_file_path)
with pdfplumber.open(pdf_file_path) as pdf:# 检查指定页面范围是否超出PDF实际页面数if pages is not None and pages > len(pdf.pages):raise PageOutOfRangeException(len(pdf.pages), pages)
# 根据是否指定了页面数，确定要解析的页面范围if pages is None:pages_to_parse = pdf.pageselse:pages_to_parse = pdf.pages[:pages]
for pdf_page in pages_to_parse:page = Page()
# 提取原始文本内容和表格raw_text = pdf_page.extract_text()tables = pdf_page.extract_tables()
# 从原始文本中移除表格内容for table_data in tables:for row in table_data:for cell in row:raw_text = raw_text.replace(cell, "", 1)
# 处理文本内容if raw_text:# 清理文本，移除空行和首尾空白字符raw_text_lines = raw_text.splitlines()cleaned_raw_text_lines = [line.strip() for line in raw_text_lines if line.strip()]cleaned_raw_text = "\n".join(cleaned_raw_text_lines)
text_content = Content(content_type=ContentType.TEXT, original=cleaned_raw_text)page.add_content(text_content)LOG.debug(f"[raw_text]\n {cleaned_raw_text}")
# 处理表格内容if tables:table = TableContent(tables)page.add_content(table)LOG.debug(f"[table]\n{table}")
book.add_page(page)
return book

import osfrom reportlab.lib import colors, pagesizes, unitsfrom reportlab.lib.styles import getSampleStyleSheet, ParagraphStylefrom reportlab.pdfbase import pdfmetricsfrom reportlab.pdfbase.ttfonts import TTFontfrom reportlab.platypus import (SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, PageBreak)
from book import Book, ContentTypefrom utils import LOG
class Writer:"""Writer类用于将书籍内容导出为不同格式的文件，目前支持PDF和Markdown格式。"""def __init__(self):pass
def save_translated_book(self, book: Book, ouput_file_format: str):"""根据指定的输出文件格式，保存翻译后的书籍内容。:param book: 书籍对象，包含翻译后的内容。:param ouput_file_format: 输出文件格式，支持"pdf"和"markdown"。:return: 保存成功则返回输出文件路径，否则返回空字符串。"""LOG.debug(ouput_file_format)
if ouput_file_format.lower() == "pdf":output_file_path = self._save_translated_book_pdf(book)elif ouput_file_format.lower() == "markdown":output_file_path = self._save_translated_book_markdown(book)else:LOG.error(f"不支持文件类型: {ouput_file_format}")return ""
LOG.info(f"翻译完成，文件保存至: {output_file_path}")
return output_file_path

def _save_translated_book_pdf(self, book: Book, output_file_path: str = None):"""将翻译后的书籍内容导出为PDF文件。:param book: 书籍对象，包含翻译后的内容。:param output_file_path: 输出PDF文件路径，默认为None，如果为None则自动生成。:return: 输出PDF文件的路径。"""output_file_path = book.pdf_file_path.replace('.pdf', f'_translated.pdf')
LOG.info(f"开始导出: {output_file_path}")
# 注册中文字体font_path = "../fonts/simsun.ttc"# 字体文件路径，请根据实际情况修改pdfmetrics.registerFont(TTFont("SimSun", font_path))
# 创建PDF文档样式simsun_style = ParagraphStyle('SimSun', fontName='SimSun', fontSize=12, leading=14)
# 创建PDF文档doc = SimpleDocTemplate(output_file_path, pagesize=pagesizes.letter)styles = getSampleStyleSheet()story = []
# 遍历页面和内容，将翻译后的内容添加到PDF中for page in book.pages:for content in page.contents:if content.status:if content.content_type == ContentType.TEXT:# 添加翻译的文本到PDFtext = content.translationpara = Paragraph(text, simsun_style)story.append(para)
elif content.content_type == ContentType.TABLE:# 添加表格到PDFtable = content.translationtable_style = TableStyle([('BACKGROUND', (0, 0), (-1, 0), colors.grey),('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke),('ALIGN', (0, 0), (-1, -1), 'CENTER'),('FONTNAME', (0, 0), (-1, 0), 'SimSun'),# 表头字体设置为 "SimSun"('FONTSIZE', (0, 0), (-1, 0), 14),('BOTTOMPADDING', (0, 0), (-1, 0), 12),('BACKGROUND', (0, 1), (-1, -1), colors.beige),('FONTNAME', (0, 1), (-1, -1), 'SimSun'),# 表格中的字体设置为 "SimSun"('GRID', (0, 0), (-1, -1), 1, colors.black)])pdf_table = Table(table.values.tolist())pdf_table.setStyle(table_style)story.append(pdf_table)# 在每个页面后添加分页符，除了最后一页if page != book.pages[-1]:story.append(PageBreak())
# 生成并保存PDF文件doc.build(story)return output_file_path

def _save_translated_book_markdown(self, book: Book, output_file_path: str = None):"""将翻译后的书籍内容导出为Markdown文件。:param book: 书籍对象，包含翻译后的内容。:param output_file_path: 输出Markdown文件路径，默认为None，如果为None则自动生成。:return: 输出Markdown文件的路径。"""output_file_path = book.pdf_file_path.replace('.pdf', f'_translated.md')
LOG.info(f"开始导出: {output_file_path}")with open(output_file_path, 'w', encoding='utf-8') as output_file:# 遍历页面和内容，将翻译后的内容添加到Markdown文件中for page in book.pages:for content in page.contents:if content.status:if content.content_type == ContentType.TEXT:# 添加翻译的文本到Markdown文件text = content.translationoutput_file.write(text + '\n\n')
elif content.content_type == ContentType.TABLE:# 添加表格到Markdown文件table = content.translationheader = '| ' + ' | '.join(str(column) for column in table.columns) + ' |' + '\n'separator = '| ' + ' | '.join(['---'] * len(table.columns)) + ' |' + '\n'body = '\n'.join(['| ' + ' | '.join(str(cell) for cell in row) + ' |' for row in table.values.tolist()]) + '\n\n'output_file.write(header + separator + body)
# 在每个页面后添加分页符（水平线），除了最后一页if page != book.pages[-1]:output_file.write('---\n\n')
return output_file_path

from langchain.chains import LLMChainfrom langchain.chat_models import ChatOpenAI #直接访问OpenAI的GPT服务import os# 加载 .env 文件from dotenv import load_dotenv, find_dotenvimport openaifrom langchain.prompts.chat import (ChatPromptTemplate,SystemMessagePromptTemplate,HumanMessagePromptTemplate,)from utils import LOG
class TranslationChain:"""TranslationChain 类用于创建和管理一个语言翻译链。参数:- model_name: str, 指定用于翻译的 OpenAI 模型名称，默认为 "gpt-3.5-turbo"。- verbose: bool, 是否在执行过程中输出详细信息，默认为 True。"""def __init__(self, model_name: str = "gpt-3.5-turbo", verbose: bool = True):# 从环境变量中加载 OpenAI 的 API Key 和 URLload_dotenv(find_dotenv())openai.api_key = os.getenv('OPENAI_API_KEY')openai.api_base = os.getenv('OPENAI_API_URL')model = os.getenv('OPENAI_API_MODEL')
# 初始化翻译任务的 ChatPromptTemplate，定义系统和用户之间的对话模式template = ("""You are a translation expert, proficient in various languages. \nTranslates {source_language} to {target_language}.""")system_message_prompt = SystemMessagePromptTemplate.from_template(template)
# 初始化待翻译文本的提示模板，由用户输入human_template = "{text}"human_message_prompt = HumanMessagePromptTemplate.from_template(human_template)
# 将系统和用户提示模板组合成完整的 ChatPromptTemplatechat_prompt_template = ChatPromptTemplate.from_messages([system_message_prompt, human_message_prompt])
# 初始化 ChatOpenAI 对象，用于实际的翻译任务执行，设置 temperature 为 0 以确保结果稳定性chat = ChatOpenAI(model_name=model_name, temperature=0, verbose=verbose)
# 创建 LLMChain 对象，将聊天模型和提示模板封装起来，用于实际的对话流程执行self.chain = LLMChain(llm=chat, prompt=chat_prompt_template, verbose=verbose)
def run(self, text: str, source_language: str, target_language: str) -> (str, bool): # type: ignore"""执行翻译任务。参数:- text: str, 待翻译的文本。- source_language: str, 源语言代码。- target_language: str, 目标语言代码。返回:- result: str, 翻译后的文本。- success: bool, 任务执行是否成功。"""result = ""try:# 执行翻译流程result = self.chain.run({"text": text,"source_language": source_language,"target_language": target_language,})except Exception as e:# 记录翻译过程中出现的异常LOG.error(f"An error occurred during translation: {e}")return result, False
# 正常完成翻译，返回结果return result, True

import sysimport osimport gradio as gr
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from utils import ArgumentParser, LOGfrom translator import PDFTranslator, TranslationConfig

def translation(input_file, source_language, target_language):"""将PDF文件从源语言翻译成目标语言。参数:- input_file: 包含待翻译PDF的文件对象。- source_language: 源语言代码（字符串）。- target_language: 目标语言代码（字符串）。返回:- 翻译后PDF文件的路径（字符串）。"""# 记录翻译任务的开始，包括输入文件和语言信息LOG.debug(f"[翻译任务]\n源文件: {input_file.name}\n源语言: {source_language}\n目标语言: {target_language}")# 调用Translator类的translate_pdf方法进行翻译，并获取翻译后的文件路径output_file_path = Translator.translate_pdf(input_file.name, source_language=source_language, target_language=target_language)
return output_file_path
def launch_gradio():"""启动Gradio界面，提供用户友好的翻译服务界面。"""# 创建Gradio界面，设置功能描述、界面元素和输出iface = gr.Interface(fn=translation,title="智能体AI-Translator(PDF 电子书翻译工具)",inputs=[gr.File(label="上传PDF文件"),gr.Textbox(label="源语言（默认：英文）", placeholder="English", value="English"),gr.Textbox(label="目标语言（默认：中文）", placeholder="Chinese", value="Chinese")],outputs=[gr.File(label="下载翻译文件")],allow_flagging="never")
# 启动Gradio界面，设置为分享模式，并指定服务器地址iface.launch(share=True, server_name="0.0.0.0")
def initialize_translator():"""初始化翻译器，包括解析命令行参数和配置翻译模型。"""# 解析启动参数argument_parser = ArgumentParser()args = argument_parser.parse_arguments()# 初始化翻译配置config = TranslationConfig()config.initialize(args)# 实例化PDF翻译器，并准备进行翻译global TranslatorTranslator = PDFTranslator(config.model_name)

if __name__ == "__main__":# 初始化翻译器实例initialize_translator()# 启动Gradio翻译服务界面launch_gradio()