alice-types

Pydantic модели для протокола Алисы.


Keywords
alice, pydantic, pypi, python, yandex
License
MIT
Install
pip install alice-types==1.0.1

Documentation

AliceTypes

Библиотека моделей Pydantic-V2 для валидации протокола Я.Диалогов

Установка:

pip install alice-types

Использование

Пример с FastApi
from alice_types.response import AliceResponse
from alice_types.request import AliceRequest
from fastapi import FastAPI

app = FastAPI()


@app.post(path="/")
async def handler(alice_request: AliceRequest):
    reply = AliceResponse()

    if alice_request.is_new_session():
        reply.response.text = "Привет, скажи что-нибудь и я это повторю"
        return reply

    reply.response.text = alice_request.request.original_utterance
    return reply
Пример с AIOHttp
from alice_types.response import AliceResponse
from alice_types.request import AliceRequest
from aiohttp import web


async def handler(request):
    body = await request.json()
    alice_request = AliceRequest.model_validate(body)

    reply = AliceResponse()

    if alice_request.is_new_session():
        reply.response.text = "Привет, скажи что-нибудь и я это повторю"
    else:
        reply.response.text = alice_request.request.original_utterance

    body = reply.model_dump()
    return web.json_response(body)


app = web.Application()
app.router.add_post('/', handler)

if __name__ == "__main__":
    web.run_app(app)

Расширение типов полей

В библиотеке имеются модели с динамически расширяемыми полями:

  • request.State
    • State.extend_session_model(model: BaseModel)
    • State.extend_user_model(model: BaseModel)
    • State.extend_application_model(model: BaseModel)
  • request.RequestPurchase.payload
    • RequestPurchase.extend_payload_model(model: BaseModel)

Ограничение размера хранилищ для модели ответа

В модели AliceResponse по умолчанию установлен стандартный размер хранилищ (1 КБ), который можно изменить при необходимости:

from alice_types.response import AliceResponse

AliceResponse.set_session_state_limit_size(1024 * 8)
AliceResponse.set_user_state_limit_size(1024 * 16)
AliceResponse.set_application_state_limit_size(1024 * 32)

Публичные методы

AliceRequest
  • AliceRequest.is_ping()
  • AliceRequest.is_new_session()
  • AliceRequest.authorization_is_completed()
from alice_types.response import AliceResponse, Response
from alice_types.request import AliceRequest


async def handler(alice_request: AliceRequest):
    if alice_request.is_ping():
        return AliceResponse(
            response=Response(
                text="pong"
            )
        )
    
    elif alice_request.is_new_session():
        return AliceResponse(
            response=Response(
                text="Привет, скажи что-нибудь и я это повторю"
            )
        )
    
    elif alice_request.authorization_is_completed():
        return AliceResponse(
            response=Response(
                text="Ты авторизован это хорошо 👍"
            )
        )
    ...
AliceRequest.request.nlu.entities
  • entities.get(entity_type: SlotsType | str) - Возвращает список сущностей заданного типа.
  • entity.available() - Возвращает список доступных атрибутов объекта, у которых значение не равно None. Метод доступен у всех сущностей.
  • EntityDatetime.to_datetime(timezone: pytz.BaseTzInfo | str | None = None)
from typing import List

from alice_types.request import AliceRequest, SlotsType, EntityFio, EntityDatetime


async def handler(alice_request: AliceRequest):
    fio_entities: List[EntityFio] = alice_request.request.nlu.entities.get(SlotsType.YANDEX_FIO)
    names = []
    for entity in fio_entities:
        if "first_name" in entity.value.available():
            names.append(entity.value.first_name)

    dates = []
    datetime_entities: List[EntityDatetime] = alice_request.request.nlu.entities.get(SlotsType.YANDEX_DATETIME)
    for entity in datetime_entities:
        dates.append(
            entity.to_datetime(
                timezone=alice_request.meta.timezone  # Default: None
            )
        )
    ...
AliceRequest.request.meta.interfaces
  • interfaces.has(interface: Union[InterfaceType, str]) - Проверяет, существует ли этот интерфейс.
  • interfaces.available() - Возвращает список доступных интерфейсов.
from alice_types.request import AliceRequest, InterfaceType


async def handler(alice_request: AliceRequest):
    if alice_request.meta.interfaces.has(InterfaceType.SCREEN):
        pass

Примеры