agent-lab-sdk

Agent Lab SDK

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

Установка

pip install agent_lab_sdk

Список изменений

Ознакомиться со списком изменений между версиями agent-lab-sdk можно по ссылке

Содержание

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

---

1. Модуль agent_lab_sdk.llm

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

from agent_lab_sdk.llm import get_model, RetryConfig

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

# Получить модель GigaChat (throttled по умолчанию), использует токен из окружения по умолчанию
model = get_model("chat")

# Получить модель GigaChat без throttled-обертки
model = get_model("chat", throttled=False)

# Получить модель EmbeddingsGigaChat (throttled по умолчанию), использует токен из окружения по умолчанию
model = get_model("embeddings")

# Получить модель EmbeddingsGigaChat без throttled-обертки
model = get_model("embeddings", throttled=False)

# Передача явных параметров GigaChat, токен из окружения не использует
model = get_model(
    access_token="YOUR_TOKEN",
    timeout=60,
    scope="GIGACHAT_API_CORP"
)

# Включить retry (по умолчанию выключен) и настроить backoff
model = get_model(
    "chat",
    retry=True,
    retry_config=RetryConfig(
        retry_attempts_count=5,
        wait_min=0.2,
        wait_max=8,
    ),
)

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

throttled включен по умолчанию и включает ограничения и метрики для GigaChat и GigaChatEmbeddings. Подробнее здесь

retry отключен по умолчанию. Для включения используйте retry=True, а параметры настраивайте через retry_attempts_count или retry_config=RetryConfig(...) (только для throttled-оберток).

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

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

1.4. Agent утилиты

Модуль agent_lab_sdk.llm предоставляет функции для обновления названия треда

from agent_lab_sdk.llm import patch_thread_name, apatch_thread_name

patch_thread_name("thread-id", "Новое название")
await apatch_thread_name("thread-id", "Новое название")

---

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

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

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

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"])

# Retry отключен по умолчанию, включайте явно при необходимости
chat_with_retry = ThrottledGigaChat(
    access_token=access_token,
    retry=True,
    retry_config=RetryConfig(retry_attempts_count=3, wait_min=0.5, wait_max=4),
)

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

2.2. Метрики

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

Все метрики семафоров экспортируются с label credential_id. Если credential не задан (локальный режим), используется значение credential_id="local".

---

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 V2 File Upload

Новый v2 API для загрузки файлов через Agent Gateway с поддержкой бинарных данных и автоматическим выбором сервиса хранения.

from agent_lab_sdk.storage import upload_file, FileUploadResponse

# Загрузка из байтов
with open("document.pdf", "rb") as f:
    file_bytes = f.read()
result: FileUploadResponse = upload_file("document.pdf", file_bytes)

# Результат - Pydantic модель с информацией о файле
print(f"File ID: {result.id}")
print(f"Absolute Path: {result.absolute_path}")
print(f"Storage: {result.storage}")

Переменные окружения для V2 Upload

4.3 AGW Checkpointer

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

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, Visibility
)

class AgentState(BaseModel):
    # Основное поле ввода
    query: Annotated[str, MainInput(placeholder="Введите ваш запрос")]
    
    # Строковое поле
    title: Annotated[str, StringInput(
        default="Без названия",
        title="Заголовок",
        description="Название для вашего запроса",
        visibility=Visibility.ALWAYS  # или visibility="always"
    )]
    
    # Массив строк
    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",
        view="button"  # или "dropzone" для drag-and-drop,
        max_size_mb=15.0 # применяем ограничение максимального размера для одного файла
    )]

    # Загрузка нескольких файлов
    attachments: Annotated[List[str], FilesInput(
        title="Прикрепленные файлы",
        file_extensions=".pdf,.csv,.xlsx",
        group="Файлы",
        view="dropzone"  # область перетаскивания файлов,
        max_size_mb=15.0 # применяем ограничение максимального размера для всех файлов
    )]

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

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

Используйте параметр 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="Дополнительно"
    )]

Управление видимостью полей

Параметр visibility контролирует, когда поле отображается в интерфейсе. Доступные значения:

from agent_lab_sdk.schema import Visibility

# Enum с тремя значениями:
Visibility.ALWAYS       # "always" - поле всегда доступно для ввода (по умолчанию)
Visibility.START        # "start" - поле доступно для ввода только при старте
Visibility.AFTER_START  # "after_start" - поле доступно для ввода после старта

Пример использования:

class AgentConfig(BaseModel):
    # Всегда доступно для ввода поле
    query: Annotated[str, MainInput(
        placeholder="Введите запрос",
        visibility=Visibility.ALWAYS
    )]

    # Поле доступно для ввода только при первом запуске
    api_key: Annotated[str, StringInput(
        title="API ключ",
        description="Ключ для доступа к внешнему API",
        visibility=Visibility.START
    )]

    # Поле появляется после первого сообщения
    session_id: Annotated[str, StringInput(
        title="ID сессии",
        description="Идентификатор текущей сессии",
        visibility=Visibility.AFTER_START,
        hidden=True
    )]

Можно также передавать строковые значения напрямую:

title: Annotated[str, StringInput(
    title="Заголовок",
    visibility="always"  # эквивалентно Visibility.ALWAYS
)]

5.2. LogMessage

LogMessage — вспомогательное сообщение для потоковой передачи логов из узлов LangGraph / LangChain. Экземпляры создаются как обычные сообщения чата, но получают тип log, поэтому фронтенд может отображать их отдельно от ответов модели.

• Импортируется из agent_lab_sdk.schema.
• По умолчанию наследуется от langchain.schema.AIMessage и устанавливает additional_kwargs={"type": "log"}.
• Если установить переменную окружения IS_LOG_MESSAGE_CUSTOM=true, будет использоваться наследник BaseMessage с явным типом log.

Переменная окружения IS_LOG_MESSAGE_CUSTOM на текущий момент установлена для всех агентов в значение true

Пример использования со StreamWriter

from langgraph.graph import MessagesState
from langgraph.types import StreamWriter
from agent_lab_sdk.schema import LogMessage

async def run(state: MessagesState, writer: StreamWriter) -> MessagesState:
    writer(LogMessage("Запускаю обработку запроса"))

    # ... полезная работа здесь ...

    writer(LogMessage("Обработка завершена"))
    return state

Вызов writer(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 .

7. Модуль agent_lab_sdk.tools

7.1 Search plugin

Позволяет использовать API search plugin + llm tool + лимитирование запросов

from agent_lab_sdk.tools.search_plugin.search_plugin import SearchPlugin, SearchMode
from agent_lab_sdk.tools.search_plugin.tools import get_search_plugin_tool, get_search_image_tool
from langgraph.prebuilt import ToolNode
from langgraph.graph import StateGraph
from agent_lab_sdk.llm import get_model

plugin = SearchPlugin()
retrieval_result = plugin.retrieval("Что такое SDK?", SearchMode.actual_info_web_search)
print(retrieval_result.documents, retrieval_result.sources, retrieval_result.images)

images_result = plugin.rambler_images_proxy("Логотип питона")
for img in images_result:
    print(img.url, img.image_link)

search_tool = get_search_plugin_tool()
image_tool = get_search_image_tool()
tools = [search_tool, image_tool]

llm = get_model().bind_tools(tools)

graph = StateGraph(State)
graph.add_node("tools", ToolNode(tools), metadata={'description': 'Вызов инструментов'})

Реализованы методы POST /retrieval, GET /rambler_proxy_images

Переменные окружения для Search plugin