Интеграция Semantic Kernel (Microsoft) для AI-оркестрации

Интеграция Semantic Kernel для AI-оркестрации

Semantic Kernel (SK) — SDK Microsoft для интеграции LLM в .NET, Python и Java приложения. Ориентирован на enterprise-разработчиков, которым нужна строгая типизация, dependency injection, интеграция с Azure AI и корпоративными системами. В отличие от LangChain и LlamaIndex, SK предоставляет SDK-опыт, близкий к традиционному enterprise-разработке.

Базовая структура Semantic Kernel

import asyncio
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion, OpenAITextEmbedding
from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior
from semantic_kernel.functions import kernel_function
from semantic_kernel.prompt_template import PromptTemplateConfig

kernel = Kernel()

# Добавление AI-сервисов
kernel.add_service(OpenAIChatCompletion(
    service_id="gpt4o",
    ai_model_id="gpt-4o",
))

kernel.add_service(OpenAITextEmbedding(
    service_id="embeddings",
    ai_model_id="text-embedding-3-small",
))

# Промпт как функция ядра
prompt = """Ты — аналитик корпоративных данных.
Ответь на вопрос на основе предоставленного контекста.

Контекст: {{$context}}
Вопрос: {{$question}}"""

settings = kernel.get_prompt_execution_settings_from_service_id("gpt4o")
settings.max_tokens = 2000
settings.temperature = 0.1

analysis_function = kernel.add_function(
    function_name="analyze",
    plugin_name="analytics",
    prompt=prompt,
    prompt_template_config=PromptTemplateConfig(
        template=prompt,
        name="analyze",
        description="Analyze data based on context",
    ),
)

async def run():
    result = await kernel.invoke(
        analysis_function,
        context="Выручка Q1 2025: 45.2M, план: 48M, отклонение: -5.8%",
        question="Каковы основные причины отклонения и что рекомендуете?",
    )
    print(result)

asyncio.run(run())

Plugins: повторно используемые компоненты

from semantic_kernel.functions import kernel_function
from typing import Annotated

class FinancialPlugin:
    """Plugin для финансового анализа"""

    @kernel_function(
        name="calculate_variance",
        description="Рассчитать отклонение план-факт в процентах",
    )
    def calculate_variance(
        self,
        actual: Annotated[float, "Фактическое значение"],
        plan: Annotated[float, "Плановое значение"],
    ) -> Annotated[str, "Процент отклонения"]:
        if plan == 0:
            return "Ошибка: плановое значение равно нулю"
        variance = (actual - plan) / plan * 100
        return f"{variance:+.2f}%"

    @kernel_function(
        name="format_currency",
        description="Форматировать число как валюту",
    )
    def format_currency(
        self,
        amount: Annotated[float, "Сумма"],
        currency: Annotated[str, "Валюта (RUB, USD, EUR)"] = "RUB",
    ) -> str:
        symbols = {"RUB": "₽", "USD": "$", "EUR": "€"}
        symbol = symbols.get(currency, currency)
        return f"{symbol}{amount:,.0f}"

# Регистрация плагина
kernel.add_plugin(FinancialPlugin(), plugin_name="finance")

# Плагин из директории с YAML/txt промптами
kernel.add_plugin(parent_directory="./plugins", plugin_name="reporting")

Auto Function Calling: агентный цикл

from semantic_kernel.connectors.ai.open_ai import OpenAIChatPromptExecutionSettings
from semantic_kernel.contents import ChatHistory
from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior

# Настройки с автоматическим вызовом функций
execution_settings = OpenAIChatPromptExecutionSettings(
    service_id="gpt4o",
    function_choice_behavior=FunctionChoiceBehavior.Auto(
        auto_invoke=True,  # Автоматически вызывать функции
        maximum_auto_invoke_attempts=10,
    ),
)

chat_service = kernel.get_service("gpt4o")
chat_history = ChatHistory()
chat_history.add_system_message("""Ты — корпоративный финансовый аналитик.
Используй доступные функции для точных расчётов.
Отвечай только на основе данных.""")

chat_history.add_user_message("Рассчитай отклонение выручки: факт 42.3M, план 45.0M. Выведи в рублях.")

result = await chat_service.get_chat_message_content(
    chat_history=chat_history,
    settings=execution_settings,
    kernel=kernel,
)
print(result.content)
# Агент автоматически вызовет calculate_variance и format_currency

Memory и Vector Store

from semantic_kernel.memory.semantic_text_memory import SemanticTextMemory
from semantic_kernel.connectors.memory.chroma import ChromaMemoryStore

memory_store = ChromaMemoryStore(persist_directory="./chroma_db")
memory = SemanticTextMemory(storage=memory_store, embeddings_generator=kernel.get_service("embeddings"))

# Сохранение информации в память
await memory.save_information(
    collection="company_policies",
    id="policy_001",
    text="Политика командировочных расходов: суточные 2500 руб/день в РФ, 80 USD за рубежом.",
    description="Командировки",
)

# Поиск по памяти
results = await memory.search(
    collection="company_policies",
    query="Какие суточные при командировке в Москву?",
    limit=3,
    min_relevance_score=0.7,
)

for result in results:
    print(f"Score: {result.relevance:.3f}: {result.text}")

Интеграция с Azure AI

from semantic_kernel.connectors.ai.azure_ai_inference import AzureAIInferenceChatCompletion
from azure.ai.inference.aio import ChatCompletionsClient
from azure.identity.aio import DefaultAzureCredential

# Azure OpenAI
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
kernel.add_service(AzureChatCompletion(
    service_id="azure-gpt4o",
    deployment_name="gpt-4o",
    endpoint="https://your-endpoint.openai.azure.com",
    api_key="...",
))

# Azure AI Foundry (Phi, Mistral, Llama через Azure)
client = ChatCompletionsClient(
    endpoint="https://your-model.inference.ai.azure.com",
    credential=DefaultAzureCredential(),
)
kernel.add_service(AzureAIInferenceChatCompletion(
    service_id="phi-4",
    ai_model_id="phi-4",
    client=client,
))

Практический кейс: .NET enterprise-приложение с AI

Контекст: крупная логистическая компания (.NET/C# backend) интегрировала SK для создания AI-ассистента диспетчера.

Плагины:

• ShipmentPlugin — запросы к TMS (транспортная система)
• RoutePlugin — расчёт маршрутов, стоимости, сроков
• CustomerPlugin — данные клиентов, история заказов
• AlertPlugin — отправка уведомлений о задержках

// C# версия
var kernel = Kernel.CreateBuilder()
    .AddAzureOpenAIChatCompletion(deploymentName, endpoint, apiKey)
    .Build();

kernel.Plugins.AddFromType<ShipmentPlugin>();
kernel.Plugins.AddFromType<RoutePlugin>();

var settings = new OpenAIPromptExecutionSettings {
    FunctionChoiceBehavior = FunctionChoiceBehavior.Auto()
};

var response = await kernel.InvokePromptAsync(
    "Где сейчас груз по накладной TN-12345? Есть ли задержки?",
    new KernelArguments(settings)
);

Результаты:

• Время ответа диспетчера на запрос клиента: 4.5 мин → 45 сек
• Интеграция в существующий .NET стек: без переработки архитектуры
• Покрытие запросов без участия диспетчера: 68%

Сроки

• Базовая интеграция SK + OpenAI/Azure: 2–4 дня
• Разработка плагинов для бизнес-логики: 1–2 недели
• Агентный цикл с auto function calling: 1 неделя
• Интеграция с корпоративными .NET системами: 2–4 недели