エージェント型AI > 構築 > トレースの実装

トレースの実装¶

OpenTelemetry（OTel）は、エージェントに包括的なオブザーバビリティを提供し、エージェントの実行をリアルタイムで監視、トレース、デバッグできます。このガイドでは、エージェントツールにカスタムトレースを追加して、詳細な実行情報を取得する方法を説明します。

OpenTelemetryによるトレースは、以下のことに役立ちます。

エージェントのパフォーマンスと実行フローを監視する。
詳細な実行トレースを追跡することで、問題をデバッグする。
ツールの実行パターンとタイミングを把握する。
ツールのカスタム属性とメタデータを確認する。

エージェントテンプレートには、CrewAI、LangGraph、Llama-Indexなどのフレームワーク向けのOpenTelemetry計装がすでに含まれています。この計装により、以下のスパンが自動的に取得されます。

エージェントの実行
ツールの起動
LLM APIの呼び出し
HTTPリクエスト

ツールにカスタムスパンやカスタム属性を追加することで、このデフォルトトレースを強化することができます。

ツールにカスタムトレースを追加する¶

OpenTelemetryのカスタムトレースをツールに追加して、ツールの実行に関する追加情報を取得します。これにより、ユースケースに固有のカスタム属性、中間出力、実行の詳細を追跡できます。

ツールにカスタムトレースを追加する基本的なパターンは以下のとおりです。

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

# Within your tool's execution
with tracer.start_as_current_span("my_custom_span_name"):
    current_span = trace.get_current_span()
    current_span.set_attribute("tool_name", "my_tool_name")
    current_span.set_attribute("gen_ai.prompt", "input passed to this step")
    current_span.set_attribute("datarobot.moderation.cost", 0.0)

    # Your tool logic here
    result = perform_tool_action()

    current_span.set_attribute("gen_ai.completion", str(result))
    # Optionally add more attributes about the result
    current_span.set_attribute("result.status", "success")
    current_span.set_attribute("result.size", len(result))

    return result

ツール例¶

OpenTelemetryのカスタムトレースをエージェントツールに追加する方法については、以下のコード例を参照してください。

CrewAILangGraphLlama-Index

import requests
from crewai.tools import BaseTool
from opentelemetry import trace

tracer = trace.get_tracer(__name__)

class WeatherTool(BaseTool):
    name: str = "weather_tool"
    description: str = (
        "Fetches the current weather for a specified city. "
        "Requires an API key from OpenWeatherMap."
    )

    def _run(self, city: str) -> str:
        with tracer.start_as_current_span("weather_tool_fetch"):
            current_span = trace.get_current_span()
            current_span.set_attribute("tool_name", "weather_tool")
            current_span.set_attribute("gen_ai.prompt", f"weather lookup for {city}")
            current_span.set_attribute("datarobot.moderation.cost", 0.0)

            # Set custom attributes
            current_span.set_attribute("weather.city", city)
            current_span.set_attribute("weather.api", "openweathermap")

            api_key = "YOUR_API_KEY"  # Replace with your API key
            base_url = "http://api.openweathermap.org/data/2.5/weather"
            params = {"q": city, "appid": api_key, "units": "metric"}

            try:
                response = requests.get(base_url, params=params, timeout=10)
                response.raise_for_status()

                data = response.json()
                weather = data['weather'][0]
                main = data['main']

                # Add result attributes
                current_span.set_attribute("weather.temperature", main['temp'])
                current_span.set_attribute("weather.condition", weather['main'])

                result = (
                    f"Current weather in {data['name']}, {data['sys']['country']}:\n"
                    f"Temperature: {main['temp']}°C (feels like {main['feels_like']}°C)\n"
                    f"Condition: {weather['main']} - {weather['description']}\n"
                    f"Humidity: {main['humidity']}%\n"
                    f"Pressure: {main['pressure']} hPa"
                )
                current_span.set_attribute("gen_ai.completion", result)

                return result

            except requests.exceptions.RequestException as e:
                current_span.set_attribute("weather.error", str(e))
                err = f"Error fetching weather data: {str(e)}"
                current_span.set_attribute("gen_ai.completion", err)
                return err

import requests
from langchain.tools import BaseTool
from opentelemetry import trace

tracer = trace.get_tracer(__name__)

