Стратегии для кода и HTML с учётом типа документа

Почему универсальное разбиение не подходит для специализированных документов

Разбиение кода Python на основе AST

import ast
import textwrap

def extract_functions(source_code: str) -> list[dict]:
    tree = ast.parse(source_code)
    chunks = []
    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            start = node.lineno - 1
            end = node.end_lineno
            lines = source_code.splitlines()[start:end]
            chunks.append({
                'name': node.name,
                'code': '\n'.join(lines),
                'start_line': node.lineno,
            })
    return chunks

Разбиение по границам классов

from langchain_text_splitters import Language, RecursiveCharacterTextSplitter

python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=1000,
    chunk_overlap=100,
)

with open('my_module.py', 'r') as f:
    source = f.read()

chunks = python_splitter.create_documents([source])
print(f'Created {len(chunks)} code chunks')

Добавление метаданных кода во фрагменты

from langchain_core.documents import Document

def chunk_python_file(filepath: str) -> list[Document]:
    with open(filepath) as f:
        source = f.read()

    functions = extract_functions(source)  # from previous example
    docs = []
    for fn in functions:
        docs.append(Document(
            page_content=fn['code'],
            metadata={
                'source': filepath,
                'function': fn['name'],
                'language': 'python',
                'start_line': fn['start_line'],
            }
        ))
    return docs

HTML: структура важнее количества символов

from bs4 import BeautifulSoup

def chunk_html_by_section(html: str) -> list[dict]:
    soup = BeautifulSoup(html, 'html.parser')
    chunks = []
    for tag in soup.find_all(['h1', 'h2', 'h3', 'p', 'li']):
        text = tag.get_text(separator=' ', strip=True)
        if len(text) > 40:  # skip trivial fragments
            chunks.append({
                'tag': tag.name,
                'text': text,
            })
    return chunks

Иерархическое разбиение HTML по заголовкам

from langchain_text_splitters import HTMLHeaderTextSplitter

headers_to_split_on = [
    ('h1', 'Header 1'),
    ('h2', 'Header 2'),
    ('h3', 'Header 3'),
]

splitter = HTMLHeaderTextSplitter(headers_to_split_on=headers_to_split_on)

with open('page.html') as f:
    html = f.read()

sections = splitter.split_text(html)
for sec in sections[:3]:
    print(sec.metadata)
    print(sec.page_content[:200])
    print('---')

Markdown: сохранение иерархии заголовков

from langchain_text_splitters import MarkdownHeaderTextSplitter

headers_to_split_on = [
    ('#', 'H1'),
    ('##', 'H2'),
    ('###', 'H3'),
]

md_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)

with open('README.md') as f:
    markdown = f.read()

docs = md_splitter.split_text(markdown)
for doc in docs[:2]:
    print('Metadata:', doc.metadata)
    print('Content:', doc.page_content[:300])
    print()

Вторичное разбиение после разбиения по заголовкам

from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter

header_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[('#', 'H1'), ('##', 'H2')]
)
char_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
)

with open('docs.md') as f:
    md = f.read()

header_chunks = header_splitter.split_text(md)
final_chunks = char_splitter.split_documents(header_chunks)
print(f'{len(final_chunks)} final chunks produced')

Разбиение PDF с учётом таблиц

import pdfplumber

def extract_pdf_chunks(pdf_path: str) -> list[dict]:
    chunks = []
    with pdfplumber.open(pdf_path) as pdf:
        for page_num, page in enumerate(pdf.pages):
            # Extract tables separately
            for table in page.extract_tables():
                rows = ['|'.join(str(c) for c in row) for row in table]
                chunks.append({
                    'type': 'table',
                    'content': '\n'.join(rows),
                    'page': page_num + 1,
                })
            # Extract prose text
            text = page.extract_text()
            if text:
                chunks.append({'type': 'text', 'content': text, 'page': page_num + 1})
    return chunks

Определение языка для смешанных корпусов

from pathlib import Path

def route_document(filepath: str) -> list[dict]:
    ext = Path(filepath).suffix.lower()
    if ext == '.py':
        return chunk_python_file(filepath)
    elif ext in ('.html', '.htm'):
        with open(filepath) as f:
            return chunk_html_by_section(f.read())
    elif ext == '.md':
        # use MarkdownHeaderTextSplitter
        return chunk_markdown(filepath)
    elif ext == '.pdf':
        return extract_pdf_chunks(filepath)
    else:
        # fallback: plain text recursive splitter
        return chunk_plain_text(filepath)

Сохранение контекста с помощью окружающих строк

def chunk_with_imports(source_code: str, fn_node, lines: list[str]) -> str:
    # Gather top-of-file imports (first block before first non-import)
    import_lines = []
    for line in lines:
        stripped = line.strip()
        if stripped.startswith('import ') or stripped.startswith('from '):
            import_lines.append(line)
        elif stripped and not stripped.startswith('#'):
            break

    fn_body = '\n'.join(lines[fn_node.lineno - 1:fn_node.end_lineno])
    return '\n'.join(import_lines) + '\n\n' + fn_body

Быстрая проверка

Итоги урока