Cloud Monitor：通過 loongsuite-util-genai 與 OpenTelemetry SDK 為調用鏈增加自訂埋點

1. 擷取 Handler 和 Tracer

通過 get_extended_telemetry_handler() 擷取 loongsuite-util-genai 的單例 Handler，通過 get_tracer(__name__) 擷取 OpenTelemetry SDK 的 Tracer。兩者分別用於建立 GenAI 語義 Span 和自訂業務 Span。

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

Handler 提供兩種使用方式：

• 上下文管理器（with handler.entry(inv) 等）：推薦方式，自動管理 Span 生命週期。

• start/stop/fail API（handler.start_entry(inv) / handler.stop_entry(inv) / handler.fail_entry(inv, error)）：適用於非同步、回調或流式等無法使用 with 語句的情境。

2. 建立 Entry Span

在請求入口處建立 Entry Span，攜帶 session_id、user_id，並通過 input_messages 記錄使用者輸入。流式響應完成後，將輸出內容拼接設定到 output_messages，再調用 stop_entry 結束 Span。這樣在控制台中能直接看到該次請求的完整輸入和最終輸出。

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 Span

通過 start_invoke_agent 建立 Agent Span，記錄 Agent 名稱、模型和描述資訊。Agent Span 是整個調用鏈的根 GenAI Span，所有後續的 ReAct Step、LLM 調用和 Tool 調用都作為它的子 Span。

invocation = InvokeAgentInvocation(
    provider="dashscope",
    agent_name="TechContentAgent",
    agent_description="技術內容產生助手",
    request_model="qwen-plus",
)
total_input_tokens = 0
total_output_tokens = 0

handler.start_invoke_agent(invocation)
try:
    # ... Agent 核心邏輯（ReAct 迴圈） ...

    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

Agent 執行完成後，將累積的 total_input_tokens 和 total_output_tokens 寫入 Agent Span，實現 Token 指標匯總統計。

4. 建立 ReAct Step Span

在每一輪 ReAct 推理迭代時建立 Step Span，傳入當前輪次 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,
    )
    # ... 處理響應 ...

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

5. 建立 Tool Span

當模型返回工具調用時，為每個 tool_call 建立 Tool Span，記錄工具名稱、調用 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 建立自訂 Span

除了 loongsuite-util-genai 提供的 GenAI 語義 Span，還可以通過 OpenTelemetry SDK 的 tracer.start_as_current_span() 建立自訂業務 Span，與 GenAI Span 混合使用。

以下樣本展示了兩種典型的自訂 Span 用法：

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]}次)" for n in duplicates)
        messages.append({
            "role": "system",
            "content": f"[系統提示] 檢測到工具被重複調用：{details}。請避免重複調用。",
        })

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：使用Golang自訂擴充建立OpenTelemetry Span。