Cloud Monitor:loongsuite-util-genai および OpenTelemetry SDK を使用したトレースのカスタマイズ

1. ハンドラおよびトレーサーの取得

loongsuite-util-genai からシングルトンハンドラを取得するには get_extended_telemetry_handler() を、OpenTelemetry SDK からトレーサーを取得するには get_tracer(__name__) を使用します。これらはそれぞれ、GenAI セマンティクススパンおよびカスタムビジネススパンを作成するために使用されます。

from opentelemetry.util.genai.extended_handler import get_extended_telemetry_handler
from opentelemetry.util.genai.extended_types import (
    ExecuteToolInvocation,
    InvokeAgentInvocation,
)
from opentelemetry.util.genai._extended_common import EntryInvocation, ReactStepInvocation
from opentelemetry.util.genai.types import Error, InputMessage, OutputMessage, Text
from opentelemetry.trace import get_tracer

handler = get_extended_telemetry_handler()
tracer = get_tracer(__name__)

ハンドラは以下の 2 つの使用パターンをサポートします。

• コンテキストマネージャー（with handler.entry(inv) など）：推奨される方法です。スパンのライフサイクルを自動的に管理します。

• start/stop/fail API（handler.start_entry(inv) / handler.stop_entry(inv) / handler.fail_entry(inv, error)）：with 文を使用できない非同期、コールバック、ストリーミングなどのシナリオに適しています。

2. Entry スパンの作成

リクエストのエントリポイントで Entry スパンを作成します。session_id および user_id を含め、input_messages を使用してユーザー入力を記録します。ストリーミング応答が完了したら、出力を連結して output_messages に設定し、stop_entry を呼び出してスパンを終了します。これにより、コンソール上でリクエストの完全な入力と最終的な出力を直接確認できます。

entry_inv = EntryInvocation(
    session_id=req.session_id or str(uuid.uuid4()),
    user_id=req.user_id or "anonymous",
    input_messages=[
        InputMessage(role="user", parts=[Text(content=req.topic)]),
    ],
)

def event_generator():
    handler.start_entry(entry_inv)
    output_chunks: list[str] = [ ]

    try:
        for chunk in run_agent_stream(topic=req.topic):
            output_chunks.append(chunk)
            yield f"data: {json.dumps({'content': chunk}, ensure_ascii=False)}\n\n"
        yield "data: [DONE]\n\n"
    except Exception as exc:
        handler.fail_entry(entry_inv, Error(message=str(exc), type=type(exc)))
        yield f"data: {json.dumps({'error': str(exc)}, ensure_ascii=False)}\n\n"
        return
    entry_inv.output_messages = [
        OutputMessage(
            role="assistant",
            parts=[Text(content="".join(output_chunks))],
            finish_reason="stop",
        ),
    ]
    handler.stop_entry(entry_inv)

3. Agent スパンの作成

start_invoke_agent を使用して、エージェント名、モデル、説明を記録する Agent スパンを作成します。Agent スパンはトレース全体のルート GenAI スパンであり、後続のすべての ReAct Step、LLM 呼び出し、およびツール呼び出しスパンはその子スパンとなります。

invocation = InvokeAgentInvocation(
    provider="dashscope",
    agent_name="TechContentAgent",
    agent_description="Technical content generation assistant",
    request_model="qwen-plus",
)
total_input_tokens = 0
total_output_tokens = 0

handler.start_invoke_agent(invocation)
try:
    # ... Core agent logic (ReAct loop) ...

    invocation.input_tokens = total_input_tokens
    invocation.output_tokens = total_output_tokens
    handler.stop_invoke_agent(invocation)
except Exception:
    handler.fail_invoke_agent(invocation, Error(message="agent failed", type=RuntimeError))
    raise

エージェントの実行が完了すると、累積された total_input_tokens および total_output_tokens が Agent スパンに書き込まれ、トークンメトリックが集約されます。

4. ReAct Step スパンの作成

各 ReAct 推論反復に対して Step スパンを作成し、現在の round を渡します。反復が終了すると、継続が必要な場合は finish_reason を continue に、最終回答の場合は stop に設定します。この例では、各反復における LLM 呼び出しは ARMS エージェントによって自動的にイベントトラッキングされるため、手動で作成する必要はありません。

