长文本处理

虽然现代 LLM 的上下文窗口越来越大（128K、200K 甚至 1M+），但处理长文本仍然面临成本、效率和效果的挑战。本文介绍长文本处理的最佳策略。

挑战与限制​

上下文窗口限制​

模型	上下文窗口	约等于字数
GPT-4o	128K	~10万字
Claude 3.5	200K	~15万字
Gemini 1.5 Pro	2M	~150万字

长上下文的问题​

1. 成本高昂： Token 数量大
2. 注意力稀释： 中间内容容易被忽略
3. 延迟增加： 处理时间更长
4. 精度下降： 关键信息可能被遗漏

处理策略​

1. 分块处理（Chunking）​

将长文本分成多个块分别处理：

def chunk_text(text, chunk_size=4000, overlap=200):  
    """将文本分成重叠的块"""  
    chunks = []  
    start = 0  
      
    while start < len(text):  
        end = start + chunk_size  
        chunk = text[start:end]  
        chunks.append({  
            "content": chunk,  
            "start": start,  
            "end": min(end, len(text))  
        })  
        start = end - overlap  
      
    return chunks  
  
def process_long_document(document, task):  
    chunks = chunk_text(document)  
    results = []  
      
    for chunk in chunks:  
        result = call_llm(f"{task}\n\n文本片段：\n{chunk['content']}")  
        results.append(result)  
      
    # 合并结果  
    return merge_results(results)

2. 递归摘要（Map-Reduce）​

先摘要各部分，再汇总：

def recursive_summarize(document, max_chunk_size=4000):  
    # 分块  
    chunks = chunk_text(document, max_chunk_size)  
      
    # Map: 每块生成摘要  
    summaries = []  
    for chunk in chunks:  
        summary = call_llm(f"请摘要以下内容（200字以内）：\n{chunk['content']}")  
        summaries.append(summary)  
      
    # 合并摘要  
    combined = "\n\n".join(summaries)  
      
    # 如果合并后仍然太长，递归处理  
    if count_tokens(combined) > max_chunk_size:  
        return recursive_summarize(combined, max_chunk_size)  
      
    # Reduce: 生成最终摘要  
    final_summary = call_llm(f"基于以下分段摘要，生成一份完整的文档摘要：\n{combined}")  
      
    return final_summary

3. 关键信息提取 + 压缩​

先提取关键信息，再处理：

def extract_and_process(document, query):  
    # 提取与查询相关的关键段落  
    relevant_parts = extract_relevant_sections(document, query)  
      
    # 压缩关键信息  
    compressed = call_llm(f"""  
    请提取以下文本中与"{query}"相关的关键信息，  
    压缩到 1000 字以内，保留关键数据和结论：  
      
    {relevant_parts}  
    """)  
      
    # 基于压缩内容回答  
    answer = call_llm(f"""  
    基于以下信息回答问题：  
      
    信息：{compressed}  
      
    问题：{query}  
    """)  
      
    return answer

4. 滑动窗口​

处理需要连续性的长文本：

def sliding_window_analysis(document, window_size=8000, step=4000):  
    """滑动窗口分析，保持上下文连续性"""  
    results = []  
      
    for start in range(0, len(document), step):  
        end = min(start + window_size, len(document))  
        window = document[start:end]  
          
        # 包含之前窗口的摘要  
        context = f"之前内容摘要：{results[-1]['summary'] if results else '无'}"  
          
        result = call_llm(f"""  
        {context}  
          
        请分析以下文本段落：  
        {window}  
        """)  
          
        results.append({  
            "start": start,  
            "end": end,  
            "analysis": result,  
            "summary": extract_summary(result)  
        })  
      
    return results

RAG 增强处理​

向量检索​

不是处理全部文本，而是检索最相关的部分：

def rag_long_document(document, query):  
    # 1. 文档分块和向量化  
    chunks = chunk_text(document, chunk_size=500)  
      
    for chunk in chunks:  
        embedding = get_embedding(chunk["content"])  
        vector_db.insert({"text": chunk["content"], "embedding": embedding})  
      
    # 2. 检索相关块  
    query_embedding = get_embedding(query)  
    relevant_chunks = vector_db.search(query_embedding, top_k=10)  
      
    # 3. 基于相关内容回答  
    context = "\n\n".join([c["text"] for c in relevant_chunks])  
      
    answer = call_llm(f"""  
    基于以下相关内容回答问题：  
      
    内容：  
    {context}  
      
    问题：{query}  
    """)  
      
    return answer

