pip install semantic-kernel

Successfully installed semantic-kernel-1.x.x ...

dotnet add package Microsoft.SemanticKernel
dotnet add package Microsoft.SemanticKernel.Connectors.OpenAI

info : Package 'Microsoft.SemanticKernel' is compatible with all the specified frameworks in project.

import asyncio
import os
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.contents.chat_history import ChatHistory

async def main():
    kernel = Kernel()
    kernel.add_service(OpenAIChatCompletion(
        service_id="chat",
        ai_model_id="gpt-4o-mini",
        api_key=os.environ["OPENAI_API_KEY"],
    ))

    history = ChatHistory()
    history.add_user_message("Define a kernel in one sentence.")

    chat = kernel.get_service("chat")
    settings = chat.get_prompt_execution_settings_class()(service_id="chat")
    response = await chat.get_chat_message_content(chat_history=history, settings=settings, kernel=kernel)
    print(response.content)

asyncio.run(main())

A kernel is the central orchestrator that holds AI services, plugins, and configuration
for invoking functions in a Semantic Kernel application.

using Microsoft.SemanticKernel;
using Microsoft.SemanticKernel.ChatCompletion;

var builder = Kernel.CreateBuilder();
builder.AddOpenAIChatCompletion(
    modelId: "gpt-4o-mini",
    apiKey: Environment.GetEnvironmentVariable("OPENAI_API_KEY")!);
Kernel kernel = builder.Build();

var chat = kernel.GetRequiredService<IChatCompletionService>();
var history = new ChatHistory();
history.AddUserMessage("Define a kernel in one sentence.");

var reply = await chat.GetChatMessageContentAsync(history, kernel: kernel);
Console.WriteLine(reply.Content);

A kernel is the orchestration root that wires AI services, plugins, and memory into a single
invocable surface for an application.

from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import (
    OpenAIChatCompletion,
    OpenAITextEmbedding,
)
import os

kernel = Kernel()

kernel.add_service(OpenAIChatCompletion(
    service_id="chat",
    ai_model_id="gpt-4o-mini",
    api_key=os.environ["OPENAI_API_KEY"],
))

kernel.add_service(OpenAITextEmbedding(
    service_id="embed",
    ai_model_id="text-embedding-3-small",
    api_key=os.environ["OPENAI_API_KEY"],
))

print(kernel.services.keys())

dict_keys(['chat', 'embed'])

from typing import Annotated
from semantic_kernel.functions import kernel_function

class TimePlugin:
    @kernel_function(description="Return the current date in ISO format.")
    def today(self) -> Annotated[str, "Current date YYYY-MM-DD"]:
        from datetime import date
        return date.today().isoformat()

    @kernel_function(description="Return the day of the week for a given ISO date.")
    def day_of_week(self, iso_date: Annotated[str, "An ISO date YYYY-MM-DD"]) -> str:
        from datetime import date
        return date.fromisoformat(iso_date).strftime("%A")

kernel.add_plugin(TimePlugin(), plugin_name="time")

result = await kernel.invoke(kernel.get_function("time", "day_of_week"), iso_date="2026-01-01")
print(result.value)

Thursday

using System.ComponentModel;
using Microsoft.SemanticKernel;

public class TimePlugin
{
    [KernelFunction("today")]
    [Description("Return the current date in ISO format.")]
    public string Today() => DateTime.UtcNow.ToString("yyyy-MM-dd");

    [KernelFunction("day_of_week")]
    [Description("Return the day of the week for a given ISO date.")]
    public string DayOfWeek([Description("An ISO date YYYY-MM-DD")] string isoDate)
        => DateTime.Parse(isoDate).DayOfWeek.ToString();
}

kernel.Plugins.AddFromObject(new TimePlugin(), "time");

plugins/
  WriterPlugin/
    ShortPoem/
      skprompt.txt
      config.json

Write a short four-line poem about {{$topic}} in the style of {{$style}}.

{
  "schema": 1,
  "description": "Generate a short poem.",
  "execution_settings": {
    "default": {
      "max_tokens": 200,
      "temperature": 0.8
    }
  },
  "input_variables": [
    {"name": "topic", "description": "Subject of the poem", "default": ""},
    {"name": "style", "description": "Author style", "default": "Robert Frost"}
  ]
}

kernel.add_plugin(parent_directory="./plugins", plugin_name="WriterPlugin")
poem = await kernel.invoke(kernel.get_function("WriterPlugin", "ShortPoem"), topic="snow", style="haiku")
print(str(poem))

White silence falls slow,
Each flake a quiet promise,
Winter writes the world.

---
name: ShortPoem
description: Generate a short poem.
model:
  api: chat
  configuration:
    type: openai
    name: gpt-4o-mini
inputs:
  topic:
    type: string
  style:
    type: string
    default: Robert Frost
---
system:
You are a poet.
user:
Write a short four-line poem about {{topic}} in the style of {{style}}.

import asyncio
import os
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion, OpenAIChatPromptExecutionSettings
from semantic_kernel.contents.chat_history import ChatHistory

kernel = Kernel()
kernel.add_service(OpenAIChatCompletion(service_id="chat", ai_model_id="gpt-4o-mini", api_key=os.environ["OPENAI_API_KEY"]))
kernel.add_plugin(TimePlugin(), plugin_name="time")

