Code RAG: индексируем код с помощью Tree-sitter и AST

Столкнулись с ситуацией: в monorepo на 500 000 строк нужно найти функцию обработки платежей, но grep выдает сотни совпадений. RAG по кодовой базе решает эту проблему, но только если чанкинг сохраняет структуру кода. Мы в таких проектах используем комбинацию Tree-sitter и AST для синтаксического разбора и разбивки на логические единицы: функции, классы, модули. Каждый чанк обогащается метаданными — именем, сигнатурой, docstring, импортами и полным путем в модульной нотации. Это позволяет семантическому поиску находить именно ту единицу кода, которая нужна, а не случайный кусок текста.

Зачем preserve структуру кода при чанкинге?

Обычный документный RAG режет текст на абзацы. Для кода это не работает: разрыв между сигнатурой и телом функции убивает контекст. Код имеет иерархию — функция внутри класса, класс внутри модуля. Мы сохраняем эту иерархию в метаданных: модульный путь, строки начала и конца, список методов для класса, декораторы для функции. Это позволяет при поиске по запросу «как реализован X» получить именно ту единицу кода, где X определён.

Как мы реализуем code-aware парсинг?

Мы построили индексер на основе Tree-sitter. Он парсит код на 50+ языков и даёт синтаксическое дерево. Для каждого узла (функция, класс, метод) извлекаем:

• имя и сигнатуру,
• docstring (если есть),
• тело функции/класса,
• декораторы и аннотации,
• список импортов (до 10).

Например, для Python используем ast для точного выделения:

import ast
from tree_sitter import Language, Parser

class CodebaseIndexer:
    def __init__(self):
        # Tree-sitter для syntax-aware парсинга
        PY_LANGUAGE = Language('build/languages.so', 'python')
        self.parser = Parser()
        self.parser.set_language(PY_LANGUAGE)

    def extract_python_units(self, file_path: str) -> list[dict]:
        """Извлечение функций и классов как отдельных единиц индексации"""
        with open(file_path, 'r', encoding='utf-8') as f:
            source = f.read()

        try:
            tree = ast.parse(source)
        except SyntaxError:
            return [{'text': source, 'type': 'file', 'file': file_path}]

        units = []
        for node in ast.walk(tree):
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                # Получение исходного кода функции
                func_source = ast.get_source_segment(source, node)
                docstring = ast.get_docstring(node)

                units.append({
                    'type': 'function',
                    'name': node.name,
                    'file': file_path,
                    'line_start': node.lineno,
                    'line_end': node.end_lineno,
                    'text': func_source,
                    'docstring': docstring or '',
                    'decorators': [ast.unparse(d) for d in node.decorator_list],
                    'signature': self._get_signature(node)
                })

            elif isinstance(node, ast.ClassDef):
                class_source = ast.get_source_segment(source, node)
                docstring = ast.get_docstring(node)

                units.append({
                    'type': 'class',
                    'name': node.name,
                    'file': file_path,
                    'line_start': node.lineno,
                    'line_end': node.end_lineno,
                    'text': class_source,
                    'docstring': docstring or '',
                    'methods': [m.name for m in ast.walk(node)
                                if isinstance(m, ast.FunctionDef)]
                })

        return units

    def _get_signature(self, func_node: ast.FunctionDef) -> str:
        args = []
        for arg in func_node.args.args:
            annotation = f": {ast.unparse(arg.annotation)}" \
                        if arg.annotation else ""
            args.append(f"{arg.arg}{annotation}")

        return_type = f" -> {ast.unparse(func_node.returns)}" \
                     if func_node.returns else ""
        return f"def {func_node.name}({', '.join(args)}){return_type}"

Обогащение метаданными: почему это важно?

Просто разбить код на чанки недостаточно. Для качественного поиска каждый чанк надо обогатить: добавить имя, сигнатуру, docstring, импорты и полный путь в модульной нотации. Это превращает плоский текст в структурированный объект, который при векторизации даёт более точные эмбеддинги. Мы формируем rich_text — комбинацию всех метаданных, которая подаётся на вход модели эмбеддингов.

