《AI Agent 高级编程》10.1 统一可观测：三框架 trace 归一 OTel

/tmp/agentbook/venv3/bin/python -c \
  "import importlib.metadata as m; print(m.version('opentelemetry-sdk'))"
# 1.44.0

"""pydantic-ai 原生 OTel：InstrumentedModel + InMemorySpanExporter。"""
import asyncio
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel
from pydantic_ai.models.instrumented import InstrumentedModel, InstrumentationSettings

exp = InMemorySpanExporter()
tp = TracerProvider()
tp.add_span_processor(SimpleSpanProcessor(exp))          # 本地 exporter，不连任何远端

def add(a: int, b: int) -> int:
    "Add two numbers"
    return a + b

agent = Agent(
    InstrumentedModel(TestModel(), InstrumentationSettings(tracer_provider=tp)),
    tools=[add], system_prompt="use tools",
)

asyncio.run(agent.run("what is 2+3"))
for s in exp.get_finished_spans():
    print(s.name, "|", s.attributes.get("gen_ai.operation.name"))

chat test                | chat
execute_tool add         | execute_tool
chat test                | chat
invoke_agent agent       | invoke_agent

from smolagents import ToolCallingAgent, tool as smtool, Model
from smolagents.models import ChatMessage, ChatMessageToolCall, ChatMessageToolCallFunction

@smtool
def add(a: int, b: int) -> int:
    """Add two numbers.
    Args:
        a: first
        b: second
    """
    return a + b

class SFFake(Model):                      # 脚本化假模型，零凭据确定性
    def __init__(self):
        super().__init__(); self.n = 0
    def generate(self, messages, stop_sequences=None, response_format=None,
                 tools_to_call_from=None, **kwargs):
        self.n += 1
        if self.n == 1:
            tc = ChatMessageToolCall(id="call_1", type="function",
                function=ChatMessageToolCallFunction(name="add", arguments={"a": 2, "b": 3}))
            return ChatMessage(role="assistant", content="", tool_calls=[tc])
        tc = ChatMessageToolCall(id="call_2", type="function",
            function=ChatMessageToolCallFunction(name="final_answer", arguments={"answer": "5"}))
        return ChatMessage(role="assistant", content="", tool_calls=[tc])

agent = ToolCallingAgent(tools=[add], model=SFFake(), max_steps=4)
agent.run("2+3?")
for st in agent.memory.steps:
    print(type(st).__name__, getattr(st, "step_number", None), getattr(st, "tool_calls", None))

TaskStep None None
ActionStep 1 [ToolCall(name='add', arguments={'a': 2, 'b': 3}, id='call_1')]
ActionStep 2 [ToolCall(name='final_answer', arguments={'answer': '5'}, id='call_2')]

"""统一中间表示 + 一次 emit，把任意框架的轨迹落成 OTel span 树。"""
from dataclasses import dataclass, field
from typing import Any

@dataclass
class NormSpan:
    name: str
    attrs: dict[str, Any] = field(default_factory=dict)
    children: list["NormSpan"] = field(default_factory=list)

def emit(root: NormSpan, tracer) -> None:
    """把 NormSpan 树递归落成 OTel span，父子关系由嵌套结构决定。"""
    def rec(n: NormSpan) -> None:
        with tracer.start_as_current_span(n.name) as sp:
            for k, v in n.attrs.items():
                sp.set_attribute(k, v)
            for c in n.children:
                rec(c)
    rec(root)

def read_pydantic_ai(spans) -> NormSpan:
    """pydantic-ai 的 span 已是 OTel，按 operation 分派即可。"""
    root = NormSpan("invoke_agent agent",
                    {"gen_ai.operation.name": "invoke_agent", "gen_ai.agent.name": "agent"})
    for s in spans:
        a = dict(s.attributes)
        op = a.get("gen_ai.operation.name")
        if op == "chat":
            root.children.append(NormSpan("chat " + str(a.get("gen_ai.request.model")), {
                "gen_ai.operation.name": "chat",
                "gen_ai.request.model": a.get("gen_ai.request.model"),
                "gen_ai.usage.input_tokens": a.get("gen_ai.usage.input_tokens"),
                "gen_ai.usage.output_tokens": a.get("gen_ai.usage.output_tokens")}))
        elif op == "execute_tool":
            root.children.append(NormSpan("execute_tool " + str(a.get("gen_ai.tool.name")), {
                "gen_ai.operation.name": "execute_tool",
                "gen_ai.tool.name": a.get("gen_ai.tool.name"),
                "gen_ai.tool.call.id": a.get("gen_ai.tool.call.id")}))
    return root


def read_smolagents(agent) -> NormSpan:
    """smolagents 只有 memory.steps，逐 ActionStep 翻译。"""
    root = NormSpan("invoke_agent smolagent",
                    {"gen_ai.operation.name": "invoke_agent", "gen_ai.agent.name": "smolagent"})
    root.children.append(NormSpan("chat fake",
                                  {"gen_ai.operation.name": "chat", "gen_ai.request.model": "fake"}))
    for st in agent.memory.steps:
        for tc in (getattr(st, "tool_calls", None) or []):
            root.children.append(NormSpan("execute_tool " + tc.name, {
                "gen_ai.operation.name": "execute_tool",
                "gen_ai.tool.name": tc.name,
                "gen_ai.tool.call.id": tc.id}))
    return root

=== 统一 emit 后，InMemorySpanExporter 里的 span 树 ===
- invoke_agent agent
  - chat test
  - execute_tool add
  - chat test
- invoke_agent smolagent
  - chat fake
  - execute_tool add
  - execute_tool final_answer
- invoke_agent minagent
  - chat fake
  - execute_tool add
span 总数: 11