追踪 OpenAI Agent🤖

MLflow Tracing 为 OpenAI Agents SDK（一个由 OpenAI 开发的多智能体框架）提供了自动追踪能力。通过调用 mlflow.openai.autolog() 函数启用 OpenAI 的自动追踪，MLflow 将捕获追踪并将其记录到当前的 MLflow Experiment 中。

import mlflow

mlflow.openai.autolog()

基本示例

以下示例展示了如何将 OpenAI Agents SDK 与 MLflow tracing 结合使用，用于简单的多语言聊天代理。三个代理协作确定输入的语言，并将其交给说该语言的相应子代理。MLflow 捕获了代理如何相互交互以及如何调用 OpenAI API。

import mlflow
import asyncio
from agents import Agent, Runner

# Enable auto tracing for OpenAI Agents SDK
mlflow.openai.autolog()

# Optional: Set a tracking URI and an experiment
mlflow.set_tracking_uri("http://localhost:5000")
mlflow.set_experiment("OpenAI Agent")

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

函数调用

OpenAI Agents SDK 支持定义可由代理调用的函数。MLflow 捕获函数调用，并显示哪些函数可供代理使用，哪些函数被调用，以及函数调用的输入和输出。

import asyncio

from agents import Agent, Runner, function_tool

# Enable auto tracing for OpenAI Agents SDK
mlflow.openai.autolog()


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

防护机制

OpenAI Agents SDK 支持定义防护机制，用于检查代理的输入和输出。MLflow 捕获防护机制检查，并显示防护机制检查背后的原因以及防护机制是否被触发。

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

# Enable auto tracing for OpenAI Agents SDK
mlflow.openai.autolog()


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

禁用自动追踪

通过调用 mlflow.openai.autolog(disable=True) 或 mlflow.autolog(disable=True)，可以全局禁用 OpenAI Agents SDK 的自动追踪功能。