agent-lab-sdk 0.1.17

SDK для работы с Agent Lab

Agent Lab SDK

Набор утилит и обёрток для упрощённой работы с LLM, Agent Gateway и метриками в проектах Giga Labs.

Установка

pip install agent_lab_sdk

Содержание

1. Модуль agent_lab_sdk.llm
2. Модуль agent_lab_sdk.llm.throttled
3. Модуль agent_lab_sdk.metrics
4. Хранилище
5. Схема
6. Сборка и публикация

---

1. Модуль agent_lab_sdk.llm

1.1. Получение модели

from agent_lab_sdk.llm import get_model

# Использует токен из окружения по умолчанию
model = get_model()

# Передача явных параметров
model = get_model(
    access_token="YOUR_TOKEN",
    timeout=60,
    scope="GIGACHAT_API_CORP"
)

если не передавать access_token, токен будет выбран через GigaChatTokenManager

1.2. Менеджеры токенов

Класс	Описание	Пример использования
AgwTokenManager	Кеширование + получение токена через Agent Gateway	token = AgwTokenManager.get_token("provider")
GigaChatTokenManager	Кеширование + получение через GigaChat OAuth с использованием пользовательских секретов	token = GigaChatTokenManager.get_token()

1.3. Переменные окружения

Переменная	Описание	Значение по умолчанию / Пример
GIGACHAT_SCOPE	Scope GigaChat API	GIGACHAT_API_PERS
GLOBAL_GIGACHAT_TIMEOUT	Таймаут запросов к GigaChat (секунды)	120
USE_TOKEN_PROVIDER_AGW	Использовать AgwTokenManager	true
GIGACHAT_CREDENTIALS	Базовые креды для GigaChat (b64(clientId:secretId))	Y2xpZW50SWQ6c2VjcmV0SWQ=
GIGACHAT_TOKEN_PATH	Путь к файлу кеша токена GigaChat	/tmp/gigachat_token.json
GIGACHAT_TOKEN_FETCH_RETRIES	Количество попыток получения токена (GigaChat)	3
USE_GIGACHAT_ADVANCED	Включает запрос токена GigaChat API в продвинутом режиме	true
GIGACHAT_BASE_URL	Базовый URL GigaChat для расширенного режима	https://ngw.devices.sberbank.ru:9443/api/v2/oauth
TOKEN_PROVIDER_AGW_URL	URL Agent Gateway для получения AGW-токена	https://agent-gateway.apps.advosd.sberdevices.ru
TOKEN_PROVIDER_AGW_DEFAULT_MAX_RETRIES	Макс. попыток запроса токена (AGW)	3
TOKEN_PROVIDER_AGW_TIMEOUT_SEC	Таймаут запроса к AGW (секунды)	5

---

2. Модуль agent_lab_sdk.llm.throttled

Позволяет ограничивать число одновременных вызовов к GigaChat и сервису эмбеддингов, автоматически собирая соответствующие метрики.

from agent_lab_sdk.llm import GigaChatTokenManager
from agent_lab_sdk.llm.throttled import ThrottledGigaChat, ThrottledGigaChatEmbeddings

access_token = GigaChatTokenManager.get_token()

# Чат с учётом ограничений
chat = ThrottledGigaChat(access_token=access_token)
response = chat.invoke("Привет!")

# Эмбеддинги с учётом ограничений
emb = ThrottledGigaChatEmbeddings(access_token=access_token)
vectors = emb.embed_documents(["Text1", "Text2"])

2.1. Переменные окружения для ограничения

Переменная	Описание	Значение по умолчанию
MAX_CHAT_CONCURRENCY	Максимум одновременных чат-запросов	100000
MAX_EMBED_CONCURRENCY	Максимум одновременных запросов эмбеддингов	100000
EMBEDDINGS_MAX_BATCH_SIZE_PARTS	Макс. размер батча частей для эмбеддингов	90

2.2. Метрики

Метрики доступны через agent_lab_sdk.metrics.get_metric:

Метрика	Описание	Тип
chat_slots_in_use	Число занятых слотов для чата	Gauge
chat_waiting_tasks	Число задач, ожидающих освобождения слота чата	Gauge
chat_wait_time_seconds	Время ожидания слота чата (секунды)	Histogram
embed_slots_in_use	Число занятых слотов для эмбеддингов	Gauge
embed_waiting_tasks	Число задач, ожидающих слота эмбеддингов	Gauge
embed_wait_time_seconds	Время ожидания слота эмбеддингов (секунды)	Histogram

---

3. Модуль agent_lab_sdk.metrics

Предоставляет удобный интерфейс для создания и управления метриками через Prometheus-клиент.

3.1. Основные функции

from agent_lab_sdk.metrics import get_metric

# Создать метрику
g = get_metric(
    metric_type="gauge",               # тип: "gauge", "counter" или "histogram"
    name="my_gauge",                   # имя метрики в Prometheus
    documentation="Моя метрика gauge"  # описание
)

# Увеличить счётчик
g.inc()

# Установить конкретное значение
g.set(42)

3.2. Пример использования в коде

