你有没有遇到过这样的情况：和 AI 聊了半小时，它突然忘了 20 分钟前说的话？或者让 AI 分析一份 100 页的技术文档，它只能处理前几页？这就是长文本处理的挑战。
为什么长文本处理这么难？
核心问题：上下文窗口限制
所有 LLM 都有"记忆容量"限制：
| 模型 | 上下文窗口 | 相当于 |
|---|---|---|
| GPT-3.5 | 4K tokens | ~3000 字 |
| GPT-4 | 8K-128K tokens | ~6000-96000 字 |
| Claude 3.5 Sonnet | 200K tokens | ~150000 字 |
| Gemini 1.5 Pro | 1M tokens | ~750000 字 |
实际问题：
- 100页PDF ≈ 50K tokens
- 1小时会议录音转文字 ≈ 20K tokens
- 一个大型项目代码库 ≈ 数百万 tokens
传统解决方案的缺陷
方案1：分段处理
# ❌ 问题：丢失全局上下文
chunks = split_text(long_text, chunk_size=4000)
for chunk in chunks:
result = process(chunk) # 每段独立处理方案2：摘要压缩
# ❌ 问题：丢失细节
summary = summarize(long_text) # 只保留主要信息方案3：只保留最近N轮对话
# ❌ 问题：忘记早期重要信息
recent_messages = messages[-10:] # 只保留最后10轮现代解决方案
策略1：分层记忆架构
像人脑一样，分为短期记忆和长期记忆：
class HierarchicalMemory:
def __init__(self):
self.working_memory = [] # 工作记忆（当前对话）
self.short_term = [] # 短期记忆（最近1小时）
self.long_term = VectorDB() # 长期记忆（永久存储）
def add_message(self, message):
"""添加消息到记忆系统"""
# 1. 工作记忆（直接可用）
self.working_memory.append(message)
# 2. 定期压缩到短期记忆
if len(self.working_memory) > 10:
summary = self.compress(self.working_memory[-10:])
self.short_term.append(summary)
self.working_memory = self.working_memory[-5:]
# 3. 定期归档到长期记忆
if len(self.short_term) > 20:
for item in self.short_term[:10]:
self.long_term.add(item)
self.short_term = self.short_term[10:]
def retrieve_context(self, query):
"""检索相关上下文"""
context = []
# 1. 工作记忆（全部）
context.extend(self.working_memory)
# 2. 短期记忆（最近）
context.extend(self.short_term[-5:])
# 3. 长期记忆（相关性检索）
relevant = self.long_term.search(query, top_k=3)
context.extend(relevant)
return context策略2：智能摘要技术
不是简单压缩，而是"结构化摘要"：
def smart_summarize(text):
"""智能摘要（保留关键信息）"""
prompt = """
分析这段对话，提取以下信息：
1. 主要讨论的议题（不超过3个）
2. 关键决策（如果有）
3. 待办事项（如果有）
4. 重要数字/日期/人名
格式：
- 议题：...
- 决策：...
- 待办：...
- 关键信息：...
"""
summary = call_llm(prompt + text)
return summary
# 示例输出
"""
议题：项目架构设计
决策：使用微服务架构
待办：下周三前完成数据库设计
关键信息：预算50万，团队5人，截止日期6月30日
"""策略3：向量数据库检索
用"语义检索"代替"关键词匹配"：
from openai import OpenAI
import chromadb
client = OpenAI()
db = chromadb.Client()
collection = db.create_collection("conversation_memory")
def add_to_memory(text, metadata):
"""添加到向量数据库"""
# 1. 生成向量
embedding = client.embeddings.create(
model="text-embedding-3-small",
input=text
).data[0].embedding
# 2. 存储到数据库
collection.add(
embeddings=[embedding],
documents=[text],
metadatas=[metadata],
ids=[metadata['id']]
)
def search_memory(query, top_k=5):
"""语义检索"""
# 1. 查询向量化
query_embedding = client.embeddings.create(
model="text-embedding-3-small",
input=query
).data[0].embedding
# 2. 检索最相关的记忆
results = collection.query(
query_embeddings=[query_embedding],
n_results=top_k
)
return results['documents'][0]实战案例：处理100页技术文档
让我们构建一个能处理超长文档的 Agent。
步骤1：文档预处理
def preprocess_document(pdf_path):
"""预处理PDF文档"""
import PyPDF2
# 1. 提取文本
with open(pdf_path, 'rb') as f:
reader = PyPDF2.PdfReader(f)
full_text = ""
for page in reader.pages:
full_text += page.extract_text() + "\n"
# 2. 智能分段（按章节）
sections = split_by_sections(full_text)
# 3. 生成向量索引
for i, section in enumerate(sections):
add_to_memory(
text=section['content'],
metadata={
'id': f"section_{i}",
'title': section['title'],
'page': section['page']
}
)
return len(sections)
def split_by_sections(text):
"""按章节分段"""
sections = []
current_section = {'title': '', 'content': '', 'page': 1}
for line in text.split('\n'):
# 检测章节标题（简单规则）
if line.strip().startswith('第') and '章' in line:
if current_section['content']:
sections.append(current_section)
current_section = {'title': line.strip(), 'content': '', 'page': 1}
else:
current_section['content'] += line + '\n'
if current_section['content']:
sections.append(current_section)
return sections步骤2：构建检索增强Agent
class LongDocumentAgent:
def __init__(self):
self.memory = HierarchicalMemory()
def ask(self, question):
"""回答关于长文档的问题"""
# 1. 检索相关章节
relevant_sections = search_memory(question, top_k=3)
# 2. 构建上下文
context = "\n\n".join([
f"【章节{i+1}】\n{section}"
for i, section in enumerate(relevant_sections)
])
# 3. 生成回答
prompt = f"""
基于以下文档内容回答问题：
{context}
问题：{question}
要求：
1. 答案必须基于文档内容
2. 引用具体章节
3. 如果文档中没有答案，明确说明
"""
response = client.chat.completions.create(
model="gpt-4-turbo",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
# 使用示例
agent = LongDocumentAgent()
answer = agent.ask("文档中提到的性能优化策略有哪些？")
print(answer)步骤3：增量学习
处理过程中持续学习：
def incremental_learning(agent, new_document):
"""增量学习新文档"""
# 1. 提取关键信息
key_points = extract_key_points(new_document)
# 2. 更新记忆
for point in key_points:
add_to_memory(
text=point['content'],
metadata={
'type': 'key_point',
'source': new_document,
'timestamp': datetime.now().isoformat()
}
)
# 3. 建立关联
connect_related_points(key_points)
def extract_key_points(text):
"""提取关键知识点"""
prompt = """
从这段文本中提取关键知识点（不超过5个）：
格式：
1. [知识点] - 简要说明
2. ...
"""
response = client.chat.completions.create(
model="gpt-4-turbo",
messages=[{"role": "user", "content": prompt + text}]
)
return parse_key_points(response.choices[0].message.content)性能优化技巧
1. 缓存常用查询
from functools import lru_cache
@lru_cache(maxsize=100)
def cached_search(query):
"""缓存常用查询"""
return search_memory(query)
# 第二次查询相同内容会直接返回缓存
result1 = cached_search("性能优化") # 慢
result2 = cached_search("性能优化") # 快（缓存）2. 批量处理
def batch_process(documents):
"""批量处理文档"""
# 1. 批量生成向量
texts = [doc['content'] for doc in documents]
embeddings = client.embeddings.create(
model="text-embedding-3-small",
input=texts
)
# 2. 批量插入数据库
collection.add(
embeddings=[e.embedding for e in embeddings.data],
documents=texts,
metadatas=[doc['metadata'] for doc in documents],
ids=[doc['id'] for doc in documents]
)3. 混合检索
结合关键词和语义检索：
def hybrid_search(query):
"""混合检索（关键词 + 语义）"""
# 1. 关键词检索（快速）
keyword_results = keyword_search(query, top_k=10)
# 2. 语义检索（精准）
semantic_results = semantic_search(query, top_k=10)
# 3. 合并去重
combined = merge_and_dedupe(keyword_results, semantic_results)
# 4. 重排序
reranked = rerank(combined, query)
return reranked[:5]成本控制
长文本处理很贵，需要优化成本：
1. 选择合适的模型
| 任务类型 | 推荐模型 | 成本 |
|---|---|---|
| 向量生成 | text-embedding-3-small | $0.02/1M tokens |
| 摘要压缩 | GPT-3.5-Turbo | $0.50/1M tokens |
| 最终回答 | GPT-4-Turbo | $10/1M tokens |
2. 优化策略
def cost_optimized_process(text):
"""成本优化处理"""
# 1. 先用便宜模型预处理
summary = cheap_summarize(text) # GPT-3.5
# 2. 只对重要内容用昂贵模型
if is_important(summary):
detailed_analysis = expensive_analyze(text) # GPT-4
else:
detailed_analysis = summary
return detailed_analysis实战建议
1. 根据场景选择策略
场景1：实时对话
- 工作记忆：最近10轮
- 短期记忆：最近1小时摘要
- 长期记忆：关键决策和事实
场景2：文档分析
- 全文索引到向量数据库
- 检索增强生成（RAG）
- 缓存常用查询
场景3：长期项目
- 持久化所有对话
- 定期归档和压缩
- 建立知识图谱
2. 监控和调优
def monitor_memory_usage():
"""监控记忆系统"""
stats = {
'working_memory_size': len(working_memory),
'short_term_size': len(short_term),
'long_term_size': db.count(),
'cache_hit_rate': cache_hit_rate(),
'avg_response_time': avg_response_time()
}
# 根据统计调优
if stats['cache_hit_rate'] < 0.3:
print("⚠️ 缓存命中率低，考虑增加缓存大小")
if stats['avg_response_time'] > 2.0:
print("⚠️ 响应时间慢，考虑优化检索")
return stats总结
长文本处理是 Agent 的"记忆宫殿"：
关键收获：
- 分层记忆（工作/短期/长期）
- 智能摘要（不是简单压缩）
- 向量检索（语义理解）
- 成本优化（模型分级）
最佳实践：
- 根据场景选择策略
- 持续监控和调优
- 平衡成本和质量
- 定期归档和清理
下一步：
- 实现一个分层记忆系统
- 用向量数据库处理你的项目文档
- 监控并优化你的记忆架构
记住：好的记忆系统不是"记住所有东西"，而是"在需要时找到需要的信息"。
系列导航：
- 上一篇：多模态Agent实战
- 下一篇：多Agent协作（敬请期待）
Views: 0