Kontrollieren Sie Ihre KI-Agenten. Bevor sie Ihr Budget kontrollieren.

# Traditional observability setup - 50+ lines of boilerplate
from langchain.callbacks import BaseCallbackHandler
from langchain.callbacks.manager import CallbackManager
import opentelemetry
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc import OTLPSpanExporter
import logging
import json
import time

class CustomAgentTracer(BaseCallbackHandler):
    def __init__(self, service_name: str):
        self.provider = TracerProvider()
        self.exporter = OTLPSpanExporter(endpoint="...")
        self.provider.add_span_processor(
            BatchSpanProcessor(self.exporter)
        )
        trace.set_tracer_provider(self.provider)
        self.tracer = trace.get_tracer(service_name)
        self.logger = logging.getLogger(__name__)
        self._spans = {}
        self._costs = {}

    def on_llm_start(self, serialized, prompts, **kwargs):
        span = self.tracer.start_span("llm_call")
        span.set_attribute("prompts", json.dumps(prompts))
        span.set_attribute("model", serialized.get("model"))
        self._spans[kwargs["run_id"]] = span
        self._costs[kwargs["run_id"]] = time.time()

    def on_llm_end(self, response, **kwargs):
        span = self._spans.pop(kwargs["run_id"])
        elapsed = time.time() - self._costs.pop(kwargs["run_id"])
        span.set_attribute("latency_ms", elapsed * 1000)
        span.set_attribute("tokens", response.usage.total)
        span.end()

    def on_chain_start(self, serialized, inputs, **kwargs):
        span = self.tracer.start_span("chain")
        self._spans[kwargs["run_id"]] = span

    def on_chain_end(self, outputs, **kwargs):
        span = self._spans.pop(kwargs["run_id"], None)
        if span: span.end()

    def on_tool_start(self, serialized, input_str, **kwargs):
        # ... more boilerplate
        pass

# Setup callback manager
tracer = CustomAgentTracer("my-agent-service")
callback_manager = CallbackManager([tracer])

# Pass to every single chain and agent...
chain = LLMChain(llm=llm, callbacks=callback_manager)