Python调用AI API分段处理长文本完整流程与重试机制
在实际项目中,AI API 往往不擅长直接处理超长文本。超过模型上下文长度、请求体过大、响应缓慢、结果不完整,都是常见问题。更合理的做法是:读取文件 → 清理内容 → 分段 → 分别处理 → 汇总结果。下面用一个完整的 Python 流程说明如何实现。from pathlib import Path
file_path = Path("example.txt")
text = file_path.read_text(encoding="utf-8")
print(text[:500])
读取文件时不要默认所有文件都是 UTF-8 编码。如果遇到编码异常,应根据实际情况处理,比如捕获 UnicodeDecodeError 后尝试其他编码。
分段前可以先做一次清理,减少重复空白、无效分隔符和明显无关内容。下面这个函数会去掉每行首尾空白,并过滤掉空行:
def clean_text(text: str) -> str:
lines =
lines =
return "\n".join(lines)
如果是日志文件,还可以根据时间、级别或关键字筛选内容,把真正需要分析的部分保留下来。
最简单、直接的分段方式是按固定字符数切分。例如每 4000 个字符一段:
def split_text(text: str, chunk_size: int = 4000) -> list:
return [
text
for i in range(0, len(text), chunk_size)
]
这种方式实现简单,但容易从句子中间或代码块中间截断,适合初步测试、结构简单的纯文本、以及边界要求不高的任务。
如果希望保留文章结构,则优先按段落切分。Markdown 文档、说明文章通常用空行分隔段落,可以按 `\n\n` 分组,再合并成不超过最大长度的块:
def split_by_paragraph(text: str, max_chars: int = 4000) -> list:
paragraphs = text.split("\n\n")
chunks = []
current = []
current_length = 0
for paragraph in paragraphs:
if current and current_length + len(paragraph) > max_chars:
chunks.append("\n\n".join(current))
current = []
current_length = 0
current.append(paragraph)
current_length += len(paragraph)
if current:
chunks.append("\n\n".join(current))
return chunks
分段处理后,建议给每一段加编号。这样既能定位失败片段,也能记录处理进度,方便后续单独重试和汇总时保持顺序:
chunks = split_by_paragraph(text)
for index, chunk in enumerate(chunks, start=1):
print(f"正在处理第 {index}/{len(chunks)} 段")
调用 AI API 时,使用 OpenAI SDK 的 chat.completions 接口即可。以下示例展示了如何逐段提交并获取总结结果:
from openai import OpenAI
client = OpenAI(
api_key="your-api-key",
base_url="https://your-api-domain.com/v1",
)
def summarize_chunk(chunk: str) -> str:
response = client.chat.completions.create(
model="your-model-name",
messages=[
{
"role": "system",
"content": "你负责提取文本中的关键事实,保持简洁,不要添加原文没有的信息。",
},
{
"role": "user",
"content": f"请总结下面这段内容:\n\n{chunk}",
},
],
)
return response.choices.message.content
实际处理时逐段调用并保存结果。最后将每段的结果拼接起来,形成完整输出:
summaries = []
for index, chunk in enumerate(chunks, start=1):
summary = summarize_chunk(chunk)
summaries.append(f"第 {index} 段:{summary}")
final_text = "\n\n".join(summaries)
print(final_text)
如果分段数量很多,最终汇总结果也可能很长。此时可以对汇总结果再做一次二次总结,或者只保留关键字段,避免再次超出上下文限制。
在长文本处理中有几个需要特别注意的点:
1. 分段不要过大。要给系统提示词和模型输出留出空间。
2. 分段不要过小。否则上下文不足,请求次数增多,还可能丢失前后文关系。
3. 保留文档结构。标题、章节、代码块尽量不要随意截断。
4. 失败后支持单段重试。不要因为一个片段失败就丢弃全部结果。
5. 注意敏感信息。上传前应检查密钥、手机号、邮箱和其他隐私内容。
这种分段处理方式适合 Markdown 文档总结、项目日志分析、代码文件说明、会议记录整理、知识库内容处理、批量文本分类等场景。如果任务需要理解全文关系,可以先分段提取信息,再做统一汇总。
长文本处理的核心不是简单截断,而是建立一个可恢复的流程:先读取和清理,再根据内容分段,逐段处理并记录编号,失败时只重试单个分段,最后汇总结果。对于 Python AI 项目来说,这种方式比一次性发送整篇文档更容易控制请求规模,也方便后续扩展缓存、限流和日志功能。
Re: Python调用AI API分段处理长文本完整流程与重试机制
楼主这流程很实用,正好最近也在折腾类似的长文本处理。固定字符切分确实简单但容易切断句子,按段落分组就灵活多了。另外编码那点提醒得很对,我经常遇到GBK或BOM的文件,不处理直接报错就很烦。 想补充一点:分段后除了编号,我还会把每段的原始内容连同结果存成JSON或临时文件,这样万一某段失败或者结果不理想,可以只重新跑那一段,不用从头再来。重试逻辑的话,我一般会加上指数退避,尤其是API偶尔限流的时候,简单重试几次往往不够。 还有个小问题想请教:如果文档里有表格或代码块,按`\n\n`切会不会还是容易从代码中间断开?楼主有没有遇到这种情况,怎么处理比较好?Re: Python调用AI API分段处理长文本完整流程与重试机制
楼主的流程很完整,从文件读取、清洗、分段到逐段调用API和汇总,思路清晰,适合实际项目落地。尤其是按段落合并而不是硬切字符,能较好保留语义结构。我有一点小建议:正文标题提到“重试机制”,示例里主要处理了失败后单独重试的思路,如果能在代码层面加一个简单的指数退避重试(比如捕获异常后等待几秒再试,最多重试3次),并把失败段落的索引记录下来,最后统一报告,会更贴合生产环境的需求。另外,对于分段后可能存在的上下文关联问题,也可以考虑让每个chunk携带前一段的结尾或摘要,帮助模型理解前后文。总之,这套流程作为基础框架很实用,感谢分享!Re: Python调用AI API分段处理长文本完整流程与重试机制
这个流程写得很实用,尤其是按段落切分和失败后单段重试的思路,实际跑长文本处理时非常关键。我补充一个小点:分段时如果遇到代码块或特殊格式,可以考虑先做占位符替换,处理完再还原,避免切到代码中间导致语法碎片。另外,重试机制里可以加上简单的指数退避,并记录每段的状态,方便断点续跑。整体很清晰,适合直接拿去改改就能用。
页:
[1]