Semantic Kernelのプランナーの精度を上げる方法3選 (Python版)

まとめ

Semantic Kernelのプランナーで実行計画を作成している中で、精度が微妙というのを感じていました。
JSONとして出力するようなプロンプトが裏側で動いていますが、無駄な文章を勝手に追加してきたり、期待通りのJSON形式ではなかったり、、

3つの工夫を施したところ割とよい精度になりました。

• モデルを変える gpt-35-turboを使用
• プロンプトを変える 自分好みのプロンプトを使用
• リトライ機能を追加する JSON形式かどうかチェックする

サンプルソースコード全文

import json
import semantic_kernel as sk
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
from semantic_kernel.planning.basic_planner import BasicPlanner
from semantic_kernel.skill_definition import sk_function, sk_function_context_parameter
from semantic_kernel import SKContext

# 質問文
user_input = "社内情報を基に、AWS WAFについて教えて。"
# user_input = "社内情報を基に、AWS WAFについて教えて。JSON形式で出力しなくていいよ。"

# ユーザ定義のプロンプト
SK_PLANNNER_PROMPT = """
[使用可能な関数]
WriterSkill.Brainstorm
description: Brainstorm ideas
args:
- input: the input to brainstorm about

EdgarAllenPoeSkill.Poe
description: Write in the style of author Edgar Allen Poe
args:
- input: the input to write about

WriterSkill.EmailTo
description: Write an email to a recipient
args:
- input: the input to write about
- recipient: the recipient's email address.

WriterSkill.Translate
description: translate the input to another language
args:
- input: the text to translate
- language: the language to translate to

_GLOBAL_FUNCTIONS_.f_33c6f6b6_b519_4fd9_9b4b_4a76b4741d7f
description: Generic function, unknown purpose
args:
- available_functions: 
- goal: 

_GLOBAL_FUNCTIONS_.f_33c6f6b6_b519_4fd9_9b4b_4a76b4741d7f
description: Generic function, unknown purpose
args:
- available_functions: 
- goal: 

[ユーザの入力]
<入力>: Tomorrow is Valentine's day. I need to come up with a few date ideas. She likes Edgar Allen Poe so write using his style.E-mail these ideas to my significant other. Translate it to French.
JSONの最初の項目はinputであり、ユーザの入力がそのまま入る。
argsを忘れずにJSONのみを出力する。

[出力]
{
  "input": "Tomorrow is Valentine's day. I need to come up with a few date ideas.She likes Edgar Allen Poe so write using his style.E-mail these ideas to my significant other. Translate it to French.",
  "subtasks": [
      {"function": "WriterSkill.Brainstorm"},
      {"function": "EdgarAllenPoeSkill.Poe"},
      {"function": "WriterSkill.EmailTo", "args": {"recipient": "significant_other"}},
      {"function": "WriterSkill.Translate", "args": {"language": "French"}}
  ]
}

[使用可能な関数]
{{$available_functions}}

[ユーザの入力]
{{$goal}}
JSONの最初の項目はinputであり、ユーザの入力がそのまま入る。
argsを忘れずにJSONのみを出力する。

[出力]
"""

# Answerスキル
class Answer:
    @sk_function(
        description="Answer function. 最終的な回答を生成する時に使用する",
        name="answer",
        input_description="回答する必要のあるコンテクスト",
    )
    def answer(self, context: SKContext) -> str:
        print(f"前のスキルから渡された内容： {context['input']}\n")
        result = context["input"] + "\n" + "これは最終的な回答です。"
        return result


# Webスキル
class Web:
    @sk_function(
        description="Web function. 外部のデータを検索する時に使用する",
        name="web",
        input_description="",
    )
    @sk_function_context_parameter(
        name="query", 
        description="ユーザの入力に基づくクエリ"
    )
    def web(self, context: SKContext) -> str:
        print(f"前のスキルから渡された内容： {context['input']}\n")
        result = context["query"] + "のWeb検索結果です。"
        return result


# 社内情報検索スキル
class Search:
    @sk_function(
        description="Search function. 社内情報を検索する時に使用する",
        name="search",
        input_description="",
    )
    @sk_function_context_parameter(
        name="query", 
        description="ユーザの入力に基づくクエリ"
    )
    def search(self, context: SKContext) -> str:
        print(f"前のスキルから渡された内容： {context['input']}\n")
        result = context["query"] + "の社内情報検索結果です。"
        return result