settings = OpenAIChatPromptExecutionSettings(service_id="chat")
settings.function_choice_behavior = FunctionChoiceBehavior.Auto()

async def ask(question: str) -> str:
    history = ChatHistory()
    history.add_user_message(question)
    chat = kernel.get_service("chat")
    reply = await chat.get_chat_message_content(chat_history=history, settings=settings, kernel=kernel)
    return reply.content

print(asyncio.run(ask("What day of the week is 2026-07-04?")))

2026-07-04 is a Saturday.

from semantic_kernel.contents.chat_history import ChatHistory

history = ChatHistory()
history.add_user_message("Explain async/await in three sentences.")

chat = kernel.get_service("chat")
async for chunk in chat.get_streaming_chat_message_content(chat_history=history, settings=settings, kernel=kernel):
    if chunk and chunk.content:
        print(chunk.content, end="", flush=True)
print()

from semantic_kernel.filters import FunctionInvocationContext
from semantic_kernel.filters.filter_types import FilterTypes

@kernel.filter(filter_type=FilterTypes.FUNCTION_INVOCATION)
async def log_calls(context: FunctionInvocationContext, next):
    print(f"-> {context.function.plugin_name}.{context.function.name}")
    await next(context)
    print(f"<- {context.function.plugin_name}.{context.function.name} = {context.result}")

from semantic_kernel.memory import VolatileMemoryStore, SemanticTextMemory

store = VolatileMemoryStore()
embed = kernel.get_service("embed")

memory = SemanticTextMemory(storage=store, embeddings_generator=embed)
await memory.save_information(collection="docs", id="1", text="The kernel routes all requests.")
await memory.save_information(collection="docs", id="2", text="Plugins expose callable functions.")

results = await memory.search(collection="docs", query="How are tools exposed?", limit=2)
for r in results:
    print(r.text, "→", r.relevance)

Plugins expose callable functions. → 0.82
The kernel routes all requests. → 0.41

docker run -it --rm -p 9001:9001 \
  -e OPENAI_API_KEY=$OPENAI_API_KEY \
  ghcr.io/microsoft/kernel-memory:latest

KernelMemory service is listening on http://+:9001

var memory = new MemoryWebClient("http://localhost:9001");
await memory.ImportDocumentAsync("readme.md", documentId: "readme");
var answer = await memory.AskAsync("What does this document cover?");
Console.WriteLine(answer.Result);

from semantic_kernel.planners import FunctionCallingStepwisePlanner

planner = FunctionCallingStepwisePlanner(service_id="chat")
result = await planner.invoke(
    kernel,
    "Tell me the day of the week for July 4th, 2026, and write a one-line poem about that day.",
)
print(result.final_answer)

from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.contents import ChatHistoryAgentThread

agent = ChatCompletionAgent(
    kernel=kernel,
    name="researcher",
    instructions="You answer questions concisely using available tools.",
)

thread = ChatHistoryAgentThread()
async for response in agent.invoke(messages="What day of the week is 2026-07-04?", thread=thread):
    print(response.content)

from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion, OpenAIChatCompletion

kernel.add_service(AzureChatCompletion(service_id="primary",  deployment_name="gpt-4o", endpoint=os.environ["AZURE_OPENAI_ENDPOINT"], api_key=os.environ["AZURE_OPENAI_KEY"]))
kernel.add_service(OpenAIChatCompletion(service_id="fallback", ai_model_id="gpt-4o-mini", api_key=os.environ["OPENAI_API_KEY"]))

try:
    settings = OpenAIChatPromptExecutionSettings(service_id="primary")
    reply = await kernel.invoke_prompt(prompt="Define vector embeddings.", settings=settings)
except Exception:
    settings = OpenAIChatPromptExecutionSettings(service_id="fallback")
    reply = await kernel.invoke_prompt(prompt="Define vector embeddings.", settings=settings)
print(reply)

from typing import Annotated
from pydantic import BaseModel
from semantic_kernel.functions import kernel_function

class Review(BaseModel):
    score: int
    summary: str

class ReviewPlugin:
    @kernel_function(description="Score and summarise a product review.")
    def review(self, text: Annotated[str, "Raw review text"]) -> Review:
        return Review(score=8, summary="Generally positive.")

builder.Services.AddScoped<Kernel>(sp =>
{
    var kb = Kernel.CreateBuilder();
    kb.AddOpenAIChatCompletion("gpt-4o-mini", builder.Configuration["OpenAI:ApiKey"]!);
    var k = kb.Build();
    k.Plugins.AddFromObject(sp.GetRequiredService<TimePlugin>(), "time");
    k.Plugins.AddFromObject(sp.GetRequiredService<UserContextPlugin>(), "user");
    return k;
});

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(endpoint="https://my-otel-collector/v1/traces")))
trace.set_tracer_provider(provider)

import functools

@functools.lru_cache(maxsize=256)
async def cached_invoke(prompt: str) -> str:
    return str(await kernel.invoke_prompt(prompt=prompt))

app.MapPost("/v1/chat/completions", async (ChatRequest req, Kernel k) =>
{
    var chat = k.GetRequiredService<IChatCompletionService>();
    var history = new ChatHistory(req.Messages);
    var reply = await chat.GetChatMessageContentAsync(history, kernel: k);
    return Results.Json(new { choices = new[] { new { message = new { role = "assistant", content = reply.Content } } } });
});