Как отслеживать взаимодействия агентов OpenAI с помощью MLflow

июля 14, 2025 · 3 min

Управление рабочими процессами агентов OpenAI с помощью MLflow

MLflow — это платформа с открытым исходным кодом для управления и отслеживания экспериментов в машинном обучении. В сочетании с OpenAI Agents SDK MLflow автоматически записывает каждое взаимодействие и вызов API, включая использование инструментов, сообщения ввода и вывода, а также промежуточные решения. Это крайне полезно для отладки, анализа производительности и воспроизводимости.

Преимущества для мультиагентных систем

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

Установка зависимостей

Для начала установите необходимые библиотеки:

pip install openai-agents mlflow pydantic pydotenv

Для работы нужен ключ OpenAI API, который можно получить на https://platform.openai.com/settings/organization/api-keys. Новым пользователям может потребоваться добавить платежные данные и внести минимальный платеж в $5 для активации доступа. Сохраните ключ в файле .env следующим образом:

OPENAI_API_KEY = <YOUR_API_KEY>

Замените <YOUR_API_KEY> на ваш ключ.

Пример 1: Мультиагентная система для вопросов по программированию и кулинарии

В скрипте multi_agent_demo.py реализован простой мультиагентный ассистент, направляющий запросы пользователей к эксперту по программированию или кулинарии. Включена функция mlflow.openai.autolog(), которая автоматически отслеживает и логирует все взаимодействия агентов, включая ввод, вывод и передачу управления.

Логи сохраняются локально в папке ./mlruns под экспериментом “Agent-Coding-Cooking”.

import mlflow, asyncio
from agents import Agent, Runner
import os
from dotenv import load_dotenv
load_dotenv()

mlflow.openai.autolog()                           # Автоматическое отслеживание вызовов OpenAI
mlflow.set_tracking_uri("./mlruns")
mlflow.set_experiment("Agent-Coding-Cooking")

coding_agent = Agent(name="Coding agent",
                     instructions="You only answer coding questions.")

cooking_agent = Agent(name="Cooking agent",
                      instructions="You only answer cooking questions.")

triage_agent = Agent(
    name="Triage agent",
    instructions="If the request is about code, handoff to coding_agent; "
                 "if about cooking, handoff to cooking_agent.",
    handoffs=[coding_agent, cooking_agent],
)

async def main():
    res = await Runner.run(triage_agent,
                           input="How do I boil pasta al dente?")
    print(res.final_output)

if __name__ == "__main__":
    asyncio.run(main())

Просмотр логов в интерфейсе MLflow

Запустите команду:

mlflow ui

Обычно интерфейс доступен по адресу http://localhost:5000. В нем можно увидеть полный путь запроса — от ввода пользователя до маршрутизации и ответа агента, что помогает в отладке и оптимизации.

Пример 2: Агент поддержки с защитными ограждениями (guardrails)

В этом примере реализован агент поддержки с защитой от ответов на медицинские вопросы. Отдельный агент-ограждение проверяет запросы и блокирует медицинские.

import mlflow, asyncio
from pydantic import BaseModel
from agents import (
    Agent, Runner,
    GuardrailFunctionOutput, InputGuardrailTripwireTriggered,
    input_guardrail, RunContextWrapper)

from dotenv import load_dotenv
load_dotenv()

mlflow.openai.autolog()
mlflow.set_tracking_uri("./mlruns")
mlflow.set_experiment("Agent-Guardrails")

class MedicalSymptons(BaseModel):
    medical_symptoms: bool
    reasoning: str


guardrail_agent = Agent(
    name="Guardrail check",
    instructions="Check if the user is asking you for medical symptons.",
    output_type=MedicalSymptons,
)


@input_guardrail
async def medical_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.medical_symptoms,
    )


agent = Agent(
    name="Customer support agent",
    instructions="You are a customer support agent. You help customers with their questions.",
    input_guardrails=[medical_guardrail],
)


async def main():
    try:
        await Runner.run(agent, "Should I take aspirin if I'm having a headache?")
        print("Guardrail didn't trip - this is unexpected")

    except InputGuardrailTripwireTriggered:
        print("Medical guardrail tripped")


if __name__ == "__main__":
    asyncio.run(main())

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

Просмотр логов ограждений

Используйте ту же команду mlflow ui для визуального анализа взаимодействий и выявления заблокированных запросов.

Интеграция OpenAI Agents SDK с MLflow облегчает мониторинг, отладку и безопасную эксплуатацию интеллектуальных мультиагентных систем.