Coze + Bot API：实现带自我反思的高质量长文翻译 Agent

最终翻译输出的结果如下，还是非常流畅与到位的：

【组装到 Bot】

现在，你可以创建一个工作流模式的单 Agent 的 Bot，并把这个工作流指定给这个 Bot，然后你就拥有一个会自我反思的翻译助手，当然这个助手还只能翻译不能太大的文本块：

长文本翻译处理

在学会了单文本块的反思与翻译以后，就可以在此基础上对长文本通过拆分与循环处理，并适当改造已有的单文本块提示，最终完成对长文本的高质量翻译。整个的处理逻辑是：

1. 对长文本的 tokens 判断，如果数量超过指定的最大限制（比如 1000），则将输入文本拆解成多个相对均匀的小文本块，并确保大小不超过限制，然后对多个文本块进行循环处理。
2. 由于目前 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

主程序

主程序的逻辑非常简单：

1. 读取需要翻译的文件内容
2. 计算 tokens、chunk_size，并进行分割，形成多个文本块
3. 调用 Bot API 循环处理多个文本块
4. 合并结果，输出翻译后的文件

这里的代码中有两种循环处理单个文本块的方式：一种不考虑上下文，另外一种考虑上下文，即每次只处理<translate_this>标签部分，并把其他部分作为上下文输入。

# 示例调用
with open('source.txt', 'r') as file:
    query = file.read().strip()

# tokens 计算、chunk_size 计算、分割 chunks
token_count_val = token_count(query)
chunk_size = calculate_chunk_size(token_count_val, 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 的稳定性和输出质量，建议关注以下几点：

1. 温度设置 (Temperature)：在 Coze 工作流配置中，适当降低 Temperature 值（如 0.3-0.5）可以减少 LLM 的随机性，提高翻译的一致性和指令遵循度。
2. 重试机制：在 Python 脚本中，建议增加针对 API 调用失败的指数退避重试逻辑，防止因网络波动或临时限流导致任务中断。
3. 上下文窗口管理：虽然使用了分块策略，但在长文本处理时，仍需注意总 Token 消耗。如果中间块过多，可以考虑在每 N 个块后重置一次会话 ID，避免历史消息占用过多上下文空间。
4. 格式清洗：LLM 有时会输出 Markdown 标记或多余的空格，建议在接收 call_coze_api 返回结果后，使用正则表达式清理非必要的格式字符，确保最终文本纯净。

这个 Agent 暂时还只能处理纯文本的输入，后续可以结合多模态的处理技术实现更复杂的文档完整翻译功能，比如图表结合的 PDF 文档。在原项目中，作者也对后续的优化提出了一些思路，比如测试更多的 LLM、增加对特定词汇表（针对特定国家或行业）的支持等，我们期待看到更多有益的尝试与测试结果。

通过上述步骤，我们成功在 Coze 平台上复现了基于自我反思机制的长文本翻译方案，并结合 Bot API 实现了自动化流程，为实际业务中的多语言内容处理提供了可行的技术路径。