通过 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。