Реализация кодирования для создания надёжных агентских рабочих процессов с помощью PydanticAI

В этом руководстве мы создаём готовый к производству агентский рабочий процесс, в котором приоритет отдаётся надёжности за счёт обеспечения строгих, типизированных выходных данных на каждом шаге. Мы используем PydanticAI для определения чётких схем ответов, подключения инструментов через внедрение зависимостей и обеспечения безопасного взаимодействия агента с внешними системами, такими как база данных, без нарушения выполнения.

Установка и настройка

1. Используем `pip` для установки необходимых библиотек:
«`
!pip -q install «pydantic-ai-slim[openai]» pydantic
«`

2. Импортируем необходимые модули:
«`
import os, json, sqlite3
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Literal, Optional, List

from pydantic import BaseModel, Field, field_validator
from pydantic_ai import Agent, RunContext, ModelRetry
«`

3. Загружаем ключ API OpenAI в удобной для Colab форме:
«`
if not os.environ.get(«OPENAIAPIKEY»):
try:
from google.colab import userdata
os.environ[«OPENAIAPIKEY»] = (userdata.get(«OPENAIAPIKEY») or «»).strip()
except Exception:
pass

if not os.environ.get(«OPENAIAPIKEY»):
import getpass
os.environ[«OPENAIAPIKEY»] = getpass.getpass(«Paste your OPENAIAPIKEY: «).strip()

assert os.environ.get(«OPENAIAPIKEY»), «OPENAIAPIKEY is required.»
«`

Определение моделей данных

«`
Priority = Literal[«low», «medium», «high», «critical»]
ActionType = Literal[«createticket», «updateticket», «queryticket», «listopentickets», «noaction»]
Confidence = Literal[«low», «medium», «high»]

class TicketDraft(BaseModel):
title: str = Field(…, minlength=8, maxlength=120)
customer: str = Field(…, minlength=2, maxlength=60)
priority: Priority
category: Literal[«billing», «bug», «feature_request», «security», «account», «other»]
description: str = Field(…, minlength=20, maxlength=1000)
expectedoutcome: str = Field(…, minlength=10, max_length=250)

class AgentDecision(BaseModel):
action: ActionType
reason: str = Field(…, minlength=20, maxlength=400)
confidence: Confidence
ticket: Optional[TicketDraft] = None
ticket_id: Optional[int] = None
followupquestions: List[str] = Field(defaultfactory=list, maxlength=5)

@fieldvalidator(«followup_questions»)
@classmethod
def short_questions(cls, v):
for q in v:
if len(q) > 140:
raise ValueError(«Each follow-up question must be <= 140 characters.")
return v
«`

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

«`
@dataclass
class SupportDeps:
db: sqlite3.Connection
tenant: str
policy: dict

def utcnowiso() -> str:
return datetime.now(timezone.utc).isoformat()

def init_db() -> sqlite3.Connection:
conn = sqlite3.connect(«:memory:», checksamethread=False)
conn.execute(«»»
CREATE TABLE tickets (
id INTEGER PRIMARY KEY AUTOINCREMENT,
tenant TEXT NOT NULL,
title TEXT NOT NULL,
customer TEXT NOT NULL,
priority TEXT NOT NULL,
category TEXT NOT NULL,
description TEXT NOT NULL,
expected_outcome TEXT NOT NULL,
status TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
«»»)
conn.commit()
return conn

def seed_ticket(db: sqlite3.Connection, tenant: str, ticket: TicketDraft, status: str = «open») -> int:
now = utcnowiso()
cur = db.execute(
«»»
INSERT INTO tickets
(tenant, title, customer, priority, category, description, expectedoutcome, status, createdat, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
«»»,
(
tenant,
ticket.title,
ticket.customer,
ticket.priority,
ticket.category,
ticket.description,
ticket.expected_outcome,
status,
now,
now,
),
)
db.commit()
return int(cur.lastrowid)

def buildagent(modelname: str) -> Agent[SupportDeps, AgentDecision]:
agent = Agent(
f»openai:{model_name}»,
output_type=AgentDecision,
output_retries=2,
instructions=(
«You are a production support triage agent.\n»
«Return an output that matches the AgentDecision schema.\n»
«Use tools when you need DB state.\n»
«Never invent ticket IDs.\n»
«If the user intent is unclear, ask concise follow-up questions.\n»
),
)
«`

Мы создаём агентскую логику для сборки модели-независимого агента PydanticAI. Регистрируем типизированные инструменты для создания, запроса, обновления и перечисления билетов, позволяя агенту взаимодействовать с внешним состоянием контролируемым образом. Также обеспечиваем проверку выходных данных, чтобы агент мог самостоятельно исправлять ошибки, когда его решения нарушают бизнес-правила.

В заключение мы показали, как агент с типизированной безопасностью может рассуждать, вызывать инструменты, проверять свои выходные данные и восстанавливаться после ошибок без ручного вмешательства. Мы сохранили логику независимой от модели, что позволяет нам заменять базовые модели LLM, сохраняя при этом те же схемы и инструменты. Это критически важно для долгосрочной поддержки.

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

1. Какие библиотеки и инструменты используются для создания агентского рабочего процесса в статье?

В статье используется библиотека PydanticAI для определения чётких схем ответов, подключения инструментов через внедрение зависимостей и обеспечения безопасного взаимодействия агента с внешними системами. Также используются библиотеки sqlite3 для работы с базой данных, os для работы с операционной системой, json для работы с форматом данных JSON и другие.

2. Какие типы данных определены для работы с билетами в статье?

Для работы с билетами в статье определены следующие типы данных:
* Priority — приоритет билета (low, medium, high, critical).
* ActionType — тип действия (createticket, updateticket, queryticket, listopentickets, noaction).
* Confidence — уровень уверенности (low, medium, high).
* TicketDraft — модель данных для создания черновика билета.
* AgentDecision — модель данных для принятия решений агентом.

3. Как обеспечивается надёжность агентского рабочего процесса в статье?

Надёжность агентского рабочего процесса обеспечивается за счёт обеспечения строгих, типизированных выходных данных на каждом шаге. Также используется PydanticAI для определения чётких схем ответов, подключения инструментов через внедрение зависимостей и обеспечения безопасного взаимодействия агента с внешними системами. Кроме того, обеспечивается проверка выходных данных, чтобы агент мог самостоятельно исправлять ошибки, когда его решения нарушают бизнес-правила.

4. Какие шаги включает в себя установка и настройка для создания агентского рабочего процесса?

Установка и настройка для создания агентского рабочего процесса включают в себя следующие шаги:
* Установка необходимых библиотек с помощью pip.
* Импорт необходимых модулей.
* Загрузка ключа API OpenAI в удобной для Colab форме.
* Определение моделей данных для работы с билетами.
* Создание агента с использованием PydanticAI.

5. Как обеспечивается взаимодействие агента с внешними системами в статье?

Взаимодействие агента с внешними системами обеспечивается за счёт использования PydanticAI для подключения инструментов через внедрение зависимостей. Это позволяет агенту взаимодействовать с внешними системами, такими как база данных, контролируемым образом.

Источник