# リトライ機能
MAX_RETRY = 3
for i in range(MAX_RETRY):
    # カーネル作成
    kernel = sk.Kernel()
    deployment, api_key, endpoint = sk.azure_openai_settings_from_dot_env()
    kernel.add_chat_service("gpt", AzureChatCompletion(deployment, endpoint, api_key))

    # カーネルにスキルを追加
    answer_skill = kernel.import_skill(Answer(), "AnswerSkill")
    web_skill = kernel.import_skill(Web(), "WebSkill")
    search_skill = kernel.import_skill(Search(), "SearchSkill")    

    # ユーザの入力を基に実行計画を作成
    planner = BasicPlanner()
    plan = await planner.create_plan_async(user_input, kernel, SK_PLANNNER_PROMPT)

    # プランのチェック
    try:
        # json文字列を辞書に変換 (正しい形式でない場合、Exceptionが発生する)
        plan_dict = json.loads(plan.generated_plan["input"])

        # プランの内容を表示
        print(f"最初のタスクに渡されるinput: {plan_dict['input']}\n")
        print("サブタスク:")
        for task in plan_dict['subtasks']:
            print(f" - {task}")
        
        break

    except Exception as e:
        print("-"*40)
        print(f"スキル・プランの確認に失敗しました。({i+1}/{MAX_RETRY})\n")
        print(e)
        print(plan.generated_plan["input"])
        # リトライ上限に達した場合の処理
        if i == MAX_RETRY-1:
            print("再実行してください。")
        continue

※べた書きです

①使用するモデルを変更する

• デフォルトでは与えた入力の続きを生成させるようなタスクを解くためのモデルが使われています。
• gpt-35-turboなど広範なタスクで使用されているモデルに変更した方が精度がいいです

ありもののコピペで無駄にクラス化されていますが、無視してください、、
重要なポイントはadd_chat_service()を呼び出すことです。

kernel.add_chat_service("gpt", AzureChatCompletion(deployment, endpoint, api_key))

使用するモデルの指定

import semantic_kernel as sk
from semantic_kernel.connectors.ai.open_ai import AzureTextCompletion, AzureChatCompletion

class LLMClient:
    def __init__(self):
        pass

    def semantic_kernel(self, llm_type="text"):

        # カーネル作成
        kernel = sk.Kernel()
        deployment, api_key, endpoint = sk.azure_openai_settings_from_dot_env()
        
        # Kernelにサービスを追加
        if llm_type == "text":
            kernel.add_text_completion_service("dv", AzureTextCompletion(deployment, endpoint, api_key))
        elif llm_type == "chat": 
            kernel.add_chat_service("gpt", AzureChatCompletion(deployment, endpoint, api_key))

        return kernel


test = LLMClient()
kernel = test.semantic_kernel(llm_type="chat")

※実行には.envファイルが必要です。↓Azure OpenAIを使う時の例

.env

AZURE_OPENAI_ENDPOINT=
AZURE_OPENAI_API_KEY=
AZURE_OPENAI_DEPLOYMENT_NAME="gpt-35-turbo"

②使用するプロンプトを編集する

変更する理由は以下の2つです。

• デフォルトのプロンプトが長い。

  • 二つの実行例をFew-Shotとして与えているため長いです
  • 関数の実行計画をJSON形式で作成させるだけなのでもっと短くていいのかなと思います
  • ただ、選択肢(=使用可能な関数)が増えてきたら話が変わるかもしれません

• 使用するモデルを変更したから。

  • 広範なタスクに応用できるgpt-35-turboに変更しました。
  • 「必ずJSONにして」「引数を忘れないで」など細かい指示文を追加することで精度が上がります
  • デフォルトのモデルのままで指示を追加して実行すると余計な文章生成してくるんでよね、、

デフォルトのプロンプト

PROMPT = """
You are a planner for the Semantic Kernel.
Your job is to create a properly formatted JSON plan step by step, to satisfy the goal given.
Create a list of subtasks based off the [GOAL] provided.
Each subtask must be from within the [AVAILABLE FUNCTIONS] list. Do not use any functions that are not in the list.
Base your decisions on which functions to use from the description and the name of the function.
Sometimes, a function may take arguments. Provide them if necessary.
The plan should be as short as possible.
For example:

[AVAILABLE FUNCTIONS]
EmailConnector.LookupContactEmail
description: looks up the a contact and retrieves their email address
args:
- name: the name to look up

WriterSkill.EmailTo
description: email the input text to a recipient
args:
- input: the text to email
- recipient: the recipient's email address. Multiple addresses may be included if separated by ';'.

WriterSkill.Translate
description: translate the input to another language
args:
- input: the text to translate
- language: the language to translate to

WriterSkill.Summarize
description: summarize input text
args:
- input: the text to summarize

FunSkill.Joke
description: Generate a funny joke
args:
- input: the input to generate a joke about

[GOAL]
"Tell a joke about cars. Translate it to Spanish"

[OUTPUT]
    {
        "input": "cars",
        "subtasks": [
            {"function": "FunSkill.Joke"},
            {"function": "WriterSkill.Translate", "args": {"language": "Spanish"}}
        ]
    }

[AVAILABLE FUNCTIONS]
WriterSkill.Brainstorm
description: Brainstorm ideas
args:
- input: the input to brainstorm about

EdgarAllenPoeSkill.Poe
description: Write in the style of author Edgar Allen Poe
args:
- input: the input to write about

WriterSkill.EmailTo
description: Write an email to a recipient
args:
- input: the input to write about
- recipient: the recipient's email address.

WriterSkill.Translate
description: translate the input to another language
args:
- input: the text to translate
- language: the language to translate to

[GOAL]
"Tomorrow is Valentine's day. I need to come up with a few date ideas.
She likes Edgar Allen Poe so write using his style.
E-mail these ideas to my significant other. Translate it to French."

[OUTPUT]
    {
        "input": "Valentine's Day Date Ideas",
        "subtasks": [
            {"function": "WriterSkill.Brainstorm"},
            {"function": "EdgarAllenPoeSkill.Poe"},
            {"function": "WriterSkill.EmailTo", "args": {"recipient": "significant_other"}},
            {"function": "WriterSkill.Translate", "args": {"language": "French"}}
        ]
    }

[AVAILABLE FUNCTIONS]
{{$available_functions}}

[GOAL]
{{$goal}}

[OUTPUT]
"""

