AI Agent開発日記 2025/06/15

2025年6月15日

やりたいこと

データを読み込むだけで、以下が自動で行われる仕組みを構築したい。

読み込んだデータがElasticsearchに自動登録される
データと指示内容がGeminiに渡され、PlaybookやTool／Schemaが自動生成される
生成された内容が自動でPlaybookやToolに反映され、管理用エージェントと紐づいて構成される

前回の調査日記

InstructionsとPlaybookを作成した後、そのPlaybookの情報を引き継ぎ、そのまま管理用Playbookに追加できるようにしたい。

やりたいこと
- 過去に作成した下記コードに加えて作成したplaybookをその管理用playbookに追加する機構を作成したい
  - bq > elastic同期
    openapi and tool作成
    instructions and playbook作成
- main.py t.pyとは別の親Playbook用Instructionsファイルに、新たに作成したPlaybookを下記の形式で追加し、その内容を親Playbookに反映させる手法のコードを作成する
  - – – 曲を検索するplaybookです
    – – – ${PLAYBOOK:有名な曲検索_playbook}
- ローカルの親Playbook用Instructionsファイルに新たに作成したPlaybookの情報を追加し、その内容を親Playbookに反映させる方法ではなく、親Playbookの現在のInstructionsを取得して、そこに新たな子Playbookを追加して反映させる方法の方が適していそうなので、その方法を採用することにして、コードを編集する
現状
- 下記のPlaybookのInstructionsを取得し、文字列を追加するコードの検証に成功したため、これを基に t.py と main.py のコードにも変更を加える

PlaybookのInstructionsを取得し、文字列を追加するコード

# Jupyter Notebook で Dialogflow CX Playbook の Instructions テキスト内容に文字列を追加して出力する簡素なコード

# 必要なライブラリをインポート
from google.cloud import dialogflowcx_v3beta1
from google.api_core.exceptions import NotFound, PermissionDenied
import google.auth

# === サンプル用の設定（必要に応じて自分の値に置き換えてください） ===
PROJECT_ID = 'your-project-id'
LOCATION_ID = 'your-location-id'  # 例: 'global' や 'us-central1'
AGENT_ID = 'your-agent-id'
PLAYBOOK_ID = 'your-playbook-id'

# 追加したい文字列（サンプル）
ADDITIONAL_STRING = """\n- ${PLAYBOOK:sample_playbook}\n- - これはサンプルの子Playbookです。"""

# Playbook のリソース名
playbook_resource_name = f'projects/{PROJECT_ID}/locations/{LOCATION_ID}/agents/{AGENT_ID}/playbooks/{PLAYBOOK_ID}'

try:
    # Dialogflow CX PlaybooksClient を初期化
    client = dialogflowcx_v3beta1.PlaybooksClient()

    # 指定されたリソース名を持つ Playbook を取得
    playbook = client.get_playbook(name=playbook_resource_name)

    # Instructions のテキスト内容を結合して取得
    instructions_text = ""
    if hasattr(playbook, 'instruction') and playbook.instruction and hasattr(playbook.instruction, 'steps') and playbook.instruction.steps:
        text_parts = []
        for step in playbook.instruction.steps:
            if hasattr(step, 'text') and step.text:
                text_parts.append(step.text)

        if text_parts:
            instructions_text = "\n".join(text_parts)

    # テキストが取得できたら追加文字列を結合して出力
    if instructions_text:
        final_output_text = instructions_text + ADDITIONAL_STRING
        print(final_output_text)
    else:
        print("Playbook から有効な Instructions テキスト内容を取得できませんでした。")

# エラーハンドリング
except NotFound:
    print(f"\nエラー: 指定された Playbook が見つかりませんでした: {playbook_resource_name}")
except PermissionDenied:
    print("\nエラー: 必要なアクセス権限がありません。IAM設定を確認してください。")
except google.auth.exceptions.DefaultCredentialsError:
    print("\nエラー: 認証情報が見つかりません。`gcloud auth application-default login` を実行してください。")
except Exception as e:
    print(f"\nエラー: Playbook 取得中に予期しないエラーが発生しました: {e}")