🤖 OpenAI Agents SDK

from agents import Agent, Runner, function_tool
from pydantic import BaseModel

# Context: estado compartilhado entre agentes
class SupportContext(BaseModel):
    user_id: str
    account_tier: str
    conversation_history: list[str] = []
    issue_type: str = ""

# Tools como funções decoradas
@function_tool
def buscar_pedido(order_id: str) -> str:
    """Busca informações de um pedido pelo ID."""
    # Lógica real aqui
    return f"Pedido {order_id}: Em trânsito, entrega prevista 2 dias"

@function_tool
def processar_reembolso(order_id: str, motivo: str) -> str:
    """Processa solicitação de reembolso."""
    return f"Reembolso para {order_id} aprovado. Prazo: 5 dias úteis"

# Agent com instructions, tools e contexto tipado
agente_billing = Agent[SupportContext](
    name="Billing Specialist",
    instructions="""Você é especialista em faturamento e pedidos.
    Sempre verifique o histórico antes de tomar ações.
    Para reembolsos acima de R$500, escale para o supervisor.""",
    tools=[buscar_pedido, processar_reembolso],
    model="gpt-4o"
)

# Runner executa o agente
async def main():
    context = SupportContext(user_id="usr_123", account_tier="premium")
    resultado = await Runner.run(
        agente_billing,
        input="Quero reembolso do pedido #4521",
        context=context
    )
    print(resultado.final_output)

from agents import Agent, handoff

# Agentes especialistas
agente_tecnico = Agent[SupportContext](
    name="Technical Support",
    instructions="""Especialista em problemas técnicos: bugs, configuração,
    integrações. Resolva issues técnicas. Para problemas de cobrança,
    faça handoff para Billing.""",
    tools=[verificar_logs, resetar_configuracao]
)

agente_billing = Agent[SupportContext](
    name="Billing Support",
    instructions="""Especialista em faturamento, reembolsos e upgrades.
    Para problemas técnicos, faça handoff para Technical.""",
    tools=[buscar_pedido, processar_reembolso]
)

agente_supervisor = Agent[SupportContext](
    name="Supervisor",
    instructions="""Lida com casos escalados e clientes insatisfeitos.
    Tem autoridade para oferecer compensações especiais.""",
    tools=[aplicar_credito, criar_ticket_vip]
)

# Agente de triagem com todos os handoffs configurados
agente_triage = Agent[SupportContext](
    name="Triage Agent",
    instructions="""Você é a entrada do suporte. Classifique o problema
    e faça handoff para o especialista correto:
    - Técnico: bugs, erros, configuração
    - Billing: pagamento, reembolso, plano
    - Supervisor: cliente muito insatisfeito, casos complexos""",
    handoffs=[
        agente_tecnico,
        agente_billing,
        agente_supervisor
    ]
)

# Execução: triage decide para quem transferir
resultado = await Runner.run(
    agente_triage,
    input="Meu pagamento foi cobrado duas vezes este mês",
    context=SupportContext(user_id="u456", account_tier="basic")
)
# Triage analisa → handoff automático para Billing
# Billing resolve com acesso ao contexto completo

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

# Schema para decisão do guardrail
class TopicCheck(BaseModel):
    is_on_topic: bool
    reason: str

# Agente classificador (usado pelo guardrail)
topic_classifier = Agent(
    name="Topic Classifier",
    instructions="Verifique se a mensagem é sobre suporte ao produto.",
    output_type=TopicCheck,
    model="gpt-4o-mini"  # modelo barato para triagem rápida
)

# Input guardrail: verifica se o input é válido
@input_guardrail
async def verificar_topico(
    ctx: RunContextWrapper, agent: Agent, input: str
) -> GuardrailFunctionOutput:
    resultado = await Runner.run(topic_classifier, input, context=ctx.context)
    check = resultado.final_output
    return GuardrailFunctionOutput(
        output_info=check,
        tripwire_triggered=not check.is_on_topic  # True = bloqueia
    )

