Создание систем принятия решений на основе контрактов с PydanticAI

Понимание систем принятия решений на основе контрактов

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

Обеспечение логической последовательности

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

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

Мы настраиваем окружение выполнения, устанавливая необходимые библиотеки и настраивая асинхронное выполнение для Google Colab. Мы безопасно загружаем API-ключ OpenAI и обеспечиваем готовность среды к обработке асинхронных вызовов агентов. Это устанавливает стабильную основу для работы агента на основе контрактов без проблем, связанных с окружением.

!pip -q install -U pydantic-ai pydantic openai nest_asyncio
 
import os
import time
import asyncio
import getpass
from dataclasses import dataclass
from typing import List, Literal
 
import nest_asyncio
nest_asyncio.apply()
 
from pydantic import BaseModel, Field, field_validator
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider
 
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if not OPENAI_API_KEY:
   try:
       from google.colab import userdata
       OPENAI_API_KEY = userdata.get("OPENAI_API_KEY")
   except Exception:
       OPENAI_API_KEY = None
if not OPENAI_API_KEY:
   OPENAI_API_KEY = getpass.getpass("Enter OPENAI_API_KEY: ").strip()

Определение модели решений

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

class RiskItem(BaseModel):
   risk: str = Field(..., min_length=8)
   severity: Literal["low", "medium", "high"]
   mitigation: str = Field(..., min_length=12)
 
class DecisionOutput(BaseModel):
   decision: Literal["approve", "approve_with_conditions", "reject"]
   confidence: float = Field(..., ge=0.0, le=1.0)
   rationale: str = Field(..., min_length=80)
   identified_risks: List[RiskItem] = Field(..., min_length=2)
   compliance_passed: bool
   conditions: List[str] = Field(default_factory=list)
   next_steps: List[str] = Field(..., min_length=3)
   timestamp_unix: int = Field(default_factory=lambda: int(time.time()))
 
   @field_validator("confidence")
   @classmethod
   def confidence_vs_risk(cls, v, info):
       risks = info.data.get("identified_risks") or []
       if any(r.severity == "high" for r in risks) and v > 0.70:
           raise ValueError("confidence too high given high-severity risks")
       return v
 
   @field_validator("decision")
   @classmethod
   def reject_if_non_compliant(cls, v, info):
       if info.data.get("compliance_passed") is False and v != "reject":
           raise ValueError("non-compliant decisions must be reject")
       return v
 
   @field_validator("conditions")
   @classmethod
   def conditions_required_for_conditional_approval(cls, v, info):
       d = info.data.get("decision")
       if d == "approve_with_conditions" and (not v or len(v) < 2):
           raise ValueError("approve_with_conditions requires at least 2 conditions")
       if d == "approve" and v:
           raise ValueError("approve must not include conditions")
       return v

Контекстуализация агентских решений

Мы внедряем контекст предприятия через типизированный объект зависимости и初始化ируем агента PydanticAI на основе OpenAI. Мы настраиваем агента на производство только структурированных выходных данных решений, которые соответствуют заранее определенному контракту. Этот шаг формализует разделение между бизнес-контекстом и логикой модели.

@dataclass
class DecisionContext:
   company_policy: str
   risk_threshold: float = 0.6
 
model = OpenAIChatModel(
   "gpt-5",
   provider=OpenAIProvider(api_key=OPENAI_API_KEY),
)
 
agent = Agent(
   model=model,
   deps_type=DecisionContext,
   output_type=DecisionOutput,
   system_prompt="""
Вы — агент анализа корпоративных решений.
Вы должны оценить риск, соответствие и неопределенность.
Все выходы должны строго соответствовать схеме DecisionOutput.
"""
)

Реализация валидаторов управления

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

@agent.output_validator
def ensure_risk_quality(result: DecisionOutput) -> DecisionOutput:
   if len(result.identified_risks) < 2:
       raise ValueError("minimum two risks required")
   if not any(r.severity in ("medium", "high") for r in result.identified_risks):
       raise ValueError("at least one medium or high risk required")
   return result
 
@agent.output_validator
def enforce_policy_controls(result: DecisionOutput) -> DecisionOutput:
   policy = CURRENT_DEPS.company_policy.lower()
   text = (
       result.rationale
       + " ".join(result.next_steps)
       + " ".join(result.conditions)
   ).lower()
   if result.compliance_passed:
       if not any(k in text for k in ["encryption", "audit", "logging", "access control", "key management"]):
           raise ValueError("missing concrete security controls")
   return result

Выполнение рабочего процесса решений

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

async def run_decision():
   global CURRENT_DEPS
   CURRENT_DEPS = DecisionContext(
       company_policy=(
           "No deployment of systems handling personal data or transaction metadata "
           "without encryption, audit logging, and least-privilege access control."
       )
   )
   prompt = """
Запрос на решение:
Разверните панель аналитики клиентов с поддержкой ИИ, используя облачного поставщика.
Система обрабатывает поведение пользователей и метаданные транзакций.
Аудит логирования не реализован, и ключи, управляемые клиентами, не определены.
"""
   result = await agent.run(prompt, deps=CURRENT_DEPS)
   return result.output
 
decision = asyncio.run(run_decision())
 
from pprint import pprint
pprint(decision.model_dump())

Переход к надежным ИИ-решениям

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