OpenAI Agents SDK

W marcu 2025 OpenAI oficjalnie wydalo Agents SDK — lekki framework do budowania agentow AI w Python. To nastepca eksperymentalnego Swarm, ale z produkcyjnym wsparciem, lepsza dokumentacja i natywna integracja z API OpenAI. W tym artykule poznasz fundamenty SDK i zbudujesz swojego pierwszego agenta.

Czym jest Agents SDK?

Agents SDK to oficjalny framework OpenAI do budowy agentow. W przeciwienstwie do surowego API, SDK oferuje:

Instalacja

# Instalacja przez pip
pip install openai-agents

# Lub z dodatkowymi zalelznosciami
pip install openai-agents[voice]  # wsparcie dla voice

# Konfiguracja klucza API
export OPENAI_API_KEY="sk-..."

# Sprawdz wersje
pip show openai-agents

Kluczowe koncepcje

Agents SDK opiera sie na pieciu glownych komponentach:

Komponent	Opis	Analogia
Agent	LLM + instrukcje + narzedzia. Podstawowa jednostka wykonawcza.	Pracownik z opisem stanowiska
Runner	Orkiestruje wykonanie agenta, zarzadza petla agent→tool→agent.	Menedzer projektu
Tools	Funkcje Python wywolywane przez agenta (function calling).	Narzedzia w skrzynce
Handoffs	Przekazywanie kontroli miedzy agentami.	Przekierowanie do specjalisty
Guardrails	Walidacja wejscia/wyjscia, filtrowanie tresci.	Ochroniarz przy wejsciu

Pierwszy agent

Najprostszy agent sklada sie z nazwy, instrukcji i modelu:

from agents import Agent, Runner

# Definicja agenta
agent = Agent(
    name="Asystent",
    instructions="""Jestes pomocnym asystentem. Odpowiadaj zwiezle po polsku.
    
Zasady:
- Badz konkretny i rzeczowy
- Jesli nie znasz odpowiedzi, powiedz to
- Uzywaj przyjaznego tonu""",
    model="gpt-4o"
)

# Uruchomienie synchroniczne
result = Runner.run_sync(agent, "Czym jest Python?")
print(result.final_output)

# Lub asynchroniczne
import asyncio

async def main():
    result = await Runner.run(agent, "Czym jest Python?")
    print(result.final_output)

asyncio.run(main())

Dodawanie narzedzi (Tools)

Narzedzia to zwykle funkcje Python z dekoratorem @function_tool. SDK automatycznie generuje schemat JSON na podstawie typow i docstringa:

from agents import Agent, Runner, function_tool

# Proste narzedzie
@function_tool
def get_weather(city: str) -> str:
    """Pobiera aktualna pogode dla miasta.
    
    Args:
        city: Nazwa miasta (np. "Warszawa", "Krakow")
    
    Returns:
        Opis pogody w formacie tekstowym
    """
    # W produkcji: wywolanie API pogodowego
    weather_data = {
        "Warszawa": "slonecznie, 22C",
        "Krakow": "pochmurno, 18C",
        "Gdansk": "deszczowo, 15C"
    }
    return f"Pogoda w {city}: {weather_data.get(city, 'brak danych')}"

# Narzedzie z walidacja
@function_tool
def calculate(expression: str) -> str:
    """Oblicza wyrazenie matematyczne.
    
    Args:
        expression: Wyrazenie do obliczenia, np. "2 + 2", "15 * 7"
    
    Returns:
        Wynik obliczenia lub komunikat o bledzie
    """
    try:
        # Bezpieczna ewaluacja (w produkcji uzyj ast.literal_eval lub sympy)
        allowed_chars = set("0123456789+-*/(). ")
        if not all(c in allowed_chars for c in expression):
            return "Blad: niedozwolone znaki"
        result = eval(expression)
        return f"Wynik: {result}"
    except Exception as e:
        return f"Blad w wyrazeniu: {str(e)}"

# Narzedzie z wyszukiwaniem
@function_tool
def search_database(query: str, limit: int = 5) -> str:
    """Przeszukuje baze danych produktow.
    
    Args:
        query: Fraza wyszukiwania
        limit: Maksymalna liczba wynikow (domyslnie 5)
    
    Returns:
        Lista znalezionych produktow
    """
    # Symulacja bazy danych
    products = ["Laptop Dell", "iPhone 15", "Sluchawki Sony"]
    results = [p for p in products if query.lower() in p.lower()][:limit]
    return f"Znaleziono: {results}" if results else "Brak wynikow"

# Agent z narzedziami
agent = Agent(
    name="Asystent",
    instructions="""Pomagasz uzytkownikom. Uzywaj narzedzi gdy potrzeba:
- get_weather: do sprawdzania pogody
- calculate: do obliczen matematycznych  
- search_database: do szukania produktow

Zawsze uzywaj odpowiedniego narzedzia zamiast zgadywac.""",
    tools=[get_weather, calculate, search_database]
)

# Test
result = Runner.run_sync(agent, "Jaka jest pogoda w Warszawie i ile to 15% z 250?")
print(result.final_output)

Handoffs — przekazywanie miedzy agentami

