追踪 OpenAI Agent🤖

MLflow 追踪功能为 OpenAI 开发的多代理框架 OpenAI Agents SDK 提供了自动追踪能力。通过调用 mlflow.openai.autolog() 函数启用 OpenAI 的自动追踪，MLflow 将捕获追踪信息并将其记录到活动的 MLflow 实验中。

显示设置代码

python

import mlflow

mlflow.openai.autolog()

基本示例

以下示例演示了如何将 OpenAI Agents SDK 与 MLflow 追踪结合使用，用于简单的多语言聊天代理。这三个代理会协作以确定输入的语言，并将输入转交给说该语言的相应子代理。MLflow 会捕获代理之间的交互以及它们对 OpenAI API 的调用。

显示基本示例代码

python

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("https://: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 会捕获函数调用，并显示代理可用的函数、已调用的函数以及函数调用的输入和输出。

显示函数调用示例代码

python

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 会捕获护栏检查，并显示护栏检查背后的推理以及护栏是否被触发。

显示护栏示例代码

python

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 的自动追踪。