エージェント型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でのトレース表示に関する情報は、デプロイされたエージェントのログとトレースを確認するを参照してください。

Map spans and attributes to the tracing table¶

On a deployment, the tracing table displays several columns derived from OpenTelemetry span attributes. The following naming conventions apply across the trace:

Tracing table column	属性	How it is derived
コスト	`datarobot.moderation.cost`	Summed across all spans in the trace.
プロンプト	`gen_ai.prompt`	If multiple spans set this attribute, the first value in trace order is used.
補完	`gen_ai.completion`	If multiple spans set this attribute, the last value in trace order is used.
ツール	`tool_name`	Every distinct `tool_name` found on any span in the trace is listed.

Surface tool names in the tracing table¶

The Tools column is populated from the span attribute tool_name. Some frameworks set it on tool spans automatically; others do not. If your traces show tool execution in the span timeline, but Tools is empty, create a span around the tool body (or use the active span) and set tool_name explicitly.

Inside a with spanCurrent span only

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

If a span is already active (for example, from upstream instrumentation), you can set the attribute on that span:

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)

For LangGraph and similar frameworks, tool calls are sometimes wired through callbacks in a way that does not add tool_name to spans; manual instrumentation can allow the name to appear in the Tools column.

ベストプラクティス¶

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

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

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

実行に関するコンテキストを提供する属性を追加します。
一貫性のある属性命名規則を使用します（例：tool.input、tool.output、tool.error）。
サイズ、カウント、ステータスなどの関連するメタデータを含めます。
To populate Cost, Prompt, Completion, and Tools in the deployment Tracing table, set datarobot.moderation.cost, gen_ai.prompt, gen_ai.completion, and tool_name on the relevant spans.

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

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