# Создаём и активируем виртуальное окружение
python3 -m venv agents-env
source agents-env/bin/activate

# Устанавливаем OpenAI Agents SDK
pip install openai-agents

# Дополнительно: трейсинг и визуализация
pip install openai-agents[tracing]

# Установка API-ключа
export OPENAI_API_KEY="sk-your-key-here"

# Проверка установки
python -c "from agents import Agent; print('SDK OK')"

import asyncio
from agents import Agent, Runner, function_tool

@function_tool
def get_weather(city: str) -> str:
    """Получение текущей погоды в городе. Используй, когда пользователь спрашивает о погоде."""
    # Имитация API-запроса к сервису погоды
    weather_data = {
        "москва": "☀️ Солнечно, +22°C, влажность 45%",
        "лондон": "🌧 Дождливо, +12°C, влажность 80%",
        "дубай": "🔥 Жарко, +41°C, влажность 15%",
    }
    return weather_data.get(city.lower(), f"Нет данных для города {city}")

@function_tool
def convert_currency(amount: float, from_currency: str, to_currency: str) -> str:
    """Конвертирует сумму из одной валюты в другую."""
    rates = {"USD": 1.0, "EUR": 0.92, "RUB": 90.5, "GBP": 0.79}
    if from_currency not in rates or to_currency not in rates:
        return "Неподдерживаемая валюта"
    result = amount * rates[to_currency] / rates[from_currency]
    return f"{amount} {from_currency} = {result:.2f} {to_currency}"

# Создаём агента с инструкцией и инструментами
assistant = Agent(
    name="Ассистент",
    instructions="Ты полезный ассистент. Отвечай на русском. Используй инструменты для погоды и валют.",
    model="gpt-4o",
    tools=[get_weather, convert_currency],
)

async def main():
    result = await Runner.run(
        assistant,
        "Какая погода в Дубае и сколько это будет в рублях — 100 USD?"
    )
    print(f"Ответ: {result.final_output}")
    print(f"Токенов: {result.usage.total_tokens}, запусков: {result.usage.requests}")

asyncio.run(main())

import asyncio
from agents import Agent, Runner, function_tool, handoff

@function_tool
def run_python(code: str) -> str:
    """Выполняет Python-код и возвращает результат."""
    try:
        namespace = {}
        exec(code, namespace)
        return f"Результат: {namespace.get('result', 'выполнено')}"
    except Exception as e:
        return f"Ошибка: {e}"

@function_tool
def analyze_data(dataset: str) -> str:
    """Анализирует набор данных и возвращает статистику."""
    import statistics
    data = [12, 45, 78, 23, 56, 89, 34]
    return f"Среднее: {statistics.mean(data):.1f}, медиана: {statistics.median(data)}"

# Специализированные агенты
code_agent = Agent(
    name="CodeMaster",
    instructions="Ты senior-разработчик. Пиши и запускай Python-код через run_python.",
    model="gpt-4o",
    tools=[run_python],
)

data_agent = Agent(
    name="DataWhiz",
    instructions="Ты аналитик данных. Используй analyze_data для анализа.",
    model="gpt-4o-mini",
    tools=[analyze_data],
)

# Диспетчер с handoffs
dispatcher = Agent(
    name="Диспетчер",
    instructions="""Ты диспетчер. Анализируй запрос:
- Вопросы о коде/программировании → передай CodeMaster
- Вопросы о данных/статистике → передай DataWhiz
- Простые вопросы — ответь сам""",
    model="gpt-4o",
    handoffs=[handoff(code_agent), handoff(data_agent)],
)

async def main():
    result = await Runner.run(
        dispatcher,
        "Напиши код, который считает сумму чисел от 1 до 100, и сохрани в переменную result"
    )
    print(f"Финальный ответ:\n{result.final_output}")
    print(f"Последний агент: {result.last_agent.name}")

asyncio.run(main())

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

class ContentCheck(BaseModel):
    is_harmful: bool
    reason: str

@input_guardrail
async def check_input_safety(context, agent, input_data) -> GuardrailFunctionOutput:
    """Проверяет входные данные на нежелательный контент."""
    # В реальном проекте: вызов moderation API или классификатора
    blocked_keywords = ["взломай", "укради", "фишинг", "ddos"]
    user_text = str(input_data).lower()
    triggered = any(kw in user_text for kw in blocked_keywords)

    return GuardrailFunctionOutput(
        output_info=ContentCheck(
            is_harmful=triggered,
            reason="Заблокировано ключевое слово" if triggered else "OK"
        ),
        tripwire_triggered=triggered,
    )

