如何高效实现带Bounding Box的PDF文本分块？求Langchain优化方案

核心问题分析 #

你当前的逐行统计分块方案存在三个关键问题：

1. 无重叠逻辑，无法实现LangChain那样的上下文连贯分块；
2. 重复计算字符长度、频繁重置列表等操作导致效率偏低；
3. 原代码存在元数据赋值错误（将未更新的metadata_dict传入Document，导致Bounding Box信息丢失）。

优化方案实现 #

1. 支持重叠的滑动窗口分块逻辑 #

通过维护滑动窗口的行列表，结合预计算的字符长度，实现精准的分块与重叠控制：

from langchain_core.documents import Document

def chunk_with_bbox(lines, metadata_dict, character_limit=1000, overlap_lines=2):
    chunk_docs = []
    if not lines:
        return chunk_docs
    
    # 预计算所有行的字符长度，避免重复计算
    precomputed_lengths = [len(line['text']) for line in lines]
    cumulative_lines = []
    current_char_count = 0

    for idx, line in enumerate(lines):
        line_len = precomputed_lengths[idx]
        
        # 若添加当前行后超过字符限制，且当前块已有内容，则生成分块
        if current_char_count + line_len > character_limit and cumulative_lines:
            # 提取块的首尾Bounding Box
            chunk_start_bbox = cumulative_lines[0]['boundingBox']
            chunk_end_bbox = cumulative_lines[-1]['boundingBox']
            # 拼接块文本（用列表推导式提升效率）
            chunk_text = "\n".join([l['text'] for l in cumulative_lines])
            # 构建元数据（修正原代码的赋值错误）
            chunk_metadata = metadata_dict.copy()
            chunk_metadata.update({
                'chunk_start_bbox': chunk_start_bbox,
                'chunk_end_bbox': chunk_end_bbox
            })
            chunk_docs.append(Document(page_content=chunk_text, metadata=chunk_metadata))
            
            # 处理重叠：保留最后N行作为下一个块的起始
            if overlap_lines > 0:
                cumulative_lines = cumulative_lines[-overlap_lines:]
                # 重新计算当前块的字符数（用预计算的长度快速求和）
                current_char_count = sum(precomputed_lengths[idx - len(cumulative_lines) + 1 : idx + 1])
            else:
                cumulative_lines = []
                current_char_count = 0
        
        # 将当前行加入累积列表
        cumulative_lines.append(line)
        current_char_count += line_len

    # 处理最后剩余的未达阈值的行
    if cumulative_lines:
        chunk_start_bbox = cumulative_lines[0]['boundingBox']
        chunk_end_bbox = cumulative_lines[-1]['boundingBox']
        chunk_text = "\n".join([l['text'] for l in cumulative_lines])
        chunk_metadata = metadata_dict.copy()
        chunk_metadata.update({
            'chunk_start_bbox': chunk_start_bbox,
            'chunk_end_bbox': chunk_end_bbox
        })
        chunk_docs.append(Document(page_content=chunk_text, metadata=chunk_metadata))
    
    return chunk_docs

# 使用示例
# item = Azure OCR返回的单页数据，包含lines列表
# metadata_dict = {'page_number': 1, ...}
# chunks = chunk_with_bbox(item['lines'], metadata_dict, character_limit=1000, overlap_lines=2)

2. 性能优化关键点 #

• 预计算字符长度：提前一次性计算所有行的字符数，避免循环中重复调用len()，减少CPU开销；
• 滑动窗口重叠：通过保留最后N行实现重叠，无需重新遍历之前的行，时间复杂度保持O(n)；
• 文本拼接优化：用列表推导式收集文本行后再调用join()，避免循环中拼接字符串产生大量中间对象；
• 并行处理多页：如果是多页PDF，可使用concurrent.futures.ThreadPoolExecutor并行处理每页的分块逻辑（每页处理相互独立）：

  from concurrent.futures import ThreadPoolExecutor
  
  def process_all_pages(pages_data, metadata_template, character_limit=1000, overlap_lines=2):
      all_chunks = []
      with ThreadPoolExecutor(max_workers=4) as executor:
          # 为每页生成任务
          futures = []
          for page_idx, page_item in enumerate(pages_data):
              page_metadata = metadata_template.copy()
              page_metadata['page_number'] = page_idx + 1
              futures.append(executor.submit(
                  chunk_with_bbox,
                  page_item['lines'],
                  page_metadata,
                  character_limit,
                  overlap_lines
              ))
          # 收集所有结果
          for future in futures:
              all_chunks.extend(future.result())
      return all_chunks
  

• 减少字典拷贝开销：如果元数据中大部分内容固定，可将固定部分作为基础，仅动态添加Bounding Box信息，避免频繁全量拷贝；
• 提前过滤空行：处理前先过滤OCR结果中的空文本行，减少无效计算。

内容的提问来源于stack exchange，提问作者AnonymousMe