# Output guardrail: verifica qualidade da resposta
@output_guardrail
async def verificar_qualidade(
    ctx: RunContextWrapper, agent: Agent, output: str
) -> GuardrailFunctionOutput:
    # Verifica se a resposta contém informações sensíveis
    has_pii = any(term in output.lower() for term in ["cpf", "senha", "cartão"])
    return GuardrailFunctionOutput(
        output_info={"has_pii": has_pii},
        tripwire_triggered=has_pii
    )

# Agente com guardrails configurados
agente_seguro = Agent[SupportContext](
    name="Secure Support Agent",
    instructions="Ajude clientes com problemas do produto.",
    input_guardrails=[verificar_topico],
    output_guardrails=[verificar_qualidade],
    tools=[buscar_pedido]
)

# Se guardrail disparar: InputGuardrailTripwireTriggered exception
try:
    resultado = await Runner.run(agente_seguro, "Quero saber o CPF do usuário X")
except Exception as e:
    print(f"Bloqueado pelo guardrail: {e}")

from agents import Runner, trace, custom_span
import asyncio

# Tracing automático — sem configuração!
resultado = await Runner.run(agente_triage, "Preciso de ajuda")
# Trace completo disponível automaticamente no portal OpenAI

# Agrupando múltiplas execuções em um único trace
with trace("pipeline_suporte_completo"):
    triagem = await Runner.run(agente_triage, "Meu acesso está bloqueado")
    follow_up = await Runner.run(
        agente_tecnico,
        "Desbloqueie o acesso do usuário",
        context=triagem.context
    )

# Custom spans para operações específicas
async def processar_com_trace(user_input: str):
    with custom_span("validacao_input"):
        input_limpo = sanitizar(user_input)

    with custom_span("execucao_agente"):
        resultado = await Runner.run(agente_billing, input_limpo)

    with custom_span("persistencia"):
        salvar_historico(resultado)

    return resultado

# Acessando dados do trace programaticamente
resultado = await Runner.run(agente_billing, "Cancele minha assinatura")
print(resultado.raw_responses)    # todas as respostas da API
print(resultado.new_messages())   # mensagens geradas nesta run
print(resultado.last_agent.name)  # último agente a executar

# 5 especialistas + 1 triage = Swarm completo
especialista_onboarding = Agent(
    name="Onboarding Specialist",
    instructions="Guia novos usuários. Ajuda com setup inicial e primeiros passos.",
    tools=[criar_tour_interativo, enviar_email_boas_vindas]
)

especialista_seguranca = Agent(
    name="Security Specialist",
    instructions="Lida com contas comprometidas, 2FA e problemas de acesso.",
    tools=[bloquear_conta, resetar_2fa, gerar_link_recuperacao]
)

especialista_integracao = Agent(
    name="Integration Specialist",
    instructions="Suporte para APIs, webhooks e integrações de terceiros.",
    tools=[verificar_webhook, testar_api, consultar_docs_api]
)

especialista_billing = Agent(
    name="Billing Specialist",
    instructions="Faturamento, upgrades, cancelamentos e reembolsos.",
    tools=[buscar_pedido, processar_reembolso, alterar_plano]
)

especialista_tecnico = Agent(
    name="Technical Specialist",
    instructions="Bugs, performance e questões técnicas complexas.",
    tools=[verificar_logs, criar_ticket, escalar_engenharia]
)

# Triage conhece todos os especialistas
triage = Agent(
    name="Triage",
    instructions="""Classifique e transfira para o especialista correto.
    Seja eficiente: uma pergunta de classificação no máximo.""",
    handoffs=[
        especialista_onboarding,
        especialista_seguranca,
        especialista_integracao,
        especialista_billing,
        especialista_tecnico
    ]
)

# O Swarm em ação
resultado = await Runner.run(
    triage,
    "Meu webhook não está sendo chamado",
    context=SupportContext(user_id="u789", account_tier="pro")
)
# Triage → handoff para Integration Specialist → resolve