Qwen 系列模型实现文本改写工具

Qwen 系列模型实现文本改写工具

文学语料库.xlsx

文学语料库_转换结果.xlsx

提示词

作为文学创作助手，请完成以下任务：
1. 对输入的文学性句子进行同义转换
2. 保持原句的核心主题、情感基调与文学体裁（诗歌/散文/小说等）
3. 通过以下方式增强表达多样性：- 替换等效但不同的文学意象（如"月光如水"→"银辉似练"）- 调整句式结构（长句拆分/短句合并）- 变换修辞手法（比喻→拟人/象征等）
4. 输出需符合文学创作专业语境，避免口语化表达
5. 直接输出改写后的句子，不包含任何解释或说明

import pandas as pd
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from openpyxl import Workbook
from openpyxl.styles import Font
from openpyxl.utils import get_column_letter
from tqdm import tqdm  # 用于显示进度条，需安装：pip install tqdmclass Config:"""配置类，集中管理所有常量参数"""BASE_URL = "http://localhost:8000/v1"API_KEY = ""MODEL_NAME = "Qwen3-8B"INPUT_EXCEL_PATH = "文学语料库.xlsx"OUTPUT_EXCEL_PATH = "文学语料库_转换结果.xlsx"BATCH_SIZE = 2  # 每处理多少条记录保存一次TIMEOUT = 60  # 请求超时时间（秒）MAX_RETRIES = 3  # 请求重试次数REQUIRED_COLUMNS = ["原文片段", "文学体裁", "主题分类"]  # 必要列名def create_http_session():"""创建带有重试机制的HTTP会话，提高请求稳定性"""session = requests.Session()retry_strategy = Retry(total=Config.MAX_RETRIES,backoff_factor=1,  # 重试间隔：1秒, 2秒, 4秒...status_forcelist=[500, 502, 503, 504]  # 需要重试的状态码)adapter = HTTPAdapter(max_retries=retry_strategy)session.mount("http://", adapter)session.mount("https://", adapter)return sessiondef build_literary_prompt(original_text):"""构建文学改写提示词，明确任务要求"""return f"""作为文学创作助手，请完成以下任务：
1. 对输入的文学性句子进行同义转换
2. 保持原句的核心主题、情感基调与文学体裁（诗歌/散文/小说等）
3. 通过以下方式增强表达多样性：- 替换等效但不同的文学意象（如"月光如水"→"银辉似练"）- 调整句式结构（长句拆分/短句合并）- 变换修辞手法（比喻→拟人/象征等）
4. 输出需符合文学创作专业语境，避免口语化表达
5. 直接输出改写后的句子，不包含任何解释或说明输入内容：{original_text}"""def adjust_column_width(worksheet):"""自动调整Excel列宽，优化可读性"""for column in worksheet.columns:max_length = 0column_letter = get_column_letter(column[0].column)for cell in column:try:if len(str(cell.value)) > max_length:max_length = len(str(cell.value))except (TypeError, ValueError):passadjusted_width = min(max_length + 2, 50)  # 限制最大宽度为50worksheet.column_dimensions[column_letter].width = adjusted_widthdef process_single_row(session, row):"""处理单行数据，调用API进行改写"""original_text = str(row["原文片段"]).strip()if not original_text:return {"status": "skip", "message": "原文片段为空"}# 构建请求参数headers = {"Content-Type": "application/json","Authorization": f"Bearer {Config.API_KEY}"}data = {"model": Config.MODEL_NAME,"messages": [{"role": "user", "content": build_literary_prompt(original_text)}],"chat_template_kwargs": {"enable_thinking": False},"top_k": 1000,"temperature": 0.9,"max_tokens": 1024,"presence_penalty": 0.5,"frequency_penalty": 0.5,"n": 1}try:response = session.post(url=f"{Config.BASE_URL}/chat/completions",headers=headers,json=data,timeout=Config.TIMEOUT)response.raise_for_status()  # 触发HTTP错误状态码的异常completion = response.json()rewritten_text = completion.get("choices", [{}])[0].get("message", {}).get("content", "")if not rewritten_text.strip():return {"status": "warning", "message": "API返回空内容", "rewritten": "无有效改写内容"}return {"status": "success", "rewritten": rewritten_text.strip()}except requests.exceptions.HTTPError as e:return {"status": "error", "message": f"HTTP错误: {str(e)}", "rewritten": f"请求失败 ({response.status_code})"}except requests.exceptions.RequestException as e:return {"status": "error", "message": f"请求异常: {str(e)}", "rewritten": f"请求错误: {str(e)}"}def initialize_workbook():"""初始化Excel工作簿和表头"""wb = Workbook()ws = wb.activews.title = "文学改写结果"# 设置表头headers = ["原始文本", "改写文本", "文学体裁", "主题分类", "处理状态"]for col_idx, header in enumerate(headers, 1):cell = ws.cell(row=1, column=col_idx)cell.value = headercell.font = Font(bold=True)return wb, wsdef main():# 初始化HTTP会话session = create_http_session()try:# 读取输入Exceltry:df = pd.read_excel(Config.INPUT_EXCEL_PATH)print(f"✅ 成功读取Excel文件，共{len(df)}条记录")except FileNotFoundError:print(f"❌ 错误：找不到输入文件 '{Config.INPUT_EXCEL_PATH}'")returnexcept Exception as e:print(f"❌ 错误：读取Excel失败 - {str(e)}")return# 检查必要列for col in Config.REQUIRED_COLUMNS:if col not in df.columns:print(f"❌ 错误：Excel缺少必要列 '{col}'")return# 初始化输出工作簿wb, ws = initialize_workbook()current_row = 2  # 从第2行开始写入数据（第1行是表头）success_count = 0error_count = 0# 分批处理数据（带进度条）for start in tqdm(range(0, len(df), Config.BATCH_SIZE), desc="处理进度"):batch = df.iloc[start:start + Config.BATCH_SIZE]for _, row in batch.iterrows():# 处理单行数据result = process_single_row(session, row)# 写入Excelws.cell(row=current_row, column=1).value = str(row["原文片段"]).strip()ws.cell(row=current_row, column=2).value = result["rewritten"]ws.cell(row=current_row, column=3).value = row["文学体裁"]ws.cell(row=current_row, column=4).value = row["主题分类"]ws.cell(row=current_row, column=5).value = result["status"]# 更新统计if result["status"] == "success":success_count += 1elif result["status"] == "error":error_count += 1current_row += 1# 每批处理完保存一次wb.save(Config.OUTPUT_EXCEL_PATH)adjust_column_width(ws)# 最终保存并调整列宽adjust_column_width(ws)wb.save(Config.OUTPUT_EXCEL_PATH)# 输出统计结果print("\n" + "="*50)print(f"处理完成！总记录数: {len(df)}")print(f"✅ 成功: {success_count} 条")print(f"⚠️ 警告: {len(df) - success_count - error_count} 条")print(f"❌ 错误: {error_count} 条")print(f"结果已保存到: {Config.OUTPUT_EXCEL_PATH}")print("="*50)finally:# 确保会话关闭session.close()if __name__ == "__main__":main()

流程

1. 准备工作：

   • 确保本地部署Qwen3-8B模型服务，API地址为http://localhost:8000/v1
   • 准备输入Excel文件“文学语料库.xlsx”，包含“原文片段”“文学体裁”“主题分类”三列数据

2. 程序启动：

   • 读取配置参数（模型名称、文件路径、超时时间等）
   • 创建带重试机制的HTTP会话，用于稳定调用API

3. 数据读取与校验：

   • 读取“文学语料库.xlsx”内容
   • 检查文件是否存在、格式是否正确，以及是否包含必要的三列数据
   • 若数据有误，提示错误信息并终止程序

4. 初始化输出文件：

   • 创建新Excel“文学语料库_转换结果.xlsx”
   • 设置表头：“原始文本”“改写文本”“文学体裁”“主题分类”“处理状态”

5. 批量处理文本：

   • 按批次（可自设置）处理输入数据
   • 对每条文本：
     • 若原文为空，标记“跳过”并记录状态
     • 否则，生成文学改写提示词（明确保持主题、变换表达等要求）
     • 调用Qwen3-8B模型API，发送提示词和原文
     • 获取模型返回的改写文本，记录结果

6. 实时保存与进度跟踪：

   • 每处理完一批数据，将结果写入输出Excel并调整列宽
   • 显示进度条，实时展示处理进度

7. 处理完成：

   • 统计总记录数、成功/警告/错误数量
   • 输出统计结果，提示文件保存路径
   • 关闭HTTP会话，程序结束