class CodeMetadataEnricher:
    def enrich(self, unit: dict) -> dict:
        unit = unit.copy()

        # Создание rich text для эмбеддинга
        # Комбинирование имени, сигнатуры, docstring и кода
        rich_text_parts = []

        if unit.get('name'):
            rich_text_parts.append(f"# {unit['name']}")

        if unit.get('signature'):
            rich_text_parts.append(f"Signature: {unit['signature']}")

        if unit.get('docstring'):
            rich_text_parts.append(f"Description: {unit['docstring']}")

        rich_text_parts.append(unit['text'])

        unit['rich_text'] = '\n\n'.join(rich_text_parts)

        # Извлечение импортов для контекста
        imports = re.findall(r'^(?:import|from)\s+\S+', unit['text'], re.MULTILINE)
        unit['imports'] = imports[:10]

        # Путь в виде breadcrumb
        parts = unit['file'].replace('\\', '/').split('/')
        unit['module_path'] = '.'.join(
            p.replace('.py', '') for p in parts if not p.startswith('.')
        )

        return unit

Индексация Git истории: что изменилось?

RAG по коду может отвечать не только на вопросы о структуре, но и об истории изменений. Мы индексируем последние 100 коммитов с diff и метаданными: автор, дата, сообщение, файлы. Это позволяет найти, когда и кем была изменена конкретная функция. Например, запрос «Кто правил calculate_total в прошлом месяце?» вернёт коммиты с этой функцией в diff.

import subprocess

class GitHistoryIndexer:
    def get_recent_changes(self, repo_path: str, n: int = 100) -> list[dict]:
        """Индексация последних коммитов с diff"""
        result = subprocess.run(
            ['git', 'log', f'-{n}', '--format=%H|%an|%ae|%ad|%s'],
            cwd=repo_path, capture_output=True, text=True
        )

        commits = []
        for line in result.stdout.strip().split('\n'):
            if not line:
                continue
            hash_, author, email, date, subject = line.split('|', 4)

            # Получение diff для этого коммита
            diff_result = subprocess.run(
                ['git', 'diff', f'{hash_}^', hash_, '--stat'],
                cwd=repo_path, capture_output=True, text=True
            )

            commits.append({
                'hash': hash_,
                'author': author,
                'date': date,
                'message': subject,
                'changes_summary': diff_result.stdout[:500],
                'text': f"Commit: {subject}\nAuthor: {author}\nDate: {date}\n\nChanges: {diff_result.stdout[:500]}"
            })

        return commits

Как оценить качество code RAG?

Хорошая метрика: при вопросе «Как реализован X?» система должна вернуть функцию или класс, который реализует X, а не просто файл с похожим названием. Для оценки мы используем golden set из 50–100 вопросов с известными ответами (конкретными функциями). Precision@3 > 0.8 — хороший результат. Ниже — сравнение стратегий чанкинга:

Смешанный чанкинг даёт выигрыш в точности в 2 раза по сравнению с файловым. Мы используем именно этот подход: каждый чанк — функция или класс, а файл становится метаданным.

Какая модель эмбеддингов подходит для кода?

Для кода лучше использовать модели, обученные на программном коде, а не на общих текстах. Ниже — сравнение популярных вариантов:

Типичные ошибки при индексации кода

• Игнорирование docstring — без них модель не понимает назначение функции, recall падает на 30%.
• Чанкинг по строкам — разрывает логические блоки, precision снижается вдвое.
• Отсутствие метаданных — только код без имени и сигнатуры даёт эмбеддинг, похожий на случайный кусок текста.
• Пропуск Git-истории — теряется информация об авторстве и контексте изменений.
• Выбор не той модели эмбеддингов — модель для документов плохо работает на коде.

Что входит в работу?

• Аудит кодовой базы: оценка размера, языков, структуры репозитория.
• Проектирование пайплайна: выбор инструментов (Tree-sitter, векторная БД, модель эмбеддингов), настройка метаданных.
• Реализация индексации: написание парсера, обогащение, векторизация, загрузка в векторную БД.
• Тестирование: проверка на golden set, итеративное улучшение чанкинга и метаданных.
• Интеграция: настройка API для поиска, интеграция с IDE, чат-ботами или внутренними инструментами.
• Деплой и мониторинг: развёртывание, логирование, метрики качества (precision, recall, latency p99).

