Интеграция OpenAI Agents SDK для построения агентов

Интеграция OpenAI Agents SDK

OpenAI Agents SDK — официальный Python SDK для построения агентов на базе моделей OpenAI. Предоставляет абстракции над Assistants API и Responses API: Agent (обёртка над моделью с инструментами), Runner (выполнение агента), Handoffs (передача между агентами), Guardrails (проверки), трассировка. SDK заменяет прямые вызовы Assistants API более типизированным и тестируемым интерфейсом.

Базовый агент

# pip install openai-agents
import asyncio
from openai import AsyncOpenAI
from agents import Agent, Runner, function_tool, RunConfig
from agents.models.openai_responses import OpenAIResponsesModel

client = AsyncOpenAI()

# Инструменты через декоратор
@function_tool
def get_weather(city: str, unit: str = "celsius") -> str:
    """Получить текущую погоду в городе.

    Args:
        city: Название города
        unit: Единица температуры (celsius/fahrenheit)
    """
    data = weather_api.get(city=city, unit=unit)
    return f"Погода в {city}: {data['temp']}°, {data['description']}"

@function_tool
def create_calendar_event(
    title: str,
    date: str,
    duration_minutes: int,
    attendees: list[str],
) -> str:
    """Создать событие в корпоративном календаре."""
    event = calendar_api.create(
        title=title,
        date=date,
        duration=duration_minutes,
        attendees=attendees,
    )
    return f"Событие создано: {event['id']}, ссылка: {event['meet_link']}"

# Создание агента
assistant = Agent(
    name="Corporate Assistant",
    instructions="""Ты — корпоративный ассистент.
Помогай сотрудникам планировать встречи, находить информацию, решать задачи.
При необходимости — используй инструменты.""",
    model="gpt-4o",
    tools=[get_weather, create_calendar_event],
)

# Запуск
async def main():
    result = await Runner.run(
        assistant,
        input="Запланируй встречу с командой на завтра в 14:00 на 1 час",
    )
    print(result.final_output)

asyncio.run(main())

Handoffs: передача между специализированными агентами

from agents import Agent, handoff, Runner

# Специализированные агенты
triage_agent = Agent(
    name="Triage",
    instructions="""Классифицируй запрос и передай соответствующему агенту.
Billing → billing_agent
Technical → tech_agent
General → general_agent""",
    model="gpt-4o-mini",
)

billing_agent = Agent(
    name="Billing Support",
    instructions="Помогай с вопросами выставления счётов, платежами, подписками.",
    model="gpt-4o",
    tools=[get_invoice, process_refund, update_payment_method],
)

tech_agent = Agent(
    name="Technical Support",
    instructions="Решай технические проблемы: API, интеграции, ошибки.",
    model="gpt-4o",
    tools=[check_api_status, get_error_logs, create_bug_report],
)

general_agent = Agent(
    name="General Support",
    instructions="Отвечай на общие вопросы о продукте.",
    model="gpt-4o-mini",
    tools=[search_docs],
)

# Настройка handoffs для triage
triage_agent.handoffs = [
    handoff(billing_agent, tool_name_override="transfer_to_billing"),
    handoff(tech_agent, tool_name_override="transfer_to_technical"),
    handoff(general_agent, tool_name_override="transfer_to_general"),
]

# При запуске: triage автоматически решает, кому передать
result = await Runner.run(
    triage_agent,
    input="Я получил двойное списание за прошлый месяц",
)
# triage → billing_agent → ответ

Guardrails: входные и выходные фильтры

from agents import Agent, InputGuardrail, OutputGuardrail, GuardrailFunctionOutput

