Создание асинхронного AI-помощника для тикетов с PydanticAI, Pydantic v2 и SQLite

Обзор AI-помощника для обработки тикетов

В этом руководстве показано, как создать полноценного помощника для работы с тикетами на базе Agentic AI и библиотеки PydanticAI. Помощник работает асинхронно, обрабатывая создание и проверку статуса тикетов через команды на естественном языке с поддержкой Google Gemini.

Настройка окружения

Для начала обновим pip и установим PydanticAI — библиотеку для создания типобезопасных AI-агентов с интеграцией Pydantic моделей и LLM.

!pip install --upgrade pip
!pip install pydantic-ai

Далее безопасно вводим API-ключ Google Gemini для аутентификации:

import os
from getpass import getpass
 
if "GEMINI_API_KEY" not in os.environ:
    os.environ["GEMINI_API_KEY"] = getpass("Введите ваш Google Gemini API ключ: ")

Устанавливаем nest_asyncio для корректной работы асинхронных функций в средах вроде Colab:

!pip install nest_asyncio

База данных и модели данных

Создаем в памяти SQLite базу для хранения тикетов с полями ID, описанием, уровнем важности, отделом и статусом.

import sqlite3
conn = sqlite3.connect(":memory:")
conn.execute("""
CREATE TABLE tickets (
    ticket_id TEXT PRIMARY KEY,
    summary   TEXT NOT NULL,
    severity  TEXT NOT NULL,
    department TEXT NOT NULL,
    status    TEXT NOT NULL
)
""")
conn.commit()

Определяем две модели Pydantic v2 для создания тикета и вывода статуса, что обеспечивает проверку и типобезопасность данных.

from pydantic import BaseModel, Field
from typing import Literal
 
class CreateTicketOutput(BaseModel):
    ticket_id: str = Field(..., description="Уникальный идентификатор тикета")
    summary: str   = Field(..., description="Описание проблемы")
    severity: Literal["low","medium","high"] = Field(..., description="Уровень срочности")
    department: str = Field(..., description="Ответственный отдел")
    status: Literal["open"] = Field("open", description="Начальный статус тикета")
 
class TicketStatusOutput(BaseModel):
    ticket_id: str = Field(..., description="Уникальный идентификатор тикета")
    status: Literal["open","in_progress","resolved"] = Field(..., description="Текущий статус тикета")

Создание агентов

Создаем два агента PydanticAI: для создания тикетов и для получения их статуса. Оба используют модель Google Gemini.

Агент создания тикета

from pydantic_ai import Agent, RunContext
import uuid
 
create_agent = Agent(
    "google-gla:gemini-2.0-flash",
    deps_type=TicketingDependencies,
    output_type=CreateTicketOutput,
    system_prompt="Вы помощник по тикетам. Используйте инструмент `create_ticket` для создания новых обращений."
)
 
@create_agent.tool
async def create_ticket(
    ctx: RunContext[TicketingDependencies],
    summary: str,
    severity: Literal["low","medium","high"],
    department: str
) -> CreateTicketOutput:
    tid = str(uuid.uuid4())
    ctx.deps.db.execute(
        "INSERT INTO tickets VALUES (?,?,?,?,?)",
        (tid, summary, severity, department, "open")
    )
    ctx.deps.db.commit()
    return CreateTicketOutput(
        ticket_id=tid,
        summary=summary,
        severity=severity,
        department=department,
        status="open"
    )

Агент получения статуса

status_agent = Agent(
    "google-gla:gemini-2.0-flash",
    deps_type=TicketingDependencies,
    output_type=TicketStatusOutput,
    system_prompt="Вы помощник по тикетам. Используйте инструмент `get_ticket_status` для получения текущего статуса."
)
 
@status_agent.tool
async def get_ticket_status(
    ctx: RunContext[TicketingDependencies],
    ticket_id: str
) -> TicketStatusOutput:
    cur = ctx.deps.db.execute(
        "SELECT status FROM tickets WHERE ticket_id = ?", (ticket_id,)
    )
    row = cur.fetchone()
    if not row:
        raise ValueError(f"Тикет с ID {ticket_id!r} не найден")
    return TicketStatusOutput(ticket_id=ticket_id, status=row[0])

Запуск агентов

Упаковываем соединение с базой в объект зависимостей и запускаем агентов с запросами на естественном языке.

deps = TicketingDependencies(db=conn)
 
create_result = await create_agent.run(
    "Мой принтер на 3 этаже выдает ошибку застревания бумаги.", deps=deps
)
print("Созданный тикет →")
print(create_result.output.model_dump_json(indent=2))
 
tid = create_result.output.ticket_id
status_result = await status_agent.run(
    f"Каков статус тикета {tid}?", deps=deps
)
print("Статус тикета →")
print(status_result.output.model_dump_json(indent=2))

Этот пример показывает, как Agentic AI и PydanticAI совместно автоматизируют процесс управления тикетами с помощью типобезопасных моделей и легковесной базы данных. Систему легко расширять, добавляя новые агенты, AI-модели или интеграции с API, сохраняя структуру и надежность данных.