LoginSignup
2
2
@sakue_103

【Azure OpenAI開発】簡易版エンタープライズサーチを自作して理解を深める (2) プロンプトエンジニアリング

Last updated at Posted at 2023-05-21

0. 全体像

今回作っていく環境は以下の通りです。

image.png

Microsoftのエンタープライズサーチのアーキテクチャの理解を深めるため、
自分で作ってみようというのがモチベーションです。

前回は、①質問を入力、③クエリを基にファイルを検索の機能を実装しました。

今回は、②ユーザの入力をクエリに変換。④ユーザの入力&ファイルの内容から回答を生成の機能を実装したので、その備忘録です。

※今回作る環境とAzure-Samplesとの差異**

  • セマンティック検索は使用しません。
  • PDFファイルを文字数で分割し、txtファイルとしてBLOBに保存する構成とします。
  • Langchainではなく、guidanceを使用します。

動作イメージ

image.png

ちょっとわかりにくいですが、以下の振る舞いを行えているのが確認できます。

  • ①ユーザの入力に応じて取得ファイルが変わっている
    • ユーザの入力を基にCognitive Search用のクエリをAzure OpenAIに生成させているためです。
    • One-Shotをして、一つだけ例を示してあげています。(Few-Shotした方が良いかも？)
  • ②ファイル検索後、ユーザの指示に従った動作を行っていることが確認できます。
    • 今回の例では、ただ単にファイルを検索してくるだけでなく、内容を要約しています。

1. ユーザの入力をクエリに変換するプロンプトの生成

guidanceを使用して、LangchainのAgents機能っぽいものを実装しようとしましたが断念、、。
検索機能しか持たないエージェントが出来上がりました。
ドキュメントがないため、Github上のguidanceのソースコードと向き合っていました。
最終的に以下のような形になりました。

[system]: あなたは事実情報に基づく回答しかできないよ。ユーザが提示するfunctionsを使ってね。
[user]: functionsと例を提示...
[assistant]: OK.そんな感じでそれらのfunctionsを使います。
[user]: <質問>
[assistant]: その質問だと、function: xxx()を実行する必要があるな
↓
プロンプト外部でfunction: xxx()を実行し、結果をプロンプトに渡す
↓
[user]: function: xxx()の結果はこんな感じ。この結果に基づいて<質問>に答えて
[assistant]: <回答>

guidanceのawait機能を使って外部でゴニョゴニョできるの地味に嬉しいかも、、？
あと、今回のような流れを実装しようとなると、githubのnotebookにはチャットベースの例しかなかったです。
他に良い方法ありそうだなあ、と思っています。

import guidance
import re

# 使用する言語モデルを指定
guidance.llm = guidance.llms.OpenAI("gpt-3.5-turbo")

# どの機能を使うか判断し、実行する関数
def exec_function(action, functions):
    # 「args=」が含まれる行を抽出
    pattern = r".*((args=.*))"  
    action = re.search(pattern, action).group()
    
    # 関数名を抽出
    func_name = action.split("(")[0]
    pattern = r"\((.*?)\)"  
    args_str = re.search(pattern, action).group()[1:-1]

    # 引数を抽出 (args={}の入力を想定している)
    pattern = r"{.*?}"  
    args_str = re.search(pattern, args_str).group()
    args = eval(args_str)
    
    # 関数の実行
    for tool in tools:
        if tool["name"].split("(")[0] == func_name:  # 関数名が一致した場合                                                    
            result = tool["func"](args)
            
    return result

    
# 情報検索用の機能を定義
def search(args:dict) -> [dict]:
    result = [{"filename": "test.txt", "content": "クラウドセキュリティに対する意識は高まっている。"}]
    return result

search_tool = {
    "name": "search(args)",
    "usecase": "should be used when information retrieval is required",
    "func": search
}

# プロンプトに渡す機能群を定義
tools = [
    search_tool
]


# 機能群を展開するプロンプト
functions_prompt = guidance('''functions:
{{#each tools}}- {{~this.name}} : {{this.usecase}} 
{{/each}}''')

# few-shot用のサンプルデータと展開するプロンプト
few_shot_data = [
    {"filename": "資格取得支援.txt", "content": "支援上限は月に3万円。"},
    {"filename": "対象資格一覧.pdf", "content": "基本情報, 応用情報, 簿記"},
]

# few-shot用のsearchの結果を展開するプロンプト
search_prompt = guidance('''search results:
{{~#each few_shot_data}}
<result>
    {{this.filename}} 
    {{this.content}}
</result>
{{/each}}''')

# few-shot用の会話例
few_shot_example = [
    {'role': 'user', 'content': f'You must use the following functions.\n\n{str(functions_prompt(tools=tools))}'},
    {'role': 'assistant', 'content': 'OK. I will use these functions from now on and answer based on factual information.'},
    {'role': 'user', 'content': """以下に例を示します。
    user: 資格取得支援プログラムの注意点は？
    assistant:
      資格取得支援プログラムについて調べて回答する必要がある。まずは、function: search(args)を使おう。
      search(args={"keyword":"資格取得 支援 制度","filename":"*"})"""},
    {'role': 'assistant', 'content': 'OK, select a function based on user input as per the example.'},
]