Handoffs pozwalaja agentowi przekazac kontrole do innego, wyspecjalizowanego agenta. To kluczowy wzorzec dla systemow multi-agent:

from agents import Agent, Runner

# Agent specjalista od sprzedazy
sales_agent = Agent(
    name="Sales",
    instructions="""Jestes specjalista ds. sprzedazy. Pomagasz w:
- Informacjach o cenach i promocjach
- Procesie zakupu
- Porownywaniu produktow
- Rabatach i ofertach specjalnych

Badz przyjazny i nastawiony na pomoc w zakupie.""",
    model="gpt-4o"
)

# Agent specjalista od supportu technicznego
support_agent = Agent(
    name="Support",
    instructions="""Jestes specjalista ds. wsparcia technicznego. Pomagasz z:
- Problemami technicznymi
- Konfiguracja produktow
- Rozwiazywaniem bledow
- Gwarancja i zwrotami

Badz cierpliwy i prowadz uzytkownika krok po kroku.""",
    model="gpt-4o"
)

# Agent specjalista od faktur
billing_agent = Agent(
    name="Billing",
    instructions="""Jestes specjalista ds. rozliczen. Pomagasz z:
- Fakturami i platnosciami
- Historia zamowien
- Zmiana danych do faktury
- Reklamacje platnosci

Badz precyzyjny i dbaj o bezpieczenstwo danych.""",
    model="gpt-4o"
)

# Agent glowny (triage) - kieruje do specjalistow
triage_agent = Agent(
    name="Triage",
    instructions="""Jestes pierwsza linia kontaktu. Twoim zadaniem jest:

1. Zrozumiec potrzebe uzytkownika
2. Przekazac do odpowiedniego specjalisty:
   - Pytania o ceny, zakupy, promocje → Sales
   - Problemy techniczne, konfiguracja → Support  
   - Faktury, platnosci, historia zamowien → Billing

3. Jesli nie jestes pewien, zapytaj uzytkownika o wiecej szczegolow.

NIE probuj sam rozwiazywac specjalistycznych pytan - przekaz do eksperta.""",
    handoffs=[sales_agent, support_agent, billing_agent]
)

# Test roznych scenariuszy
scenarios = [
    "Ile kosztuje nowy laptop?",
    "Moj produkt sie zepsul, co robic?",
    "Potrzebuje faktury za ostatnie zamowienie"
]

for query in scenarios:
    print(f"\n{'='*50}")
    print(f"User: {query}")
    result = Runner.run_sync(triage_agent, query)
    print(f"Response: {result.final_output}")

Guardrails — walidacja

Guardrails dzialaja rownolegle z agentem i moga blokowac niepozadane tresci na wejsciu lub wyjsciu:

from agents import (
    Agent, Runner, 
    InputGuardrail, OutputGuardrail,
    GuardrailFunctionOutput
)
from pydantic import BaseModel

# Model wyniku moderacji
class ModerationResult(BaseModel):
    is_appropriate: bool
    reason: str
    confidence: float

# Agent moderujacy (guardrail)
moderator = Agent(
    name="Moderator",
    instructions="""Ocen czy wiadomosc jest odpowiednia. Blokuj:
- Spam i reklamy
- Obrazliwe tresci
- Prosby o nielegalne dzialania
- Dane osobowe (PESEL, numery kart)

Zwroc JSON z polami: is_appropriate, reason, confidence""",
    output_type=ModerationResult,
    model="gpt-4o-mini"  # szybszy model dla guardrail
)

# Funkcja guardrail dla wejscia
async def input_moderation(ctx, agent, input_data):
    """Moderuje wiadomosci wejsciowe."""
    result = await Runner.run(moderator, input_data)
    output = result.final_output
    
    return GuardrailFunctionOutput(
        should_block=not output.is_appropriate,
        message=output.reason if not output.is_appropriate else None
    )

# Prostsza walidacja - bez LLM
async def simple_length_check(ctx, agent, input_data):
    """Sprawdza dlugosc wiadomosci."""
    if len(input_data) > 5000:
        return GuardrailFunctionOutput(
            should_block=True,
            message="Wiadomosc zbyt dluga (max 5000 znakow)"
        )
    return GuardrailFunctionOutput(should_block=False)

# Guardrail dla wyjscia - sprawdza odpowiedzi agenta
async def output_pii_check(ctx, agent, output_data):
    """Blokuje odpowiedzi z danymi osobowymi."""
    import re
    
    # Proste wzorce PII
    patterns = [
        r'\b\d{11}\b',  # PESEL
        r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b',  # karta
    ]
    
    for pattern in patterns:
        if re.search(pattern, output_data):
            return GuardrailFunctionOutput(
                should_block=True,
                message="Odpowiedz zawiera dane osobowe"
            )
    return GuardrailFunctionOutput(should_block=False)

# Glowny agent z guardrails
main_agent = Agent(
    name="Assistant",
    instructions="Pomagasz uzytkownikom. Odpowiadaj rzeczowo.",
    input_guardrails=[
        InputGuardrail(guardrail_function=simple_length_check),
        InputGuardrail(guardrail_function=input_moderation)
    ],
    output_guardrails=[
        OutputGuardrail(guardrail_function=output_pii_check)
    ]
)