Сроки и результаты

Ориентировочные сроки — от 2 до 4 недель в зависимости от размера кодовой базы и сложности интеграции. Результаты: полностью индексированная кодовая база с code-aware чанкингом, API для семантического поиска, документация и обучение команды (1–2 часа), поддержка в течение месяца после сдачи.

Наш опыт — 5 лет на рынке, более 20 реализованных RAG-проектов для fintech, edtech и e-commerce. Гарантируем качество: precision@3 не ниже 0.8 на вашем golden set. Свяжитесь с нами — оценим проект за 1 день и предложим архитектуру вашего code RAG. Получите консультацию по оптимизации уже на первом созвоне.

Data Engineering для ML: пайплайны, разметка и качество данных

«У нас много данных» — фраза, которая на деле часто означает «у нас много сырых логов в S3, которые никто не трогал два года». Перед тем как обучить модель, нужно понять, что вообще есть: какова структура, есть ли дубли, как часто меняется схема, насколько репрезентативна выборка.

Data Engineering для ML — не просто ETL. Это построение воспроизводимой инфраструктуры данных, которая делает обучение моделей надёжным, а переобучение — предсказуемым. По опыту нашей команды (8 лет в дата-инжиниринге, более 30 проектов в ML) каждая вторая проблема в продакшене связана не с архитектурой модели, а с качеством данных.

ETЛ-пайплайны для ML: чем отличаются от BI

ETL для аналитики и ETL для ML — разные задачи. В аналитике важна агрегация, в ML — индивидуальные записи с историей. В аналитике train/val/test split не нужен, в ML — критичен. В аналитике skew данных мешает интерпретации, в ML — напрямую влияет на качество модели.

Инструменты. Apache Spark (Wikipedia) для больших объёмов (10GB+): PySpark с DataFrames, оптимизации через partitioning и caching. dbt для трансформаций поверх DWH (Snowflake, BigQuery, Redshift) — декларативно, версионируется, тестируется. Pandas + Polars для объёмов до нескольких GB — Polars в 5-10x быстрее Pandas на типичных трансформациях.

Temporal splits. Для ML важно, что split по времени, а не случайный. Если данные временные (транзакции, события пользователей), случайный split даёт data leakage: модель видит «будущие» данные при обучении. Правило: train на периоде T1-T2, validation на T2-T3 (с gap для предотвращения leakage), test на T3-T4. Неправильный split может стоить 10–15% качества модели на валидации.

Инкрементальные пайплайны. Модель переобучается еженедельно на новых данных. Нужен пайплайн, который инкрементально добавляет новые записи к обучающей выборке, не перегружая всё с нуля. Delta Lake или Apache Iceberg — форматы с ACID-транзакциями, Change Data Capture, time travel.

Как избежать training-serving skew с помощью Feature Store

Feature Store решает проблему рассинхронизации между обучением и инференсом. Самая коварная ошибка в ML-инфраструктуре — training-serving skew: признак считается по-разному в обучении и в продакшене. Модель учится на «правильных» данных, а инференс получает другие.

Feast (open source) — офлайн store на Parquet/Delta в S3 для обучения, онлайн store на Redis для low-latency инференса (<10ms). Feature definitions как Python-код:

from feast import FeatureView, Field
from feast.types import Float32, Int64

user_features = FeatureView(
    name="user_features",
    entities=["user_id"],
    schema=[
        Field(name="purchase_count_7d", dtype=Int64),
        Field(name="avg_session_duration", dtype=Float32),
    ],
    ttl=timedelta(days=7),
    source=user_features_source,
)

Один definition, используется везде. Нет расхождений.

Потоковые признаки. Когда признак должен обновляться в реальном времени (количество транзакций за последние 10 минут), нужна потоковая обработка. Apache Kafka + Apache Flink или Kafka Streams для вычисления признаков в реальном времени → запись в онлайн store. Сложнее, дороже, нужно только когда staleness признаков критична для качества.

Разметка данных: как не потратить бюджет впустую

Разметка — самая трудоёмкая и недооцениваемая часть ML-проекта. Плохо размеченные данные не исправит никакая архитектура.