class WeatherTool(BaseTool):
    name: str = "weather_tool"
    description: str = (
        "Fetches the current weather for a specified city. "
        "Requires an API key from OpenWeatherMap."
    )

    def _run(self, city: str) -> str:
        with tracer.start_as_current_span("weather_tool_fetch"):
            current_span = trace.get_current_span()
            current_span.set_attribute("tool_name", "weather_tool")
            current_span.set_attribute("gen_ai.prompt", f"weather lookup for {city}")
            current_span.set_attribute("datarobot.moderation.cost", 0.0)

            # Set custom attributes
            current_span.set_attribute("weather.city", city)

            api_key = "YOUR_API_KEY"  # Replace with your API key
            base_url = "http://api.openweathermap.org/data/2.5/weather"
            params = {"q": city, "appid": api_key, "units": "metric"}

            try:
                response = requests.get(base_url, params=params, timeout=10)
                response.raise_for_status()

                data = response.json()

                # Add result attributes
                current_span.set_attribute("weather.temperature", data['main']['temp'])

                result = f"Temperature in {city}: {data['main']['temp']}°C"
                current_span.set_attribute("gen_ai.completion", result)
                return result

            except requests.exceptions.RequestException as e:
                current_span.set_attribute("weather.error", str(e))
                err = f"Error: {str(e)}"
                current_span.set_attribute("gen_ai.completion", err)
                return err

import requests
from llama_index.core.tools import FunctionTool
from opentelemetry import trace

tracer = trace.get_tracer(__name__)

def _weather_run(city: str) -> str:
    """Fetches the current weather for a specified city. Requires an API key from OpenWeatherMap."""
    with tracer.start_as_current_span("weather_tool_fetch"):
        current_span = trace.get_current_span()
        current_span.set_attribute("tool_name", "weather_tool")
        current_span.set_attribute("gen_ai.prompt", f"weather lookup for {city}")
        current_span.set_attribute("datarobot.moderation.cost", 0.0)

        # Set custom attributes
        current_span.set_attribute("weather.city", city)

        api_key = "YOUR_API_KEY"  # Replace with your API key
        base_url = "http://api.openweathermap.org/data/2.5/weather"
        params = {"q": city, "appid": api_key, "units": "metric"}

        try:
            response = requests.get(base_url, params=params, timeout=10)
            response.raise_for_status()

            data = response.json()

            # Add result attributes
            current_span.set_attribute("weather.temperature", data['main']['temp'])

            result = f"Temperature in {city}: {data['main']['temp']}°C"
            current_span.set_attribute("gen_ai.completion", result)
            return result

        except requests.exceptions.RequestException as e:
            current_span.set_attribute("weather.error", str(e))
            err = f"Error: {str(e)}"
            current_span.set_attribute("gen_ai.completion", err)
            return err

def WeatherTool() -> FunctionTool:
    return FunctionTool.from_defaults(
        fn=_weather_run,
        name="weather_tool",
        description=(
            "Fetches the current weather for a specified city. "
            "Requires an API key from OpenWeatherMap."
        ),
    )

ネストされたスパンの作成¶

ネストされたスパンを作成して、複数のステップからなる複雑なツールの実行を表します。

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

def complex_tool_workflow(input_data):
    with tracer.start_as_current_span("complex_tool_main"):
        current_span = trace.get_current_span()
        current_span.set_attribute("input.size", len(input_data))

        # First step in the workflow
        with tracer.start_as_current_span("data_processing"):
            processed_data = process_data(input_data)
            trace.get_current_span().set_attribute("processed_items", len(processed_data))

        # Second step in the workflow
        with tracer.start_as_current_span("data_validation"):
            validated_data = validate_data(processed_data)
            trace.get_current_span().set_attribute("validated_items", len(validated_data))

        # Third step in the workflow
        with tracer.start_as_current_span("result_generation"):
            result = generate_result(validated_data)
            current_span.set_attribute("result.size", len(result))

        return result

スパンにイベントを追加する¶

スパンにイベントを追加して、ツール実行の重要なタイミングをマークします。

from opentelemetry import trace
from datetime import datetime

tracer = trace.get_tracer(__name__)

def tool_with_events():
    with tracer.start_as_current_span("tool_execution"):
        current_span = trace.get_current_span()

        # Add an event for when processing starts
        current_span.add_event(
            "Processing started",
            {"timestamp": datetime.utcnow().isoformat()}
        )

        # Your tool logic
        intermediate_result = perform_action()

        # Add an event for mid-execution
        current_span.add_event(
            "Intermediate result ready",
            {"result_count": len(intermediate_result)}
        )

        # More processing
        final_result = complete_processing(intermediate_result)

        # Add final event
        current_span.add_event(
            "Processing completed",
            {"output_size": len(final_result)}
        )

        return final_result

エージェントにカスタムトレースを追加する¶

カスタムトレースを設定して、構成や環境の詳細を含むエージェントの起動方法を取り込むことができます。ランタイムパラメーター（環境変数など）をスパンに表示するには、以下の手順に従います。

.envファイルを更新して、以下の環境変数を含め、ローカル開発時およびモデルをパッケージ化する際に使用できるようにします。
```
EXAMPLE_ENV_VAR=my_example_value 
```

agent/model-metadata.yamlにパラメーターを追加して、エージェントの実行時に挿入できるようにします。

