MAESTRO

Описание методологии

Цепочки рассуждений — это структурированный подход к формализации экспертного мышления, основанный на триаде Event-Action-Result. Методология позволяет преобразовать сложные мыслительные процессы экспертов в понятную для LLM форму, создавая основу для интеллектуальных агентных систем.

Event обозначает исходное событие, факт или действие, которое инициирует процесс рассуждения. Action отражает ответную деятельность системы или эксперта, включая уточнения, анализ и назначение процедур. Result показывает конечный вывод на каждом шаге, будь то полученная информация, промежуточное заключение или корректировка стратегии.

Библиотека MMAR CARL

Для практической реализации цепочек рассуждений разработана специализированная библиотека MMAR CARL (Collaborative Agent Reasoning Library) — универсальный инструмент для построения систем экспертного мышления с поддержкой:

• RAG-подобного извлечения контекста — автоматическое извлечение релевантной информации из входных данных для каждого шага рассуждений
• Параллельного выполнения на основе DAG — автоматическая оптимизация последовательности выполнения с учётом зависимостей между шагами
• Мультиязычности — встроенная поддержка русского и английского языков
• Универсальной архитектуры — применимость к любой предметной области

Ключевые компоненты

• StepDescription — формализует шаг рассуждений с указанием цели, вопросов для анализа, запросов для извлечения контекста и зависимостей от предыдущих шагов.
• ReasoningChain — управляет последовательностью выполнения шагов с автоматической параллелизацией независимых операций.
• ReasoningContext — содержит исходные данные, конфигурацию LLM и историю выполнения.

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

import asyncio
from mmar_carl import (
    ReasoningChain, StepDescription, ReasoningContext,
    Language, ContextSearchConfig
)
from mmar_llm import EntrypointsAccessor, EntrypointsConfig
import json

# Создание EntrypointsAccessor
def create_entrypoints(entrypoints_path: str):
    with open(entrypoints_path, encoding="utf-8") as f:
        config_data = json.load(f)
    entrypoints_config = EntrypointsConfig.model_validate(config_data)
    return EntrypointsAccessor(entrypoints_config)

# Определение цепочки медицинских рассуждений
CLINICAL_REASONING = [
    StepDescription(
        number=1,
        title="Сбор жалоб и анамнеза",
        aim="Систематизировать клиническую картину",
        reasoning_questions="Какие симптомы указывают на возможные диагнозы?",
        step_context_queries=[
            "основные жалобы",
            "длительность симптомов",
            "факторы риска"
        ],
        stage_action="Формирование предварительных гипотез",
        example_reasoning="Характер и динамика симптомов определяют направление диагностического поиска"
    ),
    StepDescription(
        number=2,
        title="Анализ объективных данных",
        aim="Оценить результаты обследований",
        reasoning_questions="Какие объективные признаки подтверждают гипотезы?",
        dependencies=[1],  # Зависит от шага 1
        step_context_queries=[
            "результаты осмотра",
            "лабораторные показатели",
            "инструментальные данные"
        ],
        stage_action="Сопоставить субъективные и объективные данные",
        example_reasoning="Лабораторные показатели коррелируют с клинической картиной"
    ),
    StepDescription(
        number=3,
        title="Формирование диагностического заключения",
        aim="Установить окончательный диагноз",
        reasoning_questions="Какой диагноз наиболее вероятен?",
        dependencies=[1, 2],  # Зависит от шагов 1 и 2
        step_context_queries=[
            "клинические рекомендации",
            "дифференциальная диагностика",
            "критерии постановки диагноза"
        ],
        stage_action="Синтезировать диагностическое заключение",
        example_reasoning="Диагноз устанавливается на основе совокупности клинических и лабораторных данных"
    )
]

# Конфигурация поиска контекста
search_config = ContextSearchConfig(
    strategy="vector",  # Семантический поиск
    vector_config={
        "similarity_threshold": 0.75,
        "max_results": 5
    }
)

