Semantic Kernel実践ガイド - Microsoftのエンタープライズ級AIオーケストレーション

from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion

# Kernel初期化
kernel = Kernel()

# LLMサービス追加
kernel.add_service(
    OpenAIChatCompletion(
        service_id="gpt-4",
        api_key="your-api-key",
        model_id="gpt-4"
    )
)

# シンプルなプロンプト実行
result = await kernel.invoke_prompt("Tokyoの観光スポットを3つ教えて")
print(result)

using Microsoft.SemanticKernel;
using Microsoft.SemanticKernel.Connectors.OpenAI;

// Kernel構築
var builder = Kernel.CreateBuilder();
builder.Services.AddOpenAIChatCompletion(
    "gpt-4",
    "your-api-key"
);
var kernel = builder.Build();

// プロンプト実行
var result = await kernel.InvokePromptAsync("Tokyoの観光スポットを3つ教えて");
Console.WriteLine(result);

from semantic_kernel.functions import kernel_function

class MathPlugin:
    @kernel_function(
        name="Add",
        description="2つの数値を加算します"
    )
    def add(self, a: int, b: int) -> int:
        return a + b
    
    @kernel_function(
        name="Multiply",
        description="2つの数値を乗算します"
    )
    def multiply(self, a: int, b: int) -> int:
        return a * b

# Pluginをkernelに追加
kernel.add_plugin(MathPlugin(), plugin_name="Math")

# 関数呼び出し
result = await kernel.invoke_function(
    plugin_name="Math",
    function_name="Add",
    a=10,
    b=20
)
print(result)  # 30

from semantic_kernel.connectors.search_engine import BingConnector

# Bing検索プラグイン
bing = BingConnector(api_key="bing-api-key")
kernel.add_plugin(bing, plugin_name="BingSearch")

# LLMが自動でツール選択
result = await kernel.invoke_prompt(
    "2025年のAI技術トレンドを調べて要約して",
    functions=kernel.plugins["BingSearch"]
)

from semantic_kernel.planners import SequentialPlanner

# プランナー作成
planner = SequentialPlanner(kernel)

# 複雑なタスクを自動分解
plan = await planner.create_plan("顧客データを分析し、レポートをメールで送信して")

# プラン実行
result = await plan.invoke()

1. データベースから顧客データを取得（DBPlugin.GetCustomers）
2. データ分析を実行（AnalyticsPlugin.Analyze）
3. レポート生成（ReportPlugin.GenerateReport）
4. メール送信（EmailPlugin.SendEmail）

from semantic_kernel.agents import Agent, AgentGroupChat
from autogen import AssistantAgent, UserProxyAgent

# Semantic Kernelエージェント
researcher = Agent(
    name="Researcher",
    instructions="ウェブから最新情報を収集します",
    kernel=kernel
)

writer = Agent(
    name="Writer",
    instructions="収集した情報を元に記事を作成します",
    kernel=kernel
)

# エージェント協調
chat = AgentGroupChat(agents=[researcher, writer])

# タスク実行
result = await chat.invoke("AIエージェントの最新トレンドについて記事を書いて")

from semantic_kernel.agents import SequentialAgentFlow

flow = SequentialAgentFlow()
flow.add_agent(data_collector)
flow.add_agent(analyzer)
flow.add_agent(reporter)

result = await flow.run("月次レポートを作成")

manager = Agent(
    name="Manager",
    instructions="タスクを分解し、各エージェントに割り当てます",
    kernel=kernel
)

workers = [researcher_agent, coder_agent, reviewer_agent]

result = await manager.orchestrate(
    task="新機能の設計と実装",
    available_agents=workers
)

from semantic_kernel.reliability import RetryHandler
from semantic_kernel.security import ContentFilter

# コンテンツフィルタリング
kernel.add_filter(ContentFilter(
    block_harmful_content=True,
    pii_detection=True
))

# リトライポリシー
kernel.add_handler(RetryHandler(
    max_retries=3,
    backoff_factor=2.0
))

# プロンプトテンプレート
template = """
あなたは{{$role}}です。
タスク: {{$task}}
制約: {{$constraints}}

回答:
"""

result = await kernel.invoke_prompt_template(
    template,
    role="データアナリスト",
    task="売上データを分析",
    constraints="機密情報は含めない"
)

from semantic_kernel.memory import VolatileMemoryStore

# メモリストア
memory = VolatileMemoryStore()
kernel.register_memory_store(memory)

# コンテキスト保存
await kernel.memory.save_information(
    collection="customer_interactions",
    text="顧客Aは製品Bに興味を示した",
    id="interaction_001"
)

# コンテキスト取得
relevant_info = await kernel.memory.search(
    collection="customer_interactions",
    query="製品Bの購入履歴",
    limit=5
)

from semantic_kernel.agents import Agent

# サポートエージェント
support_agent = Agent(
    name="CustomerSupport",
    instructions="""
    あなたはカスタマーサポート担当です。
    - 顧客の質問に丁寧に回答
    - 必要に応じてFAQやドキュメントを検索
    - 解決できない場合は人間にエスカレーション
    """,
    kernel=kernel
)

# FAQプラグイン
class FAQPlugin:
    @kernel_function(description="FAQを検索")
    async def search_faq(self, query: str) -> str:
        # Vector DBから検索
        results = await vector_db.search(query, top_k=3)
        return "\n".join(results)

kernel.add_plugin(FAQPlugin(), "FAQ")

# サポート対応
response = await support_agent.invoke("返品ポリシーを教えてください")