|
|
import gradio as gr |
|
|
import torch |
|
|
import gc, os |
|
|
|
|
|
import PyPDF2 |
|
|
import chardet |
|
|
os.environ["RWKV_V7_ON"] = '1' |
|
|
os.environ["RWKV_JIT_ON"] = '1' |
|
|
os.environ["RWKV_CUDA_ON"] = '1' |
|
|
from rwkv.model import RWKV |
|
|
from rwkv.utils import PIPELINE, PIPELINE_ARGS |
|
|
|
|
|
device = torch.device("cuda" if torch.cuda.is_available() else "cpu") |
|
|
ctx_limit = 4096 |
|
|
gen_limit = 4096 |
|
|
|
|
|
|
|
|
|
|
|
title_v6 = "RWKV_v7_G1_0.4B_Translate_ctx4096_20250620" |
|
|
model_path_v6 = "/home/alic-li/rwkv7-g1-tanslate/RWKV_v7_G1_1.5B_Translate_ctx4096_20250714_latest.pth" |
|
|
model_v6 = RWKV(model=model_path_v6.replace('.pth',''), strategy='cuda fp16') |
|
|
pipeline_v6 = PIPELINE(model_v6, "rwkv_vocab_v20230424") |
|
|
|
|
|
args = model_v6.args |
|
|
|
|
|
penalty_decay = 0.996 |
|
|
|
|
|
def evaluate( |
|
|
ctx, |
|
|
token_count=200, |
|
|
temperature=1.0, |
|
|
top_p=0.7, |
|
|
presencePenalty = 0.1, |
|
|
countPenalty = 0.1, |
|
|
): |
|
|
args = PIPELINE_ARGS(temperature = max(0.2, float(temperature)), top_p = float(top_p), |
|
|
alpha_frequency = countPenalty, |
|
|
alpha_presence = presencePenalty, |
|
|
token_ban = [], |
|
|
token_stop = [0]) |
|
|
ctx = ctx.strip() |
|
|
all_tokens = [] |
|
|
out_last = 0 |
|
|
out_str = '' |
|
|
occurrence = {} |
|
|
state = None |
|
|
for i in range(int(token_count)): |
|
|
|
|
|
input_ids = pipeline_v6.encode(ctx)[-ctx_limit:] if i == 0 else [token] |
|
|
out, state = model_v6.forward(input_ids, state) |
|
|
for n in occurrence: |
|
|
out[n] -= (args.alpha_presence + occurrence[n] * args.alpha_frequency) |
|
|
|
|
|
token = pipeline_v6.sample_logits(out, temperature=args.temperature, top_p=args.top_p) |
|
|
if token in args.token_stop: |
|
|
break |
|
|
all_tokens += [token] |
|
|
for xxx in occurrence: |
|
|
occurrence[xxx] *= penalty_decay |
|
|
|
|
|
ttt = pipeline_v6.decode([token]) |
|
|
www = 1 |
|
|
if ttt in ' \t0123456789': |
|
|
www = 0 |
|
|
|
|
|
|
|
|
if token not in occurrence: |
|
|
occurrence[token] = www |
|
|
else: |
|
|
occurrence[token] += www |
|
|
|
|
|
tmp = pipeline_v6.decode(all_tokens[out_last:]) |
|
|
if '\ufffd' not in tmp: |
|
|
out_str += tmp |
|
|
yield out_str.strip() |
|
|
out_last = i + 1 |
|
|
del out |
|
|
del state |
|
|
gc.collect() |
|
|
torch.cuda.empty_cache() |
|
|
yield out_str.strip() |
|
|
|
|
|
def translate_english_to_chinese(english_text, token_count, temperature, top_p, presence_penalty, count_penalty): |
|
|
if not english_text.strip(): |
|
|
return "Chinese:\n请输入英文内容。" |
|
|
|
|
|
full_prompt = f"English: {english_text}\n\nChinese:" |
|
|
for output in evaluate(full_prompt, token_count, temperature, top_p, presence_penalty, count_penalty): |
|
|
yield output |
|
|
|
|
|
def translate_chinese_to_chinses(Chinese_text, token_count, temperature, top_p, presence_penalty, count_penalty): |
|
|
if not Chinese_text.strip(): |
|
|
return "Chinses:\n请输入中文内容。" |
|
|
|
|
|
full_prompt = f"Chinese: {Chinese_text}\n\nEnglish:" |
|
|
for output in evaluate(full_prompt, token_count, temperature, top_p, presence_penalty, count_penalty): |
|
|
yield output |
|
|
|
|
|
def extract_text_from_file(file_path): |
|
|
if file_path.endswith(".txt"): |
|
|
with open(file_path, "rb") as f: |
|
|
raw_data = f.read() |
|
|
result = chardet.detect(raw_data) |
|
|
encoding = result["encoding"] or "utf-8" |
|
|
try: |
|
|
return raw_data.decode(encoding) |
|
|
except UnicodeDecodeError: |
|
|
return raw_data.decode("utf-8", errors="replace") |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
elif file_path.endswith(".pdf"): |
|
|
text = "" |
|
|
with open(file_path, "rb") as f: |
|
|
reader = PyPDF2.PdfReader(f) |
|
|
for page in reader.pages: |
|
|
text += page.extract_text() + "\n" |
|
|
return text |
|
|
|
|
|
else: |
|
|
with open(file_path, "rb") as f: |
|
|
raw_data = f.read() |
|
|
result = chardet.detect(raw_data) |
|
|
encoding = result["encoding"] or "utf-8" |
|
|
try: |
|
|
return raw_data.decode(encoding) |
|
|
except UnicodeDecodeError: |
|
|
return raw_data.decode("utf-8", errors="replace") |
|
|
|
|
|
def batch_translate_english_to_chinese(file, token_count, temperature, top_p, presence_penalty, count_penalty): |
|
|
if file is None: |
|
|
yield "请先上传文件", None |
|
|
return |
|
|
|
|
|
try: |
|
|
content = extract_text_from_file(file.name) |
|
|
except Exception as e: |
|
|
yield f"读取文件失败：{str(e)}", None |
|
|
return |
|
|
|
|
|
entries = [e.strip() for e in content.split('\n\n') if e.strip()] |
|
|
if len(entries) == 1: |
|
|
entries = [e.strip() for e in content.split('\n') if e.strip()] |
|
|
|
|
|
translated_texts = [] |
|
|
|
|
|
for idx, entry in enumerate(entries): |
|
|
full_prompt = f"English: {entry}\n\nChinese:" |
|
|
result = "" |
|
|
for output in evaluate(full_prompt, token_count, temperature, top_p, presence_penalty, count_penalty): |
|
|
result = output |
|
|
translated_texts.append(f"{result}") |
|
|
yield "\n".join(translated_texts), None |
|
|
|
|
|
output_text = "\n".join(translated_texts) |
|
|
output_path = f"{file.name}_translate.txt" |
|
|
with open(output_path, "w", encoding="utf-8") as f: |
|
|
f.write(output_text) |
|
|
|
|
|
yield "", output_path |
|
|
|
|
|
with gr.Blocks(title="RWKV_v7_G1_1.5B_Translate_ctx4096 English -> Chinese") as demo: |
|
|
with gr.Tab("English To Chinses"): |
|
|
gr.HTML(f"<div style='text-align:center;'><h1>RWKV_v7_G1_1.5B_Translate_ctx4096_2025062 English -> Chinese</h1></div>") |
|
|
with gr.Row(): |
|
|
with gr.Column(): |
|
|
english_input = gr.Textbox( |
|
|
label="英文输入(注意不能有空行)", |
|
|
lines=20, |
|
|
placeholder="请输入英文内容...", |
|
|
value="ROCm is an open-source stack, composed primarily of open-source software, designed for graphics processing unit (GPU) computation. ROCm consists of a collection of drivers, development tools, and APIs that enable GPU programming from low-level kernel to end-user applications.\n" |
|
|
"With ROCm, you can customize your GPU software to meet your specific needs.You can develop, collaborate, test, and deploy your applications in a free, open source, integrated, and secure software ecosystem. ROCm is particularly well-suited to GPU-accelerated high-performance computing (HPC), artificial intelligence (AI), scientific computing, and computer aided design (CAD).\n" |
|
|
"ROCm is powered by AMD’s Heterogeneous-computing Interface for Portability (HIP), an open-source software C++ GPU programming environment and its corresponding runtime. HIP allows ROCm developers to create portable applications on different platforms by deploying code on a range of platforms, from dedicated gaming GPUs to exascale HPC clusters.\n" |
|
|
"ROCm supports programming models, such as OpenMP and OpenCL, and includes all necessary open source software compilers, debuggers, and libraries. ROCm is fully integrated into machine learning (ML) frameworks, such as PyTorch and TensorFlow." |
|
|
) |
|
|
|
|
|
with gr.Column(): |
|
|
chinese_output = gr.Textbox( |
|
|
label="中文输出", |
|
|
lines=20, |
|
|
placeholder="翻译结果将显示在此处", |
|
|
value="" |
|
|
) |
|
|
|
|
|
with gr.Row(): |
|
|
translate_btn = gr.Button("Translate", variant="primary") |
|
|
clear_btn = gr.Button("Clear", variant="secondary") |
|
|
stop_btn = gr.Button("Stop", variant="stop") |
|
|
|
|
|
with gr.Accordion("Advanced Settings", open=False): |
|
|
token_count = gr.Slider(10, gen_limit, label="Max Tokens", step=10, value=gen_limit) |
|
|
temperature = gr.Slider(0.2, 2.0, label="Temperature", step=0.1, value=1.0) |
|
|
top_p = gr.Slider(0.0, 1.0, label="Top P", step=0.05, value=0) |
|
|
presence_penalty = gr.Slider(0.0, 1.0, label="Presence Penalty", step=0.1, value=0) |
|
|
count_penalty = gr.Slider(0.0, 1.0, label="Count Penalty", step=0.1, value=0) |
|
|
|
|
|
translate_event = translate_btn.click( |
|
|
fn=translate_english_to_chinese, |
|
|
inputs=[english_input, token_count, temperature, top_p, presence_penalty, count_penalty], |
|
|
outputs=[chinese_output] |
|
|
) |
|
|
|
|
|
clear_btn.click( |
|
|
fn=lambda: ("", ""), |
|
|
inputs=[], |
|
|
outputs=[english_input, chinese_output] |
|
|
) |
|
|
|
|
|
stop_btn.click( |
|
|
fn=None, |
|
|
inputs=None, |
|
|
outputs=None, |
|
|
cancels=[translate_event] |
|
|
) |
|
|
with gr.Tab("Chinses To English"): |
|
|
gr.HTML(f"<div style='text-align:center;'><h1>RWKV_v7_G1_1.5B_Translate_ctx4096 Chinses -> English</h1></div>") |
|
|
with gr.Row(): |
|
|
with gr.Column(): |
|
|
chinese_input = gr.Textbox( |
|
|
label="中文输入(注意不能有空行)", |
|
|
lines=20, |
|
|
placeholder="请输入中文内容...", |
|
|
value="ROCm是一个开源栈，主要由开源软件组成，旨在用于图形处理单元（GPU）计算。ROCm由一系列驱动程序、开发工具和API组成，这些工具和API允许从低级内核到最终用户应用程序对GPU进行编程。" |
|
|
"使用ROCm，您可以根据您的特定需求定制GPU软件。您可以在一个免费、开源、集成和安全的软件生态系统中开发、协作、测试和部署应用程序。ROCm特别适合GPU加速的高性能计算（HPC）、人工智能（AI）、科学计算和计算机辅助设计（CAD）。" |
|
|
"ROCm由AMD的可移植性图形处理接口（HIP）驱动，这是一个开源的C++ GPU编程环境及其相应的运行时。HIP允许ROCm开发者在不同平台上创建可移植应用程序，通过在从专用游戏GPU到exascale HPC集群的各种平台上部署代码来实现这一目标。" |
|
|
"ROCm支持编程模型，如OpenMP和OpenCL，并包含所有必要的开源软件编译器、调试器和库。ROCm完全集成到机器学习（ML）框架中，如PyTorch和TensorFlow。" |
|
|
) |
|
|
|
|
|
with gr.Column(): |
|
|
english_output = gr.Textbox( |
|
|
label="英文输出", |
|
|
lines=20, |
|
|
placeholder="翻译结果将显示在此处", |
|
|
value="" |
|
|
) |
|
|
|
|
|
with gr.Row(): |
|
|
translate_btn = gr.Button("Translate", variant="primary") |
|
|
clear_btn = gr.Button("Clear", variant="secondary") |
|
|
stop_btn = gr.Button("Stop", variant="stop") |
|
|
|
|
|
with gr.Accordion("Advanced Settings", open=False): |
|
|
token_count = gr.Slider(10, gen_limit, label="Max Tokens", step=10, value=gen_limit) |
|
|
temperature = gr.Slider(0.2, 2.0, label="Temperature", step=0.1, value=1.0) |
|
|
top_p = gr.Slider(0.0, 1.0, label="Top P", step=0.05, value=0) |
|
|
presence_penalty = gr.Slider(0.0, 1.0, label="Presence Penalty", step=0.1, value=0) |
|
|
count_penalty = gr.Slider(0.0, 1.0, label="Count Penalty", step=0.1, value=0) |
|
|
|
|
|
translate_event = translate_btn.click( |
|
|
fn=translate_chinese_to_chinses, |
|
|
inputs=[chinese_input, token_count, temperature, top_p, presence_penalty, count_penalty], |
|
|
outputs=[english_output] |
|
|
) |
|
|
|
|
|
clear_btn.click( |
|
|
fn=lambda: ("", ""), |
|
|
inputs=[], |
|
|
outputs=[chinese_input, english_output] |
|
|
) |
|
|
|
|
|
stop_btn.click( |
|
|
fn=None, |
|
|
inputs=None, |
|
|
outputs=None, |
|
|
cancels=[translate_event] |
|
|
) |
|
|
with gr.Tab("Batch Translate from File"): |
|
|
gr.Markdown("## 上传英文文本文件进行批量翻译") |
|
|
|
|
|
with gr.Row(): |
|
|
file_input = gr.File(label="上传英文文本文件") |
|
|
|
|
|
with gr.Accordion("Advanced Settings (for Batch)", open=False): |
|
|
batch_token_count = gr.Slider(10, gen_limit, label="Max Tokens", step=10, value=gen_limit) |
|
|
batch_temperature = gr.Slider(0.2, 2.0, label="Temperature", step=0.1, value=1.0) |
|
|
batch_top_p = gr.Slider(0.0, 1.0, label="Top P", step=0.05, value=0) |
|
|
batch_presence_penalty = gr.Slider(0.0, 1.0, label="Presence Penalty", step=0.1, value=0) |
|
|
batch_count_penalty = gr.Slider(0.0, 1.0, label="Count Penalty", step=0.1, value=0) |
|
|
|
|
|
batch_translate_btn = gr.Button("Batch Translate", variant="primary") |
|
|
batch_clear_btn = gr.Button("Clear", variant="secondary") |
|
|
stop_btn = gr.Button("Stop", variant="stop") |
|
|
|
|
|
batch_output_text = gr.Textbox(label="翻译进度", lines=10) |
|
|
batch_output_file = gr.File(label="下载翻译结果 (.txt)") |
|
|
|
|
|
def preview_file_content(file): |
|
|
if file is None: |
|
|
return "" |
|
|
try: |
|
|
with open(file.name, "r", encoding="utf-8") as f: |
|
|
content = f.read() |
|
|
except Exception as e: |
|
|
return f"无法读取文件：{str(e)}" |
|
|
|
|
|
entries = [e.strip() for e in content.split('\n\n') if e.strip()] |
|
|
if len(entries) == 1: |
|
|
entries = [e.strip() for e in content.split('\n') if e.strip()] |
|
|
return "\n\n---\n\n".join(entries[:3]) + ("\n\n..." if len(entries) > 3 else "") |
|
|
|
|
|
batch_event = batch_translate_btn.click( |
|
|
fn=batch_translate_english_to_chinese, |
|
|
inputs=[ |
|
|
file_input, |
|
|
batch_token_count, |
|
|
batch_temperature, |
|
|
batch_top_p, |
|
|
batch_presence_penalty, |
|
|
batch_count_penalty |
|
|
], |
|
|
outputs=[batch_output_text, batch_output_file] |
|
|
) |
|
|
|
|
|
batch_clear_btn.click( |
|
|
fn=lambda: ("", None), |
|
|
inputs=[], |
|
|
outputs=[batch_output_text, batch_output_file] |
|
|
) |
|
|
stop_btn.click( |
|
|
fn=None, |
|
|
inputs=None, |
|
|
outputs=None, |
|
|
cancels=[batch_event] |
|
|
) |
|
|
|
|
|
demo.queue(max_size=10, default_concurrency_limit=1) |
|
|
demo.launch(share=False) |