# Test
try:
    result = Runner.run_sync(main_agent, "Normalne pytanie")
    print(result.final_output)
except Exception as e:
    print(f"Zablokowano: {e}")

Kontekst i pamiec

Agents SDK umozliwia przekazywanie kontekstu miedzy wywolaniami:

from agents import Agent, Runner
from dataclasses import dataclass

# Kontekst uzytkownika
@dataclass
class UserContext:
    user_id: str
    name: str
    preferences: dict
    history: list

# Agent z dynamicznymi instrukcjami
def create_personalized_agent(ctx: UserContext) -> Agent:
    return Agent(
        name="PersonalAssistant",
        instructions=f"""Jestes osobistym asystentem dla {ctx.name}.

Preferencje uzytkownika:
- Jezyk: {ctx.preferences.get('language', 'polski')}
- Styl: {ctx.preferences.get('style', 'formalny')}

Historia ostatnich interakcji:
{chr(10).join(ctx.history[-5:])}

Dostosuj odpowiedzi do preferencji uzytkownika.""",
        model="gpt-4o"
    )

# Uzycie
user_ctx = UserContext(
    user_id="user-123",
    name="Jan",
    preferences={"language": "polski", "style": "nieformalny"},
    history=["Pytal o pogode", "Szukal restauracji"]
)

agent = create_personalized_agent(user_ctx)
result = Runner.run_sync(agent, "Co mi polecasz na dzisiaj?")
print(result.final_output)

Streaming odpowiedzi

from agents import Agent, Runner

agent = Agent(
    name="Writer",
    instructions="Piszesz dlugie, szczegolowe odpowiedzi.",
    model="gpt-4o"
)

# Streaming synchroniczny
for event in Runner.run_streamed_sync(agent, "Napisz krotkie opowiadanie"):
    if event.type == "raw_response_event":
        if hasattr(event.data, 'delta') and event.data.delta:
            print(event.data.delta, end="", flush=True)

# Streaming asynchroniczny
async def stream_response():
    async for event in Runner.run_streamed(agent, "Napisz wiersz"):
        if event.type == "raw_response_event":
            if hasattr(event.data, 'delta') and event.data.delta:
                print(event.data.delta, end="", flush=True)

Tracing i debugowanie

SDK automatycznie loguje wykonanie do OpenAI Dashboard lub mozna uzyc wlasnego tracera:

from agents import Runner, trace
from agents.tracing import set_tracing_export_api_key

# Wlacz eksport do OpenAI Dashboard
set_tracing_export_api_key("sk-...")

# Trace z nazwa workflow
with trace("customer-support-workflow"):
    result = Runner.run_sync(agent, "Zapytanie uzytkownika")

# Trace z metadata
with trace("order-processing", metadata={"user_id": "123", "order_id": "456"}):
    result = Runner.run_sync(agent, "Sprawdz status zamowienia")

# Trace dostepny w OpenAI Dashboard: platform.openai.com

Structured Output

Agent moze zwracac typowane dane zamiast tekstu:

from agents import Agent, Runner
from pydantic import BaseModel
from typing import Optional
from enum import Enum

class Priority(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

class TicketAnalysis(BaseModel):
    """Analiza zgloszenia klienta."""
    category: str
    priority: Priority
    sentiment: str
    summary: str
    suggested_action: str
    requires_human: bool
    confidence: float

# Agent z structured output
analyst = Agent(
    name="TicketAnalyst",
    instructions="""Analizujesz zgloszenia klientow. Dla kazdego okresl:
- category: typ problemu (technical, billing, sales, complaint)
- priority: low/medium/high/critical
- sentiment: positive/neutral/negative
- summary: krotkie podsumowanie (max 100 znakow)
- suggested_action: co zrobic
- requires_human: czy wymaga czlowieka
- confidence: pewnosc analizy 0.0-1.0""",
    output_type=TicketAnalysis,
    model="gpt-4o"
)

# Test
ticket = "Moj komputer sie zepsul i stracilam wszystkie dane! Potrzebuje pomocy TERAZ!"
result = Runner.run_sync(analyst, ticket)

# result.final_output to teraz TicketAnalysis
analysis: TicketAnalysis = result.final_output
print(f"Kategoria: {analysis.category}")
print(f"Priorytet: {analysis.priority}")
print(f"Wymaga czlowieka: {analysis.requires_human}")

Kiedy uzywac Agents SDK?

Podsumowanie

Funkcja	API	Opis
Tworzenie agenta	Agent(name, instructions, tools)	Podstawowy agent
Uruchomienie sync	Runner.run_sync(agent, input)	Synchroniczne wykonanie
Uruchomienie async	await Runner.run(agent, input)	Asynchroniczne wykonanie
Narzedzie	@function_tool	Dekorator dla funkcji
Handoff	Agent(handoffs=[…])	Lista agentow do przekazania
Guardrail	InputGuardrail, OutputGuardrail	Walidacja wejscia/wyjscia
Structured output	Agent(output_type=Model)	Typowane odpowiedzi