Label Studio — open source, поддерживает разметку изображений (bounding box, polygon, segmentation), текста (NER, классификация), аудио, видео. Поднимается за 10 минут через Docker. Для небольших команд — первый выбор.

Оценка качества разметки. Inter-annotator agreement — насколько согласны разметчики между собой. Cohen's Kappa > 0.8 — хорошо, 0.6-0.8 — приемлемо, < 0.6 — задача неоднозначна или инструкция плохая. Пересечение разметок (10-20% примеров размечают два независимых аннотатора) — обязательная практика.

Active learning. Не размечать случайные примеры, а выбирать те, на которых модель наиболее неуверена (low confidence, high uncertainty). Позволяет добиться того же качества при 50-70% объёма разметки. Modals, Prodigy, Label Studio поддерживают active learning workflows. На одном из проектов для NLP мы сократили бюджет на разметку в 2,5 раза за счёт active learning.

Синтетические данные. Когда реальных данных мало или получить их дорого. Для CV: рендеринг в Blender/Unity с реалистичными текстурами (domain randomization). Для NLP: parafrase через LLM, backtranslation. Риск: модель обучается на distribution синтетических данных, а не реальных — нужна осторожность и проверка на реальном holdout.

Качество данных: валидация и мониторинг

Great Expectations — de facto стандарт для data validation в ML-пайплайнах. Expectations — это декларативные утверждения о данных: «колонка age содержит значения от 0 до 120», «колонка user_id не содержит null», «распределение amount не отклоняется более чем на 20% от baseline». Запускается в пайплайне, при провале — блокирует прохождение.

Pandera — Pythonic alternative для pandas/polars DataFrames. Schema-based validation с type hints:

import pandera as pa

schema = pa.DataFrameSchema({
    "user_id": pa.Column(int, nullable=False),
    "score": pa.Column(float, pa.Check.between(0, 1)),
    "label": pa.Column(str, pa.Check.isin(["positive", "negative", "neutral"])),
})

Data freshness. Модель ожидает данные за последние N дней. ETL упал, данные не обновились — модель использует устаревшие признаки. Мониторинг свежести данных: timestamp последней записи в каждой таблице, алерт при задержке > порога.

Дедупликация. Дубликаты в обучающей выборке завышают метрики (одни и те же примеры в train и val) и искажают веса модели. MinHash LSH для приближённой дедупликации больших датасетов. Для точной — хэш по нормализованному контенту.

Инструменты валидации: сравнение

Хранилища и форматы

Для большинства ML-проектов на старте: Parquet в S3 + DVC для версионирования. Delta Lake или Iceberg — когда появляется потребность в инкрементальных обновлениях или time travel.

Что входит в проект по дата-инжинирингу для ML

Мы предоставляем полный цикл:

• Аудит существующих данных и пайплайнов (1 неделя).
• Проектирование архитектуры: выбор инструментов, форматов, способов разметки.
• Реализация ETL/ELT пайплайна с валидацией и мониторингом.
• Документация кода и процессов (model card, data card).
• Обучение вашей команды работе с пайплайном.
• SLA на сопровождение и поддержку.

Как мы строим пайплайн: пошагово

1. Аудит существующих данных. Профилирование: ydata-profiling (бывший pandas-profiling) генерирует HTML-репорт со статистиками, дистрибуциями, корреляциями, missing values за минуты.
2. Проектирование пайплайна. Определяем источники данных, частоту обновления, требования к latency признаков, объёмы.
3. Реализация и тестирование. Unit-тесты на трансформации, integration-тесты на пайплайн, data validation через Great Expectations.
4. Деплой и мониторинг. Алерты на freshness, quality checks, аномалии в объёмах данных.

Почему стоит доверить это нам

Мы занимаемся дата-инжинирингом и ML с 2016 года. За это время реализовали более 40 проектов — от построения пайплайнов для NLP-моделей до разметки датасетов для компьютерного зрения. Гарантируем воспроизводимость пайплайнов и полную прозрачность процессов. В каждом проекте используем инструменты с открытым исходным кодом, чтобы вы не были привязаны к вендору.

Свяжитесь с нами для бесплатного аудита ваших данных — оценим текущий пайплайн и предложим roadmap. Закажите построение ML-пайплайна под ключ.