async def pii_detection_guardrail(ctx, agent, input) -> GuardrailFunctionOutput:
    """Проверяет входящий запрос на наличие PII"""
    pii_check_agent = Agent(
        name="PII Checker",
        instructions="Проверь, содержит ли текст персональные данные (номера карт, паспортов, СНИЛС).",
        model="gpt-4o-mini",
        output_type={"contains_pii": bool, "pii_types": list[str]},
    )

    result = await Runner.run(pii_check_agent, input=input)
    contains_pii = result.final_output.get("contains_pii", False)

    return GuardrailFunctionOutput(
        output_info=result.final_output,
        tripwire_triggered=contains_pii,  # Блокируем при наличии PII
    )

async def content_safety_guardrail(ctx, agent, output) -> GuardrailFunctionOutput:
    """Проверяет выходной текст на compliance"""
    violations = await compliance_checker.check(output)
    return GuardrailFunctionOutput(
        output_info=violations,
        tripwire_triggered=len(violations) > 0,
    )

# Агент с guardrails
safe_agent = Agent(
    name="Safe Assistant",
    instructions="Отвечай на вопросы о финансовых продуктах.",
    model="gpt-4o",
    input_guardrails=[InputGuardrail(guardrail_function=pii_detection_guardrail)],
    output_guardrails=[OutputGuardrail(guardrail_function=content_safety_guardrail)],
)

Structured Output с типизацией

from pydantic import BaseModel
from typing import Literal

class CustomerAnalysis(BaseModel):
    customer_id: str
    churn_risk: Literal["low", "medium", "high"]
    churn_probability: float
    key_risk_factors: list[str]
    recommended_actions: list[str]
    priority_contact: bool

analysis_agent = Agent(
    name="Churn Analyst",
    instructions="""Анализируй данные клиента и оцени риск оттока.
Учитывай: активность за последние 30 дней, NPS, количество обращений в поддержку,
использование функций продукта.""",
    model="gpt-4o",
    tools=[get_customer_activity, get_support_history, get_product_usage],
    output_type=CustomerAnalysis,
)

result = await Runner.run(
    analysis_agent,
    input=f"Проанализируй клиента ID: {customer_id}",
)
analysis: CustomerAnalysis = result.final_output
print(f"Риск оттока: {analysis.churn_risk} ({analysis.churn_probability:.0%})")

Трассировка и мониторинг

from agents.tracing import set_tracing_provider
from agents.tracing.opentelemetry import OpenTelemetryTracingProvider
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

# Интеграция с OpenTelemetry
otlp_exporter = OTLPSpanExporter(endpoint="http://jaeger:4318/v1/traces")
set_tracing_provider(OpenTelemetryTracingProvider(exporter=otlp_exporter))

# Или встроенная трассировка OpenAI
from agents import trace

async def run_with_trace():
    with trace("customer_support_session"):
        result = await Runner.run(
            triage_agent,
            input="Проблема с подключением к API",
            run_config=RunConfig(
                trace_include_sensitive_data=False,  # Не логируем PII
                workflow_name="customer_support",
            ),
        )
    return result

Практический кейс: мульти-агентная система онбординга SaaS

Задача: автоматизация онбординга новых B2B клиентов. Типичный онбординг занимал 2 недели и требовал координации 3 отделов.

Архитектура агентов:

• Onboarding Coordinator (triage): принимает запрос, маршрутизирует по этапам
• Account Setup Agent: настройка аккаунта, роли, SSO
• Integration Agent: помощь с API-интеграцией, генерация примеров кода
• Training Agent: персонализированный обучающий контент
• Success Agent: follow-up, мониторинг adoption metrics

Handoff-цепочка: Coordinator → Account Setup → Integration → Training → Success

Результаты:

• Time-to-value (первая успешная интеграция): 14 дней → 3 дня
• Вопросы техподдержки в первые 30 дней: -54%
• 30-day activation rate: 61% → 84%
• Engagement score за первый месяц: +31%

Сроки

• Базовый агент с инструментами: 2–4 дня
• Handoff-архитектура с 3–5 агентами: 1–2 недели
• Guardrails и safety checks: 3–5 дней
• Трассировка и мониторинг: 3–5 дней
• Production-деплой: 1 неделя