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

# 0. 全体像

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/393221/bf10fde2-acb7-266e-5732-eb9649370899.png)


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

https://github.com/Azure-Samples/azure-search-openai-demo/

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

https://qiita.com/sakue_103/items/95087cb95f1936a8791b

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



## ※今回作る環境とAzure-Samplesとの差異**
- セマンティック検索は使用しません。
- PDFファイルを文字数で分割し、txtファイルとしてBLOBに保存する構成とします。
- Langchainではなく、guidanceを使用します。


## 動作イメージ

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/393221/57a7f406-4744-e883-d050-d99ce1364180.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にはチャットベースの例しかなかったです。
他に良い方法ありそうだなあ、と思っています。

```py:
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](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/393221/184cf457-9c83-82f0-5908-45515abc031f.png)


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

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

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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/393221/d9b8c368-12cf-2b62-32bf-1cf88a56e4e6.png)


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



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

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

```py:
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]をインストールします。

```sh:
pip install flask[async]
```

flask[async]を使用して以下のような設定を行いました。
- guidanceを実行する関数の呼び出し時に**await**
- awaitを使用する関数に**async**

```py:
@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'.**