Go to Qiita Advent Calendar 2024 Top
LoginSignup
4
5

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@faunsu(yuichiro)

Semantic Kernel の使い方

Last updated at Posted at 2024-11-24

この記事は？

この記事は Semantic Kernel の使い方を順次追記していく私の個人的なメモです。今後、新たな機能等を試した際に、順次書き進めていきます。

Semantic Kernel とは

以下公式レポジトリ README の書き出しの日本語訳です。

Semantic Kernel は、OpenAI、Azure OpenAI、Hugging Face などの大規模言語モデル（LLM）を、C#、Python、Java といった従来のプログラミング言語と統合するための SDK です。Semantic Kernel は、プラグインを定義し、それらを数行のコードで連結できる仕組みを提供することで、この統合を実現しています。

主な特徴

  • マルチプラットフォーム対応: C#、Python、Java で使用可能
  • プラグインベースの設計: カスタム関数やスキルを簡単に統合
  • AI サービスとのシームレスな接続: OpenAI や Azure OpenAI Service の活用が容易

公式リポジトリ

環境構築

この記事で紹介するコードは以下の手順で環境を構築しています。Semantic Kernel のバージョンは執筆時点で最新の 1.16.0 を使用しています。

必要なツール

  • Python 3.10 以上
  • uv: Python プロジェクト管理ツール

セットアップ手順

プロジェクト初期化

uv init .

Semantic Kernel を追加

uv add semantic-kernel

以下のコードで .env ファイルから環境変数を読み込みます。

config.py
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    """read .env file"""

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
    OPENAI_COMPLETION_DEPLOYMENT_NAME: str
    OPENAI_COMPLETION_ENDPOINT: str
    OPENAI_COMPLETION_API_KEY: str
    SERVICE_ID: str = "default"

app_settings = Settings()  # type: ignore

.env ファイル例:

.env
OPENAI_COMPLETION_DEPLOYMENT_NAME=your-deployment-name
OPENAI_COMPLETION_ENDPOINT=https://your-endpoint.openai.azure.com/
OPENAI_COMPLETION_API_KEY=your-api-key
SERVICE_ID=default

Code

Chat Completion

以下は Semantic Kernel を使用して簡単なチャット推論を行う Python コード例です。この例では Azure OpenAI Service を利用しています。

app.chat_completion.py

import asyncio

from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.function_choice_behavior import (
    FunctionChoiceBehavior,
)
from semantic_kernel.connectors.ai.open_ai import AzureChatPromptExecutionSettings
from semantic_kernel.connectors.ai.open_ai.services.azure_chat_completion import AzureChatCompletion
from semantic_kernel.contents import ChatHistory
from semantic_kernel.functions import KernelArguments

from app.libs.config import app_settings


async def main() -> None:
    kernel = Kernel()
    kernel.add_service(
        service=AzureChatCompletion(
            service_id=app_settings.SERVICE_ID,
            api_key=app_settings.OPENAI_COMPLETION_API_KEY,
            deployment_name=app_settings.OPENAI_COMPLETION_DEPLOYMENT_NAME,
            endpoint=app_settings.OPENAI_COMPLETION_ENDPOINT,
        )
    )

    settings = kernel.get_prompt_execution_settings_from_service_id(service_id=app_settings.SERVICE_ID)

    if isinstance(settings, AzureChatPromptExecutionSettings):
        settings.function_choice_behavior = FunctionChoiceBehavior.Auto(auto_invoke=True)

    service = kernel.get_service(service_id=app_settings.SERVICE_ID)

    if not isinstance(service, AzureChatCompletion):
        raise Exception("Invalid Value")

    history = ChatHistory()
    history.add_user_message("hello")

    result = await service.get_chat_message_contents(
        chat_history=history,
        settings=settings,
        kernel=kernel,
        arguments=KernelArguments(settings=settings),
    )

    if not result:
        raise Exception("result is None")

    print(result[0].content)


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

