GLM-4-9B-Chat-1M实战教程使用Python requests调用vLLM API实现批量翻译任务1. 项目概述与准备工作今天我们来学习如何使用Python的requests库调用vLLM部署的GLM-4-9B-Chat-1M模型实现高效的批量翻译任务。这个模型支持1M的上下文长度相当于约200万中文字符特别适合处理长文本和多语言翻译需求。1.1 环境要求与前置准备在开始之前请确保你已经完成以下准备工作已通过vLLM成功部署GLM-4-9B-Chat-1M模型模型服务正常运行可通过cat /root/workspace/llm.log确认安装Python 3.8环境安装requests库pip install requests1.2 模型特点与优势GLM-4-9B-Chat-1M模型有几个显著特点让它特别适合翻译任务超长上下文支持1M tokens可处理超长文档多语言支持覆盖26种语言包括中文、英文、日语、韩语、德语等高质量翻译在语义理解和语言生成方面表现优异批量处理能力可同时处理多个翻译任务2. 基础API调用方法让我们先从最简单的单个翻译请求开始了解如何与vLLM API进行交互。2.1 基本的API请求结构import requests import json # vLLM API的基础URL根据你的实际部署地址修改 API_URL http://localhost:8000/v1/completions def translate_single_text(text, target_language英文): 翻译单条文本 # 构建提示词 prompt f请将以下内容翻译成{target_language}\n{text} # 请求参数 payload { model: glm-4-9b-chat-1m, prompt: prompt, max_tokens: 1000, temperature: 0.1, stop: [。, !, ?, \n\n] } # 设置请求头 headers { Content-Type: application/json } try: # 发送请求 response requests.post(API_URL, jsonpayload, headersheaders) response.raise_for_status() # 解析响应 result response.json() translated_text result[choices][0][text].strip() return translated_text except requests.exceptions.RequestException as e: print(f请求失败: {e}) return None except KeyError as e: print(f解析响应失败: {e}) return None # 使用示例 if __name__ __main__: text_to_translate 今天天气真好适合出去散步。 translated translate_single_text(text_to_translate, 英语) print(f原文: {text_to_translate}) print(f翻译: {translated})2.2 处理API响应在实际使用中我们需要更健壮地处理API响应def safe_translate(text, target_language, max_retries3): 带重试机制的翻译函数 for attempt in range(max_retries): try: result translate_single_text(text, target_language) if result and result ! text: # 确保不是返回了原文 return result else: print(f第{attempt1}次尝试翻译结果异常重试中...) except Exception as e: print(f第{attempt1}次尝试失败: {e}) print(所有重试均失败) return None3. 批量翻译实现方案现在我们来实现真正的批量翻译功能这是本教程的核心部分。3.1 简单的批量翻译函数import time from typing import List, Dict def batch_translate_simple(texts: List[str], target_language: str 英文) - List[str]: 简单的批量翻译实现 results [] for i, text in enumerate(texts): print(f正在翻译第 {i1}/{len(texts)} 条文本...) translated safe_translate(text, target_language) if translated: results.append(translated) else: results.append() # 翻译失败时返回空字符串 # 添加短暂延迟避免请求过于频繁 time.sleep(0.5) return results # 使用示例 texts [ 你好世界, 今天是个好天气。, 人工智能正在改变世界。, 学习编程很有趣。 ] translations batch_translate_simple(texts, 英语) for orig, trans in zip(texts, translations): print(f原文: {orig}) print(f翻译: {trans}) print(- * 50)3.2 高效的批量处理方案对于大量文本我们可以使用更高效的处理方式import concurrent.futures import threading class BatchTranslator: def __init__(self, api_url, max_workers5): self.api_url api_url self.max_workers max_workers self.lock threading.Lock() self.success_count 0 self.fail_count 0 def translate_with_stats(self, text, target_language): 带统计信息的翻译函数 result safe_translate(text, target_language) with self.lock: if result: self.success_count 1 else: self.fail_count 1 return result def parallel_batch_translate(self, texts, target_language英文): 并行批量翻译 self.success_count 0 self.fail_count 0 with concurrent.futures.ThreadPoolExecutor(max_workersself.max_workers) as executor: # 提交所有翻译任务 future_to_text { executor.submit(self.translate_with_stats, text, target_language): text for text in texts } results [] for future in concurrent.futures.as_completed(future_to_text): try: result future.result() results.append(result) except Exception as e: print(f翻译任务异常: {e}) results.append() print(f翻译完成成功: {self.success_count}, 失败: {self.fail_count}) return results # 使用示例 translator BatchTranslator(http://localhost:8000/v1/completions) large_texts [文本 str(i) for i in range(20)] # 示例文本列表 results translator.parallel_batch_translate(large_texts, 英语)4. 高级功能与优化技巧4.1 支持多种文件格式让我们扩展功能支持从不同文件格式读取和保存翻译结果import pandas as pd import json import csv class FileTranslator: def __init__(self, translator): self.translator translator def translate_csv(self, input_file, output_file, text_column, target_language): 翻译CSV文件中的文本列 # 读取CSV文件 df pd.read_csv(input_file) # 提取需要翻译的文本 texts_to_translate df[text_column].tolist() # 执行批量翻译 translations self.translator.parallel_batch_translate(texts_to_translate, target_language) # 添加翻译结果到DataFrame df[translation] translations # 保存结果 df.to_csv(output_file, indexFalse, encodingutf-8-sig) print(f翻译完成结果已保存到: {output_file}) def translate_json(self, input_file, output_file, target_language): 翻译JSON文件中的文本 with open(input_file, r, encodingutf-8) as f: data json.load(f) # 假设JSON中是文本列表 if isinstance(data, list): translations self.translator.parallel_batch_translate(data, target_language) # 创建包含原文和翻译的结果 result [ {original: orig, translation: trans} for orig, trans in zip(data, translations) ] with open(output_file, w, encodingutf-8) as f: json.dump(result, f, ensure_asciiFalse, indent2) print(fJSON翻译完成结果已保存到: {output_file}) # 使用示例 translator BatchTranslator(http://localhost:8000/v1/completions) file_translator FileTranslator(translator) # 翻译CSV文件 # file_translator.translate_csv(input.csv, output.csv, content, 英语)4.2 翻译质量优化技巧为了提高翻译质量我们可以优化提示词和参数设置def create_optimized_prompt(text, target_language, contextNone): 创建优化的翻译提示词 base_prompt f请将以下内容专业准确地翻译成{target_language}。 要求 1. 保持原文的意思和语气 2. 使用自然流畅的目标语言表达 3. 专业术语保持准确 4. 文化特定内容适当本地化 待翻译内容 {text} if context: base_prompt f上下文信息{context}\n\n base_prompt return base_prompt def high_quality_translate(text, target_language, contextNone): 高质量翻译函数 prompt create_optimized_prompt(text, target_language, context) payload { model: glm-4-9b-chat-1m, prompt: prompt, max_tokens: 1500, temperature: 0.3, # 稍高的温度值让翻译更自然 top_p: 0.9, frequency_penalty: 0.1, presence_penalty: 0.1 } headers {Content-Type: application/json} try: response requests.post(API_URL, jsonpayload, headersheaders) response.raise_for_status() result response.json() translated_text result[choices][0][text].strip() # 清理可能的多余内容 if 翻译 in translated_text: translated_text translated_text.split(翻译)[-1].strip() return translated_text except Exception as e: print(f高质量翻译失败: {e}) return None4.3 处理长文档翻译利用GLM-4-9B-Chat-1M的长文本能力我们可以处理完整文档def translate_long_document(document, target_language, chunk_size5000): 翻译长文档自动分块处理 # 简单的文本分块实际应用中可能需要更智能的分块 chunks [document[i:ichunk_size] for i in range(0, len(document), chunk_size)] print(f文档已分成 {len(chunks)} 个块进行翻译) translations [] for i, chunk in enumerate(chunks): print(f正在翻译第 {i1}/{len(chunks)} 块...) # 为每个块添加上下文信息 context f这是长文档的第{i1}部分请保持翻译的连贯性。 translated_chunk high_quality_translate(chunk, target_language, context) if translated_chunk: translations.append(translated_chunk) else: translations.append(f[第{i1}块翻译失败]) time.sleep(1) # 避免请求过于频繁 # 合并翻译结果 full_translation \n\n.join(translations) return full_translation5. 错误处理与性能监控5.1 完善的错误处理机制class RobustTranslator: def __init__(self, api_url, timeout30): self.api_url api_url self.timeout timeout self.session requests.Session() # 配置会话 self.session.headers.update({ Content-Type: application/json, User-Agent: GLM-Translator/1.0 }) def translate_with_retry(self, text, target_language, max_retries3): 带重试和超时控制的翻译函数 prompt create_optimized_prompt(text, target_language) payload { model: glm-4-9b-chat-1m, prompt: prompt, max_tokens: 1000, temperature: 0.1 } for attempt in range(max_retries): try: response self.session.post( self.api_url, jsonpayload, timeoutself.timeout ) if response.status_code 200: result response.json() translated_text result[choices][0][text].strip() return translated_text elif response.status_code 429: wait_time 2 ** attempt # 指数退避 print(f速率限制等待 {wait_time} 秒后重试...) time.sleep(wait_time) else: print(fHTTP错误 {response.status_code}: {response.text}) break except requests.exceptions.Timeout: print(f第{attempt1}次尝试超时) except requests.exceptions.ConnectionError: print(f第{attempt1}次连接错误) except Exception as e: print(f第{attempt1}次尝试失败: {e}) if attempt max_retries - 1: time.sleep(1) # 重试前等待 return None5.2 性能监控与统计import time from dataclasses import dataclass from datetime import datetime dataclass class TranslationStats: total_requests: int 0 successful_requests: int 0 failed_requests: int 0 total_chars_translated: int 0 start_time: datetime None end_time: datetime None def start_timing(self): self.start_time datetime.now() def end_timing(self): self.end_time datetime.now() property def duration(self): if self.start_time and self.end_time: return (self.end_time - self.start_time).total_seconds() return 0 property def success_rate(self): if self.total_requests 0: return 0 return self.successful_requests / self.total_requests property def chars_per_second(self): if self.duration 0: return 0 return self.total_chars_translated / self.duration class MonitoredTranslator: def __init__(self, api_url): self.api_url api_url self.stats TranslationStats() self.base_translator RobustTranslator(api_url) def translate_with_monitoring(self, text, target_language): 带性能监控的翻译 self.stats.total_requests 1 self.stats.total_chars_translated len(text) start_time time.time() result self.base_translator.translate_with_retry(text, target_language) end_time time.time() if result: self.stats.successful_requests 1 print(f翻译成功 - 耗时: {end_time - start_time:.2f}秒) else: self.stats.failed_requests 1 print(翻译失败) return result def print_stats(self): 打印统计信息 print(\n *50) print(翻译统计报告) print(*50) print(f总请求数: {self.stats.total_requests}) print(f成功请求: {self.stats.successful_requests}) print(f失败请求: {self.stats.failed_requests}) print(f成功率: {self.stats.success_rate:.2%}) print(f总翻译字符数: {self.stats.total_chars_translated}) print(f总耗时: {self.stats.duration:.2f}秒) print(f字符/秒: {self.stats.chars_per_second:.2f}) print(*50)6. 完整示例与总结6.1 完整的批量翻译脚本#!/usr/bin/env python3 GLM-4-9B-Chat-1M 批量翻译脚本 支持多种文件格式和并行处理 import argparse import pandas as pd from monitored_translator import MonitoredTranslator def main(): parser argparse.ArgumentParser(descriptionGLM-4-9B-Chat-1M 批量翻译工具) parser.add_argument(input, help输入文件路径) parser.add_argument(output, help输出文件路径) parser.add_argument(--target, default英语, help目标语言) parser.add_argument(--format, choices[csv, json, txt], help文件格式) parser.add_argument(--column, helpCSV文件中要翻译的列名) parser.add_argument(--workers, typeint, default3, help并行工作线程数) args parser.parse_args() # 初始化翻译器 translator MonitoredTranslator(http://localhost:8000/v1/completions) translator.stats.start_timing() try: if args.format csv or args.input.endswith(.csv): # 处理CSV文件 df pd.read_csv(args.input) if args.column and args.column in df.columns: texts df[args.column].tolist() # 批量翻译 translations [] for text in texts: translated translator.translate_with_monitoring(str(text), args.target) translations.append(translated if translated else ) # 保存结果 df[translation] translations df.to_csv(args.output, indexFalse, encodingutf-8-sig) else: print(请指定有效的列名) return elif args.format txt or args.input.endswith(.txt): # 处理文本文件 with open(args.input, r, encodingutf-8) as f: content f.read() # 简单按行分割可根据需要调整 texts [line.strip() for line in content.split(\n) if line.strip()] translations [] for text in texts: translated translator.translate_with_monitoring(text, args.target) translations.append(translated if translated else ) # 保存结果 with open(args.output, w, encodingutf-8) as f: for orig, trans in zip(texts, translations): f.write(f原文: {orig}\n) f.write(f翻译: {trans}\n) f.write(- * 40 \n) print(f翻译完成结果已保存到: {args.output}) except Exception as e: print(f处理过程中发生错误: {e}) finally: translator.stats.end_timing() translator.print_stats() if __name__ __main__: main()6.2 使用示例与总结要使用这个批量翻译脚本你可以这样调用# 翻译CSV文件 python batch_translator.py input.csv output.csv --format csv --column content --target 英语 # 翻译文本文件 python batch_translator.py document.txt translated.txt --format txt --target 日语通过本教程你学会了基础API调用如何使用requests库与vLLM部署的GLM模型交互批量处理技术实现高效的多文本并行翻译文件格式支持处理CSV、JSON、TXT等多种格式的文件质量优化通过优化提示词和参数提升翻译质量错误处理完善的异常处理和重试机制性能监控实时监控翻译性能和统计信息GLM-4-9B-Chat-1M凭借其1M的超长上下文能力和多语言支持为批量翻译任务提供了强大的技术基础。结合Python的并发处理能力你可以构建出高效、稳定的翻译流水线满足各种实际应用需求。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。