AI Agent開発日記 2025/06/07

2025年6月7日

やりたいこと

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

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

前回の調査日記

Agentbuilderの既存playbookを編集するpythonコードを作成する

やりたいこと
- 下記の Python クライアントライブラリドキュメントを参考にして、既存playbookを編集するpythonコードを作成したい
現状
- ツールの参照設定がない状態の Playbook に対して編集できるコードは作成済み。
- ただし、Playbook にツールの参照設定がある場合、事前に Web UI でその参照を削除しないと 400 エラーが発生するという問題があるため、解決したい。
  - 以下のコードを用意して対応が出来た。
    - playbookのtoolの参照設定を解除する
      playbookに特定のtoolを参照させる

参考ドキュメント

playbookのtoolの参照設定を解除し
playbookに特定のtoolを参照させるコード

import google.auth
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
import re

# 必要情報（すべてプレースホルダです）
PROJECT_ID = 'your-project-id'
LOCATION_ID = 'global'
AGENT_ID = 'your-agent-id'
PLAYBOOK_ID = 'your-playbook-id'

# 削除対象の Tool ID（存在しないものも含めておく）
TOOL_IDS_TO_REMOVE_LIST = [
    'tool-id-to-remove-1',
    'tool-id-to-remove-2'
]

def get_tool_resource_name(project_id, location_id, agent_id, tool_id):
    return f'projects/{project_id}/locations/{location_id}/agents/{agent_id}/tools/{tool_id}'

def get_tool_name_id_map(project_id, location_id, agent_id, dialogflow_service):
    parent = f'projects/{project_id}/locations/{location_id}/agents/{agent_id}'
    tool_map = {}
    try:
        request = dialogflow_service.projects().locations().agents().tools().list(parent=parent)
        response = request.execute()
        for tool in response.get('tools', []):
            if 'displayName' in tool and 'name' in tool:
                tool_map[tool['displayName']] = tool['name']
        return tool_map
    except HttpError as e:
        print(f"ツール取得失敗: {e}")
        return None

def remove_specific_referenced_tools(project_id, location_id, agent_id, playbook_id, tool_ids_to_remove):
    try:
        credentials, _ = google.auth.default()
        dialogflow_service = build('dialogflow', 'v3beta1', credentials=credentials)
        playbook_name = f'projects/{project_id}/locations/{location_id}/agents/{agent_id}/playbooks/{playbook_id}'

        tool_names_to_remove_set = {
            get_tool_resource_name(project_id, location_id, agent_id, tid)
            for tid in tool_ids_to_remove if tid
        }

        tool_name_to_resource_map = get_tool_name_id_map(project_id, location_id, agent_id, dialogflow_service)
        if tool_name_to_resource_map is None:
            return

        current_playbook = dialogflow_service.projects().locations().agents().playbooks().get(
            name=playbook_name
        ).execute()

        existing_referenced_tools = current_playbook.get('referencedTools', [])
        current_goal = current_playbook.get('goal')
        instruction_obj = current_playbook.get('instruction')
        current_instruction_text = instruction_obj.get('steps', [{}])[0].get('text') if instruction_obj else None

        updated_referenced_tools = [
            tool for tool in existing_referenced_tools if tool not in tool_names_to_remove_set
        ]

        def remove_tool_ref_from_text(text):
            if not text:
                return text, False
            modified = False
            tool_refs = re.findall(r'\$\{TOOL:([^}]+)\}', text)
            for ref in set(tool_refs):
                res = tool_name_to_resource_map.get(ref)
                if res in tool_names_to_remove_set:
                    text = re.sub(rf'\s*\$\{{TOOL:{re.escape(ref)}\}}\s*', '', text).strip()
                    modified = True
            return text, modified

        updated_goal, goal_modified = remove_tool_ref_from_text(current_goal)
        updated_instruction_text, instruction_modified = remove_tool_ref_from_text(current_instruction_text)

        update_body = {}
        update_mask_parts = []

        if set(updated_referenced_tools) != set(existing_referenced_tools):
            update_body['referencedTools'] = updated_referenced_tools
            update_mask_parts.append('referencedTools')

        if goal_modified:
            update_body['goal'] = updated_goal
            update_mask_parts.append('goal')

        if instruction_modified and instruction_obj:
            updated_steps = instruction_obj['steps']
            updated_steps[0]['text'] = updated_instruction_text
            update_body['instruction'] = {'steps': updated_steps}
            update_mask_parts.append('instruction.steps')

        if update_mask_parts:
            request = dialogflow_service.projects().locations().agents().playbooks().patch(
                name=playbook_name,
                updateMask=','.join(update_mask_parts),
                body=update_body
            )
            request.execute()
            print("更新完了: プレイブックの参照ツールを削除しました。")
        else:
            print("変更がないため更新不要です。")

    except HttpError as e:
        print(f"APIエラー: {e}")
    except Exception as e:
        print(f"エラー: {e}")

# 実行例
remove_specific_referenced_tools(
    project_id=PROJECT_ID,
    location_id=LOCATION_ID,
    agent_id=AGENT_ID,
    playbook_id=PLAYBOOK_ID,
    tool_ids_to_remove=TOOL_IDS_TO_REMOVE_LIST
)