代码在最后

前言

在企业级RAG（检索增强生成）系统中，表格数据的处理是一个关键挑战。本文基于RAG Challenge竞赛获奖方案中的tables_serialization.py模块，深入分析如何利用LLM实现高质量的表格序列化，将HTML表格转换为结构化的、上下文独立的信息块。

1. 模块架构概览

1.1 核心组件

tables_serialization.py模块包含三个主要类：

• TableSerializer: 表格序列化主类，支持同步/异步处理

• TableSerialization: 配置和模型定义类

• TqdmLoggingHandler: 进度日志处理器

1.2 技术栈

# 核心依赖
from openai import OpenAI  # OpenAI API调用
from pydantic import BaseModel  # 数据模型定义
from concurrent.futures import ThreadPoolExecutor  # 并行处理
from queue import Queue  # 消息队列
import asyncio  # 异步支持

2. 日志处理机制

2.1 消息队列设计

message_queue = Queue()
​
class TqdmLoggingHandler(logging.Handler):
    def emit(self, record):
        try:
            msg = self.format(record)
            message_queue.put((record.levelno, msg))
        except Exception:
            self.handleError(record)
​
def process_messages():
    while not message_queue.empty():
        level, msg = message_queue.get_nowait()
        tqdm.write(msg)

设计亮点：

1. 线程安全：使用Queue确保多线程环境下的日志安全

2. 进度条兼容：通过tqdm.write避免日志干扰进度显示

3. 异步处理：非阻塞式消息处理

3. TableSerializer类深度解析

3.1 上下文提取

def _get_table_context(self, json_report, target_table_index):
    # 获取表格所在页的上下文文本（前后各最多3个块）
    table_info = next(table for table in json_report["tables"] 
                     if table["table_id"] == target_table_index)
    page_num = table_info["page"]
    
    # 定位目标表格位置
    current_table_position = -1
    for i, block in enumerate(page_content):
        if block["type"] == "table" and block.get("table_id") == target_table_index:
            current_table_position = i
            break
            
    # 提取上下文
    context_before = page_content[start_position:current_table_position]
    context_after = page_content[current_table_position + 1:current_table_position + 4]

智能上下文提取：

1. 精确定位：通过table_id准确找到目标表格

2. 边界处理：考虑页面开始和结束的特殊情况

3. 上下文平衡：合理控制上下文范围

3.2 序列化请求构造

def _send_serialization_request(self, table, context_before, context_after):
    user_prompt = ""
    
    if context_before:
        user_prompt += f'Here is additional text before the table that might be relevant (or not):\n"""{context_before}"""\n\n'
    
    user_prompt += f'Here is a table in HTML format:\n"""{table}"""'
    
    if context_after:
        user_prompt += f'\n\nHere is additional text after the table that might be relevant (or not):\n"""{context_after}"""'

提示工程特点：

1. 结构化输入：清晰的三段式提示结构

2. 上下文融合：智能整合表格周边信息

3. 格式保持：保留HTML格式确保结构完整

4. 并行处理机制

4.1 线程池执行器

def process_directory_parallel(self, input_dir: Path, max_workers: int = 5):
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        with tqdm(total=len(json_files)) as pbar:
            futures = []
            for json_file in json_files:
                future = executor.submit(self.process_file, json_file)
                future.add_done_callback(lambda p: pbar.update(1))
                futures.append(future)

并行优化：

1. 资源控制：可配置的worker数量

2. 进度监控：实时进度条显示

3. 错误处理：完善的异常捕获机制

4.2 异步处理支持

async def async_serialize_tables(self, json_report: dict) -> dict:
    queries = []
    table_indices = []
    
    # 批量构建请求
    for table in json_report["tables"]:
        table_index = table["table_id"]
        table_indices.append(table_index)
        
        context_before, context_after = self._get_table_context(
            json_report, table_index)
        queries.append(self._build_query(
            table["html"], context_before, context_after))
    
    # 异步处理所有请求
    results = await AsyncOpenaiProcessor().process_structured_ouputs_requests(
        queries=queries,
        response_format=TableSerialization.TableBlocksCollection
    )

异步处理优势：

1. 批量处理：一次性处理多个表格

2. 资源效率：避免同步等待浪费

3. 可扩展性：支持大规模并发处理

5. 数据模型设计

5.1 序列化模型

