Semantic Kernel setup tutorial: choosing a model on n4n.ai

Semantic Kernel choosing model n4n.ai tutorial starts with a practical reality: the framework is powerful but the documentation assumes you already know which model fits your workload. This post walks you from a blank directory to a working kernel that streams tokens, calls functions, and falls back automatically when a provider hiccups — all against a single OpenAI-compatible endpoint.

Prerequisites

Python 3.10+ (3.11 or 3.12 recommended)
An n4n.ai API key — get one at https://n4n.ai
pip install semantic-kernel openai python-dotenv

Create a project directory and a virtual environment:

mkdir sk-n4n-tutorial && cd sk-n4n-tutorial
python -m venv .venv
source .venv/bin/activate
pip install semantic-kernel openai python-dotenv

Configure the kernel

Semantic Kernel’s Kernel class is a container for services, plugins, and middleware. For n4n.ai, you register an OpenAIChatCompletion service pointed at the n4n.ai base URL. The SDK treats it like any other OpenAI-compatible endpoint.

Create kernel_setup.py:

import os
from dotenv import load_dotenv
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.connectors.ai.open_ai.prompt_execution_settings.open_ai_prompt_execution_settings import OpenAIChatPromptExecutionSettings

load_dotenv()

API_KEY = os.getenv("N4N_API_KEY")
BASE_URL = "https://api.n4n.ai/v1"  # OpenAI-compatible endpoint

if not API_KEY:
    raise RuntimeError("Set N4N_API_KEY in .env or environment")

kernel = Kernel()

# Register the chat completion service
chat_service = OpenAIChatCompletion(
    ai_model_id="gpt-4o-mini",  # logical model id; n4n.ai routes to the best available
    api_key=API_KEY,
    base_url=BASE_URL,
)
kernel.add_service(chat_service)

print("Kernel services:", list(kernel.get_all_services().keys()))

Create a .env file:

N4N_API_KEY=sk-your-key-here

Run it:

python kernel_setup.py

Expected output:

Kernel services: ['openai_chat_completion']

The service is registered. Now let’s talk about model selection — this is where most tutorials go vague.

Choosing a model on n4n.ai

n4n.ai exposes 240+ models behind one endpoint. You don’t pick a provider; you pick a capability profile and let the gateway route. The ai_model_id you pass to OpenAIChatCompletion is a logical identifier. Common choices:

Logical id	Use case	Notes
`gpt-4o-mini`	Default general purpose, low latency, cheap	Great for classification, extraction, simple chat
`gpt-4o`	Complex reasoning, code, long context	Higher cost, stronger reasoning
`claude-3.5-sonnet`	Long-form writing, analysis, 200k context	Strong at nuance and instruction following
`llama-3.1-70b-instruct`	Open-weight preference, on-par with GPT-4 class	Good for data-sensitive workloads
`gemini-1.5-pro`	Massive context (1M+), multimodal	When you need to stuff entire codebases

You can also pass routing directives via the extra_headers parameter on the service if you want to pin a provider or enable specific fallbacks. For most cases, the default routing — automatic fallback when a provider is rate-limited or degraded — is the right call.

Update kernel_setup.py to make model selection configurable:

import os
from dotenv import load_dotenv
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion

load_dotenv()

API_KEY = os.getenv("N4N_API_KEY")
BASE_URL = "https://api.n4n.ai/v1"

# Choose your model here — change one line to swap capability profile
MODEL_ID = os.getenv("SK_MODEL_ID", "gpt-4o-mini")

if not API_KEY:
    raise RuntimeError("Set N4N_API_KEY in .env or environment")

kernel = Kernel()

chat_service = OpenAIChatCompletion(
    ai_model_id=MODEL_ID,
    api_key=API_KEY,
    base_url=BASE_URL,
)
kernel.add_service(chat_service)

print(f"Registered model: {MODEL_ID}")
print("Kernel services:", list(kernel.get_all_services().keys()))

Run with different models:

SK_MODEL_ID=claude-3.5-sonnet python kernel_setup.py
SK_MODEL_ID=llama-3.1-70b-instruct python kernel_setup.py

Your first completion: non-streaming

Semantic Kernel uses PromptExecutionSettings to control sampling parameters. Create first_completion.py:

import asyncio
import os
from dotenv import load_dotenv
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.connectors.ai.open_ai.prompt_execution_settings.open_ai_prompt_execution_settings import OpenAIChatPromptExecutionSettings
from semantic_kernel.contents import ChatHistory

load_dotenv()

API_KEY = os.getenv("N4N_API_KEY")
BASE_URL = "https://api.n4n.ai/v1"
MODEL_ID = os.getenv("SK_MODEL_ID", "gpt-4o-mini")