Function Calling

プラグインを1つだけ登録して function calling します。ポイントは、本来以下のような処理をプログラマ側でしなければならない所、ほぼ自動でやってくれる点でしょうか。（ちゃんと実装していくともっと色々書かないといけないけど）

  1. 関数宣言の準備
  2. ユーザーの入力 + 1 による推論
  3. 2 の結果を元に関数の実行
  4. 3 の結果を再度 2 で得ている tool call id を添えて推論
  5. ユーザーへの応答
import asyncio

from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.function_choice_behavior import (
    FunctionChoiceBehavior,
)
from semantic_kernel.connectors.ai.open_ai import AzureChatPromptExecutionSettings
from semantic_kernel.connectors.ai.open_ai.services.azure_chat_completion import AzureChatCompletion
from semantic_kernel.contents import ChatHistory
from semantic_kernel.core_plugins.math_plugin import MathPlugin
from semantic_kernel.functions import KernelArguments

from app.libs.config import app_settings


async def main() -> None:
    kernel = Kernel()
    kernel.add_service(
        service=AzureChatCompletion(
            service_id=app_settings.SERVICE_ID,
            api_key=app_settings.OPENAI_COMPLETION_API_KEY,
            deployment_name=app_settings.OPENAI_COMPLETION_DEPLOYMENT_NAME,
            endpoint=app_settings.OPENAI_COMPLETION_ENDPOINT,
        )
    )

    # 計算用のプラグインを追加
    kernel.add_plugin(plugin=MathPlugin(), plugin_name="math")

    settings = kernel.get_prompt_execution_settings_from_service_id(service_id=app_settings.SERVICE_ID)

    if isinstance(settings, AzureChatPromptExecutionSettings):
        settings.function_choice_behavior = FunctionChoiceBehavior.Auto(auto_invoke=True)

    service = kernel.get_service(service_id=app_settings.SERVICE_ID)

    if not isinstance(service, AzureChatCompletion):
        raise Exception("Invalid Value")

    history = ChatHistory()
    history.add_user_message("3 + 3 = ?")

    # 内部で function call + 最終的な推論を行う
    result = await service.get_chat_message_contents(
        chat_history=history,
        settings=settings,
        kernel=kernel,
        arguments=KernelArguments(settings=settings),
    )

    if not result:
        raise Exception("result is None")

    print(result[0].content)


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

Speech to Text

Semantic Kernel を使用して、Whisper による Speech to text を行う Python コード例です。 wav を直接取り込んでも良いですが、 pyaudio を用いて録音する部分も記載しています。

app.stt.py

import asyncio
import os
import tempfile
import wave

import pyaudio  # type: ignore
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import AzureAudioToText
from semantic_kernel.contents import AudioContent

from app.libs.config import app_settings


async def main() -> None:
    kernel = Kernel()
    kernel.add_service(
        service=AzureAudioToText(
            service_id=f"{app_settings.SERVICE_ID}_audio",
            deployment_name="whisper",
            api_key=app_settings.OPENAI_COMPLETION_API_KEY,
            endpoint=app_settings.OPENAI_COMPLETION_ENDPOINT,
        )
    )

    service = kernel.get_service(service_id=f"{app_settings.SERVICE_ID}_audio")
    if not isinstance(service, AzureAudioToText):
        raise Exception("Invalid Value")

    wav_file_path: str = record_audio_to_tempfile(5)

    try:
        result = await service.get_text_content(AudioContent.from_audio_file(path=wav_file_path))
        print(result)
    finally:
        print(f"一時ファイル '{wav_file_path}' を削除します...")
        os.remove(wav_file_path)