class SerializedInformationBlock(BaseModel):
    subject_core_entity: str = Field(
        description="A primary focus of what this block is about")
    information_block: str = Field(description=(
        "Detailed information about the chosen core subject"))
​
class TableBlocksCollection(BaseModel):
    subject_core_entities_list: List[str]
    relevant_headers_list: List[str]
    information_blocks: List[SerializedInformationBlock]

模型特点：

1. 结构化定义：使用Pydantic确保数据验证

2. 层次化组织：清晰的数据层次结构

3. 完整性保证：强制要求必要字段

6. 实践建议

6.1 性能优化

1. 批量处理：

   • 使用异步模式处理大量表格

   • 合理设置并行度

   • 监控资源使用

2. 内存管理：

   • 及时清理临时文件

   • 控制上下文大小

   • 使用生成器处理大数据

3. 错误处理：

   • 完善的日志记录

   • 异常重试机制

   • 状态恢复支持

6.2 部署注意事项

1. 环境配置：

   • 设置合适的OpenAI API密钥

   • 配置足够的并发限制

   • 准备充足的存储空间

2. 监控告警：

   • 设置处理超时告警

   • 监控API调用限制

   • 跟踪处理成功率

总结

tables_serialization.py模块通过结合LLM能力和高效的并行处理机制，提供了一个强大的表格序列化解决方案。其模块化设计、完善的错误处理和灵活的配置选项，使其能够适应各种企业级应用场景。通过合理的部署和优化，可以构建出高性能、可靠的表格处理系统。

import os
import json
import asyncio
from pathlib import Path
from dotenv import load_dotenv
from typing import Optional, List, Union, Literal
from pydantic import BaseModel, Field
from openai import OpenAI
from src.api_requests import BaseOpenaiProcessor, AsyncOpenaiProcessor
import tiktoken
from tqdm import tqdm
import logging
import threading
from concurrent.futures import ThreadPoolExecutor
from queue import Queue
import time

message_queue = Queue()

class TqdmLoggingHandler(logging.Handler):
    def emit(self, record):
        try:
            msg = self.format(record)
            message_queue.put((record.levelno, msg))
        except Exception:
            self.handleError(record)

def process_messages():
    while not message_queue.empty():
        level, msg = message_queue.get_nowait()
        tqdm.write(msg)