kernel = Kernel()
chat_service = OpenAIChatCompletion(
    ai_model_id=MODEL_ID,
    api_key=API_KEY,
    base_url=BASE_URL,
)
kernel.add_service(chat_service)

# Execution settings — temperature, max_tokens, etc.
settings = OpenAIChatPromptExecutionSettings(
    temperature=0.2,
    max_tokens=300,
)

history = ChatHistory()
history.add_system_message("You are a concise senior engineer. Answer in 3 sentences max.")
history.add_user_message("Why does Semantic Kernel separate the kernel from the AI service?")

async def main():
    result = await kernel.get_chat_message_content(
        chat_history=history,
        settings=settings,
    )
    print(result.content)

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

Run it:

python first_completion.py

Expected output (varies by model):

The kernel is an orchestration layer that manages plugins, memory, and planning — it doesn't know how to generate tokens. The AI service is a pluggable backend that only knows how to complete prompts. This separation lets you swap models, add fallbacks, or route by capability without rewriting your business logic.

Streaming completions

Streaming is essential for UX. Semantic Kernel exposes an async generator via get_streaming_chat_message_contents. Create streaming.py:

import asyncio
import os
import sys
from dotenv import load_dotenv
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.connectors.ai.open_ai.prompt_execution_settings.open_ai_prompt_execution_settings import OpenAIChatPromptExecutionSettings
from semantic_kernel.contents import ChatHistory
from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent

load_dotenv()

API_KEY = os.getenv("N4N_API_KEY")
BASE_URL = "https://api.n4n.ai/v1"
MODEL_ID = os.getenv("SK_MODEL_ID", "gpt-4o-mini")

kernel = Kernel()
chat_service = OpenAIChatCompletion(
    ai_model_id=MODEL_ID,
    api_key=API_KEY,
    base_url=BASE_URL,
)
kernel.add_service(chat_service)

settings = OpenAIChatPromptExecutionSettings(
    temperature=0.3,
    max_tokens=500,
)

history = ChatHistory()
history.add_system_message("You are a senior engineer. Explain concepts clearly with code snippets where helpful.")
history.add_user_message("Show me a Python decorator that retries a function with exponential backoff.")

async def main():
    print("Streaming response:\n")
    full_response = ""
    async for chunk in kernel.get_streaming_chat_message_contents(
        chat_history=history,
        settings=settings,
    ):
        # chunk is a list of StreamingChatMessageContent (usually length 1)
        for msg in chunk:
            if msg.content:
                print(msg.content, end="", flush=True)
                full_response += msg.content
    print("\n\n--- Done ---")

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

Run it:

python streaming.py

Expected output (streaming, so you see tokens appear):

Streaming response:

Here's a retry decorator with exponential backoff:

```python
import time
import random
from functools import wraps
from typing import Callable, Type, Tuple

def retry(
    max_attempts: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    exponential_base: float = 2.0,
    jitter: bool = True,
    exceptions: Tuple[Type[Exception], ...] = (Exception,),
):
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_attempts - 1:
                        raise
                    delay = min(base_delay * (exponential_base ** attempt), max_delay)
                    if jitter:
                        delay *= (0.5 + random.random())
                    time.sleep(delay)
            return None
        return wrapper
    return decorator

Usage:

@retry(max_attempts=5, base_delay=1.0, exceptions=(ConnectionError, TimeoutError))
def flaky_api_call():
    ...

Key points: jitter prevents thundering herd, max_delay caps backoff, and you specify which exceptions trigger retry.

— Done —


## Function calling (tools)

Semantic Kernel calls plugins "functions." You can register native Python functions and let the model invoke them. Create `function_calling.py`:

```python
import asyncio
import json
import os
from typing import Annotated
from dotenv import load_dotenv
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.connectors.ai.open_ai.prompt_execution_settings.open_ai_prompt_execution_settings import OpenAIChatPromptExecutionSettings
from semantic_kernel.contents import ChatHistory
from semantic_kernel.functions import kernel_function

load_dotenv()

API_KEY = os.getenv("N4N_API_KEY")
BASE_URL = "https://api.n4n.ai/v1"
MODEL_ID = os.getenv("SK_MODEL_ID", "gpt-4o-mini")

kernel = Kernel()
chat_service = OpenAIChatCompletion(
    ai_model_id=MODEL_ID,
    api_key=API_KEY,
    base_url=BASE_URL,
)
kernel.add_service(chat_service)

