很多人熟悉的吴恩达老师前段时间发布了一个开源项目translation-agent，提出了一种利用LLM进行自我反思并完善的自动化长文翻译智能体，并给出了简单的原型代码，在国内也看到了开源RAG平台FastGPT对此流程的实现（一键生成高质量长文翻译，吴恩达新方法颠覆传统！）。于是我们尝试在Coze平台上类似的实现一个翻译智能体，本文将简单分享这个过程。
Translation-agent工作流程
我们都知道借助LLM与提示词可以很简单的让模型将一段文本从一种语言翻译成另外一种语言，而且一般质量还不错。但有时也会存在一些问题：
翻译仍然会存在一些准确性、流畅度、文字风格等方面的不足
涉及到长文本时，比如超出模型的最大输出tokens，处理较麻烦
Translation-agent采用的工作流程大致如下：
对短文本块（不超过最大允许的tokens数量）的翻译过程为：
通过提示词让LLM将文本从一种语言翻译到另外一种语言
让LLM对翻译的结果进行反思，提出各方面的改进建议
再次让LLM根据生成的改进建议对初始的翻译进行完善并输出
而对于长文本来说，只需要通过相应算法将长文本分割成相对均匀的小“文本块”，再对多个小文本块进行循环处理即可（循环的单文本块处理方法并不完全等同于直接的短文本块的一次性处理，后面会看到）。
接下来我们在Coze平台上来模拟实现这个翻译Agent，为了简化处理，这里直接把翻译语言固定成从英文翻译为中文，你当然可以根据需要把这两种语言做参数化，在使用时格式化到提示词中即可。
单文本块的翻译与反思
我们首先来构建单文本块的翻译与反思工作流。在Coze中创建一个新的工作流，然后开始配置：
【初始翻译】
创建大模型节点，并作如下配置，输入一个源文本，输出一个翻译文本，注意这里的提示词是比较简单的：
【反思】
将初始翻译的结果输入到反思节点，并配置反思的大模型提示。在原项目中，除了源语言和目标语言之外，还有一个国家的参数(country)，这个参数用来生成更符合该国语言风格的翻译，此处我们也暂时忽略。输入参数是源文本与初步翻译的结果，输出参数是反思的建议：
【改进翻译】
通过模型的反思得到建议后，通过再次输入LLM要求根据反思建议进行改进，输出更加完善的翻译结果即可，这里需输入源文本、初始翻译结果以及改进建议三个参数：
【试运行】
我们选择一段Llama3的介绍英文来测试，使用国内的智谱GLM-4模型：
这里观察到在运行期间LLM给首次翻译提出的改进建议，客观的说，有的建议可能并非完全必要，但整体而言还是在上下文准确性、流畅度、减少重复上有相当的意义：
最终翻译输出的结果如下，还是非常流畅与到位的：
【组装到Bot】
现在，你可以创建一个工作流模式的单Agent的Bot，并把这个工作流指定给这个Bot，然后你就拥有一个会自我反思的翻译助手，当然这个助手还只能翻译不能太大的文本块：
长文本翻译处理
在学会了单文本块的反思与翻译以后，就可以在此基础上对长文本通过拆分与循环处理，并适当改造已有的单文本块提示，最终完成对长文本的高质量翻译。整个的处理逻辑是：
对长文本的tokens判断，如果数量超过指定的最大限制（比如1000），则将输入文本拆解成多个相对均匀的小文本块，并确保大小不超过限制，然后对多个文本块进行循环处理。很显然，多文本块的处理应该是可以兼容单个文本块的处理（原项目中做了区分）。
由于目前Coze的工作流编排暂时还不支持循环，所以无法直接编排循环处理的工作流，这里我们借助Coze的Bot API，将此过程编码实现，通过循环调用前面的单文本块翻译的Bot来实现。
对长文本下的单文本块处理并不简单等同于直接的短文本块处理，主要原因是为了给长文本下的单个文本块增加足够的上下文，使得翻译的更加准确且一致。主要区别就是在输入完整的源文本（source_text标签部分）中标识了本次需要翻译的文本块（translate_this）标签，并提示LLM只对translate_this部分做处理，这样既提供了足够的上下文，又只翻译了本次处理的文本块。以初始化翻译为例，你需要对上面创建的单文本块bot提示做如下修改：
你是一个把英文文本转化成简体中文的翻译助手。
源文本如下，以XML标签<SOURCE_TEXT>和</SOURCE_TEXT>分隔。
你只需要翻译源文本中以<TRANSLATE_THIS>和</TRANSLATE_THIS>分隔的部分；您可以将其余部分作为上下文，但不要翻译其他文本。
不要输出任何除指定部分的翻译之外的内容。不要有多余解释。不要重复原文。
--------
英文文本：{{source_text}}
中文文本：
再次重申，仅翻译<TRANSLATE_THIS>和</TRANSLATE_THIS>之间的文本。
反思与完善的环节处理方式类似。通过这样处理后的Bot既可以支持直接的短文本块处理（给全部文本添加<TRANSLATE_THIS>标签）；也可以支持长文本下的单个文本块循环处理（给每次需要处理的文本块增加<TRANSLATE_THIS>标签）
【Bot API准备】
现在的Coze平台支持API调用，通过API可以访问云端创建的Bot与工作流、对话、管理知识库等，大大扩展了Coze的应用场景。这里我们先把上面创建的单文本块的Bot发布成API，首先需要申请API调用的令牌：
然后将自己的Bot进行发布，发布时选择“Bot as API”的模式，并等待审核通过即可，现在你就可以通过API调用你自己的Bot免费用Coze平台的大模型：
【调用Bot】
在客户端创建一个简单的函数来调用自己的Bot API，参考Coze的文档实现即可，由于我们做了简化，只需要传入一个需要翻译的源文本即可。注意由于Coze采用的异步API，所以需要多次调用：
import requests
import time
import json
import re
import tiktoken
from langchain_text_splitters import TokenTextSplitter
ACCESS_TOKEN = "***"
BOT_ID = "***"
USER_ID = "***"
def call_coze_api(query):
url = 'https://api.coze.cn/v3/chat'
headers = {
'Authorization': f'Bearer {ACCESS_TOKEN}',
'Content-Type': 'application/json'
}
payload = {
"bot_id": BOT_ID,
"user_id": USER_ID,
"stream": False,
"auto_save_history": True,
"additional_messages": [
{
"role": "user",
"content": query,
"content_type": "text"
}
]
}
response = requests.post(url, headers=headers, json=payload)
if response.status_code == 200:
response_data = response.json().get('data')
id = response_data['id']
conversation_id = response_data['conversation_id']
retrieve_url = f'https://api.coze.cn/v3/chat/retrieve?chat_id={id}&conversation_id={conversation_id}'
while True:
print('Starting retrieve the response...')
retrieve_response = requests.get(retrieve_url, headers=headers)
if retrieve_response.status_code == 200:
retrieve_data = retrieve_response.json().get('data')
if retrieve_data['status'] == 'completed':
result_url = f'https://api.coze.cn/v3/chat/message/list?chat_id={id}&conversation_id={conversation_id}'
result_response = requests.get(result_url, headers=headers)
if result_response.status_code == 200:
result_data = result_response.json().get('data')
result_dict = next((item for item in result_data if item['type'] == 'answer'), None)
return result_dict.get('content')
else:
print(f"Error: {result_response.status_code}, {result_response.text}")
return {"error": result_response.status_code, "message": result_response.text}
else:
print(f"Error: {retrieve_response.status_code}, {retrieve_response.text}")
return {"error": retrieve_response.status_code, "message": retrieve_response.text}
time.sleep(10) # Add a delay to avoid excessive API calls
else:
print(f"Error: {response.status_code}, {response.text}")
return {"error": response.status_code, "message": response.text}【几个辅助函数】
实现tokens计算、chunk_size计算，以及chunks分割的三个辅助函数。基本思想是：
tokens计算：借助tiktoken库实现
chunk_size计算：根据最大tokens限制来计算每个chunk的size
chunks分割：输入text与计算出的chunk_size，借助langchain做分割，并保证句子完整性
def token_count(text):
encoding = tiktoken.get_encoding("gpt2")
tokens = encoding.encode(text)
token_count = len(tokens)
return token_count
def calculate_chunk_size(token_count: int, token_limit: int) -> int:
if token_count <= token_limit:
return token_count
num_chunks = (token_count + token_limit - 1) // token_limit
chunk_size = token_count // num_chunks
remaining_tokens = token_count % token_limit
if remaining_tokens > 0:
chunk_size += remaining_tokens // num_chunks
return chunk_size
def get_text_chunks(text,chunk_size) :
splitter = TokenTextSplitter(
chunk_size=int(chunk_size),
chunk_overlap=0,
)
initial_chunks = splitter.split_text(text)
# 定义不同语言的句子分隔符
sentence_delimiters = re.compile(r'[。！？.!?]')
# 进一步处理每个初步分割块
output = []
current_chunk = initial_chunks[0] if initial_chunks else ''
for i in range(1, len(initial_chunks)):
sentences = sentence_delimiters.split(initial_chunks[i])
if sentences:
current_chunk += sentences[0] # 拼接第一个句子到当前块
output.append(current_chunk.strip()) # 将当前块加入输出数组
current_chunk = ''.join(sentences[1:]) # 剩余的句子作为新的当前块
# 将最后一个块加入输出数组
if current_chunk.strip():
output.append(current_chunk.strip())
return output【主程序】
主程序的逻辑非常简单：
读取需要翻译的文件内容
计算tokens、chunk_size，并进行分割，形成多个文本块
调用Bot API循环处理多个文本块
合并结果，输出翻译后的文件
这里的代码中有两种循环处理单个文本块的方式：一种不考虑上下文，另外一种考虑上下文，即每次只处理<translate_this>标签部分，并把其他部分作为上下文输入。
# 示例调用
with open('source.txt', 'r') as file:
query = file.read().strip()
#tokens计算、chunk_size计算、分割chunks
token_count = token_count(query)
chunk_size = calculate_chunk_size(token_count, 800)
chunks = get_text_chunks(query,chunk_size)
results = []
for index in range(len(chunks)):
tagged_text = chunks[index]
"""
如果需要增加上下文，采用如下方式；注意需要提示词配合
tagged_text = (
"".join(chunks[0:index])
+ "<TRANSLATE_THIS>"
+ chunks[index]
+ "</TRANSLATE_THIS>"
+ "".join(chunks[index + 1 :])
)
"""
result = call_coze_api(tagged_text)
result_dict =json.loads(result)
results.append(result_dict.get('output'))
output = " ".join(results)
with open('target.txt', 'w') as file:
file.write(output)这样就实现了一个带有自我反思与提升功能的长文本翻译Agent。整个过程逻辑还是比较清楚的，在实际测试中遇到的主要问题是LLM遵循性的问题，偶尔会出现不够遵循指令，比如在输出中出现多余的解释说明，特别是在完善环节以及多文本块处理时，偶尔会有错误的输出。
这个Agent暂时还只能处理纯文本的输入，后续可以结合多模态的处理技术实现更复杂的文档完整翻译功能，比如图表结合的PDF文档。在原项目中，作者也对后续的优化提出了一些思路，比如测试更多的LLM、增加对特定词汇表（针对特定国家或行业）的支持等，我们期待看到更多有益的尝试与测试结果。
