【Prompt Flow】カスタムPython環境の作成 & Function callingの実行の備忘録

背景

以前、ChatGPTを使って以下のようなものを作りました。
「 決算短信などのPDFファイルから分析に必要なデータを抽出し、Excelとして出力してくれる 」

機能的には気に入っていますが、実行がとても面倒くさいです
「 Pythonスクリプトを実行した後に、ChatGPTに入力して、出力をコピペして、、、 」といった感じですね。
そこで、自動化・パイプライン化したいなあと思い始めたのが今回の背景です。

自動化しようと思ったときに、Prompt Flowを使えるのではないかということで実装してみました。(Pythonだけでも実装可能だと思います。)
Prompt Flowを使えば、Variantsを使って並列で色々なプロンプトを試せますし評価も可能です。

やること

今回やることの全体像は以下の通りです。

• 入力
  • ユーザがPrompt Flow対してファイル名を入力
• Prompt Flowの処理
  • ファイルをダウンロードし、内容を文字列として読み込む
  • 文字列を入力として、分析対象のデータを表形式で出力させる
  • 表形式の文字列を入力として、数値データなどを抽出して辞書型として出力させる
    （Function callingで実装）
  • 辞書型のデータを基にExcelファイルを作成し、Blobにアップロード
• 出力
  • アップロード先のURL

背景で示した図ではCode InterpreterにExcelを出力させていました。
Prompt Flowの実装ではFunction Callingを使って辞書型のデータを生成し、それを使ってExcelを作成しています。

また、何段階かに分けて目的の達成を行うようなフローとなっています。一気に回答を得ようとするとプロンプトエンジニアリング要素が高くなってしまうため、細かく段階を分けてなるべくプロンプトが複雑にならないようにしています。

実装の手順

• 独自のPython環境を用いてPrompt Flowのランタイムを作成
• Prompt Flowで必要な処理を実装

今回、PDFの読み込みやExcelの作成にPythonライブラリを使用します。
これらをPrompt Flow上で使用する際は必要なライブラリを追加する必要があります。

公式ドキュメントにて、「 DockerFile作成 → ACRにイメージ格納 → イメージを基にランタイム作成 」といった手順が紹介されていました。
今回はこの方法で独自のPython環境を作成し、それを用いてランタイム(Prompt Flowの実行環境のようなもの)を作成していきます。

ランタイムの作成後、Prompt Flowで必要な処理を実装することでパイプラインを作っていきます。

① 独自のPython環境を用いてPrompt Flowのランタイムを作成

まずは、環境作成を行います。
必要なPythonライブラリを含めた環境を作成し、それを基にPrompt Flowのランタイムを作成します。

最終的なフォルダ構成

最終的には以下のようなフォルダ構成となります。
シェルスクリプトは任意です。Azure CLI(azコマンド)を使って色々とデプロイしていくため、それらをシェルスクリプトにて行っています。

|--image_build
|  |--requirements.txt
|  |--Dockerfile
|  |--environment-build.yaml
|  |--environment.yaml
|  |--(create_image.sh)
|  |--(create_env.sh)

requirements.txtとDockerfile

必要なPythonライブラリ達をrequirements.txtに記述し、
Dockerfileでpip install -r requirements.txtが実行されるようにします。
また、Azure OpenAIへの接続情報が環境変数に設定されるようにしておきます。

requirements.txt

openai
pypdf
pandas
openpyxl

Dockerfile