step_inv = ReactStepInvocation(round=iteration + 1)
handler.start_react_step(step_inv)

try:
    response = client.chat.completions.create(
        model="qwen-plus",
        messages=messages,
        tools=TOOL_DEFINITIONS,
    )
    # ... Process the response ...

    step_inv.finish_reason = "stop"  # or "continue"
    handler.stop_react_step(step_inv)
except Exception:
    handler.fail_react_step(step_inv, Error(message="step failed", type=RuntimeError))
    raise

5. Tool スパンの作成

モデルがツール呼び出しを返した場合、各 tool_call に対して Tool スパンを作成し、ツール名、呼び出し ID、入力パラメーター、および結果を記録します。

tool_inv = ExecuteToolInvocation(
    tool_name=tool_call.function.name,
    tool_call_id=tool_call.id,
    tool_call_arguments=tool_call.function.arguments,
    tool_type="function",
)
handler.start_execute_tool(tool_inv)
try:
    result = dispatch_tool(tool_name, tool_call.function.arguments)
    tool_inv.tool_call_result = result
except Exception as exc:
    handler.fail_execute_tool(tool_inv, error=Error(message=str(exc), type=type(exc)))
    raise
else:
    handler.stop_execute_tool(tool_inv)

6. OpenTelemetry SDK を使用したカスタムスパンの作成

loongsuite-util-genai が提供する GenAI セマンティクススパンに加えて、OpenTelemetry SDK の tracer.start_as_current_span() メソッドを使用してカスタムビジネススパンを作成し、GenAI スパンと組み合わせて使用できます。

以下の例は、カスタムスパンの代表的なユースケースを 2 つ示しています。

def _check_duplicate_tools(
    tool_usage_counter: Counter,
    messages: list[dict[str, Any]],
) -> None:
    duplicates = [name for name, count in tool_usage_counter.items() if count > 1]
    has_duplicates = len(duplicates) > 0

    with tracer.start_as_current_span("duplicate_tool_detection") as span:
        span.set_attributes({
            "gen_ai.loop_detection.detected": has_duplicates,
            "gen_ai.loop_detection.duplicate_tools": str(duplicates) if has_duplicates else "[ ]",
            "gen_ai.loop_detection.total_calls": sum(tool_usage_counter.values()),
            "gen_ai.loop_detection.unique_tools": len(tool_usage_counter),
        })

    if has_duplicates:
        details = ", ".join(f"{n}({tool_usage_counter[n]} calls)" for n in duplicates)
        messages.append({
            "role": "system",
            "content": f"[System Hint] Duplicate tool calls detected: {details}. Please avoid repeating the call.",
        })

def _check_response_loop(
    current_content: str | None,
    previous_content: str | None,
) -> bool:
    cur = (current_content or "").strip()
    prev = (previous_content or "").strip()

    with tracer.start_as_current_span("response_loop_detection") as span:
        if not prev or not cur:
            span.set_attributes({
                "gen_ai.loop_detection.is_loop": False,
                "gen_ai.loop_detection.reason": "no_text_content",
            })
            return False

        is_identical = cur == prev
        longer = max(len(cur), len(prev))
        common_prefix_len = sum(1 for a, b in zip(cur, prev) if a == b)
        overlap_ratio = common_prefix_len / longer if longer > 0 else 0.0
        is_loop = is_identical or overlap_ratio > 0.8

        span.set_attributes({
            "gen_ai.loop_detection.is_loop": is_loop,
            "gen_ai.loop_detection.is_identical": is_identical,
            "gen_ai.loop_detection.overlap_ratio": round(overlap_ratio, 2),
            "gen_ai.loop_detection.current_length": len(cur),
            "gen_ai.loop_detection.previous_length": len(prev),
        })
        return is_loop

他の言語向けのカスタムイベントトラッキング

• Java： OpenTelemetry Java SDK を使用したトレースへのカスタムイベントトラッキングの追加。

• Go： Go でのカスタム拡張を使用した OpenTelemetry スパンの作成。