# TableSerializer：表格序列化主流程类，支持同步/异步LLM表格结构化
class TableSerializer(BaseOpenaiProcessor):
    def __init__(self, preserve_temp_files: bool = True):
        super().__init__()
        self.preserve_temp_files = preserve_temp_files
        os.makedirs('./temp', exist_ok=True)
        
        self.logger = logging.getLogger('TableSerializer')
        self.logger.setLevel(logging.INFO)
        
        self.logger.handlers.clear()
        
        handler = TqdmLoggingHandler()
        handler.setFormatter(logging.Formatter('%(levelname)s: %(message)s'))
        self.logger.addHandler(handler)
        
        self.logger.propagate = False

    def _get_table_context(self, json_report, target_table_index):
        # 获取表格所在页的上下文文本（前后各最多3个块）
        table_info = next(table for table in json_report["tables"] if table["table_id"] == target_table_index)
        page_num = table_info["page"]
        
        page_content = next(
            (page["content"] for page in json_report["content"] if page["page"] == page_num),
            []
        )
        
        if not page_content:
            self.logger.warning(f"Page {page_num} not found for table {target_table_index}")
            return "", ""

        # 定位目标表格在页面中的位置
        current_table_position = -1
        for i, block in enumerate(page_content):
            if block["type"] == "table" and block.get("table_id") == target_table_index:
                current_table_position = i
                break

        # 查找前一个表格位置
        previous_table_position = -1
        for i in range(current_table_position-1, -1, -1):
            if page_content[i]["type"] == "table":
                previous_table_position = i
                break

        # 查找下一个表格位置
        next_table_position = -1
        for i in range(current_table_position + 1, len(page_content)):
            if page_content[i]["type"] == "table":
                next_table_position = i
                break

        # 获取当前表格上方的块
        start_position = previous_table_position + 1 if previous_table_position != -1 else 0
        context_before = page_content[start_position:current_table_position]

        # 获取当前表格下方的块
        context_after = []
        if next_table_position == -1:
            # 没有下一个表格，取后3个块
            context_after = page_content[current_table_position + 1:current_table_position + 4]
        else:
            # 有下一个表格，取到下一个表格前最多3个块
            blocks_between = next_table_position - (current_table_position + 1)
            if blocks_between > 3:
                context_after = page_content[current_table_position + 1:current_table_position + 4]
            elif blocks_between > 1:
                context_after = page_content[current_table_position + 1:current_table_position + blocks_between]

        context_before = "\n".join(block.get("text", "") for block in context_before if "text" in block)
        context_after = "\n".join(block.get("text", "") for block in context_after if "text" in block)

        return context_before, context_after

    def _send_serialization_request(self, table, context_before, context_after):
        # 构造LLM表格序列化请求，拼接上下文和表格HTML
        user_prompt = ""
        
        if context_before:
            user_prompt += f'Here is additional text before the table that might be relevant (or not):\n"""{context_before}"""\n\n'
        
        user_prompt += f'Here is a table in HTML format:\n"""{table}"""'
        
        if context_after:
            user_prompt += f'\n\nHere is additional text after the table that might be relevant (or not):\n"""{context_after}"""'
        
        system_prompt = TableSerialization.system_prompt
        reponse_schema = TableSerialization.TableBlocksCollection

        answer_dict = self.send_message(
            model='gpt-4o-mini-2024-07-18',
            temperature=0,
            system_content=system_prompt,
            human_content=user_prompt,
            is_structured=True,
            response_format=reponse_schema
        )

        input_message = user_prompt + system_prompt + str(reponse_schema.schema())
        input_tokens = self.count_tokens(input_message)
        output_tokens = self.count_tokens(str(answer_dict))

        result = answer_dict
        return result
    
    def _serialize_table(self, json_report: dict, target_table_index: int) -> dict:
        # 序列化单个表格，获取上下文并调用LLM
        context_before, context_after = self._get_table_context(json_report, target_table_index)
        table_info = next(table for table in json_report["tables"] if table["table_id"] == target_table_index)
        table_content = table_info["html"]
        result = self._send_serialization_request(
            table=table_content,
            context_before=context_before,
            context_after=context_after
        )
        return result

    def serialize_tables(self, json_report: dict) -> dict:
        """批量处理报告中所有表格，序列化结果写入table['serialized']"""
        for table in json_report["tables"]:
            table_index = table["table_id"]
            # 获取当前表格的序列化结果
            serialization_result = self._serialize_table(
                json_report=json_report,
                target_table_index=table_index
            )
            # 写入序列化结果
            table["serialized"] = serialization_result
        return json_report

    async def async_serialize_tables(
        self, 
        json_report: dict,
        requests_filepath: str = './temp_async_llm_requests.jsonl',
        results_filepath: str = './temp_async_llm_results.jsonl'
    ) -> dict:
        """异步批量处理报告中所有表格，适合大规模并发"""
        queries = []
        table_indices = []
        
        for table in json_report["tables"]:
            table_index = table["table_id"]
            table_indices.append(table_index)
            
            context_before, context_after = self._get_table_context(json_report, table_index)
            table_info = next(table for table in json_report["tables"] if table["table_id"] == table_index)
            table_content = table_info["html"]
            
            # 构造异步请求query
            query = ""
            if context_before:
                query += f'Here is additional text before the table that might be relevant (or not):\n"""{context_before}"""\n\n'
            query += f'Here is a table in HTML format:\n"""{table_content}"""'
            if context_after:
                query += f'\n\nHere is additional text after the table that might be relevant (or not):\n"""{context_after}"""'
            
            queries.append(query)

        results = await AsyncOpenaiProcessor().process_structured_ouputs_requests(
            model='gpt-4o-mini-2024-07-18',
            temperature=0,
            system_content=TableSerialization.system_prompt,
            queries=queries,
            response_format=TableSerialization.TableBlocksCollection,
            preserve_requests=False,
            preserve_results=False,
            logging_level=20,
            requests_filepath=requests_filepath,
            save_filepath=results_filepath,
        )

        # Add results back to json_report
        for table_index, result in zip(table_indices, results):
            table_info = next(table for table in json_report["tables"] if table["table_id"] == table_index)
            
            new_table = {}
            for key, value in table_info.items():
                new_table[key] = value
                if key == "html":
                    new_table["serialized"] = result["answer"]
            
            for i, table in enumerate(json_report["tables"]):
                if table["table_id"] == table_index:
                    json_report["tables"][i] = new_table

        return json_report

    def process_file(self, json_path: Path) -> None:
        try:
            with open(json_path, 'r', encoding='utf-8') as f:
                json_report = json.load(f)
            
            thread_id = threading.get_ident()
            requests_filepath = f'./temp/async_llm_requests_{thread_id}.jsonl'
            results_filepath = f'./temp/async_llm_results_{thread_id}.jsonl'
            
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
            try:
                updated_report = loop.run_until_complete(self.async_serialize_tables(
                    json_report,
                    requests_filepath=requests_filepath,
                    results_filepath=results_filepath
                ))
            finally:
                loop.close()
                try:
                    os.remove(requests_filepath)
                    os.remove(results_filepath)
                except FileNotFoundError:
                    pass
            
            with open(json_path, 'w', encoding='utf-8') as f:
                json.dump(updated_report, f, indent=2, ensure_ascii=False)
                
        except json.JSONDecodeError as e:
            self.logger.error("JSON Error in %s: %s", json_path.name, str(e))
            raise
        except Exception as e:
            self.logger.error("Error processing %s: %s", json_path.name, str(e))
            raise

    def process_directory_parallel(self, input_dir: Path, max_workers: int = 5):
        """Process JSON files in parallel using thread pool.
        
        Args:
            input_dir: Path to directory containing JSON files
            max_workers: Maximum number of threads to use
        """
        self.logger.info("Starting parallel table serialization...")
        
        json_files = list(input_dir.glob("*.json"))
        
        if not json_files:
            self.logger.warning("No JSON files found in %s", input_dir)
            return

        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            with tqdm(
                total=len(json_files),
                desc="Processing files",
                mininterval=1.0,
                maxinterval=5.0,
                smoothing=0.3
            ) as pbar:
                futures = []
                for json_file in json_files:
                    future = executor.submit(self.process_file, json_file)
                    future.add_done_callback(lambda p: pbar.update(1))
                    futures.append(future)
                
                while futures:
                    process_messages()
                    
                    done_futures = []
                    for future in futures:
                        if future.done():
                            done_futures.append(future)
                            try:
                                future.result()
                            except Exception as e:
                                self.logger.error(str(e))
                    
                    for future in done_futures:
                        futures.remove(future)
                    
                    time.sleep(0.1)

        process_messages()
        self.logger.info("Table serialization completed!")


