5장: Semantic Kernel -- 엔터프라이즈 AI 오케스트레이션

from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import (
    AzureChatCompletion,
    AzureTextEmbedding,
)
 
# Kernel 생성
kernel = Kernel()
 
# AI 서비스 등록
kernel.add_service(
    AzureChatCompletion(
        service_id="chat",
        deployment_name="gpt-4o",
        endpoint="https://myorg.openai.azure.com/",
        api_key="your-api-key",
    )
)
 
kernel.add_service(
    AzureTextEmbedding(
        service_id="embedding",
        deployment_name="text-embedding-3-small",
        endpoint="https://myorg.openai.azure.com/",
        api_key="your-api-key",
    )
)

using Microsoft.SemanticKernel;
 
var builder = Kernel.CreateBuilder();
 
builder.AddAzureOpenAIChatCompletion(
    deploymentName: "gpt-4o",
    endpoint: "https://myorg.openai.azure.com/",
    apiKey: "your-api-key"
);
 
builder.AddAzureOpenAITextEmbeddingGeneration(
    deploymentName: "text-embedding-3-small",
    endpoint: "https://myorg.openai.azure.com/",
    apiKey: "your-api-key"
);
 
var kernel = builder.Build();

from semantic_kernel.functions import kernel_function
 
class WeatherPlugin:
    """날씨 정보를 제공하는 플러그인"""
 
    @kernel_function(
        name="get_weather",
        description="특정 도시의 현재 날씨를 조회합니다",
    )
    def get_weather(self, city: str) -> str:
        # 실제 날씨 API 호출 로직
        return f"{city}의 현재 기온은 18도이며 맑음입니다."
 
    @kernel_function(
        name="get_forecast",
        description="특정 도시의 3일간 날씨 예보를 조회합니다",
    )
    def get_forecast(self, city: str, days: int = 3) -> str:
        return f"{city}의 {days}일 예보: 맑음 → 흐림 → 비"
 
# Kernel에 플러그인 등록
kernel.add_plugin(WeatherPlugin(), plugin_name="Weather")

from semantic_kernel.functions import KernelFunctionFromPrompt
 
# 프롬프트 기반 함수 정의
summarize_function = KernelFunctionFromPrompt(
    function_name="summarize",
    plugin_name="TextUtils",
    prompt="""
다음 텍스트를 {{$max_sentences}}문장으로 요약해주세요.
 
텍스트:
{{$input}}
 
요약:
""",
    description="텍스트를 지정된 문장 수로 요약합니다",
)
 
kernel.add_function(
    plugin_name="TextUtils",
    function=summarize_function,
)
 
# 실행
result = await kernel.invoke(
    function=summarize_function,
    input="긴 텍스트 내용...",
    max_sentences="3",
)

from semantic_kernel.connectors.ai.open_ai import (
    OpenAIChatPromptExecutionSettings,
)
from semantic_kernel.connectors.ai.function_choice_behavior import (
    FunctionChoiceBehavior,
)
 
settings = OpenAIChatPromptExecutionSettings(
    function_choice_behavior=FunctionChoiceBehavior.Auto()
)
 
result = await kernel.invoke_prompt(
    prompt="서울의 내일 날씨를 알려줘",
    settings=settings,
)
# LLM이 자동으로 Weather.get_forecast를 호출

from semantic_kernel.planners import FunctionCallingStepwisePlanner
 
planner = FunctionCallingStepwisePlanner(
    service_id="chat",
    max_iterations=10,
)
 
# 복합 목표 제시
result = await planner.invoke(
    kernel=kernel,
    question="서울의 내일 날씨를 확인하고, 날씨에 맞는 옷차림을 추천해주세요.",
)
# 플래너가 자동으로:
# 1. Weather.get_forecast("서울") 호출
# 2. 결과를 바탕으로 옷차림 추천 생성