def record_audio_to_tempfile(duration: int) -> str:
    """音声を録音して一時ファイルに保存"""

    FORMAT = pyaudio.paInt16  # 16ビットフォーマット
    CHANNELS = 1  # モノラル
    RATE = 44100  # サンプルレート（44.1kHz）
    CHUNK = 1024  # バッファサイズ

    # 録音開始
    audio = pyaudio.PyAudio()
    stream = audio.open(format=FORMAT, channels=CHANNELS, rate=RATE, input=True, frames_per_buffer=CHUNK,)
    print("録音開始...")

    frames = []
    try:
        frames = [stream.read(num_frames=CHUNK) for _ in range(int(RATE / CHUNK * duration))]
    finally:
        print("録音終了。")
        stream.stop_stream()
        stream.close()
        audio.terminate()

    with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as temp_file:
        with wave.open(f=temp_file.name, mode="wb") as wf:
            wf.setnchannels(nchannels=CHANNELS)
            wf.setsampwidth(sampwidth=audio.get_sample_size(format=FORMAT))
            wf.setframerate(framerate=RATE)
            wf.writeframes(data=b"".join(frames))
        return temp_file.name


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

Multi Agent

複数の Agent を定義してグループチャットで自動的に会話し、結論を返却できます。

from pydantic_core import Url
from semantic_kernel import Kernel
from semantic_kernel.agents import Agent, AgentGroupChat
from semantic_kernel.agents.open_ai import AzureAssistantAgent
from semantic_kernel.agents.strategies.termination.termination_strategy import TerminationStrategy
from semantic_kernel.contents import AuthorRole, ChatMessageContent

from app.libs.config import app_settings

# 終了ロジックのカスタム
class ApprovalTerminationStrategy(TerminationStrategy):
    async def should_agent_terminate(self, agent: Agent, history: list[ChatMessageContent]) -> bool:
        return False

# Agent 切り替えロジックをカスタムする場合
# from semantic_kernel.agents.strategies.selection.selection_strategy import SelectionStrategy
# class ApprovalSelectionStrategy(SelectionStrategy):
#     async def next(self, agents: list[Agent], history: list[ChatMessageContent]) -> Agent:
#         return await super().next(agents, history)


async def main():
    kernel = Kernel()
    agent_a = await AzureAssistantAgent.create(
        # id="asst_********************", # 既存 Agent に接続する場合
        service_id=app_settings.SERVICE_ID + "_a",
        kernel=kernel,
        api_key=app_settings.OPENAI_COMPLETION_API_KEY,
        deployment_name=app_settings.OPENAI_COMPLETION_DEPLOYMENT_NAME,
        api_version="2024-05-01-preview",
        endpoint=Url(app_settings.OPENAI_COMPLETION_ENDPOINT),
        enable_code_interpreter=False,
        enable_file_search=False,
        instructions="あなたは漫才師であり、ボケ担当です。ボケてください。ツッコミ担当から返事をします。",
        max_completion_tokens=200,
    )

    agent_b = await AzureAssistantAgent.create(
        # id="asst_********************", # 既存 Agent に接続する場合
        service_id=app_settings.SERVICE_ID + "_b",
        kernel=kernel,
        api_key=app_settings.OPENAI_COMPLETION_API_KEY,
        deployment_name=app_settings.OPENAI_COMPLETION_DEPLOYMENT_NAME,
        api_version="2024-05-01-preview",
        endpoint=Url(app_settings.OPENAI_COMPLETION_ENDPOINT),
        enable_code_interpreter=False,
        enable_file_search=False,
        instructions="あなたは漫才師であり、ツッコミ担当です。雑談をしつつ、ボケが来たら突っ込みます。",
        max_completion_tokens=200,
    )

    group_chat = AgentGroupChat(
        agents=[agent_a, agent_b],
        termination_strategy=ApprovalTerminationStrategy(maximum_iterations=3, automatic_reset=False),
    )
    await group_chat.add_chat_message(ChatMessageContent(role=AuthorRole.USER, content="お題は「バナナ」です。"))

    async for message_content in group_chat.invoke():
        print(">>", message_content.content)


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

... 以降工事中 - 順次記載 ...

4
5
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
4
5

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?