# Создание цепочки рассуждений
chain = ReasoningChain(
    steps=CLINICAL_REASONING,
    search_config=search_config,
    max_workers=2,
    enable_progress=True
)

# Медицинские данные пациента
patient_data = """
Пациент: мужчина, 45 лет
Основные жалобы: боли в грудной клетке давящего характера,
одышка при физической нагрузке
Длительность симптомов: 2 недели
Факторы риска: курение 20 лет, артериальная гипертензия
Результаты осмотра: АД 160/95, ЧСС 88, хрипы в лёгких отсутствуют
Лабораторные показатели: холестерин ЛПНП 4.9 ммоль/л, тропонин отрицательный
Инструментальные данные: ЭКГ - признаки гипертрофии левого желудочка
"""

# Инициализация контекста
entrypoints = create_entrypoints("entrypoints.json")
context = ReasoningContext(
    outer_context=patient_data,
    entrypoints=entrypoints,
    entrypoint_key="gigachat-2-max",
    language=Language.RUSSIAN,
    retry_max=3
)

# Выполнение цепочки рассуждений
result = chain.execute(context)

# Получение результатов
print("=== Диагностическое заключение ===")
print(result.get_final_output())

print("\n=== Результаты по шагам ===")
for step_num, step_result in result.step_results.items():
    print(f"\nШаг {step_num}: {step_result}")

Конфигурация поиска контекста

CARL поддерживает два режима извлечения релевантной информации:

Подстроковый поиск

from mmar_carl import ContextSearchConfig, ReasoningChain

search_config = ContextSearchConfig(
    strategy="substring",
    substring_config={
        "case_sensitive": False,  # Регистронезависимый поиск
        "min_word_length": 3,     # Минимальная длина слова
        "max_matches_per_query": 5  # Максимум результатов на запрос
    }
)

chain = ReasoningChain(
    steps=steps,
    search_config=search_config
)

Векторный поиск

search_config = ContextSearchConfig(
    strategy="vector",
    vector_config={
        "similarity_threshold": 0.7,
        "max_results": 5
    }
)

Преимущества использования CARL

• 🎯 Формализация экспертного мышления — структурированное представление сложных рассуждений
• ⚡ Автоматическая оптимизация — параллельное выполнение независимых шагов
• 🔍 Интеллектуальное извлечение контекста — RAG-подобный поиск релевантной информации
• 🌍 Многоязычность — встроенная поддержка русского и английского
• 🏗️ Универсальность — применимость к любой предметной области (медицина, юриспруденция, финансы)
• ⚙️ Готовность к продуктовой разработке — обработка ошибок, повторные попытки, мониторинг

Пример 1: Базовая обработка результата

result = chain.execute(context)

# Проверка успешности
if result.success:
    print("✓ Цепочка выполнена успешно")
    print(f"Общее время: {result.total_execution_time:.2f}с")
    print(f"\nФинальный результат:\n{result.get_final_output()}")
else:
    print("✗ Выполнение завершилось с ошибками")
    for failed_step in result.get_failed_steps():
        print(f"  Шаг {failed_step.step_number}: {failed_step.error_message}")
                                            

Пример 2: Детальный анализ выполнения

result = chain.execute(context)

print("=== СТАТИСТИКА ВЫПОЛНЕНИЯ ===")
print(f"Всего шагов: {len(result.step_results)}")
print(f"Успешных: {len(result.get_successful_steps())}")
print(f"Неуспешных: {len(result.get_failed_steps())}")
print(f"Общее время: {result.total_execution_time:.2f}с")

# Детальная информация по каждому шагу
print("\n=== РЕЗУЛЬТАТЫ ПО ШАГАМ ===")
for step in result.step_results:
    status = "✓ УСПЕХ" if step.success else "✗ ОШИБКА"
    print(f"\n[{status}] Шаг {step.step_number}: {step.step_title}")
    print(f"Время: {step.execution_time:.2f}с")

    if step.success:
        # Показать первые 200 символов результата
        preview = step.result[:200] + "..." if len(step.result) > 200 else step.result
        print(f"Результат: {preview}")
    else:
        print(f"Ошибка: {step.error_message}")
                                            

