在实际的大模型应用中，我们经常需要处理录音转写、会议记录、客服对话、访谈内容等长文本。
最直接的方式，是将完整文本一次性提交给大模型，让模型生成摘要。但随着文本长度增加，这种方案容易出现上下文超限、信息遗漏、调用成本增加以及输出不稳定等问题。
本文介绍一种通用的长文本处理方案：先对原始文本进行分段，再分别提取结构化信息，最后进行合并、去重和摘要生成。
一、直接处理长文本存在的问题
假设一段录音转写文本包含几万字，如果直接提交给大模型，通常会遇到以下问题。
- 上下文长度限制
每个大模型都有上下文窗口限制，输入文本、系统提示词、用户提示词和模型输出都会占用Token。
当文本超过限制时，可能出现：
请求直接失败
文本被截断
模型输出不完整
前后文信息无法同时参与分析
- 重要信息被忽略
即使文本没有超过上下文限制，模型也不一定能稳定识别所有信息。
当一段重要内容只在长文本中出现一次时，可能被大量重复对话和无效内容淹没。
- 调用成本较高
将完整文本多次提交给模型，会重复消耗大量Token。如果还需要分别生成摘要、标签和分析结果，整体调用成本会明显增加。
- 输出结果难以校验
如果只生成一段自然语言摘要，很难判断：
哪些信息没有被提取
某个结论来自哪段原文
模型是否进行了无依据推断
多次提及的信息是否产生冲突
因此，长文本处理不能只依赖一次大模型调用。
二、整体解决方案
可以将长文本处理拆分为四个阶段：
原始长文本
↓
文本分段
↓
分段结构化提取
↓
结果合并与去重
↓
最终摘要生成
这种方案本质上类似MapReduce：
Map阶段：分别处理每一个文本片段
Reduce阶段：合并所有片段的提取结果
最终摘要不再直接基于原始长文本生成，而是基于已经提取并整理好的结构化信息生成。
三、第一步：文本分段
- 固定长度分段
最简单的方式是按照字符数或Token数量切分。
def split_text(text, chunk_size=4000):
chunks = []
for start in range(0, len(text), chunk_size):
chunks.append(text[start:start + chunk_size])
return chunks
这种方式实现简单，但可能在一句话或一个完整话题中间进行切分。
例如：
片段1：客户目前最担心的是装修过程中
片段2：不断增加合同之外的费用。
两个片段分别处理时，可能无法提取完整信息。
- 按句子分段
更合理的方式是先按照句号、问号、换行等边界拆分，再将多个句子组合成符合长度要求的片段。
import re
def split_by_sentence(text, max_length=4000):
sentences = re.split(r'(?<=[。！？\n])', text)
chunks = []
current_chunk = ""
for sentence in sentences:
if len(current_chunk) + len(sentence) <= max_length:
current_chunk += sentence
else:
if current_chunk:
chunks.append(current_chunk)
current_chunk = sentence
if current_chunk:
chunks.append(current_chunk)
return chunks
这种方式能够尽量保持句子的完整性。
- 设置上下文重叠
即使按照句子分段，跨片段信息仍然可能被拆开。可以在相邻片段之间保留一定长度的重叠内容。
片段1：第1句～第50句
片段2：第46句～第95句
片段3：第91句～第140句
重叠部分可以帮助模型理解片段边界附近的上下文。
def split_with_overlap(text, chunk_size=4000, overlap=500):
if overlap >= chunk_size:
raise ValueError("overlap必须小于chunk_size")
chunks = []
start = 0
while start < len(text):
end = min(start + chunk_size, len(text))
chunks.append(text[start:end])
if end == len(text):
break
start = end - overlap
return chunks
需要注意，重叠内容可能被重复提取，因此后续必须进行去重。
四、第二步：分段结构化提取
文本完成分段后，对每个片段分别调用大模型。
相比直接要求模型生成摘要，建议先要求模型提取结构化事实。
例如，定义统一的输出格式：
{
"facts": [
{
"type": "时间",
"value": "计划两个月内启动",
"evidence": "我们想最近两个月把这个事情定下来",
"speaker": "客户"
}
],
"questions": [],
"risks": [],
"pendingItems": []
}
其中：
type：信息类型
value：整理后的信息
evidence：对应的原始文本
speaker：信息来源
segmentId：所在片段编号
建议保留evidence字段。这样既方便检查模型结果，也可以在页面上实现点击分析结论跳转原文。
分段处理示例：
def extract_chunk(llm_client, chunk, segment_id):
prompt = f"""
你是一名文本信息提取助手。
请从下面的文本中提取明确出现的事实、问题、风险和待确认事项。
要求：
- 只能使用文本中出现的信息。
- 不得补充原文没有的内容。
- 每条结果必须保留原文依据。
- 无法确定的信息放入pendingItems。
- 严格按照JSON格式返回。
片段编号：{segment_id}
文本：
{chunk}
"""
return llm_client.generate(prompt)
五、第三步：并行处理文本片段
各个文本片段之间通常没有调用依赖，因此可以并行请求模型，减少总处理时间。
from concurrent.futures import ThreadPoolExecutor, as_completed
def process_chunks(llm_client, chunks, max_workers=4):
results = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {
executor.submit(
extract_chunk,
llm_client,
chunk,
index
): index
for index, chunk in enumerate(chunks)
}
for future in as_completed(futures):
segment_id = futures[future]
try:
result = future.result()
results.append({
"segmentId": segment_id,
"status": "success",
"result": result
})
except Exception as error:
results.append({
"segmentId": segment_id,
"status": "failed",
"error": str(error)
})
return sorted(results, key=lambda item: item["segmentId"])
并发数量不能无限增加，需要结合以下因素设置：
模型接口的并发限制
每分钟请求数限制
Token速率限制
单次请求平均耗时
服务端资源情况
如果模型服务限制每秒请求数，还需要配合限流器使用。
六、第四步：结果合并与去重
所有片段处理完成后，会得到多组结构化结果。由于片段之间存在重叠，相同信息可能被提取多次。
例如：
[
{
"type": "预算",
"value": "预算约15万元"
},
{
"type": "预算",
"value": "整体预算大概15万"
}
]
虽然文字不同，但表达的是同一个事实。
- 完全匹配去重
对完全相同的内容，可以直接使用集合去重。
def exact_deduplicate(items):
unique_items = []
seen = set()
for item in items:
key = (
item.get("type", "").strip(),
item.get("value", "").strip()
)
if key not in seen:
seen.add(key)
unique_items.append(item)
return unique_items
- 规则归一化
对于金额、日期、手机号等结构化字段，可以先标准化，再进行比较。
例如：
十五万元
15万元
150000元
可以统一转换为：
{
"amount": 150000,
"currency": "CNY"
}
- 语义相似度去重
自然语言内容可以通过向量相似度判断是否重复。
基本流程如下：
提取结果
↓
生成文本向量
↓
计算余弦相似度
↓
相似度超过阈值
↓
合并为同一条信息
余弦相似度计算公式为：
similarity(A,B)=
∥A∥∥B∥
A⋅B
Python示例：
import numpy as np
def cosine_similarity(vector_a, vector_b):
vector_a = np.array(vector_a)
vector_b = np.array(vector_b)
denominator = (
np.linalg.norm(vector_a) *
np.linalg.norm(vector_b)
)
if denominator == 0:
return 0.0
return np.dot(vector_a, vector_b) / denominator
相似度阈值需要根据数据进行调整。阈值过低，可能把不同信息错误合并；阈值过高，则可能无法识别表达方式不同的重复内容。
七、冲突信息处理
长文本中可能出现前后不一致的信息。
例如：
前半段：预算大概15万元。
后半段：如果包含所有柜子，18万元以内可以接受。
这两条信息不一定互相矛盾，可能分别表示基础预算和预算上限。
但下面的情况可能存在真实冲突：
片段1：计划8月份开始装修。
片段5：装修暂时不着急，可能要等到明年。
遇到冲突时，不建议直接让系统随机选择一条。可以保留所有候选信息，并附带来源和时间：
{
"field": "计划装修时间",
"status": "conflict",
"candidates": [
{
"value": "8月份",
"segmentId": 1,
"evidence": "计划8月份开始装修"
},
{
"value": "可能明年",
"segmentId": 5,
"evidence": "可能要等到明年"
}
]
}
如果对话本身存在时间顺序，可以采用“后续明确表达优先”的规则，但仍应保留历史值，避免丢失信息变化。
八、第五步：生成最终摘要
完成提取、去重和冲突处理后，再将整理好的结构化数据提交给大模型生成最终摘要。
此时输入内容已经从几万字的原始文本，压缩为少量高价值信息。
import json
def generate_summary(llm_client, merged_data):
prompt = f"""
请根据下面的结构化信息生成摘要。
要求：
- 不得添加输入中不存在的信息。
- 优先展示已确认的事实。
- 存在冲突的信息需要明确说明。
- 不要按照原文顺序进行流水账式复述。
- 摘要控制在300字以内。
结构化信息：
{json.dumps(merged_data, ensure_ascii=False)}
"""
return llm_client.generate(prompt)
与直接对原文进行摘要相比，这种方式具有以下优势：
输入Token明显减少
摘要中的事实更容易控制
每条结论可以追溯到原文
冲突信息不会被自动掩盖
方便对关键字段进行单独校验
九、失败重试设计
大模型接口可能出现超时、限流、网络异常或输出格式错误，因此需要设置重试机制。
import time
def call_with_retry(func, max_retries=3):
for attempt in range(max_retries):
try:
return func()
except Exception:
if attempt == max_retries - 1:
raise
wait_seconds = 2 ** attempt
time.sleep(wait_seconds)
这里使用指数退避：
第一次失败：等待1秒
第二次失败：等待2秒
第三次失败：等待4秒
并不是所有错误都应该重试：
错误类型 是否重试
网络超时 是
服务端5xx错误 是
请求频率限制 是，延迟后重试
API密钥错误 否
参数格式错误 否
输入超过上下文限制 否，应重新分段
模型返回非法JSON 可以修复或重新生成
分段任务还需要独立记录状态。某个片段失败时，只重试失败片段，不需要重新处理全部文本。
十、结构化输出校验
大模型即使被要求返回JSON，也可能输出Markdown代码块、说明文字或不完整字段。
可以使用数据模型对结果进行校验：
from typing import List
from pydantic import BaseModel, Field
class Fact(BaseModel):
type: str
value: str
evidence: str
speaker: str = ""
segment_id: int = Field(alias="segmentId")
class ExtractResult(BaseModel):
facts: List[Fact] = []
questions: List[str] = []
risks: List[str] = []
pending_items: List[str] = Field(
default=[],
alias="pendingItems"
)
解析模型输出：
def validate_result(data):
return ExtractResult.model_validate(data)
如果校验失败，可以：
尝试清理Markdown代码块；
对JSON进行基础修复；
将错误信息反馈给模型重新生成；
多次失败后将片段标记为异常。
十一、完整流程示例
def process_long_text(llm_client, text):
# 1. 文本分段
chunks = split_with_overlap(
text=text,
chunk_size=4000,
overlap=500
)
# 2. 分段并行提取
chunk_results = process_chunks(
llm_client=llm_client,
chunks=chunks,
max_workers=4
)
# 3. 收集成功结果
extracted_items = []
for result in chunk_results:
if result["status"] == "success":
extracted_items.extend(
result["result"].get("facts", [])
)
# 4. 合并与去重
unique_items = exact_deduplicate(extracted_items)
# 5. 生成最终摘要
summary = generate_summary(
llm_client=llm_client,
merged_data={"facts": unique_items}
)
return {
"chunkCount": len(chunks),
"facts": unique_items,
"summary": summary
}
实际项目中，还可以继续加入：
Token数量计算
调用成本统计
分段结果缓存
任务状态持久化
失败片段补偿
语义向量去重
原文证据定位
模型输出质量评分
十二、总结
大模型处理长文本时，直接将完整内容提交给模型虽然实现简单，但在准确性、成本、稳定性和可追溯性方面存在一定限制。
一种更加稳定的处理方式是：
合理分段
→ 分段提取
→ 并行调用
→ 结果去重
→ 冲突处理
→ 结构校验
→ 最终摘要
这套方案不只适用于录音转写，还可以用于：
会议纪要生成
客服对话分析
访谈资料整理
合同信息提取
长篇文档摘要
多份报告综合分析
长文本处理的重点，不是单纯扩大模型的输入长度，而是通过合理的任务拆分，让模型每次只处理一个清晰、可验证的小任务，再将多个局部结果组合成稳定的最终结果。
