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())