Пример 3: Работа с метаданными

result = chain.execute(context)

# Получение статистики выполнения
if "execution_stats" in result.metadata:
    stats = result.metadata["execution_stats"]
    print("=== СТАТИСТИКА ПАРАЛЛЕЛИЗАЦИИ ===")
    print(f"Параллельных батчей: {stats['parallel_batches']}")
    print(f"Всего шагов: {stats['total_steps']}")
    print(f"Выполнено: {stats['executed_steps']}")
    print(f"Ошибок: {stats['failed_steps']}")
                                            

Пример 4: Доступ к истории выполнения

result = chain.execute(context)

# Получить историю как список
for i, entry in enumerate(result.history, 1):
    print(f"\n--- Запись {i} ---")
    print(entry)

# Или как одну строку
full_text = result.get_full_output()
                                            

Пример 5: Обработка ошибок

result = chain.execute(context)

if not result.success:
    print("⚠ Обнаружены ошибки в выполнении цепочки")

    # Получить все неуспешные шаги
    failed_steps = result.get_failed_steps()

    for step in failed_steps:
        print(f"\nОшибка на шаге {step.step_number}: {step.step_title}")
        print(f"Сообщение: {step.error_message}")

        # Вывести последнюю успешную запись в истории
        if step.updated_history:
            print(f"Последняя история: {step.updated_history[-1][:100]}...")

    # Получить частичные результаты успешных шагов
    successful_steps = result.get_successful_steps()
    print(f"\nУспешно завершено {len(successful_steps)} из {len(result.step_results)} шагов")
                                            

Пример 6: Интеграция с системой логирования

import logging

logger = logging.getLogger(__name__)

result = chain.execute(context)

# Логирование результатов
logger.info(f"Chain execution completed: success={result.success}")
logger.info(f"Total execution time: {result.total_execution_time:.2f}s")

for step in result.step_results:
    if step.success:
        logger.debug(f"Step {step.step_number} succeeded in {step.execution_time:.2f}s")
    else:
        logger.error(f"Step {step.step_number} failed: {step.error_message}")

# Сохранение финального результата
logger.info(f"Final output: {result.get_final_output()}")
                                            

Описание

AuthService — это полнофункциональный модуль аутентификации, который предоставляет гибкую систему авторизации для различных типов клиентов. Сервис поддерживает два основных сценария работы: авторизация конечных пользователей через email и пароль, и авторизация клиентских приложений через пару client_id/client_secret.

Для защиты от автоматических атак и брутфорса система интегрирована с Google reCAPTCHA, которая может быть включена или отключена в конфигурации. Все пароли хэшируются с использованием современного алгоритма хэширования перед сохранением в базу данных.

Сервис использует JWT (JSON Web Tokens) для управления сессиями пользователей. Токены содержат полезную нагрузку с информацией о клиенте, пользователе и типе авторизации, а также временем expiration. Каждый токен сохраняется в базе данных для возможности принудительного завершения сессий.

Основные методы API

class AuthService:
    async def register(self, data: V1RegisterRequest, external_id: str | None = None) -> list
    async def login(self, form: LoginForm) -> V1JWTResponse
    async def login_client(self, form: ClientLoginForm) -> V1JWTResponse
    async def identify(self, auth_header: str) -> dict
    async def get_info_about_me(self, auth_header: str) -> V1Me
    async def refresh(self, auth_header: str) -> V1JWTResponse
    async def logout(self, auth_header: str) -> V1JWTResponse

Типы авторизации

Сервис поддерживает два механизма авторизации:

• Bearer Token — для авторизации пользователей через JWT-токены
• Basic Auth — для авторизации клиентских приложений через client_id/client_secret

Пользовательская авторизация:

# Заголовок для пользовательской авторизации
Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...

Клиентская авторизация:

# Заголовок для клиентской авторизации  
Authorization: Basic Y2xpZW50X2lkOmNsaWVudF9zZWNyZXQ=

Меры безопасности