from agent_lab_sdk.metrics import get_metric
import time

# Счётчик HTTP-запросов с метками
reqs = get_metric(
    metric_type="counter",
    name="http_requests_total",
    documentation="Всего HTTP-запросов",
    labelnames=["method", "endpoint"]
)
reqs.labels("GET", "/api").inc()

# Гистограмма задержек
lat = get_metric(
    metric_type="histogram",
    name="http_request_latency_seconds",
    documentation="Длительность HTTP-запроса",
    buckets=[0.1, 0.5, 1.0, 5.0]
)
with lat.time():
    time.sleep(0.5)

print(reqs.collect())
print(lat.collect())

4. Хранилище

4.1 SD Ассетница

функция store_file_in_sd_asset сохраняет base64‑файл в хранилище S3 и отдаёт публичную ссылку на файл

from agent_lab_sdk.storage import store_file_in_sd_asset

store_file_in_storage("my-agent-name-filename.png", file_b64, "giga-agents")

4.2 AGW Checkpointer

AGW поддерживает langgraph checkpoint API и в SDK представлен AsyncAGWCheckpointSaver, который позволяет сохранять состояние графа в AGW напрямую.

5. Схема

5.1. Типы входных данных

Модуль agent_lab_sdk.schema.input_types предоставляет фабричные функции для создания аннотированных типов полей, которые могут использоваться в Pydantic моделях для описания интерфейса агентов.

Основные типы полей

from typing import List, Annotated
from pydantic import BaseModel, Field
from agent_lab_sdk.schema import (
    MainInput, StringInput, StringArrayInput, NumberInput,
    SelectInput, CheckboxInput, FileInput, FilesInput, SelectOption
)

class AgentState(BaseModel):
    # Основное поле ввода
    query: Annotated[str, MainInput(placeholder="Введите ваш запрос")]
    
    # Строковое поле
    title: Annotated[str, StringInput(
        default="Без названия",
        title="Заголовок",
        description="Название для вашего запроса"
    )]
    
    # Массив строк
    keywords: Annotated[List[str], StringArrayInput(
        placeholder="Добавьте ключевые слова...",
        title="Ключевые слова",
        description="Список ключевых слов для поиска",
        group="Параметры"
    )]
    
    # Числовое поле
    temperature: Annotated[float, NumberInput(
        default=0.7,
        title="Температура",
        description="Параметр креативности модели (0.0 - 1.0)",
        hidden=True
    )]
    
    # Выпадающий список
    mode: Annotated[str, SelectInput(
        title="Режим работы",
        items=[
            SelectOption(label="Быстрый", value="fast").model_dump(),
            SelectOption(label="Точный", value="precise").model_dump()
        ],
        default="fast",
        group="Настройки"
    )]
    
    # Чекбокс
    save_history: Annotated[bool, CheckboxInput(
        title="Сохранять историю",
        description="Сохранять диалог для последующего анализа",
        default=True,
        group="Опции"
    )]
    
    # Загрузка одного файла
    document: Annotated[str, FileInput(
        title="Документ",
        file_extensions=".pdf,.docx,.txt"
    )]
    
    # Загрузка нескольких файлов
    attachments: Annotated[List[str], FilesInput(
        title="Прикрепленные файлы",
        file_extensions=".pdf,.csv,.xlsx",
        group="Файлы"
    )]

Доступные фабричные функции

Тип	Описание	Основные параметры
MainInput	Основное поле ввода	placeholder
StringInput	Текстовое поле	default, title, description, hidden
StringArrayInput	Массив строк	placeholder, title, description, group, hidden
NumberInput	Числовое поле	default, title, description, hidden
SelectInput	Выпадающий список	items, title, group, default, hidden
CheckboxInput	Чекбокс	title, group, description, default, hidden
FileInput	Загрузка одного файла	title, file_extensions, group, hidden
FilesInput	Загрузка нескольких файлов	title, file_extensions, group, hidden

Группировка полей

Используйте параметр group для логической группировки полей в интерфейсе:

class TaskConfig(BaseModel):
    # Группа "Основные параметры"
    task_type: Annotated[str, SelectInput(
        title="Тип задачи",
        items=[...],
        group="Основные параметры"
    )]
    
    priority: Annotated[str, SelectInput(
        title="Приоритет",
        items=[...],
        group="Основные параметры"
    )]
    
    # Группа "Дополнительно"
    notifications: Annotated[bool, CheckboxInput(
        title="Уведомления",
        group="Дополнительно"
    )]
    
    tags: Annotated[List[str], StringArrayInput(
        placeholder="Теги...",
        group="Дополнительно"
    )]

5.2. LogMessage

TODO: описание LogMessage

6. Сборка и публикация

1. Установка twine

pip install --upgrade build twine

2. Собрать и загрузить в pypi

перед обновлением сборки нужно не забыть поменять версию в pyproject.toml

python -m build && python -m twine upload dist/*

3. Ссылка на проект pypi

https://pypi.org/project/agent-lab-sdk/

4. установка локально в editable mode. Предварительно может потребоваться выбрать необходимое окружение

pip install -e .