Трассировка агентов OpenAI

MLflow Tracing обеспечивает автоматическую возможность трассировки для мультиагентной платформы SDK OpenAI, разработанной OpenAI. Включив автоматическую трассировку для OpenAI с помощью вызова функции mlflow.openai.autolog, MLflow будет отслеживать и регистрировать их в активном эксперименте MLflow.

import mlflow

mlflow.openai.autolog()

Замечание

В бессерверных вычислительных кластерах автоматическая отладка платформ трассировки genAI не включена автоматически. Необходимо явно включить автологирование, вызвав соответствующую mlflow.<library>.autolog() функцию для конкретных интеграции, которые требуется трассировать.

Предпосылки

Чтобы использовать трассировку MLflow с пакетом SDK для агентов OpenAI, необходимо установить MLflow, пакет SDK OpenAI и библиотеку openai-agents .

Развитие

Для сред разработки инсталлируйте полный пакет MLflow с дополнительными компонентами Databricks, openai, и openai-agents.

pip install --upgrade "mlflow[databricks]>=3.1" openai openai-agents

Полный mlflow[databricks] пакет включает все функции для локальной разработки и экспериментирования в Databricks.

Производство

Для рабочих развертываний установите mlflow-tracing, openai и openai-agents.

pip install --upgrade mlflow-tracing openai openai-agents

Пакет mlflow-tracing оптимизирован для использования в рабочей среде.

Замечание

MLflow 3 настоятельно рекомендуется для оптимальной трассировки с помощью агентов OpenAI.

Перед выполнением примеров необходимо настроить среду:

Для пользователей вне ноутбуков Databricks: задайте переменные среды Databricks:

export DATABRICKS_HOST="https://your-workspace.cloud.databricks.com"
export DATABRICKS_TOKEN="your-personal-access-token"

Для пользователей записных книжек Databricks: эти учетные данные устанавливаются автоматически.

Ключи API. Убедитесь, что ключ API OpenAI настроен. В рабочих средах используйте секреты Шлюза ИИ Мозаики или Databricks вместо жестко закодированных значений для безопасного управления ключами API.

export OPENAI_API_KEY="your-openai-api-key"

Базовый пример

В следующем примере показано, как использовать пакет SDK для агентов OpenAI с трассировкой MLflow для простых агентов чата с несколькими языками. Три агента сотрудничают, чтобы определить язык ввода и передать задачу соответствующему подагенту, который говорит на этом языке. MLflow записывает, как агенты взаимодействуют друг с другом и выполняют вызовы к API 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())

Вызов функции

Пакет SDK для агентов OpenAI поддерживает определение функций, которые могут вызываться агентом. MLflow записывает вызовы функций и отображает функции, доступные агенту, которые из них вызываются, и входные и выходные данные вызовов функций.

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())

Предупреждение

В рабочих средах используйте секреты Шлюза ИИ Мозаики или Databricks вместо жестко закодированных значений для безопасного управления ключами API.

Ограждения

Пакет SDK для агентов OpenAI поддерживает определение ограничений, которые можно использовать для проверки входных и выходных данных агента. MLflow фиксирует проверки ограничений и отображает обоснование этих проверок, а также был ли нарушен ограничитель.

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())

Отключение автоматической трассировки

Автоматическая трассировка для агентов OpenAI в SDK может быть отключена глобально путем вызова mlflow.openai.autolog(disable=True) или mlflow.autolog(disable=True).