FROM mcr.microsoft.com/azureml/promptflow/promptflow-runtime:latest
COPY ./* ./
RUN pip install -r requirements.txt

ENV AZURE_OPENAI_API_KEY="APIキー"
ENV AZURE_OPENAI_ENDPOINT="エンドポイント"
ENV AZURE_OPENAI_API_VERSION="2023-07-01-preview"
ENV AZURE_OPENAI_DEPLOYMENT_NAME="gpt-35-turbo"

ENV AZURE_STORAGE_CONNECTION_STRING="接続文字列"
ENV STORAGE_CONTAINER_NAME="コンテナ名"

※ ハードコーディングではなく、ランタイム時に設定した方がよさそうです。

environment_build.yaml

Machine Learning Studioに環境(Dockerイメージ)を作成するためのyamlファイルです。

$schema: https://azuremlschemas.azureedge.net/latest/environment.schema.json
name: <任意のイメージ名>
build:
  path: .

ランタイムのベースとなるイメージを作成

まず、Azure CLIでmlを追加します。

az extension add --name ml

次に、イメージを作成します。

create_image.sh

ws_name="ワークスペース名"
rg_name="リソースグループ名"
sub_id="サブスクリプションID"

az login
az account set --subscription $sub_id
az configure --defaults workspace=$ws_name group=$rg_name

az ml environment create -f environment_build.yaml --subscription $sub_id -g $rg_name -w $ws_name

上記のコマンド群を実行した後、イメージが作成されているか確認します。
Machine Learning Studioから環境をクリックします。

カスタム環境タブに移動し、環境一覧を確認します。

今回environment_build.yamlにてイメージ名をcustom-python-env-imageと指定しています。
名前をクリックするとステータスを確認できます。成功となっていればOKです。

また、後の手順でACRのイメージIDを使用するため控えておきます。

environment.yaml

さきほど作成したイメージからランタイム用の環境を作成するためのyamlファイルです。

environment.yaml

$schema: https://azuremlschemas.azureedge.net/latest/environment.schema.json
name: custom-python-env
image: <ACRのイメージID>
inference_config:
  liveness_route:
    port: 8080
    path: /health
  readiness_route:
    port: 8080
    path: /health
  scoring_route:
    port: 8080
    path: /score

ランタイム用の環境を作成

create_env.sh

ws_name="ワークスペース名"
rg_name="リソースグループ名"
sub_id="サブスクリプションID"

az ml environment create -f environment.yaml --subscription $sub_id -g $rg_name -w $ws_name

上記コマンド群を実行した後、環境を確認します。
environment.yamlで指定した名前で環境が作成されていればOKです。

ランタイムを作成

Machine Learning Studioから該当のワークスペースへ移動し、プロンプトフローをクリックします。

ランタイムタブに移動し、作成をクリックし、コンピューティングインスタンスランタイムをクリックします。

以上で作成は終了です。動作確認をしてみます。
プロンプトフローをクリックし、Pythonをクリックします。
すると、Pythonのコンポーネントが追加されるのでコードを以下のように編集します。(ライブラリのimportと環境変数の読み込みを行う)

編集後のコード

from promptflow import tool
import pypdf
import os

# The inputs section will change based on the arguments of the tool function, after you save the code
# Adding type to arguments and return value will help the system show the types properly
# Please update the function name/signature per need
@tool
def my_python_tool() -> str:
  return os.getenv("AZURE_OPENAI_API_VERSION")

入力の検証と解析をクリックし、右上の再生ボタンをクリックして実行します。

出力という欄に実行結果が表示されるので確認します。さきほど設定した環境変数が出力されているのでよさそうです。importもエラーが出ていませんね。

② Prompt Flowで必要な処理を実装

さきほどまでの手順でPython環境ができたので、あとはPython関数の定義やプロンプトの定義を行うことで処理フローを作っていきます。

実装した処理フロー

今回実装したフローは以下の通りです。全てを記載するととても長くなってしまうので、ポイントを絞って以降で解説したいと思います。

• inputs
  • ファイル名を入力として受け取る
• find_keywords
  • ファイル名を基にBlobストレージからファイルをダウンロード
  • ファイルの中身を文字列として読み込む
  • 必要なデータが含まれる部分だけを抽出するための正規表現を定義
    • ファイルの中身をそのままモデルに入力してしまうとトークン数や精度面でデメリットがある
• extract_target_data
  • 分析対象となるデータのみを抽出し、表形式で出力させるためのプロンプトを定義
    • （以降で詳細解説）
• extract_data_function_calling
  • 表形式の文字列から辞書型のデータを抽出するためのFunction callingを実装
    • （以降で詳細解説）
• create_excel
  • 辞書型のデータを用いてExcelを作成し、Blobストレージへアップロード
• outputs
  • アップロード先のURL

必要な情報を抽出するためのプロンプト

プロンプトの定義

system:
あなたはテキストファイルから情報を抽出するスペシャリストです。
回答は表形式で、必ず表のみを出力することを忘れないでください。

user:
# ゴール
テキストから「会社名と決算期」に該当しそうなデータを抽出する
テキストから今期の${バランスシート}に該当しそうなデータを抽出する
テキストから今期の${損益計算書}に該当しそうなデータを抽出する
テキストから今期の${キャッシュフロー計算書}に該当しそうなデータを抽出する
テキストから「直近に公表されている予想から修正があるか抽出する
テキストから「株式分割があるか」抽出する

# 変数の定義
${バランスシート}：流動資産合計・固定(非流動)資産合計・流動負債合計・固定(非流動)負債合計・純資産(資本)合計
${損益計算書}：売上高・売上原価・販管費・経常利益
${キャッシュフロー計算書}：営業活動・投資活動・財務活動

# 実行のプロセス
テキストファイルを読み込む (今期のデータがどこにあるか入念に確認する)
変数の定義を確認する
変数の定義に示されたデータを抽出し、表形式で結果を出力する


それでは、実行のプロセスに従って表形式の成果物を作成してください。
必ず全ての表を埋めて成果物を作成すること。
以下はテキストファイルです。必ず必要なデータが含まれているので入念に探してください。

###

{{user_input}}

定義している情報をざっくりまとめると以下の通りです。

• 役割 (system)
  - テキストから情報を抽出することを伝える
  - どうやって出力してほしいか伝える

• ゴール
  - なにを達成してほしいか伝える

• 変数の定義
  - どのデータを抽出してほしいか伝える

• 実行のプロセス
  - どのようにしてゴールを達成すべきか伝える

• 実行をお願いする
  - これまでの定義に従って実行してほしいと伝える

色々と試した結果、以上のようなプロンプトがよさそうです。
Prompt Flowでは評価機能が備わっているので、今後その機能を用いて評価したいと思います。

また、このプロンプトで辞書型として文字列を出力させ、それをパースするといった方法もありだと思います。
ただ、おしゃべりかつランダム性が高めなGPTの出力をパースするためには、実装面・プロンプトエンジニアリング面で少し面倒です、、。
そこで今回はFunction callingを使って、構造化データを抽出するといった構成にしています。

Function callingの実装

（以降の説明はPrompt Flowとは無関係です。Function callingの説明です。）

Functnion callingは動的に実行する関数を決定する機能というイメージが強いですが、
「 テキストから構造化データを生成する 」というタスクにも使用できます。

もう少し具体的に説明すると、Function callingはユーザの入力に応じて、

• ①どの関数を
• ②どのような引数で実行するのか

を出力してくれます。この②を使うことでユーザの入力から構造化データを取り出せるわけですね。
入出力の一例を示します。

入力

テキストファイルから情報を抽出し、表形式で結果を出力します。
"|会社名 |決算期 | |--------------|--------------| |ＫＤＤＩ株式会社|2023年3月期 | |バランスシート項目|金額 | |-----------------|------------| |流動資産合計 |3,667,028 | |固定(非流動)資産合計|7,417,350 | |流動負債合計 |4,015,953 | |固定(非流動)負債合計|1,557,762 | |純資産(資本)合計 |5,510,663 | |損益計算書項目 |金額 | |--------------|-----------| |売上高 |4,182,893 | |売上原価 |2,351,364 | |販管費 |1,037,312 | |経常利益 |843,420 | |キャッシュフロー計算書項目|金額 | |-----------------------|-----------| |営業活動によるキャッシュ・フロー|842,440 | |投資活動によるキャッシュ・フロー|-567,964| |財務活動によるキャッシュ・フロー|-524,954| |予想修正の有無|あり | |--------------|-------------| |株式分割の有無|なし | |--------------|-------------|"

出力

{
  "予想修正の有無": "有",
  "会社名": "ＫＤＤＩ株式会社",
  "営業活動キャッシュフロー": 842440,
  "固定負債合計": 1557762,
  "売上原価":2351364
  "売上高":4182893
  ... , 
  "販管費": 1037312
}

上記のような適当な入力であっても、該当しそうな箇所からデータを抽出してきてくれます。

実装する場合は以下のようになります。Functioin callingで関数呼び出しがある場合は「 引数 」に該当する情報を取得しています。

引数部分を抽出する例

# Function calling
response = openai.ChatCompletion.create(
  engine = os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"),
  messages = messages,
  functions=functions_metadata,  # 使用したい関数をリストで指定
  function_call="auto",
  temperature=0,
)

# 関数呼び出し有無を判定
if response["choices"][0]["message"].get("function_call"):
  msg = response["choices"][0]["message"]
  return json.loads(msg["function_call"]["arguments"])  # 関数呼び出しがある場合、引数部分をreturn

以下にPrompt Flow上に実装したコード全文を載せておきます。

実装例

from promptflow import tool
import os
import openai
import json

# Azure OpenAIの設定
openai.api_type = "azure"
openai.api_key = os.getenv("AZURE_OPENAI_API_KEY")
openai.api_base = os.getenv("AZURE_OPENAI_ENDPOINT")
openai.api_version = os.getenv("AZURE_OPENAI_API_VERSION")


# システムのプロンプト
SYSTEM_PROMPT = """
あなたはユーザを助けるアシスタントです。
ユーザの入力に正しく回答を出力するために、ステップバイステップで慎重に考えることができます。
まずはゴール達成のためになにが必要かを考え、自分の思考と行動を説明します。
"""

messages = [
    {"role":"system", "content": SYSTEM_PROMPT},
]

@tool
def main(input1: str) -> str:
  # Function Callingを使って、必要なデータを抽出
  result = exec_function_calling(input1)

  # データをチェック
  result = check_data(result)

  return result


def exec_function_calling(input1:str):
  # 使用したい関数を定義
  functions_metadata = [CreateFinancialDataDict.metadata]
  
  # ユーザの入力にサフィックスを追加しメッセージに追加
  SUFFIX = """
  ###

  上記のテキストから財務データの辞書を生成して。
  """
  messages.append({"role": "user", "content": input1 + SUFFIX})

  # 推論実行
  response = openai.ChatCompletion.create(
      engine = os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"),
      messages = messages,
      functions=functions_metadata,
      function_call="auto",
      temperature=0,
  )

  # 関数呼び出し有無を判定
  if response["choices"][0]["message"].get("function_call"):
      msg = response["choices"][0]["message"]
      return json.loads(msg["function_call"]["arguments"])
  else:
      return None
  

def check_data(data:dict) -> dict:
  # キーを満たしているか確認
  keys = [
      "営業活動キャッシュフロー",
      "投資活動キャッシュフロー",
      "財務活動キャッシュフロー",
  ]

  # キーがない場合は値をNoneで追加
  for key in keys:
      if key not in data.keys():
          data[key] = None

  return data

# Function calling用関数のクラスを定義
class CreateFinancialDataDict:
  metadata = {
      "name": "create_financial_data_dict",
      "description": "会社の財務データを格納する辞書を作成する関数",
      "parameters": {
          "type": "object",
          "properties": {
              "会社名": {
                  "type": "string",
                  "description": "会社の名前"
              },
              "決算期": {
                  "type": "string",
                  "description": "会社の決算期"
              },
              "売上高": {
                  "type": "number",
                  "description": "会社の売上高"
              },
              "売上原価": {
                  "type": "number",
                  "description": "会社の売上原価"
              },
              "販管費": {
                  "type": "number",
                  "description": "会社の販管費"
              },
              "経常利益": {
                  "type": "number",
                  "description": "会社の経常利益"
              },
              "流動資産合計": {
                  "type": "number",
                  "description": "会社の流動資産合計"
              },
              "固定資産合計": {
                  "type": "number",
                  "description": "会社の固定資産合計"
              },
              "流動負債合計": {
                  "type": "number",
                  "description": "会社の流動負債合計"
              },
              "固定負債合計": {
                  "type": "number",
                  "description": "会社の固定負債合計"
              },
              "純資産合計": {
                  "type": "number",
                  "description": "会社の純資産合計"
              },
              "営業活動キャッシュフロー": {
                  "type": ["number", "null"],
                  "description": "会社の営業活動キャッシュフロー"
              },
              "投資活動キャッシュフロー": {
                  "type": ["number", "null"],
                  "description": "会社の投資活動キャッシュフロー"
              },
              "財務活動キャッシュフロー": {
                  "type": ["number", "null"],
                  "description": "会社の財務活動キャッシュフロー"
              },
              "株式分割の有無": {
                  "type": "string",
                  "enum": ["有", "無"],
                  "description": "株式分割の有無"
              },
              "予想修正の有無": {
                  "type": "string",
                  "enum": ["有", "無"],
                  "description": "予想修正の有無"
              }
          },
          "required": [
              "会社名", 
              "決算期", 
              "売上高", 
              "売上原価", 
              "販管費", 
              "経常利益", 
              "流動資産合計", 
              "固定資産合計", 
              "流動負債合計", 
              "固定負債合計", 
              "純資産合計", 
              "株式分割の有無",
              "予想修正の有無"
          ]
      }
  }

Prompt Flow上でFunction callingを実行する

Prompt Flow上で実行するには、Pythonコンポーネントを追加し、そこにソースコードを書き込むことで実行できます。

少しややこしいポイントとしては、Prompt Flow上の接続という機能とは別でAzure OpenAIと接続しているところかと思います。

Prompt Flowには、Azure OpenAIやCognitive Searchなどと接続するための機能が備わっていますが、その機能ではなくソースコード上で接続情報を定義し、リクエストを飛ばしています。
そのため、「 Prompt Flow自体とAOAIの接続 」と「 Prompt FlowのPythonコンポーネントとAOAIの接続 」が混在していることになります。図にすると以下のようなイメージですね。

バグ？エラー？集

プロンプトを定義するコンポーネント

なにがトリガーかは分かりませんが、Prompt Flow上で色々と操作を行うと何かの拍子に以下のようなエラー文が出力されました。（何度か遭遇）

プロンプトを定義するコンポーネントは電球マークなのですが、スパナ?のマークになっているので、裏側で他のコンポーネントと認識されている？(要調査)

まとめ

めんどくさい実行手順をパイプライン化するために、独自のPython環境を作成しPrompt Flowで様々な処理を実装してつなぎ合わせてみました。
独自のPython環境を作成することで、Prompt Flowでやれることの幅が一気に広がりますね。

また、プロンプトの評価機能や複雑のプロンプトを並列に実験できるVariantsなどの機能をまだまだ使いこなせていないので、次回以降詳細に踏み込んでいきたいと思います。