Python调用AI API分段处理长文本完整流程与重试机制

脚本专家

在实际项目中，AI API 往往不擅长直接处理超长文本。超过模型上下文长度、请求体过大、响应缓慢、结果不完整，都是常见问题。更合理的做法是：读取文件 → 清理内容 → 分段 → 分别处理 → 汇总结果。下面用一个完整的 Python 流程说明如何实现。

2. from pathlib import Path
4. file_path = Path("example.txt")
5. text = file_path.read_text(encoding="utf-8")
6. print(text[:500])

复制代码

读取文件时不要默认所有文件都是 UTF-8 编码。如果遇到编码异常，应根据实际情况处理，比如捕获 UnicodeDecodeError 后尝试其他编码。

分段前可以先做一次清理，减少重复空白、无效分隔符和明显无关内容。下面这个函数会去掉每行首尾空白，并过滤掉空行：

2. def clean_text(text: str) -> str:
3.     lines = [line.strip() for line in text.splitlines()]
4.     lines = [line for line in lines if line]
5.     return "\n".join(lines)

复制代码

如果是日志文件，还可以根据时间、级别或关键字筛选内容，把真正需要分析的部分保留下来。

最简单、直接的分段方式是按固定字符数切分。例如每 4000 个字符一段：

2. def split_text(text: str, chunk_size: int = 4000) -> list[str]:
3.     return [
4.         text[i:i + chunk_size]
5.         for i in range(0, len(text), chunk_size)
6.     ]

复制代码

这种方式实现简单，但容易从句子中间或代码块中间截断，适合初步测试、结构简单的纯文本、以及边界要求不高的任务。

如果希望保留文章结构，则优先按段落切分。Markdown 文档、说明文章通常用空行分隔段落，可以按 `\n\n` 分组，再合并成不超过最大长度的块：

2. def split_by_paragraph(text: str, max_chars: int = 4000) -> list[str]:
3.     paragraphs = text.split("\n\n")
4.     chunks = []
5.     current = []
6.     current_length = 0
8.     for paragraph in paragraphs:
9.         if current and current_length + len(paragraph) > max_chars:
10.             chunks.append("\n\n".join(current))
11.             current = []
12.             current_length = 0
13.         current.append(paragraph)
14.         current_length += len(paragraph)
16.     if current:
17.         chunks.append("\n\n".join(current))
19.     return chunks

复制代码

分段处理后，建议给每一段加编号。这样既能定位失败片段，也能记录处理进度，方便后续单独重试和汇总时保持顺序：

2. chunks = split_by_paragraph(text)
3. for index, chunk in enumerate(chunks, start=1):
4.     print(f"正在处理第 {index}/{len(chunks)} 段")

复制代码

调用 AI API 时，使用 OpenAI SDK 的 chat.completions 接口即可。以下示例展示了如何逐段提交并获取总结结果：

2. from openai import OpenAI
4. client = OpenAI(
5.     api_key="your-api-key",
6.     base_url="https://your-api-domain.com/v1",
7. )
9. def summarize_chunk(chunk: str) -> str:
10.     response = client.chat.completions.create(
11.         model="your-model-name",
12.         messages=[
13.             {
14.                 "role": "system",
15.                 "content": "你负责提取文本中的关键事实，保持简洁，不要添加原文没有的信息。",
16.             },
17.             {
18.                 "role": "user",
19.                 "content": f"请总结下面这段内容：\n\n{chunk}",
20.             },
21.         ],
22.     )
23.     return response.choices[0].message.content

复制代码

实际处理时逐段调用并保存结果。最后将每段的结果拼接起来，形成完整输出：

2. summaries = []
3. for index, chunk in enumerate(chunks, start=1):
4.     summary = summarize_chunk(chunk)
5.     summaries.append(f"第 {index} 段：{summary}")
7. final_text = "\n\n".join(summaries)
8. print(final_text)

复制代码

如果分段数量很多，最终汇总结果也可能很长。此时可以对汇总结果再做一次二次总结，或者只保留关键字段，避免再次超出上下文限制。

在长文本处理中有几个需要特别注意的点：

1. 分段不要过大。要给系统提示词和模型输出留出空间。
2. 分段不要过小。否则上下文不足，请求次数增多，还可能丢失前后文关系。
3. 保留文档结构。标题、章节、代码块尽量不要随意截断。
4. 失败后支持单段重试。不要因为一个片段失败就丢弃全部结果。
5. 注意敏感信息。上传前应检查密钥、手机号、邮箱和其他隐私内容。

这种分段处理方式适合 Markdown 文档总结、项目日志分析、代码文件说明、会议记录整理、知识库内容处理、批量文本分类等场景。如果任务需要理解全文关系，可以先分段提取信息，再做统一汇总。

长文本处理的核心不是简单截断，而是建立一个可恢复的流程：先读取和清理，再根据内容分段，逐段处理并记录编号，失败时只重试单个分段，最后汇总结果。对于 Python AI 项目来说，这种方式比一次性发送整篇文档更容易控制请求规模，也方便后续扩展缓存、限流和日志功能。

热心网友2

楼主这流程很实用，正好最近也在折腾类似的长文本处理。固定字符切分确实简单但容易切断句子，按段落分组就灵活多了。另外编码那点提醒得很对，我经常遇到GBK或BOM的文件，不处理直接报错就很烦。 想补充一点：分段后除了编号，我还会把每段的原始内容连同结果存成JSON或临时文件，这样万一某段失败或者结果不理想，可以只重新跑那一段，不用从头再来。重试逻辑的话，我一般会加上指数退避，尤其是API偶尔限流的时候，简单重试几次往往不够。 还有个小问题想请教：如果文档里有表格或代码块，按`\n\n`切会不会还是容易从代码中间断开？楼主有没有遇到这种情况，怎么处理比较好？

热心网友2

楼主的流程很完整，从文件读取、清洗、分段到逐段调用API和汇总，思路清晰，适合实际项目落地。尤其是按段落合并而不是硬切字符，能较好保留语义结构。我有一点小建议：正文标题提到“重试机制”，示例里主要处理了失败后单独重试的思路，如果能在代码层面加一个简单的指数退避重试（比如捕获异常后等待几秒再试，最多重试3次），并把失败段落的索引记录下来，最后统一报告，会更贴合生产环境的需求。另外，对于分段后可能存在的上下文关联问题，也可以考虑让每个chunk携带前一段的结尾或摘要，帮助模型理解前后文。总之，这套流程作为基础框架很实用，感谢分享！

热心网友2

这个流程写得很实用，尤其是按段落切分和失败后单段重试的思路，实际跑长文本处理时非常关键。我补充一个小点：分段时如果遇到代码块或特殊格式，可以考虑先做占位符替换，处理完再还原，避免切到代码中间导致语法碎片。另外，重试机制里可以加上简单的指数退避，并记录每段的状态，方便断点续跑。整体很清晰，适合直接拿去改改就能用。