跳转至

Parser 子系统

Parser 子系统负责把不同格式的文档字节流转换为可分块的纯文本，服务于 RAG 摄取流程。

概览

解析器提供以下能力：

  • 统一接口：BaseDocumentParser 抽象基类
  • 内置解析器：文本、PDF、DOCX
  • 自动选择：按 MIME 类型选择解析器
  • 易扩展：新增解析器并注册即可

支持格式

格式 扩展名 MIME 类型 解析器
纯文本 .txt text/plain TextDocumentParser
Markdown .md, .markdown text/markdown TextDocumentParser
PDF .pdf application/pdf PDFDocumentParser
DOCX .docx application/vnd.openxmlformats-officedocument.wordprocessingml.document DocxDocumentParser

架构

BaseDocumentParser 接口

所有解析器都实现以下抽象接口：

from abc import ABC, abstractmethod
from dataclasses import dataclass

@dataclass
class ParseResult:
    """Result of parsing a document."""
    text: str                          # Extracted text content
    metadata: dict[str, Any]           # Parser-specific metadata
    page_count: Optional[int] = None   # Number of pages
    char_count: int = 0                # Character count

class BaseDocumentParser(ABC):
    """Abstract base class for document parsers."""

    @abstractmethod
    def parse(self, file_data: bytes, filename: str) -> ParseResult:
        """Parse document and extract text."""
        pass

    @property
    @abstractmethod
    def supported_content_types(self) -> list[str]:
        """List of supported MIME types."""
        pass

    @property
    @abstractmethod
    def supported_extensions(self) -> list[str]:
        """List of supported file extensions."""
        pass

解析器注册表

通过注册表实现自动选择：

from ai_service.services.parsers import get_parser, is_supported_content_type

parser = get_parser("application/pdf")
result = parser.parse(file_data, "document.pdf")

if is_supported_content_type("application/pdf"):
    print("PDF is supported")

内置解析器

TextDocumentParser

用于纯文本与 Markdown。

特性：

  • 多编码兼容（UTF-8、Latin-1、CP1252）
  • 换行符归一化
  • 编码自动检测

示例：

from ai_service.services.parsers import TextDocumentParser

parser = TextDocumentParser()
result = parser.parse(file_data, "document.txt")

print(f"Text: {result.text}")
print(f"Encoding: {result.metadata['encoding']}")
print(f"Lines: {result.metadata['line_count']}")

PDFDocumentParser

用于 PDF 文档。

特性：

  • 按页提取文本
  • 解析元数据
  • 统计页数

示例：

from ai_service.services.parsers import PDFDocumentParser

parser = PDFDocumentParser()
result = parser.parse(file_data, "document.pdf")

print(f"Pages: {result.page_count}")
print(f"Text: {result.text}")

DocxDocumentParser

用于 Word 文档。

特性：

  • 段落级提取
  • 保留基础结构
  • 提取文档元信息

示例：

from ai_service.services.parsers import DocxDocumentParser

parser = DocxDocumentParser()
result = parser.parse(file_data, "document.docx")

print(f"Text: {result.text}")

自定义解析器

步骤 1：实现 BaseDocumentParser

from ai_service.services.parsers.base import BaseDocumentParser, ParseResult

class CSVDocumentParser(BaseDocumentParser):
    """Parser for CSV files."""

    @property
    def supported_content_types(self) -> list[str]:
        return ["text/csv", "application/csv"]

    @property
    def supported_extensions(self) -> list[str]:
        return [".csv"]

    def parse(self, file_data: bytes, filename: str) -> ParseResult:
        """Parse CSV file and convert to text."""
        import csv
        from io import StringIO

        text_data = file_data.decode("utf-8")
        reader = csv.reader(StringIO(text_data))
        rows = list(reader)

        text_lines = []
        for row in rows:
            text_lines.append(" | ".join(row))

        text = "\n".join(text_lines)

        metadata = {
            "filename": filename,
            "row_count": len(rows),
            "column_count": len(rows[0]) if rows else 0,
        }

        return ParseResult(text=text, metadata=metadata)

步骤 2：注册解析器

在 ai_service/services/parsers/__init__.py 中注册：

from ai_service.services.parsers.csv_parser import CSVDocumentParser

_PARSER_REGISTRY: dict[str, type[BaseDocumentParser]] = {
    # ... existing parsers ...
    "text/csv": CSVDocumentParser,
    "application/csv": CSVDocumentParser,
}

_EXTENSION_TO_CONTENT_TYPE: dict[str, str] = {
    # ... existing mappings ...
    ".csv": "text/csv",
}

步骤 3：验证解析器

from ai_service.services.parsers import get_parser

csv_data = b"Name,Age,City\nJohn,30,NYC\nJane,25,LA"
parser = get_parser("text/csv")
result = parser.parse(csv_data, "data.csv")

print(result.text)

最佳实践

错误处理

try:
    parser = get_parser(content_type)
    result = parser.parse(file_data, filename)
except ValueError as e:
    print(f"Parse error: {e}")

编码检测

encodings = ["utf-8", "utf-8-sig", "latin-1", "cp1252"]
for encoding in encodings:
    try:
        text = file_data.decode(encoding)
        break
    except UnicodeDecodeError:
        continue

元数据建议

metadata = {
    "filename": filename,
    "parser_version": "1.0",
    "encoding": "utf-8",
    "page_count": 10,
    "author": "John Doe",
}

常见问题

问题：文本编码错误

解决：在 TextDocumentParser 中使用编码回退。

问题：PDF 乱码

原因：扫描件或非标准字体

解决：需要 OCR 或替换解析器。

问题：DOCX 内容缺失

原因：部分表格或页眉未完全解析

解决：根据文档结构定制解析器。

关联文档