# Define a plugin with native functions
class WeatherPlugin:
    @kernel_function(
        name="get_weather",
        description="Get current weather for a location",
    )
    def get_weather(
        self,
        location: Annotated[str, "City and state, e.g. 'San Francisco, CA'"],
        unit: Annotated[str, "Temperature unit: 'celsius' or 'fahrenheit'"] = "fahrenheit",
    ) -> Annotated[str, "Weather report as JSON string"]:
        # In production, call a real API. Here we mock.
        mock_data = {
            "San Francisco, CA": {"temp": 58, "condition": "foggy", "humidity": 85},
            "New York, NY": {"temp": 72, "condition": "sunny", "humidity": 45},
            "Seattle, WA": {"temp": 61, "condition": "rainy", "humidity": 90},
        }
        data = mock_data.get(location, {"temp": 70, "condition": "unknown", "humidity": 50})
        if unit == "celsius":
            data = {**data, "temp": round((data["temp"] - 32) * 5 / 9)}
        return json.dumps({**data, "location": location, "unit": unit})

# Register the plugin
kernel.add_plugin(WeatherPlugin(), plugin_name="weather")

settings = OpenAIChatPromptExecutionSettings(
    temperature=0.1,
    max_tokens=400,
    tool_choice="auto",  # let the model decide when to call functions
)

history = ChatHistory()
history.add_system_message(
    "You have access to a weather tool. Use it when users ask for current weather. "
    "Respond naturally with the tool result."
)
history.add_user_message("What's the weather in San Francisco right now?")

async def main():
    print("Calling model with function access...\n")
    result = await kernel.get_chat_message_content(
        chat_history=history,
        settings=settings,
    )
    print(result.content)

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

Run it:

python function_calling.py

Expected output:

Calling model with function access...

San Francisco is currently foggy at 58°F with 85% humidity. Classic June weather — the marine layer is doing its thing.

The model invoked weather.get_weather, received the JSON, and synthesized a natural response. Check the raw tool call by enabling debug logging:

import logging
logging.basicConfig(level=logging.DEBUG)

You’ll see the tool_calls and tool messages in the chat history.

Automatic fallback in action

One reason to use n4n.ai is automatic fallback when a provider is rate-limited or degraded. You don’t write retry logic — the gateway handles it. To verify, you can force a fallback by using a model id that maps to multiple providers (most do) and checking response headers.

Create check_fallback.py:

import asyncio
import os
from dotenv import load_dotenv
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.connectors.ai.open_ai.prompt_execution_settings.open_ai_prompt_execution_settings import OpenAIChatPromptExecutionSettings
from semantic_kernel.contents import ChatHistory

load_dotenv()

API_KEY = os.getenv("N4N_API_KEY")
BASE_URL = "https://api.n4n.ai/v1"
MODEL_ID = os.getenv("SK_MODEL_ID", "gpt-4o-mini")

kernel = Kernel()
chat_service = OpenAIChatCompletion(
    ai_model_id=MODEL_ID,
    api_key=API_KEY,
    base_url=BASE_URL,
)
kernel.add_service(chat_service)

settings = OpenAIChatPromptExecutionSettings(temperature=0, max_tokens=50)
history = ChatHistory()
history.add_user_message("Reply with exactly: OK")

async def main():
    # Access the underlying client to inspect response headers
    client = chat_service.client
    response = await client.chat.completions.create(
        model=MODEL_ID,
        messages=[{"role": "user", "content": "Reply with exactly: OK"}],
        max_tokens=10,
        temperature=0,
    )
    print("Response:", response.choices[0].message.content)
    print("Headers:")
    for k, v in response._response.headers.items():
        if k.startswith("x-") or k in ("server", "via"):
            print(f"  {k}: {v}")

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

Run it:

python check_fallback.py

Expected output includes headers like:

Response: OK
Headers:
  x-provider: anthropic
  x-model: claude-3.5-sonnet
  x-fallback-attempted: false
  server: n4n.ai

The x-provider and x-model headers tell you which upstream actually served the request. If the primary is degraded, you’ll see x-fallback-attempted: true and a different provider. This is the gateway honoring provider cache-control hints and routing directives without your code changing.

Per-token usage metering

n4n.ai returns usage in the standard OpenAI format. Semantic Kernel surfaces it via the metadata on the response. Create usage_tracking.py:

import asyncio
import os
from dotenv import load_dotenv
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.connectors.ai.open_ai.prompt_execution_settings.open_ai_prompt_execution_settings import OpenAIChatPromptExecutionSettings
from semantic_kernel.contents import ChatHistory

load_dotenv()

API_KEY = os.getenv("N4N_API_KEY")
BASE_URL = "https://api.n4n.ai/v1"
MODEL_ID = os.getenv("SK_MODEL_ID", "gpt-4o-mini")

