长文本处理指南
概述
DeepSeek 支持处理长文本内容,具备强大的长上下文理解能力。本指南将帮助您有效处理长文档、大型代码库、学术论文等长文本场景。
模型能力
上下文长度支持
- DeepSeek-V3: 支持最长 128K tokens
- DeepSeek-Coder: 支持最长 64K tokens
- DeepSeek-Chat: 支持最长 32K tokens
处理能力
- 长文档理解和总结
- 大型代码库分析
- 学术论文研究
- 法律文档审查
- 技术文档编写
基础长文本处理
1. 文档总结
python
import openai
client = openai.OpenAI(
api_key="your-api-key",
base_url="https://api.deepseek.com"
)
def summarize_long_document(document_text):
response = client.chat.completions.create(
model="deepseek-chat",
messages=[
{
"role": "system",
"content": "你是一个专业的文档分析师,擅长提取关键信息和生成简洁的摘要。"
},
{
"role": "user",
"content": f"""
请对以下长文档进行总结:
{document_text}
要求:
1. 提取主要观点和结论
2. 保留重要的数据和事实
3. 生成结构化的摘要
4. 控制在500字以内
"""
}
],
max_tokens=1000,
temperature=0.3
)
return response.choices[0].message.content
2. 关键信息提取
python
def extract_key_information(document_text, extraction_type="all"):
extraction_prompts = {
"people": "提取文档中提到的所有人名、职位和相关信息",
"dates": "提取文档中的所有日期、时间和时间相关信息",
"numbers": "提取文档中的所有数字、统计数据和量化信息",
"locations": "提取文档中提到的所有地点、地址和位置信息",
"all": "提取文档中的关键信息,包括人名、日期、地点、数字等"
}
prompt = extraction_prompts.get(extraction_type, extraction_prompts["all"])
response = client.chat.completions.create(
model="deepseek-chat",
messages=[
{
"role": "user",
"content": f"""
{prompt}
文档内容:
{document_text}
请以结构化的JSON格式返回结果。
"""
}
],
response_format={"type": "json_object"}
)
return response.choices[0].message.content
3. 文档问答
python
def document_qa(document_text, question):
response = client.chat.completions.create(
model="deepseek-chat",
messages=[
{
"role": "system",
"content": "基于提供的文档内容回答问题,确保答案准确且有依据。"
},
{
"role": "user",
"content": f"""
文档内容:
{document_text}
问题:{question}
请基于文档内容回答问题,如果文档中没有相关信息,请明确说明。
"""
}
]
)
return response.choices[0].message.content
高级长文本处理
1. 分块处理策略
python
def chunk_text(text, chunk_size=4000, overlap=200):
"""将长文本分块处理"""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
# 如果不是最后一块,尝试在句号处分割
if end < len(text):
last_period = text.rfind('.', start, end)
if last_period > start:
end = last_period + 1
chunk = text[start:end]
chunks.append(chunk)
start = end - overlap
return chunks
def process_long_text_in_chunks(text, task="summarize"):
chunks = chunk_text(text)
results = []
for i, chunk in enumerate(chunks):
print(f"处理第 {i+1}/{len(chunks)} 块...")
if task == "summarize":
result = summarize_chunk(chunk)
elif task == "analyze":
result = analyze_chunk(chunk)
results.append(result)
# 合并结果
return combine_results(results, task)
def summarize_chunk(chunk):
response = client.chat.completions.create(
model="deepseek-chat",
messages=[
{
"role": "user",
"content": f"请简要总结以下文本的主要内容:\n\n{chunk}"
}
],
max_tokens=300
)
return response.choices[0].message.content
def combine_results(results, task):
combined_text = "\n\n".join(results)
response = client.chat.completions.create(
model="deepseek-chat",
messages=[
{
"role": "user",
"content": f"""
以下是对长文档各部分的{task}结果:
{combined_text}
请将这些结果整合成一个连贯、完整的最终{task}。
"""
}
]
)
return response.choices[0].message.content
2. 层次化分析
python
def hierarchical_analysis(document_text):
"""层次化文档分析"""
# 第一层:整体结构分析
structure_analysis = analyze_document_structure(document_text)
# 第二层:章节内容分析
section_analyses = []
sections = extract_sections(document_text)
for section in sections:
analysis = analyze_section(section)
section_analyses.append(analysis)
# 第三层:综合分析
comprehensive_analysis = synthesize_analysis(
structure_analysis,
section_analyses
)
return {
"structure": structure_analysis,
"sections": section_analyses,
"comprehensive": comprehensive_analysis
}
def analyze_document_structure(text):
response = client.chat.completions.create(
model="deepseek-chat",
messages=[
{
"role": "user",
"content": f"""
分析以下文档的整体结构:
{text[:2000]}...
请识别:
1. 文档类型和主题
2. 主要章节和结构
3. 逻辑组织方式
4. 写作风格和特点
"""
}
]
)
return response.choices[0].message.content
3. 多轮对话处理
python
class LongTextConversation:
def __init__(self, document_text):
self.document = document_text
self.conversation_history = []
self.context_summary = self.create_initial_summary()
def create_initial_summary(self):
"""创建文档的初始摘要"""
response = client.chat.completions.create(
model="deepseek-chat",
messages=[
{
"role": "user",
"content": f"""
请为以下文档创建一个详细的摘要,用于后续的问答对话:
{self.document}
摘要应包括主要观点、关键信息和重要细节。
"""
}
]
)
return response.choices[0].message.content
def ask_question(self, question):
"""基于文档进行问答"""
# 构建对话上下文
messages = [
{
"role": "system",
"content": f"""
你是一个文档分析助手。以下是文档摘要:
{self.context_summary}
请基于文档内容回答用户问题。
"""
}
]
# 添加对话历史
messages.extend(self.conversation_history)
# 添加当前问题
messages.append({
"role": "user",
"content": question
})
response = client.chat.completions.create(
model="deepseek-chat",
messages=messages
)
answer = response.choices[0].message.content
# 更新对话历史
self.conversation_history.append({
"role": "user",
"content": question
})
self.conversation_history.append({
"role": "assistant",
"content": answer
})
# 保持对话历史在合理长度内
if len(self.conversation_history) > 10:
self.conversation_history = self.conversation_history[-10:]
return answer
# 使用示例
doc_chat = LongTextConversation(long_document)
answer1 = doc_chat.ask_question("这个文档的主要结论是什么?")
answer2 = doc_chat.ask_question("有哪些支持这个结论的证据?")
特定应用场景
1. 学术论文分析
python
def analyze_academic_paper(paper_text):
analysis_tasks = [
"提取论文的研究问题和假设",
"总结研究方法和实验设计",
"分析主要发现和结论",
"评估论文的贡献和局限性",
"识别相关工作和引用"
]
results = {}
for task in analysis_tasks:
response = client.chat.completions.create(
model="deepseek-chat",
messages=[
{
"role": "user",
"content": f"""
{task}
论文内容:
{paper_text}
"""
}
]
)
task_key = task.split("和")[0].replace("提取", "").replace("总结", "").replace("分析", "").replace("评估", "").replace("识别", "")
results[task_key] = response.choices[0].message.content
return results
2. 法律文档审查
python
def legal_document_review(legal_text):
review_aspects = {
"key_terms": "识别和解释关键法律术语",
"obligations": "提取各方的权利和义务",
"risks": "识别潜在的法律风险和问题",
"compliance": "检查合规性要求",
"recommendations": "提供修改建议和注意事项"
}
review_results = {}
for aspect, prompt in review_aspects.items():
response = client.chat.completions.create(
model="deepseek-chat",
messages=[
{
"role": "system",
"content": "你是一个专业的法律文档分析师,请提供准确、专业的分析。"
},
{
"role": "user",
"content": f"""
{prompt}
法律文档:
{legal_text}
"""
}
]
)
review_results[aspect] = response.choices[0].message.content
return review_results
3. 代码库分析
python
def analyze_codebase(code_files):
"""分析大型代码库"""
# 合并所有代码文件
combined_code = "\n\n".join([
f"// File: {filename}\n{content}"
for filename, content in code_files.items()
])
analysis_prompts = {
"architecture": "分析代码架构和设计模式",
"dependencies": "识别模块依赖关系",
"quality": "评估代码质量和潜在问题",
"documentation": "检查文档完整性",
"security": "识别安全漏洞和风险",
"performance": "分析性能瓶颈和优化机会"
}
results = {}
for aspect, prompt in analysis_prompts.items():
response = client.chat.completions.create(
model="deepseek-coder",
messages=[
{
"role": "user",
"content": f"""
{prompt}
代码库内容:
{combined_code}
请提供详细的分析报告。
"""
}
]
)
results[aspect] = response.choices[0].message.content
return results
性能优化技巧
1. Token 使用优化
python
def optimize_token_usage(text, max_tokens=100000):
"""优化token使用"""
# 估算token数量(粗略估算:1 token ≈ 4 字符)
estimated_tokens = len(text) // 4
if estimated_tokens <= max_tokens:
return text
# 如果超出限制,进行智能截取
return smart_truncate(text, max_tokens)
def smart_truncate(text, max_tokens):
"""智能截取文本"""
target_length = max_tokens * 4 # 转换为字符数
if len(text) <= target_length:
return text
# 尝试在段落边界截取
paragraphs = text.split('\n\n')
truncated = ""
for paragraph in paragraphs:
if len(truncated + paragraph) <= target_length:
truncated += paragraph + '\n\n'
else:
break
return truncated.strip()
2. 并行处理
python
import asyncio
import aiohttp
async def process_multiple_documents(documents):
"""并行处理多个文档"""
async def process_single_doc(doc_id, content):
try:
result = await async_summarize(content)
return {"doc_id": doc_id, "result": result}
except Exception as e:
return {"doc_id": doc_id, "error": str(e)}
tasks = [
process_single_doc(doc_id, content)
for doc_id, content in documents.items()
]
results = await asyncio.gather(*tasks)
return results
async def async_summarize(text):
"""异步文档总结"""
# 这里需要使用支持异步的HTTP客户端
# 示例使用同步客户端的异步包装
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None,
lambda: summarize_long_document(text)
)
3. 缓存机制
python
import hashlib
import json
from functools import wraps
def cache_result(func):
"""缓存处理结果"""
cache = {}
@wraps(func)
def wrapper(text, *args, **kwargs):
# 创建缓存键
cache_key = hashlib.md5(
(text + str(args) + str(kwargs)).encode()
).hexdigest()
if cache_key in cache:
return cache[cache_key]
result = func(text, *args, **kwargs)
cache[cache_key] = result
return result
return wrapper
@cache_result
def cached_summarize(text):
return summarize_long_document(text)
质量控制
1. 结果验证
python
def validate_summary_quality(original_text, summary):
"""验证摘要质量"""
validation_prompt = f"""
评估以下摘要的质量:
原文长度:{len(original_text)} 字符
摘要长度:{len(summary)} 字符
原文:{original_text[:500]}...
摘要:{summary}
请从以下方面评估:
1. 准确性:摘要是否准确反映原文内容
2. 完整性:是否涵盖了主要观点
3. 简洁性:是否去除了冗余信息
4. 连贯性:摘要是否逻辑清晰
给出1-10分的评分和改进建议。
"""
response = client.chat.completions.create(
model="deepseek-chat",
messages=[{"role": "user", "content": validation_prompt}]
)
return response.choices[0].message.content
2. 一致性检查
python
def consistency_check(text, multiple_results):
"""检查多次处理结果的一致性"""
consistency_prompt = f"""
检查以下多个处理结果的一致性:
原文:{text[:200]}...
结果1:{multiple_results[0]}
结果2:{multiple_results[1]}
结果3:{multiple_results[2]}
请分析:
1. 结果之间的相似性
2. 主要差异点
3. 哪个结果更准确
4. 一致性评分(1-10)
"""
response = client.chat.completions.create(
model="deepseek-chat",
messages=[{"role": "user", "content": consistency_prompt}]
)
return response.choices[0].message.content
常见问题解答
Q: 如何处理超长文档?
A:
- 使用分块处理策略
- 采用层次化分析方法
- 利用摘要和关键信息提取
- 考虑并行处理提高效率
Q: 如何保证长文本处理的准确性?
A:
- 使用多轮验证机制
- 实施结果一致性检查
- 采用人工审核关键结果
- 建立质量评估标准
Q: 处理速度如何优化?
A:
- 实施智能缓存机制
- 使用并行处理技术
- 优化token使用策略
- 选择合适的模型
Q: 如何处理不同类型的长文档?
A:
- 根据文档类型调整处理策略
- 使用专门的提示词模板
- 采用领域特定的分析方法
- 建立文档类型识别机制
最佳实践总结
- 预处理优化:清理文本、标准化格式
- 分块策略:合理分割、保持上下文
- 并行处理:提高处理效率
- 质量控制:验证结果、确保准确性
- 缓存机制:避免重复处理
- 错误处理:优雅处理异常情况
相关资源
技术支持
如需技术支持,请联系我们的技术支持团队。