runtimeParameterDefinitions:
  - fieldName: EXAMPLE_ENV_VAR
    type: string
    defaultValue: SET_VIA_PULUMI_OR_MANUALLY

agent/agent/config.pyのConfigクラスにパラメーター（例：example_env_var: str = ""）を追加して、エージェントコードで値をconfig.example_env_varとして使用できるようにします。

infra/infra/llm.pyを更新して、ランタイムパラメーターをカスタムモデル環境に転送します。

custom_model_runtime_parameters = [
    # ...existing parameters...
    datarobot.CustomModelRuntimeParameterValueArgs(
        key="EXAMPLE_ENV_VAR",
        type="string",
        value=os.environ.get("EXAMPLE_ENV_VAR"),
    ),
]

設定読み込みコードをスパンでラップし、set_attributeとadd_eventを使用して値を添付します。プロパティ名はテンプレートによって異なります（例：CrewAIテンプレートではagent_planner）。

@property
def agent_planner(self) -> Any:
    with tracer.start_as_current_span("config_variables"):
        current_span = trace.get_current_span()
        current_span.set_attribute("config.example_env_var", config.example_env_var)
        current_span.add_event("config attribute set on span")
    # ...agent code continued...

エージェントをデプロイして実行すると、トレースビジュアライザーは config.example_env_var=my_example_valueなどの属性を持つconfig_variablesスパンを表示します。これにより、ランタイムパラメーターやその他の環境値が正しく読み込まれたことを簡単に確認できます。

デプロイ後のDataRobot UIでのトレース表示に関する情報は、デプロイされたエージェントのログとトレースを確認するを参照してください。

スパンと属性をトレーステーブルにマッピングする¶

デプロイでは、トレーステーブルにOpenTelemetryのスパン属性から派生したいくつかの列が表示されます。トレース全体に以下の命名規則が適用されます。

トレーステーブルの列	属性	派生方法
コスト	`datarobot.moderation.cost`	トレース内のすべてのスパンにおける合計。
プロンプト	`gen_ai.prompt`	複数のスパンがこの属性を設定している場合、トレース順で最初の値が使用されます。
補完	`gen_ai.completion`	複数のスパンがこの属性を設定している場合、トレース順で最後の値が使用されます。
ツール	`tool_name`	トレース内のいずれかのスパンで見つかった一意の`tool_name`がすべてリストされます。

トレーステーブルにツール名を表示する¶

ツール列には、スパン属性tool_nameの値が入力されます。一部のフレームワークではこれがツールスパンに自動的に設定されますが、設定されないものもあります。トレースのスパンタイムラインにツールの実行が表示されているのにツールが空の場合は、ツール本体を囲むスパンを作成（またはアクティブなスパンを使用）し、tool_nameを明示的に設定します。

withスパン内現在のスパンのみ

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

def my_tool_impl(query: str) -> str:
    with tracer.start_as_current_span("my_tool"):
        span = trace.get_current_span()
        # Attributes that map to deployment Tracing table columns (see above)
        span.set_attribute("tool_name", "my_tool_name")
        span.set_attribute("gen_ai.prompt", query)
        span.set_attribute("datarobot.moderation.cost", 0.0)  # numeric; summed per trace
        # ... tool logic ...
        result = "result"
        span.set_attribute("gen_ai.completion", result)
        return result

スパンがすでにアクティブである場合（例：アップストリームの計装によるもの）、そのスパンに属性を設定できます。

from opentelemetry import trace

span = trace.get_current_span()
span.set_attribute("tool_name", "my_tool_name")
span.set_attribute("gen_ai.prompt", "user input or request text")
span.set_attribute("gen_ai.completion", "model or tool output text")
span.set_attribute("datarobot.moderation.cost", 0.0)

LangGraphや同様のフレームワークでは、ツール呼び出しは、スパンにtool_nameを追加しない方法でコールバックを介して接続されることがあります。手動で計装を行うことで、ツール列に名前を表示させることができます。

ベストプラクティス¶

わかりやすいスパン名を使用する：

スパンには明確でわかりやすい名前を使います（たとえば、"span1"ではなく"weather_fetch"）。
必要に応じて、スパン名にツール名を含めます。

意味のある属性を設定する：

実行に関するコンテキストを提供する属性を追加します。
一貫性のある属性命名規則を使用します（例：tool.input、tool.output、tool.error）。
サイズ、カウント、ステータスなどの関連するメタデータを含めます。
デプロイのトレーステーブル内のコスト、プロンプト、補完、およびツールに値を入力するには、関連するスパンにdatarobot.moderation.cost、gen_ai.prompt、gen_ai.completion、およびtool_nameを設定します。

スパンを適切に使用する：

すべてのコード行ではなく、重要な操作のためにスパンを作成します。
各スパンは、意味のある作業単位を表す必要があります。
サブ操作を表すには、ネストされたスパンを使用します。