kernel = Kernel()
chat_service = OpenAIChatCompletion(
    ai_model_id=MODEL_ID,
    api_key=API_KEY,
    base_url=BASE_URL,
)
kernel.add_service(chat_service)

settings = OpenAIChatPromptExecutionSettings(temperature=0.3, max_tokens=200)
history = ChatHistory()
history.add_system_message("You are a concise engineer.")
history.add_user_message("Explain the difference between a mutex and a semaphore in 3 sentences.")

async def main():
    result = await kernel.get_chat_message_content(
        chat_history=history,
        settings=settings,
    )
    print("Response:", result.content)
    print("\nUsage metadata:", result.metadata)

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

Run it:

python usage_tracking.py

Expected output:

Response: A mutex allows only one thread to access a resource at a time and has ownership — only the locking thread can unlock it. A semaphore maintains a counter permitting N concurrent accesses and has no ownership; any thread can signal. Mutexes are for mutual exclusion; semaphores are for signaling and resource pooling.

Usage metadata: {'usage': {'prompt_tokens': 42, 'completion_tokens': 67, 'total_tokens': 109}, 'model': 'gpt-4o-mini', 'finish_reason': 'stop'}

The usage object gives you prompt, completion, and total tokens — useful for cost tracking, rate limiting your own callers, or alerting.

Putting it together: a reusable kernel factory

You’ll want a single place to construct the kernel with your preferred defaults. Create kernel_factory.py:

import os
from functools import lru_cache
from dotenv import load_dotenv
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.connectors.ai.open_ai.prompt_execution_settings.open_ai_prompt_execution_settings import OpenAIChatPromptExecutionSettings

load_dotenv()

@lru_cache(maxsize=4)
def create_kernel(model_id: str | None = None) -> Kernel:
    """Create a configured Kernel instance. Cached by model_id."""
    api_key = os.getenv("N4N_API_KEY")
    base_url = "https://api.n4n.ai/v1"
    model_id = model_id or os.getenv("SK_MODEL_ID", "gpt-4o-mini")

    if not api_key:
        raise RuntimeError("N4N_API_KEY not set")

    kernel = Kernel()
    chat_service = OpenAIChatCompletion(
        ai_model_id=model_id,
        api_key=api_key,
        base_url=base_url,
    )
    kernel.add_service(chat_service)
    return kernel

def default_settings(**overrides) -> OpenAIChatPromptExecutionSettings:
    """Sensible defaults for most chat workloads."""
    defaults = dict(
        temperature=0.2,
        max_tokens=1000,
        tool_choice="auto",
    )
    defaults.update(overrides)
    return OpenAIChatPromptExecutionSettings(**defaults)

Now your application code stays clean:

from kernel_factory import create_kernel, default_settings
from semantic_kernel.contents import ChatHistory

kernel = create_kernel("claude-3.5-sonnet")
settings = default_settings(temperature=0.1, max_tokens=500)

history = ChatHistory()
history.add_system_message("You are a principal engineer.")
history.add_user_message("When would you choose a semaphore over a mutex?")

result = await kernel.get_chat_message_content(chat_history=history, settings=settings)
print(result.content)

Common pitfalls

Pitfall: Forgetting await on streaming. get_streaming_chat_message_contents returns an async generator. You must async for it — iterating synchronously yields nothing.

Pitfall: Passing model in execution settings. The model is bound to the service at registration (ai_model_id). Setting model in OpenAIChatPromptExecutionSettings is ignored.

Pitfall: Assuming function calling works on all models. Only models with tool-calling support (most frontier models, some open weights) will invoke functions. If the model doesn’t support tools, the kernel returns a text response saying it can’t call functions. Check x-model header to confirm.

Pitfall: Hardcoding base_url in multiple files. Use the factory pattern above. One change propagates everywhere.

What’s next

You now have a working Semantic Kernel setup against n4n.ai with:

Model selection via a single environment variable
Streaming and non-streaming completions
Native function calling with automatic tool dispatch
Visibility into provider routing and fallback via response headers
Per-token usage for metering

From here, explore:

Planners — let the kernel decompose a goal into a sequence of function calls
Memory — semantic memory with vector stores for RAG
Filters — middleware for logging, guardrails, or prompt rewriting
Multi-modal — pass images to gpt-4o, claude-3.5-sonnet, or gemini-1.5-pro via ChatHistory.add_image_message

The kernel is just the orchestration layer. The model capability profile you choose determines what’s possible. Start with gpt-4o-mini for speed, graduate to claude-3.5-sonnet or gpt-4o when reasoning depth matters, and reach for gemini-1.5-pro when context window is the constraint. All behind one endpoint, one API key, zero provider management.