prompt = guidance('''
{{#system~}}
You are an assistant who answers user input based on factual information in Japanese.
You can use user-directed functions as needed.
{{~/system}}

{{#each few_shot_example}}
{{#if (== this.role "user")}}
{{#user}} {{this.content}} {{/user}}
{{else}}
{{#assistant}} {{this.content}} {{/assistant}}
{{/if}}
{{/each}}

{{#user~}}
That was great, now let's do another one.
{{~/user}}

{{#assistant~}}
OK. what is the question?
{{~/assistant}}

{{#user~}}
Remember to use the functions above.
user: {{user_input}}
{{~/user}}

{{#assistant~}}
{{gen "action" max_tokens=64 temperature=0}}
{{~/assistant}}

{{#user~}}
{{set 'results' (await 'results')}}
{{#each results}}
results:
<result>
{{this.filename}}
{{this.content}}
</result>
Answer 「{{user_input}}」 based on the above results.
Avoid mentioning things you didn't write in the results above.
{{/each}}
{{~/user}}

{{#assistant~}}
{{gen "answer" max_tokens=256 temperature=0}}
{{~/assistant}}
''')


user_input = "クラウドのセキュリティについて教えて"

executed_prompt = prompt(
    # functionsに関する引数
    tools=tools,
    functions_prompt=functions_prompt,
    search=search,
    # search()のfew-shotに関する引数
    few_shot_data=few_shot_data,
    search_prompt=search_prompt,
    # 会話例
    few_shot_example=few_shot_example,
    # ユーザの入力
    user_input=user_input,
)

上記コードにより以下のようなプロンプトが生成されます。
青がユーザ指定、緑がモデルの出力です。

image.png

モデルがsearch(args={"keyword":"クラウド セキュリティ","filename":"*"})を実行したいと言っているので、外部でそれを実行してあげます。
実行結果をプロンプトに挿入してあげると、await 'results'以降のプロンプトの処理が行われます。

# モデルの出力の中からアクションを抽出
action = executed_prompt["action"]

# アクションと関数を用いて、必要な処理を実行
results = exec_function(action, tools)

# awaitに実行結果を挿入
executed_prompt = executed_prompt(results=results)

image.png

ファイルの内容とユーザの質問に基づいて回答できていることを確認できます。

2. アプリケーションに組み込む

上記のコードではsearch()に該当するのは、テスト用のファイルを取得する関数でしたが、
ここをCognitive Searchへの検索処理に置き換えます。

def search(self, args:dict) -> list[dict]:
    print(f"!!! search() args = {args} !!!")
    api_key = os.getenv("SEARCH_API_KEY")
    endpoint = os.getenv("SEARCH_ENDPOINT")

    headers = {
        "api-key": api_key,
        "Content-Type": "application/json"
    }

    keyword, filter_file_name = args["keyword"], args["filename"]

    query = {
        "search": keyword,       # 検索ワード
        "searchMode": "any",     # 部分一致 (デフォルト)
        "top": files_top_n,      # 上位n件
        "filter": f"search.ismatch('{filter_file_name}', 'metadata_storage_name')"  # ファイル名にfile_nameが含まれているか
    }

    response = requests.post(endpoint, headers=headers, json=query)
    data = response.json()["value"]
    results = []

    for row in data:
        content = row["content"]
        file_name = row["metadata_storage_name"]
        results.append(
            {"content": content, "filename": file_name}
        )

    return results

あとは、フロントエンドでユーザの入力を受け取って、バックエンドで上記の処理を実行するだけです。

3. まとめ

プロンプト生成で力尽きたので、フロントエンド・バックエンドの部分は全然まとめられませんでした、、
機会があれば詳細にまとめようかと思います。

次回以降は、非機能要件的な部分に触れようかなあと思っています。
管理者目線で、ユーザの利用履歴を収集・分析してみたり？

4. 追記: flask async

バックエンド側で非同期処理の設定をしないとStatic Web Apps経由で動きませんでした。

まず、flask[async]をインストールします。

pip install flask[async]

flask[async]を使用して以下のような設定を行いました。

  • guidanceを実行する関数の呼び出し時にawait
  • awaitを使用する関数にasync 
@app.route('/api/search', methods=['POST'])
async def search():
    question = request.json['question']
    response, search_results = await guidance_client.exec_prompt(question)
    
    # file名を取り出して情報ソースをresponseに追加
    source_info = "\n\nsource: "
    for result in search_results:
        source_info += result["filename"] + ","

    response += source_info

    response = {'message': response}

    return jsonify(response)

これにより、以下のようなエラーを回避できました。

There is no current event loop in thread 'threadpoolexecutor-1_0'.

2
2
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
2
2