用 OpenTelemetry 看 AI Agent 的每一步：LoongSuite GenAI 语义规范实战

Trace: user_query_001
├── Span: agent.plan          (gen_ai.agent.step.type=planning)
│   ├── Span: gen_ai.invoke   (model=gpt-4o, step=1)
│   └── Span: tool.search     (tool.name=web_search, step=2)
├── Span: agent.execute       (gen_ai.agent.step.type=execution)
│   ├── Span: gen_ai.invoke   (model=gpt-4o, step=3)
│   └── Span: tool.code_run   (tool.name=python_executor, step=4)
└── Span: agent.respond       (gen_ai.agent.step.type=response)
    └── Span: gen_ai.invoke   (model=gpt-4o, step=5)

pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp

"""
LoongSuite GenAI 语义规范埋点示例
演示一个两步推理 Agent：先检索，再生成回复
运行前需启动 OTel Collector 接收数据
"""

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource

# 1. 初始化 Tracer — 标记服务身份
resource = Resource.create({
    "service.name": "my-agent-app",
    "service.version": "0.1.0",
})

provider = TracerProvider(resource=resource)
# 将 OTLP exporter 指向你的 Collector 地址
provider.add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint="localhost:4317"))
)
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("my-agent-app", "0.1.0")


# 2. 模拟模型调用，按规范挂属性
def call_llm(prompt: str, step_name: str, step_index: int) -> str:
    """模拟一次 LLM 调用，埋入 LoongSuite GenAI 语义属性"""
    with tracer.start_as_current_span("gen_ai.invoke") as span:
        # OTel GenAI 基础属性
        span.set_attribute("gen_ai.system", "openai")
        span.set_attribute("gen_ai.request.model", "gpt-4o")
        span.set_attribute("gen_ai.request.max_tokens", 1024)
        # LoongSuite 扩展属性：关联 Agent 步骤
        span.set_attribute("gen_ai.agent.step.name", step_name)
        span.set_attribute("gen_ai.agent.step.type", "reasoning")
        span.set_attribute("gen_ai.agent.step.index", step_index)
        # 消息来源
        span.set_attribute("gen_ai.message.source", "agent")
        span.set_attribute("gen_ai.message.role", "assistant")
        # 模拟返回
        span.set_attribute("gen_ai.response.finish_reason", "stop")
        span.set_attribute("gen_ai.usage.input_tokens", len(prompt))
        span.set_attribute("gen_ai.usage.output_tokens", 150)
        return f"[模拟回复] 基于 {prompt[:20]}... 生成的回答"


# 3. 模拟工具调用
def search_tool(query: str, step_index: int) -> str:
    """模拟一次检索工具调用"""
    with tracer.start_as_current_span("tool.search") as span:
        span.set_attribute("tool.name", "web_search")
        span.set_attribute("tool.input", query)
        span.set_attribute("gen_ai.agent.step.name", "retrieve_context")
        span.set_attribute("gen_ai.agent.step.type", "tool_call")
        span.set_attribute("gen_ai.agent.step.index", step_index)
        return f"[检索结果] 与 {query} 相关的 3 条文档片段"


# 4. 组装 Agent 主流程
def run_agent(user_query: str) -> str:
    """两步推理 Agent：检索 → 生成"""
    with tracer.start_as_current_span("agent.run") as root_span:
        root_span.set_attribute("gen_ai.agent.type", "react")
        root_span.set_attribute("gen_ai.agent.input", user_query)

        # Step 1: 检索上下文
        context = search_tool(user_query, step_index=1)

        # Step 2: 基于上下文生成回复
        augmented_prompt = f"上下文：{context}\n问题：{user_query}"
        answer = call_llm(augmented_prompt, step_name="generate_answer", step_index=2)

        root_span.set_attribute("gen_ai.agent.output", answer)
        return answer


# 5. 运行
if __name__ == "__main__":
    result = run_agent("LoongSuite GenAI 规范和 OTel 有什么关系？")
    print(f"Agent 回复：{result}")

from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

exporter = OTLPSpanExporter(
    endpoint="https://your-arms-endpoint:4317",
    headers={"Authentication": "your-arms-token"},
)