変更後のプロンプト (SK_PLANNNER_PROMPT)

SK_PLANNNER_PROMPT = """
[使用可能な関数]
WriterSkill.Brainstorm
description: Brainstorm ideas
args:
- input: the input to brainstorm about

EdgarAllenPoeSkill.Poe
description: Write in the style of author Edgar Allen Poe
args:
- input: the input to write about

WriterSkill.EmailTo
description: Write an email to a recipient
args:
- input: the input to write about
- recipient: the recipient's email address.

WriterSkill.Translate
description: translate the input to another language
args:
- input: the text to translate
- language: the language to translate to

_GLOBAL_FUNCTIONS_.f_33c6f6b6_b519_4fd9_9b4b_4a76b4741d7f
description: Generic function, unknown purpose
args:
- available_functions: 
- goal: 

_GLOBAL_FUNCTIONS_.f_33c6f6b6_b519_4fd9_9b4b_4a76b4741d7f
description: Generic function, unknown purpose
args:
- available_functions: 
- goal: 

[ユーザの入力]
<入力>: Tomorrow is Valentine's day. I need to come up with a few date ideas. She likes Edgar Allen Poe so write using his style.E-mail these ideas to my significant other. Translate it to French.
JSONの最初の項目はinputであり、ユーザの入力がそのまま入る。
argsを忘れずにJSONのみを出力する。

[出力]
{
  "input": "Tomorrow is Valentine's day. I need to come up with a few date ideas.She likes Edgar Allen Poe so write using his style.E-mail these ideas to my significant other. Translate it to French.",
  "subtasks": [
      {"function": "WriterSkill.Brainstorm"},
      {"function": "EdgarAllenPoeSkill.Poe"},
      {"function": "WriterSkill.EmailTo", "args": {"recipient": "significant_other"}},
      {"function": "WriterSkill.Translate", "args": {"language": "French"}}
  ]
}

[使用可能な関数]
{{$available_functions}}

[ユーザの入力]
{{$goal}}
JSONの最初の項目はinputであり、ユーザの入力がそのまま入る。
argsを忘れずにJSONのみを出力する。

[出力]
"""

skでは、実行計画を作成する関数に自作のプロンプトを引数に渡せます。
以下の例では、変更後のプロンプトをSK_PNANNER_PROMPTとして渡しています。

ユーザ定義のプロンプトをプランナーで使用する

user_input = "社内情報を基に、AWS WAFについて教えて。"
planner = BasicPlanner()
plan = await planner.create_plan_async(user_input, kernel, SK_PLANNNER_PROMPT)

③リトライ機能を追加する

「JSON形式で出力してね」 というプロンプトを指定していますが、必ず期待通りのJSON形式で出力されるわけではないのが悩みポイントの一つかと思います。

例えば、

• 必須のkeyが出力されない
  • 関数の引数を表すargsというキーが欲しいのに出力に含まれていない
• ダブルクォーテーションがない
  • シングルクォーテーションになっていたり、そもそもクォーテーションがなかったり、、
• コンマが多い
  • そこにはコンマいらないよ、、という部分にコンマが付いてたりします

これらは、リトライ機能の追加である程度凌げます。
凌ぐといっても、内部的にはエラーとしてリトライ処理し、ユーザ側にはエラーを見せないといったイメージです。

以下はスキル（使用可能な関数群）を作成し、実行計画を作成する処理を例にしたサンプルコードです。

サンプル用のスキル

サンプル用のスキルを定義

