Cloud Monitor:Customize traces with loongsuite-util-genai and the OpenTelemetry SDK

1. Get the handler and tracer

Use get_extended_telemetry_handler() to obtain the singleton Handler from loongsuite-util-genai, and get_tracer(__name__) to obtain the Tracer from the OpenTelemetry SDK. They are used to create GenAI semantic Spans and custom business Spans, respectively.

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__)

The handler supports two usage patterns:

• Context manager (with handler.entry(inv), etc.): This is the recommended method. It automatically manages the span lifecycle.

• start/stop/fail API (handler.start_entry(inv) / handler.stop_entry(inv) / handler.fail_entry(inv, error)): Suitable for scenarios such as asynchronous, callback, or streaming where you cannot use with statements.

2. Create an Entry span

At the request entry point, create an Entry Span. Include the session_id and user_id, and record the user input by using input_messages. After the streaming response is complete, concatenate the output, set it to output_messages, and then call stop_entry to end the Span. This allows you to directly see the full input and final output of the request in the console.

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. Create an Agent span

Use start_invoke_agent to create an Agent Span that records the agent name, model, and description. The Agent Span is the root GenAI Span of the entire trace, and all subsequent ReAct Step, LLM call, and Tool call spans are its child spans.

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

After an agent completes its execution, it writes the accumulated total_input_tokens and total_output_tokens to the Agent span to aggregate token metrics.

4. Create a ReAct Step span

Create a Step Span for each ReAct inference iteration and pass the current round. When an iteration ends, set finish_reason to continue if the iteration needs to continue, or to stop if it is the final answer. In the example, the LLM calls in each iteration are automatically instrumented by the ARMS agent, so you do not need to create them manually.

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. Create a Tool span

When the model returns a tool call, create a Tool Span for each tool_call, recording the tool name, call ID, input parameters, and result.

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. Create custom spans with the OpenTelemetry SDK

In addition to the GenAI semantic Spans provided by loongsuite-util-genai, you can use the tracer.start_as_current_span() method from the OpenTelemetry SDK to create custom business Spans and use them together with GenAI Spans.

The following examples show two typical use cases for custom spans:

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

Custom instrumentation for other languages

• Java: Add custom instrumentation to traces with the OpenTelemetry Java SDK.

• Go: Create OpenTelemetry spans with custom extensions in Go.