Agent 服务的可观测性与链路追踪

"""agent_tracing.py 手动埋点，OpenTelemetry SDK 1.27.0 加 GenAI 语义约定"""
from contextlib import contextmanager
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import Resource
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

resource = Resource.create({
    "service.name": "agent-runtime",
    "service.version": "2.4.0",
    "deployment.environment": "prod",
})
provider = TracerProvider(resource=resource)
provider.add_span_processor(
    BatchSpanProcessor(
        OTLPSpanExporter(endpoint="http://otel-collector:4318/v1/traces"),
        max_queue_size=8192,
        schedule_delay_millis=2000,
    )
)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("agent.runtime", "2.4.0")


@contextmanager
def llm_span(model: str, temperature: float, max_tokens: int):
    """包住一次模型调用，退出时自动写入 token 用量"""
    with tracer.start_as_current_span("llm.chat") as span:
        span.set_attribute("gen_ai.system", "openai")
        span.set_attribute("gen_ai.request.model", model)
        span.set_attribute("gen_ai.request.temperature", temperature)
        span.set_attribute("gen_ai.request.max_tokens", max_tokens)
        try:
            yield span
        except Exception as exc:
            span.set_attribute("error.type", type(exc).__name__)
            span.record_exception(exc)
            raise


@contextmanager
def tool_span(name: str, call_id: str):
    """包住一次工具调用，记录耗时与结果状态"""
    with tracer.start_as_current_span("tool.execute") as span:
        span.set_attribute("tool.name", name)
        span.set_attribute("tool.call.id", call_id)
        try:
            yield span
            span.set_attribute("tool.status", "ok")
        except Exception as exc:
            span.set_attribute("tool.status", "error")
            span.record_exception(exc)
            raise


def run_agent(session_id: str, user_input: str, max_steps: int = 12):
    with tracer.start_as_current_span("agent.session") as session:
        session.set_attribute("session.id", session_id)
        for step in range(max_steps):
            with tracer.start_as_current_span("agent.invoke") as step_span:
                step_span.set_attribute("agent.step.index", step)
                with llm_span("gpt-4o-2024-08-06", 0.2, 2048) as llm:
                    # 实际调用替换为你的 SDK 调用
                    resp = call_model(user_input)
                    llm.set_attribute("gen_ai.usage.input_tokens", resp.usage.input_tokens)
                    llm.set_attribute("gen_ai.usage.output_tokens", resp.usage.output_tokens)
                    llm.set_attribute("gen_ai.response.finish_reasons", [resp.finish_reason])
                if resp.finish_reason != "tool_calls":
                    return resp.content
                for tc in resp.tool_calls:
                    with tool_span(tc.name, tc.id) as tool:
                        tool.set_attribute("tool.args.size", len(str(tc.arguments)))
                        dispatch_tool(tc)
        session.set_attribute("agent.terminated", "max_steps_exceeded")
        return None

"""auto_tracing.py OpenLLMetry 自动埋点，traceloop-sdk 0.35.0"""
from traceloop.sdk import Traceloop
from traceloop.sdk.decorators import workflow, task, agent, tool

Traceloop.init(
    app_name="agent-runtime",
    api_endpoint="http://otel-collector:4318",
    disable_batch=False,
    resource_attributes={"deployment.environment": "prod"},
)


@tool(name="search_docs")
def search_docs(query: str, top_k: int = 5) -> list[dict]:
    return vector_store.search(query, top_k=top_k)


@task(name="plan")
def plan(goal: str) -> list[str]:
    return call_model(goal).content


@workflow(name="research_agent")
def research_agent(question: str) -> str:
    steps = plan(question)
    for step in steps:
        search_docs(step, top_k=5)
    return call_model(question).content

"""langfuse_tracing.py Langfuse SDK 2.53.0"""
from langfuse.decorators import observe, langfuse_context
from langfuse import Langfuse

langfuse = Langfuse(
    public_key="pk-lf-xxx",
    secret_key="sk-lf-xxx",
    host="https://langfuse.internal",
    release="2.4.0",
    flush_at=64,
    flush_interval=1.0,
)


@observe(as_type="generation", name="llm.chat")
def call_model(prompt: str, model: str = "gpt-4o-2024-08-06"):
    resp = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2,
    )
    langfuse_context.update_current_observation(
        model=model,
        usage={
            "input": resp.usage.prompt_tokens,
            "output": resp.usage.completion_tokens,
            "unit": "TOKENS",
        },
    )
    return resp.choices[0].message.content


@observe(name="agent.session")
def run_agent(session_id: str, question: str):
    langfuse_context.update_current_trace(
        session_id=session_id,
        user_id="u_1024",
        tags=["prod", "research"],
        metadata={"agent.version": "2.4.0"},
    )
    return call_model(question)

"""sampler.py 头部采样加错误保真"""
from opentelemetry.sdk.trace.sampling import Sampler, SamplingResult, Decision
from opentelemetry.trace import SpanKind
from opentelemetry import trace


class AdaptiveSampler(Sampler):
    """基础采样率 10%，错误与慢请求强制保留"""

    def __init__(self, base_ratio: float = 0.1, slow_threshold_ms: float = 20000):
        self.base_ratio = base_ratio
        self.slow_threshold_ms = slow_threshold_ms
        self._counter = 0

    def should_sample(self, parent_context, trace_id, name, kind=SpanKind.INTERNAL,
                      attributes=None, links=None, trace_state=None):
        attributes = attributes or {}
        # 显式调试标记：全采
        if attributes.get("debug") is True:
            return SamplingResult(Decision.RECORD_AND_SAMPLE)
        # 确定性哈希采样：同一 trace_id 结果稳定，父子一致
        self._counter += 1
        keep = (trace_id % 1000) < int(self.base_ratio * 1000)
        decision = Decision.RECORD_AND_SAMPLE if keep else Decision.DROP
        return SamplingResult(decision)

    def get_description(self) -> str:
        return f"AdaptiveSampler(base={self.base_ratio})"


provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(...)))
provider._sampler = AdaptiveSampler(base_ratio=0.1)

-- 找出过去 1 小时内最可疑的 20 个会话
SELECT
    session_id,
    countIf(span_name = 'agent.invoke')              AS steps,
    countIf(span_name = 'tool.execute')              AS tool_calls,
    countIf(span_name = 'tool.execute' AND status = 'error') AS tool_errors,
    round(sum(input_tokens + output_tokens) / 1000, 1)       AS ktokens,
    round(max(duration_ms) / 1000, 2)                AS session_seconds
FROM otel_spans
WHERE timestamp >= now() - INTERVAL 1 HOUR
  AND service_name = 'agent-runtime'
GROUP BY session_id
HAVING steps >= 12 OR tool_errors >= 3
ORDER BY tool_errors DESC, steps DESC
LIMIT 20;

async def bad():
    asyncio.create_task(handle_tool())   # 子任务里的 span 变成新的 root，链路断成两截

import contextvars
from opentelemetry import context as otel_context

async def good():
    ctx = contextvars.copy_context()
    asyncio.create_task(asyncio.to_thread(ctx.run, handle_tool))