from semantic_kernel.agents import ChatCompletionAgent, AgentGroupChat
from semantic_kernel.agents.strategies import (
    TerminationStrategy,
    SelectionStrategy,
)
 
# 에이전트 정의
researcher = ChatCompletionAgent(
    kernel=kernel,
    service_id="chat",
    name="Researcher",
    instructions="당신은 정보를 조사하는 연구원입니다.",
)
 
writer = ChatCompletionAgent(
    kernel=kernel,
    service_id="chat",
    name="Writer",
    instructions="당신은 조사 결과를 정리하는 작가입니다.",
)
 
reviewer = ChatCompletionAgent(
    kernel=kernel,
    service_id="chat",
    name="Reviewer",
    instructions="당신은 문서의 품질을 검토하는 리뷰어입니다.",
)
 
# 그룹 채팅으로 협업
group_chat = AgentGroupChat(
    agents=[researcher, writer, reviewer],
    termination_strategy=TerminationStrategy(max_iterations=10),
)
 
async for message in group_chat.invoke("AI 오케스트레이션에 대한 보고서를 작성해주세요"):
    print(f"[{message.name}]: {message.content}")

from semantic_kernel.connectors.ai.open_ai import (
    OpenAIChatPromptExecutionSettings,
)
 
settings = OpenAIChatPromptExecutionSettings(
    max_tokens=4096,
    temperature=0.7,
)
 
# 실행 후 토큰 사용량 확인
result = await kernel.invoke_prompt(
    prompt="AI 오케스트레이션을 설명해주세요",
    settings=settings,
)
 
# 메타데이터에서 토큰 정보 추출
metadata = result.metadata
if metadata:
    usage = metadata.get("usage")
    print(f"입력 토큰: {usage.prompt_tokens}")
    print(f"출력 토큰: {usage.completion_tokens}")
    print(f"총 토큰: {usage.total_tokens}")

from semantic_kernel.filters import FunctionInvocationFilter
 
class RBACFilter(FunctionInvocationFilter):
    """역할 기반 접근 제어 필터"""
 
    def __init__(self, allowed_roles: dict[str, list[str]]):
        self.allowed_roles = allowed_roles
 
    async def on_function_invocation(self, context, next_handler):
        user_role = context.arguments.get("user_role", "viewer")
        function_name = context.function.name
        plugin_name = context.function.plugin_name
 
        full_name = f"{plugin_name}.{function_name}"
 
        if full_name not in self.allowed_roles.get(user_role, []):
            raise PermissionError(
                f"역할 '{user_role}'은 '{full_name}' 함수에 접근할 수 없습니다."
            )
 
        await next_handler(context)
 
# 필터 등록
kernel.add_filter(
    filter_type="function_invocation",
    filter=RBACFilter(
        allowed_roles={
            "admin": ["Weather.get_weather", "Weather.get_forecast", "DB.query"],
            "user": ["Weather.get_weather"],
            "viewer": [],
        }
    ),
)

from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient
 
# Azure Key Vault를 통한 안전한 자격 증명 관리
credential = DefaultAzureCredential()
secret_client = SecretClient(
    vault_url="https://myorg-vault.vault.azure.net/",
    credential=credential,
)
 
api_key = secret_client.get_secret("openai-api-key").value
 
kernel.add_service(
    AzureChatCompletion(
        service_id="chat",
        deployment_name="gpt-4o",
        endpoint="https://myorg.openai.azure.com/",
        api_key=api_key,
    )
)

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
 
# SK는 OpenTelemetry를 통해 텔레메트리 데이터를 생성
provider = TracerProvider()
trace.set_tracer_provider(provider)
 
# Azure Application Insights와 연동
from azure.monitor.opentelemetry.exporter import (
    AzureMonitorTraceExporter,
)
from opentelemetry.sdk.trace.export import BatchSpanExporter
 
exporter = AzureMonitorTraceExporter(
    connection_string="InstrumentationKey=..."
)
provider.add_span_processor(BatchSpanExporter(exporter))