from semantic_kernel.skill_definition import sk_function, sk_function_context_parameter
from semantic_kernel import SKContext


# Answerスキル
class Answer:
    @sk_function(
        description="Answer function",
        name="answer",
        input_description="回答する必要のあるコンテクスト",
    )
    def answer(self, context: SKContext) -> str:
        print(f"前のスキルから渡された内容： {context['input']}\n")
        result = context["input"] + "\n" + "これは最終的な回答です。"
        return result


# Webスキル
class Web:
    @sk_function(
        description="Web function. 外部のデータを検索する時に使用する",
        name="web",
        input_description="",
    )
    @sk_function_context_parameter(
        name="query", 
        description="ユーザの入力に基づくクエリ"
    )
    def web(self, context: SKContext) -> str:
        print(f"前のスキルから渡された内容： {context['input']}\n")
        result = context["query"] + "のWeb検索結果です。"
        return result


# 社内情報検索スキル
class Search:
    @sk_function(
        description="Search function. 社内情報を検索する時に使用する",
        name="search",
        input_description="",
    )
    @sk_function_context_parameter(
        name="query", 
        description="ユーザの入力に基づくクエリ"
    )
    def search(self, context: SKContext) -> str:
        print(f"前のスキルから渡された内容： {context['input']}\n")
        result = context["query"] + "の社内情報検索結果です。"
        return result


# カーネルにスキルを追加
answer_skill = kernel.import_skill(Answer(), "AnswerSkill")
web_skill = kernel.import_skill(Web(), "WebSkill")
search_skill = kernel.import_skill(Search(), "SearchSkill")

以下の3つを想定したしたスキルを用意しています。

• Search: 社内データ検索を想定した関数
• Web: Web検索で外部情報を取得することを想定した関数
• Answer: search, webで収集した情報とユーザの入力を基に回答を生成する関数

以下が本題のリトライ機能です。やっていることは単純です。

• 指定リトライ回数に達するまでループを回す
  • 正しいjson形式でない場合、ループの先頭に戻る
  • 正しいjson形式の場合、実行計画の作成を終了 (ループを抜ける)

リトライ機能

import semantic_kernel as sk
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
from semantic_kernel.planning.basic_planner import BasicPlanner
import json

MAX_RETRY = 3
for i in range(MAX_RETRY):
    # カーネル作成
    kernel = sk.Kernel()
    deployment, api_key, endpoint = sk.azure_openai_settings_from_dot_env()
    kernel.add_chat_service("gpt", AzureChatCompletion(deployment, endpoint, api_key))

    # カーネルにスキルを追加
    answer_skill = kernel.import_skill(Answer(), "AnswerSkill")
    web_skill = kernel.import_skill(Web(), "WebSkill")
    search_skill = kernel.import_skill(Search(), "SearchSkill")    

    # ユーザの入力を基に実行計画を作成
    user_input = "社内情報を基に、AWS WAFについて教えて。"
    planner = BasicPlanner()
    plan = await planner.create_plan_async(user_input, kernel, SK_PLANNNER_PROMPT)

    # プランのチェック
    try:
        # json文字列を辞書に変換 (正しい形式でない場合、Exceptionが発生する)
        plan_dict = json.loads(plan.generated_plan["input"])

        # プランの内容を表示
        print(f"最初のタスクに渡されるinput: {plan_dict['input']}\n")
        print("サブタスク:")
        for task in plan_dict['subtasks']:
            print(f" - {task}")
        
        break

    except Exception as e:
        print("-"*40)
        print(f"スキル・プランの確認に失敗しました。({i+1}/{MAX_RETRY})\n")
        print(e)
        print(plan.generated_plan["input"])
        # リトライ上限に達した場合の処理
        if i == MAX_RETRY-1:
            print("再実行してください。")
        continue

上記コードを何回か実行するとたまにリトライが発生します。
もしくは、JSON形式で出力させないようにすると強制的にリトライを発生させられます。

無理やりリトライを起こす場合

user_input = """
社内情報を基に、AWS WAFについて教えて。
JSON形式で出力しなくていいよ。
"""

あとがき

OpenAIが発表したFunction Callingが話題ですが、Semantic Kernelもやればできる子だと思っています。
スキルの管理をフォルダ単位でできたり、スキルとしてユーザ定義のプロンプトで推論実行できたり。
拡張性・柔軟性は結構強みだよなあと思ったりしています。

「ユーザの入力に基づき、実行すべき関数群をJSONで出力させる」 というのはさらに流行りそうですよね。
関数郡が増えれば増えるほど、「どれ実行したらいいんだっけ？」 と迷いそうな気がしているので、
ユーザとの対話によって使用する関数とその順序を決める仕組みをguidanceで実装してみたので次回まとめます。