The Semantic Kernel Masterclass

import asyncio
import semantic_kernel as sk
from semantic_kernel.connectors.ai.google.google_ai import GoogleAIChatCompletion

async def main():
    # 1. Initialize the Kernel (The container)
    kernel = sk.Kernel()

    # 2. Add the Brain (LLM Service)
    kernel.add_service(
        GoogleAIChatCompletion(
            service_id="gemini",
            ai_model_id="gemini-1.5-pro",
            api_key="YOUR_API_KEY"
        )
    )

if __name__ == "__main__":
    asyncio.run(main())

# Creating a function from a prompt template
summarize = kernel.create_function_from_prompt(
    function_name="Summarize",
    plugin_name="Writer",
    prompt="Summarize this text in 3 bullet points: {{$input}}"
)

# Call it exactly like a native function (inside async main)
result = await kernel.invoke(summarize, input="Semantic Kernel is...")
print(result)

from semantic_kernel.functions import kernel_function
from typing import Annotated

class MathPlugin:
    @kernel_function(
        description="Adds two numbers together",
        name="Add"
    )
    def add(
        self, 
        number1: Annotated[float, "The first number"], 
        number2: Annotated[float, "The second number"]
    ) -> float:
        return number1 + number2

# Import the plugin into the kernel
kernel.add_plugin(MathPlugin(), plugin_name="Math")

from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior

# Configure execution settings
settings = kernel.get_prompt_execution_settings_from_service_id("gemini")
settings.function_choice_behavior = FunctionChoiceBehavior.Auto()

# Now, if you ask a math question, SK automatically calls your Python code
result = await kernel.invoke_prompt(
    prompt="How much is 50 plus 25?",
    settings=settings
)
# Result: 75 (calculated by code, not guessed by LLM)

from semantic_kernel.contents import ChatHistory

# 1. Create a History Object
history = ChatHistory()
history.add_system_message("You are a helpful assistant.")

# 2. Add User Message
history.add_user_message("My name is Vinoth.")

# 3. Invoke Kernel with History
# The Kernel sends the WHOLE history to the LLM
result = await kernel.invoke_prompt(
    prompt=None, # Not needed if using history
    chat_history=history
)

# 4. Save Assistant Response back to History
history.add_assistant_message(str(result))

# Use precise import path for safety
from semantic_kernel.connectors.ai.google.google_ai.services.google_ai_text_embedding import GoogleAITextEmbedding

kernel.add_service(
    GoogleAITextEmbedding(service_id="embedding", ai_model_id="embedding-001")
)

from semantic_kernel.memory import VolatileMemoryStore, SemanticTextMemory

# Setup Memory
memory = SemanticTextMemory(storage=VolatileMemoryStore(), embeddings_generator=kernel.get_service("embedding"))

# Save a fact
await memory.save_information(collection="HR_Policies", id="info1", text="Refunds are only allowed within 30 days.")

# Search (The 'Retrieval' in RAG)
results = await memory.search("HR_Policies", "Can I get a refund?", limit=1)
print(results[0].text)

class PolicyPlugin:
    def __init__(self, memory):
        self.memory = memory

    @kernel_function(description="Searches company policies for answers", name="SearchPolicy")
    async def search_policy(self, query: str) -> str:
        results = await self.memory.search("HR_Policies", query, limit=1)
        return results[0].text if results else "No policy found."

# Add to kernel
kernel.add_plugin(PolicyPlugin(memory), plugin_name="Policy")

from semantic_kernel.planners import FunctionCallingStepwisePlanner

# Create the planner
planner = FunctionCallingStepwisePlanner(service_id="gemini")

# Give it a complex goal
goal = "Check my calendar for tomorrow and book a meeting room if I'm free."

# The planner figures out it needs:
# 1. CalendarPlugin.GetEvents()
# 2. Logic check
# 3. ComponentPlugin.BookRoom()
result = await planner.invoke(kernel, goal)

from semantic_kernel.filters import Filter, AutoFunctionInvocationContext

class SafeGuardFilter(Filter):
    async def on_auto_function_invocation(self, context, next):
        # NOTE: 'on_auto_...' triggers only when the LLM effectively *chooses* a tool.
        # Use 'on_function_invocation' if you want to catch manual calls too.
        
        # PRE-HOOK: Check for dangerous parameters
        if "delete_database" in context.function.name:
            raise SecurityError("Action Blocked: Database deletion is not allowed.")
        
        # Execute the function
        await next(context)
        
        # POST-HOOK: Log the result
        print(f"Function {context.function.name} executed successfully.")

# Register the filter
kernel.add_filter(SafeGuardFilter())

import logging

# Set up standard Python logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)

# Now, when you run the kernel, you will see logs like:
# [INFO] Function SearchPolicy execution time: 0.45s
# [INFO] Rendered Prompt: "You are a helpful assistant..."
# [INFO] Usage: 154 tokens

# Conceptual: A Manager Agent deciding which 'worker' to call
@kernel_function(description="Delegates task to a coder or writer")
async def manager_delegate(self, task: str) -> str:
    if "code" in task:
        # Pass the task to the independent 'Coder Kernel'
        return await coder_kernel.invoke(..., input=task)
    elif "blog" in task:
        # Pass the task to the independent 'Writer Kernel'
        return await writer_kernel.invoke(..., input=task)
    return "I can handle this myself."