Instrumentation()

1
from respan import Instrumentation
2
3
class Instrumentation(Protocol):
4
    name: str
5
6
    def activate(self) -> None:
7
        """Start intercepting spans and injecting them into the OTEL pipeline."""
8
        ...
9
10
    def deactivate(self) -> None:
11
        """Stop intercepting spans and clean up resources."""
12
        ...

1
from respan import Respan
2
from respan_tracing.utils.span_factory import build_readable_span, inject_span
3
4
class MySDKInstrumentor:
5
    name = "my-sdk"
6
7
    def activate(self) -> None:
8
        # Register a callback with the vendor SDK
9
        import my_sdk
10
        my_sdk.on_completion(self._on_completion)
11
12
    def deactivate(self) -> None:
13
        import my_sdk
14
        my_sdk.remove_callback(self._on_completion)
15
16
    def _on_completion(self, event):
17
        span = build_readable_span(
18
            name=f"my-sdk.completion",
19
            trace_id=event.trace_id,
20
            attributes={
21
                "llm.request.type": "chat",
22
                "gen_ai.request.model": event.model,
23
                "gen_ai.usage.prompt_tokens": event.input_tokens,
24
                "gen_ai.usage.completion_tokens": event.output_tokens,
25
                "traceloop.span.kind": "task",
26
                "traceloop.entity.name": "completion",
27
            },
28
        )
29
        inject_span(span)
30
31
# Use it
32
respan = Respan(
33
    api_key="your-api-key",
34
    instrumentations=[MySDKInstrumentor()],
35
)