class TableSerialization:
        
    system_prompt = (
        "You are a table serialization agent.\n"
        "Your task is to create a set of contextually independent blocks of information based on the provided table and surrounding text.\n"
        "These blocks must be totally context-independent because they will be used as separate chunk to populate database."
    )

    class SerializedInformationBlock(BaseModel):
        "A single self-contained information block enriched with comprehensive context"

        subject_core_entity: str = Field(description="A primary focus of what this block is about. Usually located in a row header. If one row in the table doesn't make sense without neighboring rows, you can merge information from neighboring rows into one block")
        information_block: str = Field(description=(
    "Detailed information about the chosen core subject from tables and additional texts. Information SHOULD include:\n"
    "1. All related header information\n"
    "2. All related units and their descriptions\n"
    "    2.1. If header is Total, always write additional context about what this total represents in this block!\n"
    "3. All additional info for context enrichment to make ensure complete context-independency if it present in whole table. This can include:\n"
    "    - The name of the table\n"
    "    - Additional footnotes\n"
    "    - The currency used\n"
    "    - The way amounts are presented\n"
    "    - Anything else that can make context even slightly richer\n"
    "SKIPPING ANY VALUABLE INFORMATION WILL BE HEAVILY PENALIZED!"
    ))

    class TableBlocksCollection(BaseModel):
        """Collection of serialized table blocks with their core entities and header relationships"""

        subject_core_entities_list: List[str] = Field(
            description="A complete list of core entities. Keep in mind, empty headers are possible - they should also be interpreted and listed (Usually it's a total or something similar). In most cases each row header represents a core entity")
        relevant_headers_list: List[str] = Field(description="A list of ALL headers relevant to the subject. These headers will serve as keys in each information block. In most cases each column header represents a core entity")
        information_blocks: List["TableSerialization.SerializedInformationBlock"] = Field(description="Complete list of fully described context-independent information blocks")