Observability for Semantic Kernel (Python) with Opik

1
import asyncio
2
import os
3
4
# REQUIRED: Enable Semantic Kernel diagnostics
5
# Option 1: Include sensitive data (prompts and completions)
6
os.environ["SEMANTICKERNEL_EXPERIMENTAL_GENAI_ENABLE_OTEL_DIAGNOSTICS_SENSITIVE"] = (
7
    "true"
8
)
9
10
# Option 2: Hide sensitive data (prompts and completions)
11
# os.environ["SEMANTICKERNEL_EXPERIMENTAL_GENAI_ENABLE_OTEL_DIAGNOSTICS"] = "true"
12
13
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
14
from opentelemetry.sdk.resources import Resource
15
from opentelemetry.sdk.trace import TracerProvider
16
from opentelemetry.sdk.trace.export import BatchSpanProcessor
17
from opentelemetry.semconv.resource import ResourceAttributes
18
from opentelemetry.trace import set_tracer_provider
19
from semantic_kernel import Kernel
20
from semantic_kernel.connectors.ai.function_choice_behavior import (
21
    FunctionChoiceBehavior,
22
)
23
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
24
from semantic_kernel.connectors.ai.prompt_execution_settings import (
25
    PromptExecutionSettings,
26
)
27
from semantic_kernel.functions.kernel_arguments import KernelArguments
28
from semantic_kernel.functions.kernel_function_decorator import kernel_function
29
30
31
class BookingPlugin:
32
    @kernel_function(
33
        name="find_available_rooms",
34
        description="Find available conference rooms for today.",
35
    )
36
    def find_available_rooms(
37
        self,
38
    ) -> list[str]:
39
        return ["Room 101", "Room 201", "Room 301"]
40
41
    @kernel_function(
42
        name="book_room",
43
        description="Book a conference room.",
44
    )
45
    def book_room(self, room: str) -> str:
46
        return f"Room {room} booked."
47
48
49
def set_up_tracing():
50
    # Create a resource to represent the service/sample
51
    resource = Resource.create(
52
        {ResourceAttributes.SERVICE_NAME: "semantic-kernel-app"}
53
    )
54
55
    exporter = OTLPSpanExporter()
56
57
    # Initialize a trace provider for the application. This is a factory for creating tracers.
58
    tracer_provider = TracerProvider(resource=resource)
59
    # Span processors are initialized with an exporter which is responsible
60
    # for sending the telemetry data to a particular backend.
61
    tracer_provider.add_span_processor(BatchSpanProcessor(exporter))
62
    # Sets the global default tracer provider
63
    set_tracer_provider(tracer_provider)
64
65
66
# This must be done before any other telemetry calls
67
set_up_tracing()
68
69
70
async def main():
71
    # Create a kernel and add a service
72
    kernel = Kernel()
73
    kernel.add_service(OpenAIChatCompletion(ai_model_id="gpt-4.1"))
74
    kernel.add_plugin(BookingPlugin(), "BookingPlugin")
75
76
    answer = await kernel.invoke_prompt(
77
        "Reserve a conference room for me today.",
78
        arguments=KernelArguments(
79
            settings=PromptExecutionSettings(
80
                function_choice_behavior=FunctionChoiceBehavior.Auto(),
81
            ),
82
        ),
83
    )
84
    print(answer)
85
86
87
if __name__ == "__main__":
88
    asyncio.run(main())