特定任务的处理​

1. 文档问答​

def document_qa(document, questions):  
    # 创建文档索引  
    index = create_document_index(document)  
      
    answers = []  
    for question in questions:  
        # 检索相关段落  
        relevant = index.search(question, top_k=5)  
          
        # 生成答案  
        answer = call_llm(f"""  
        根据以下文档内容回答问题。如果文档中没有相关信息，请说明。  
          
        文档内容：  
        {relevant}  
          
        问题：{question}  
        """)  
          
        answers.append({"question": question, "answer": answer})  
      
    return answers

2. 文档摘要​

def hierarchical_summary(document):  
    """层级摘要：章节摘要 → 全文摘要"""  
      
    # 按章节分割  
    chapters = split_by_chapters(document)  
      
    # 每章摘要  
    chapter_summaries = []  
    for chapter in chapters:  
        summary = call_llm(f"""  
        请为以下章节生成摘要（100字以内）：  
          
        标题：{chapter['title']}  
        内容：{chapter['content']}  
        """)  
        chapter_summaries.append(f"## {chapter['title']}\n{summary}")  
      
    # 全文摘要  
    overall_summary = call_llm(f"""  
    基于以下章节摘要，生成一份 300 字的全文摘要：  
      
    {chr(10).join(chapter_summaries)}  
    """)  
      
    return {  
        "chapter_summaries": chapter_summaries,  
        "overall_summary": overall_summary  
    }

3. 文档翻译​

def translate_long_document(document, target_lang="中文"):  
    # 按段落分割  
    paragraphs = document.split("\n\n")  
      
    translated = []  
    for i, para in enumerate(paragraphs):  
        if not para.strip():  
            translated.append("")  
            continue  
          
        # 翻译时提供上下文  
        context = translated[-2:] if len(translated) >= 2 else []  
          
        translation = call_llm(f"""  
        翻译以下段落为{target_lang}，保持与上文的连贯性。  
          
        上文参考：  
        {chr(10).join(context)}  
          
        待翻译段落：  
        {para}  
          
        只输出翻译结果。  
        """)  
          
        translated.append(translation)  
      
    return "\n\n".join(translated)

4. 代码库分析​

def analyze_codebase(code_files):  
    # 1. 先分析各文件结构  
    file_analyses = []  
    for file in code_files:  
        analysis = call_llm(f"""  
        分析以下代码文件，提取：  
        - 主要功能  
        - 关键类/函数  
        - 依赖关系  
          
        文件：{file['path']}  
        代码：  
        ```  
        {file['content']}  
        ```  
        """)  
        file_analyses.append({"file": file['path'], "analysis": analysis})  
      
    # 2. 综合分析整体架构  
    overall = call_llm(f"""  
    基于以下各文件分析，描述整体代码架构：  
      
    {json.dumps(file_analyses, ensure_ascii=False, indent=2)}  
    """)  
      
    return overall

性能优化​

1. 并行处理​

import asyncio  
  
async def process_chunks_parallel(chunks, task, max_concurrent=5):  
    semaphore = asyncio.Semaphore(max_concurrent)  
      
    async def process_chunk(chunk):  
        async with semaphore:  
            return await async_llm_call(f"{task}\n\n{chunk['content']}")  
      
    tasks = [process_chunk(chunk) for chunk in chunks]  
    results = await asyncio.gather(*tasks)  
      
    return results

2. 缓存中间结果​

import hashlib  
  
cache = {}  
  
def process_with_cache(document, task):  
    cache_key = hashlib.md5((document + task).encode()).hexdigest()  
      
    if cache_key in cache:  
        return cache[cache_key]  
      
    result = process_long_document(document, task)  
    cache[cache_key] = result  
      
    return result

最佳实践​

1. 选择合适的分块大小： 通常 500-2000 tokens
2. 保留重叠： 分块时保持 10-20% 重叠
3. 使用 RAG： 大文档优先考虑检索增强
4. 分层处理： 先提取要点，再深入分析
5. 利用长上下文模型： Gemini、Claude 适合处理超长文本
6. 监控成本： 长文本处理成本高，注意控制

---

本页内容来自文档知识库整理，已按 ArmNet 字元服务 服务命名统一更新。