Создание асинхронного помощника по билетам с использованием Agentic AI и Pydantic v2

Руководство по созданию асинхронного помощника по билетам на основе ИИ

В этом руководстве мы создадим помощника по билетам, использующего ИИ, с помощью библиотеки PydanticAI. Мы определим правила данных с помощью моделей Pydantic v2, сохраним билеты в базе данных SQLite и сгенерируем уникальные идентификаторы с помощью модуля uuid. Два агента, один для создания билетов и один для проверки статуса, будут использовать Google Gemini для интерпретации ваших запросов на естественном языке и вызова функций нашей базы данных.

Установка необходимых библиотек

Сначала обновите ваш pip и установите PydanticAI:

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

Настройка API ключа

Проверьте, установлен ли переменная окружения GEMINI_API_KEY. Если нет, введите ваш ключ API Google Gemini:

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

Установка дополнительных пакетов

Установите пакет nest_asyncio, который позволяет использовать асинхронные функции в средах, таких как Colab:

!pip install nest_asyncio

Создание базы данных

Настройте базу данных SQLite в памяти и определите таблицу для билетов:

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()

Определение зависимостей и моделей

Создайте класс зависимостей для передачи соединения с базой данных и определите модели для вывода:

@dataclass
class TicketingDependencies:
    db: sqlite3.Connection

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="Текущий статус билета")

Создание агента для создания билетов

Создайте агента, который будет регистрировать новые билеты:

create_agent = Agent(
    "google-gla:gemini-2.0-flash",
    deps_type=TicketingDependencies,
    output_type=CreateTicketOutput,
    system_prompt="Вы помощник по билетам. Используйте инструмент `create_ticket` для регистрации новых проблем."
)

@create_agent.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.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("Мой принтер на 3-м этаже показывает ошибку зажима бумаги.", deps=deps)
print("Созданный билет →", create_result.json(indent=2))

status_result = await status_agent(f"Какой статус у билета {create_result.ticket_id}? ", deps=deps)
print("Статус билета →", status_result.json(indent=2))

Заключение

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

Практические рекомендации

• Ищите процессы, которые можно автоматизировать, и моменты взаимодействия с клиентами, где ИИ может добавить наибольшую ценность.
• Определите важные ключевые показатели эффективности (KPI), чтобы убедиться, что ваши инвестиции в ИИ приносят положительный результат.
• Выбирайте инструменты, которые соответствуют вашим потребностям и позволяют настраивать их под ваши цели.
• Начните с небольшого проекта, собирайте данные о его эффективности и постепенно расширяйте использование ИИ в вашей работе.

Если вам нужна помощь в управлении ИИ в бизнесе, свяжитесь с нами по адресу hello@itinai.ru. Чтобы быть в курсе последних новостей ИИ, подписывайтесь на наш Telegram.

Посмотрите практический пример решения на основе ИИ: продажный бот, предназначенный для автоматизации взаимодействия с клиентами круглосуточно и управления взаимодействиями на всех этапах клиентского пути.