Rastreando agentes OpenAI
O MLflow Tracing fornece o recurso de rastreamento automático para o OpenAI Agents SDK, uma estrutura de vários agentes desenvolvida pela OpenAI. Ao ativar o rastreamento automático
para OpenAI chamando a função mlflow.openai.autolog
, MLflow capturará os rastreamentos e log os enviará para o experimento MLflow ativo.
import mlflow
mlflow.openai.autolog()
Pré-requisitos
Para usar o MLflow Tracing com o OpenAI Agents SDK, o senhor precisa instalar o MLflow, o OpenAI SDK e a biblioteca openai-agents
.
- Development
- Production
Para ambientes de desenvolvimento, instale o pacote completo do MLflow com os extras do Databricks, openai
, e openai-agents
:
pip install --upgrade "mlflow[databricks]>=3.1" openai openai-agents
O pacote completo do mlflow[databricks]
inclui todos os recursos para desenvolvimento local e experimentação no Databricks.
Para implantações de produção, instale mlflow-tracing
, openai
e openai-agents
:
pip install --upgrade mlflow-tracing openai openai-agents
O pacote mlflow-tracing
é otimizado para uso na produção.
O MLflow 3 é altamente recomendado para obter a melhor experiência de rastreamento com os agentes da OpenAI.
Antes de executar os exemplos, você precisará configurar seu ambiente:
Para usuários fora do Databricks Notebook : Defina seu Databricks variável de ambiente:
export DATABRICKS_HOST="https://your-workspace.cloud.databricks.com"
export DATABRICKS_TOKEN="your-personal-access-token"
Para usuários do Databricks Notebook : Essas credenciais são definidas automaticamente para o senhor.
API chave : Certifique-se de que o OpenAI API key esteja configurado:
export OPENAI_API_KEY="your-openai-api-key"
Exemplo básico
O exemplo a seguir demonstra como usar o SDK do OpenAI Agents com rastreamento do MLflow para agentes de bate-papo simples em vários idiomas. Os três agentes colaboram para determinar o idioma da entrada e da entrega ao subagente apropriado que fala o idioma. O MLflow captura como os agentes interagem uns com os outros e fazem chamadas para a API da OpenAI.
import mlflow
import asyncio
from agents import Agent, Runner
import os
# Ensure your OPENAI_API_KEY is set in your environment
# os.environ["OPENAI_API_KEY"] = "your-openai-api-key" # Uncomment and set if not globally configured
# Enable auto tracing for OpenAI Agents SDK
mlflow.openai.autolog() # This covers agents if using openai module for LLM calls
# If agents have their own autolog, e.g., mlflow.agents.autolog(), prefer that.
# Set up MLflow tracking to Databricks
mlflow.set_tracking_uri("databricks")
mlflow.set_experiment("/Shared/openai-agent-demo")
# Define a simple multi-agent workflow
spanish_agent = Agent(
name="Spanish agent",
instructions="You only speak Spanish.",
)
english_agent = Agent(
name="English agent",
instructions="You only speak English",
)
triage_agent = Agent(
name="Triage agent",
instructions="Handoff to the appropriate agent based on the language of the request.",
handoffs=[spanish_agent, english_agent],
)
async def main():
result = await Runner.run(triage_agent, input="Hola, ¿cómo estás?")
print(result.final_output)
# If you are running this code in a Jupyter notebook, replace this with `await main()`.
if __name__ == "__main__":
asyncio.run(main())
Chamada de função
O OpenAI Agents SDK suporta a definição de funções que podem ser chamadas pelo agente. O MLflow captura as chamadas de função e exibe quais funções estão disponíveis para o agente, quais delas são chamadas e as entradas e saídas das chamadas de função.
import asyncio
from agents import Agent, Runner, function_tool
import mlflow
import os
# Ensure your OPENAI_API_KEY is set in your environment
# os.environ["OPENAI_API_KEY"] = "your-openai-api-key" # Uncomment and set if not globally configured
# Enable auto tracing for OpenAI Agents SDK
mlflow.openai.autolog() # Assuming underlying LLM calls are via OpenAI
# Set up MLflow tracking to Databricks if not already configured
# mlflow.set_tracking_uri("databricks")
# mlflow.set_experiment("/Shared/openai-agent-function-calling-demo")
@function_tool
def get_weather(city: str) -> str:
return f"The weather in {city} is sunny."
agent = Agent(
name="Hello world",
instructions="You are a helpful agent.",
tools=[get_weather],
)
async def main():
result = await Runner.run(agent, input="What's the weather in Tokyo?")
print(result.final_output)
# The weather in Tokyo is sunny.
# If you are running this code in a Jupyter notebook, replace this with `await main()`.
if __name__ == "__main__":
asyncio.run(main())
Corrimãos
O OpenAI Agents SDK suporta a definição de guardrails que podem ser usados para verificar a entrada e a saída do agente. O MLflow captura as verificações do guarda-corpo e exibe o raciocínio por trás da verificação do guarda-corpo e se o guarda-corpo foi acionado.
from pydantic import BaseModel
from agents import (
Agent,
GuardrailFunctionOutput,
InputGuardrailTripwireTriggered,
RunContextWrapper,
Runner,
TResponseInputItem,
input_guardrail,
)
import mlflow
import os
# Ensure your OPENAI_API_KEY is set in your environment
# os.environ["OPENAI_API_KEY"] = "your-openai-api-key" # Uncomment and set if not globally configured
# Enable auto tracing for OpenAI Agents SDK
mlflow.openai.autolog() # Assuming underlying LLM calls are via OpenAI
# Set up MLflow tracking to Databricks if not already configured
# mlflow.set_tracking_uri("databricks")
# mlflow.set_experiment("/Shared/openai-agent-guardrails-demo")
class MathHomeworkOutput(BaseModel):
is_math_homework: bool
reasoning: str
guardrail_agent = Agent(
name="Guardrail check",
instructions="Check if the user is asking you to do their math homework.",
output_type=MathHomeworkOutput,
)
@input_guardrail
async def math_guardrail(
ctx: RunContextWrapper[None], agent: Agent, input
) -> GuardrailFunctionOutput:
result = await Runner.run(guardrail_agent, input, context=ctx.context)
return GuardrailFunctionOutput(
output_info=result.final_output,
tripwire_triggered=result.final_output.is_math_homework,
)
agent = Agent(
name="Customer support agent",
instructions="You are a customer support agent. You help customers with their questions.",
input_guardrails=[math_guardrail],
)
async def main():
# This should trip the guardrail
try:
await Runner.run(agent, "Hello, can you help me solve for x: 2x + 3 = 11?")
print("Guardrail didn't trip - this is unexpected")
except InputGuardrailTripwireTriggered:
print("Math homework guardrail tripped")
# If you are running this code in a Jupyter notebook, replace this with `await main()`.
if __name__ == "__main__":
asyncio.run(main())
Desativar o rastreamento automático
O rastreamento automático para o OpenAI Agents SDK pode ser desativado globalmente chamando mlflow.openai.autolog(disable=True)
ou mlflow.autolog(disable=True)
.