@output_guardrail
async def check_output_safety(context, agent, output) -> GuardrailFunctionOutput:
    """Проверяет ответ агента: нет ли утечки системного промпта."""
    output_text = str(output).lower()
    sensitive_patterns = ["system prompt", "инструкция системы", "ты должен"]
    leaked = any(p in output_text for p in sensitive_patterns)

    return GuardrailFunctionOutput(
        output_info={"blocked": leaked},
        tripwire_triggered=leaked,
    )

# Агент с guardrails
safe_agent = Agent(
    name="Безопасный ассистент",
    instructions="Ты безопасный ассистент. Помогай только с легальными задачами.",
    model="gpt-4o",
    input_guardrails=[check_input_safety],
    output_guardrails=[check_output_safety],
)

async def main():
    try:
        result = await Runner.run(
            safe_agent,
            "Помоги написать легальный скрипт для автоматизации отчётов"
        )
        print(f"✅ OK: {result.final_output[:200]}")
    except Exception as e:
        print(f"🚫 Guardrail сработал: {e}")

asyncio.run(main())

import asyncio
import os
from agents import Agent, Runner, function_tool, trace

# Включаем трейсинг (альтернативно: export OPENAI_AGENTS_TRACE=1)
os.environ["OPENAI_AGENTS_TRACE"] = "1"

@function_tool
def search_knowledge_base(query: str) -> str:
    """Поиск по внутренней базе знаний."""
    kb = {
        "api": "REST API доступно по /api/v1, аутентификация через Bearer token",
        "deploy": "Деплой через Docker: docker build -t app . && docker push",
        "db": "PostgreSQL 15, хост db.internal, порт 5432, репликация streaming",
    }
    for key, value in kb.items():
        if key in query.lower():
            return value
    return "Ничего не найдено по запросу"

@function_tool
def calculate_metrics(expression: str) -> str:
    """Вычисляет математическое выражение."""
    try:
        return f"Результат: {eval(expression)}"
    except:
        return "Ошибка вычисления"

support_agent = Agent(
    name="SupportBot",
    instructions="Ты агент поддержки. Используй базу знаний и калькулятор.",
    model="gpt-4o",
    tools=[search_knowledge_base, calculate_metrics],
)

async def main():
    # Явный трейс с метаданными
    with trace("User Session #4821", metadata={"user_id": "u_9912", "plan": "pro"}):
        result = await Runner.run(
            support_agent,
            "Расскажи про API аутентификацию и посчитай 156 * 34 + 1200"
        )
        print(f"Ответ: {result.final_output}")

        # Детальная статистика выполнения
        print(f"\n📊 Статистика трейса:")
        print(f"  Запросов к LLM: {result.usage.requests}")
        print(f"  Входных токенов: {result.usage.input_tokens}")
        print(f"  Выходных токенов: {result.usage.output_tokens}")
        print(f"  Всего токенов: {result.usage.total_tokens}")

asyncio.run(main())

import asyncio
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from agents import Agent, Runner, function_tool
import json

app = FastAPI(title="Agent API", version="1.0")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])

class ChatRequest(BaseModel):
    message: str
    session_id: str = "default"

@function_tool
def get_server_status() -> str:
    """Возвращает статус сервера."""
    return "🟢 Все системы работают: CPU 23%, RAM 58%, диски OK"

prod_agent = Agent(
    name="API-Agent",
    instructions="Ты агент, работающий через API. Отвечай кратко на русском.",
    model="gpt-4o-mini",
    tools=[get_server_status],
)

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    async def event_generator():
        with trace(req.session_id):
            streaming = Runner.run_streamed(prod_agent, req.message)
            async for event in streaming.stream_events():
                if event.type == "raw_response_event":
                    yield f"data: {json.dumps({'type': 'text', 'content': event.data.delta})}\n\n"
                elif event.type == "run_item_stream_event":
                    yield f"data: {json.dumps({'type': 'tool', 'name': event.item.name})}\n\n"
            yield "data: [DONE]\n\n"
    return StreamingResponse(event_generator(), media_type="text/event-stream")

@app.get("/health")
async def health():
    return {"status": "ok", "agent": "ready"}

# Запуск: uvicorn server:app --host 0.0.0.0 --port 8000
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)