agent-observability

Files (1)

• SKILL.md

SKILL.md (first lines)

# Agent Observability & Monitoring Skill

## When to Use

Implement this skill when your multi-agent system needs observability across agent spans, LLM calls, token costs, latency, error rates, and alerting. Covers the full stack: trace export → metrics collection → structured logging → dashboards → incident response.

## OpenTelemetry Tracing

### Span Creation & Context Propagation

```python
from opentelemetry import trace
from opentelemetry.trace import SpanKind, Status, StatusCode
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import Resource
from opentelemetry.propagate import inject, extract
import json

resource = Resource.create({"service.name": "agent-orchestrator"})
provider = TracerProvider(resource=resource)
provider.add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint="http://jaeger:4317"))
)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)

class AgentTracer:
    def __init__(self, agent_name: str):
        self.agent_name = agent_name
        self.tracer = trace.get_tracer(agent_name)

    async def trace_agent_run(self, task: dict) -> dict:
        with self.tracer.start_as_current_span(
            f"agent.{self.agent_name}.run",
            kind=SpanKind.SERVER,
            attributes={
                "agent.name": self.agent_name,
                "task.id": task.get("id"),
                "task.type": task.get("type", "unknown"),
            },
        ) as span:
            try:
                result = await self._execute(task)
                span.set_attribute("result.status", "success")
                span.set_attribute("result.length", len(str(result)))
                return result
            except Exception as e:
                span.set_status(Status(StatusCode.ERROR, str(e)))
                span.record_exception(e)
                raise

    async def _execute(self, task: dict) -> dict:
        with self.tracer.start_as_current_span(
            f"agent.{self.agent_name}.execute",
            kind=SpanKind.INTERNAL,
            attributes={"task.priority": task.get("priority", 0)},
        ) as span:
            spans = await self._llm_call(task["prompt"])
            return spans
```

### Agent Span Hierarchy

```python
async def orchestrate_workflow(request: dict, headers: dict) -> dict:
    ctx = extract(headers)
    with tracer.start_as_current_span(
        "orchestrator", context=ctx, kind=SpanKind.SERVER
    ) as root:
        root.set_attribute("request.id", request["id"])

        with tracer.start_as_current_span("agent.hermes.run") as hermes_span:
            hermes_result = await hermes.process(request["query"])
            hermes_span.set_attribute("tokens", hermes_result["tokens"])

        with tracer.start_as_current_span("agent.aphrodite.run") as aphro_span:
            aphro_result = await aphrodite.render(hermes_result["data"])
            aphro_span.set_attribute("components", len(aphro_result["ui"]))

        root.set_attribute("total.duration_ms", root.end_time - root.start_time)
        return {"result": hermes_result, "ui": aphro_result}
```

## LLM Call Tracing with LangSmith

```python
from langsmith import Client, traceable
from langsmith.run_trees import RunTree
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

client = Client()

@traceable(
    project_name="agent-hermes",
    tags=["production", "hermes-v2"],
    metadata={"agent": "hermes", "version": "2.1.0"},
)
async def agent_llm_call(prompt: str, context: dict) -> str:
    llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
    messages = [
        SystemMessage(content="You are Hermes, a backend agent."),
        HumanMessage(content=prompt),
    ]
    response = await llm.ainvoke(messages)
    return response.content

@traceable(run_type="chain", name="hermes_full_flow")
async def hermes_plan_and_generate(task: str) -> dict:
    plan = await agent_llm_call(f"Plan implementation for: {task}")
    code = await agent_llm_call(f"Generate code for plan: {plan}")
    return {"plan": plan, "code": code}
```

### Custom Traces & Feedback

```python

From ils15/pantheon-legacy, MIT.

More from ils15/pantheon-legacy