• Хэширование паролей — все пароли хэшируются перед сохранением
• reCAPTCHA — защита от автоматической регистрации и брутфорса
• JWT с expiration — ограниченное время жизни токенов
• Валидация email — приведение email к нижнему регистру
• Проверка активности пользователя — только активные пользователи могут авторизоваться
• Client Secret Hole — дополнительный механизм для генерации client_secret на основе client_id

Модели данных

class V1RegisterRequest:
    email: str
    password: str
    full_name: str
    where_from: str
    why_access: str
    captcha: str
    client_id: str

class LoginForm:
    email: str
    password: str

class ClientLoginForm:
    client_id: str
    client_secret: str

class V1JWTResponse:
    access_token: str
    token_type: str = "bearer"

Обработка ошибок

Сервис выбрасывает стандартизированные исключения:

• FailedToRegisterException — ошибка регистрации (непройденная капча, некорректные данные)
• InvalidCredentialsException — неверные учетные данные
• TokenExpiredException — истек срок действия токена
• InvalidTokenException — некорректный формат токена

Все исключения автоматически преобразуются в соответствующие HTTP-статусы (400, 401) при использовании в FastAPI.

Описание

FLAME: Flexible LLM-Assisted Moderation Engine — это современное решение для эффективной и гибкой модерации контента в больших языковых моделях, также предложенное нашей командой. Традиционные системы модерации сосредоточены на фильтрации входных запросов. В отличие от них FLAME анализирует выходные данные модели, что позволяет надежнее защищаться от jailbreak-атак, включая методы типа Best-of-N. Система легко настраивается под конкретные нужды, позволяя определять запрещенные темы и оперативно обновлять правила модерации без необходимости сложного переобучения.

Принцип работы FLAME основан на использовании n-грамм и правил для классификации текста. Система преобразует сообщения в нормализованные n-граммы и сравнивает их с заранее подготовленным с помощью LLM списком запрещенных фраз. Благодаря легковесной архитектуре, FLAME требует минимальных вычислительных ресурсов — до 0.1 ядра CPU и 100 МБ оперативной памяти при работе в 1 поток. Это позволяет обрабатывать сообщения всего за 2-5 мс, обеспечивая высокую производительность даже в условиях интенсивной нагрузки.

Сервис уже успел пройти проверку на тестах и реальное внедрение в СберЗдоровье, СберМедИИ и ЦРТ Сбера. FLAME демонстрирует хорошие метрики: точность (precision) достигает 98.7%, а полнота (recall) — 90.9%. В ходе испытаний система снизила успешность jailbreak-атак в 2-9 раз по сравнению со встроенными механизмами модерации таких моделей, как GPT-4o-mini, DeepSeek-v3 и других. Низкий уровень ложных срабатываний и устойчивость к современным угрозам делают FLAME надежным решением для модерации в чат-системах и других LLM-приложениях.

API сервиса (доступен по GRPC протоколу)

class BinaryClassifiersAPI:
def get_classifiers(self) -> list[str]:
    raise NotImplementedError

def evaluate(self, *, classifier: str | None = None, text: str) -> bool:
    raise NotImplementedError
                                

Детали использования

Для вызова надо выбрать классификатор. Список классификаторов можно получить от сервиса:

classifiers: list[str] = moderator.get_classifiers()

На данный момент доступны классификаторы:

• black — черный список (политика, религия, одиозные личности)
• greet — приветствия
• receipt — рецептурные лекарства
• child — дети

Расширение классификаторов и создание новых

Для пополнения списков запрещённых фраз в FLAME используется специальный пайплайн генерации: с помощью немодерируемого LLM создаются многочисленные вариации сообщений по заданным запрещённым темам, которые затем разбиваются на n-граммы (до 3 слов), нормализуются и фильтруются -- в финальный чёрный список попадают только те n-граммы, которые встречаются достаточно часто и при этом не вызывают ложных срабатываний на коллекции безопасных диалогов. Тот же пайплайн можно использовать и для создания классификаторов по другим темам.