メインコンテンツまでスキップ

OpenAIエージェントのトレース

オートログによるOpenAIトレース

MLflow Tracingは、OpenAIが開発したマルチエージェントフレームワークであるOpenAI Agents SDKの自動トレース機能を提供します。自動トレースを有効にする OpenAI の場合、 mlflow.openai.autolog 関数を呼び出すと、 MLflow はトレースをキャプチャし、アクティブな MLflow エクスペリメントに記録します。

Python
import mlflow

mlflow.openai.autolog()

前提 条件​

OpenAI Agents SDKでMLflow Tracingを使用するには、 MLflow、OpenAISDK、およびopenai-agents ライブラリをインストールする必要があります。

開発環境の場合は、Databricks extras、 openai、 openai-agentsを含む完全な MLflow パッケージをインストールします。

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

フル mlflow[databricks] パッケージには、Databricks でのローカル開発と実験のためのすべての機能が含まれています。

注記

MLflow 3 は、OpenAI エージェントで最適なトレース エクスペリエンスを実現するため、強くお勧めします。

例を実行する前に、環境を構成する必要があります。

Databricks ノートブックの外部ユーザーの場合 : Databricks 環境変数を設定します。

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

Databricks ノートブック内のユーザーの場合 : これらの資格情報は自動的に設定されます。

API キー : OpenAI API キーが構成されていることを確認します。本番運用環境では、安全な キー管理のために、ハードコードされた値の代わりにMosaic AI Gateway またはDatabricksシークレットAPI を使用します。

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

基本的な例​

次の例は、OpenAI Agents SDK と MLflow トレースを使用して、単純な多言語チャット エージェントを実現する方法を示しています。3 つのエージェントは協力して、入力の言語を決定し、その言語を話す適切なサブエージェントに引き継ぎます。MLflow は、エージェントが相互に対話し、OpenAI API を呼び出す方法をキャプチャします。

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

関数呼び出し​

OpenAI Agents SDK は、エージェントが呼び出すことができる関数の定義をサポートしています。MLflow は関数呼び出しをキャプチャし、エージェントが使用できる関数、呼び出された関数、および関数呼び出しの入力と出力を表示します。

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

オートログによるOpenAIトレース

警告

本番運用環境では、安全な キー管理のために、ハードコードされた値の代わりにMosaic AI Gateway またはDatabricksシークレットAPI を使用します。

ガードレール​

OpenAI Agents SDK は、エージェントの入力と出力を確認するために使用できるガードレールの定義をサポートしています。MLflow は、ガードレール チェックをキャプチャし、ガードレール チェックの背後にある理由と、ガードレールがトリップしたかどうかを表示します。

Python
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トレース

自動トレースを無効にする​

OpenAI Agents SDK の自動トレースは、 mlflow.openai.autolog(disable=True) または mlflow.autolog(disable=True)を呼び出すことでグローバルに無効にできます。

に最終更新