やりたいこと
データを読み込むだけで、以下が自動で行われる仕組みを構築したい。
- 読み込んだデータがElasticsearchに自動登録される
- データと指示内容がGeminiに渡され、PlaybookやTool／Schemaが自動生成される
- 生成された内容が自動でPlaybookやToolに反映され、管理用エージェントと紐づいて構成される
前回の調査日記
あわせて読みたい
AI Agent開発日記 2025/06/28
やりたいこと データを読み込むだけで、以下が自動で行われる仕組みを構築したい。 読み込んだデータがElasticsearchに自動登録される データと指示内容がGeminiに渡さ...
目次
bqとelasticsearchの同期の値をコンソールから入力できるように調整する
- 現状
- 下記コードで可能になった
コード
import requests
import json
import os
import argparse
import time
import unicodedata
import sys
try:
import vars
except ImportError:
vars = None
print("⚠️ 警告: vars.py が見つかりません。APIキーが設定されない可能性があります。")
# --- グローバル設定 ---
PORT = os.getenv("PORT", "8080")
BASE_URL = f"http://localhost:{PORT}"
APP_PREFIX = "integration_service"
# --- デフォルトテスト設定値 (サンプル用に汎用化) ---
DEFAULT_BQ_DATASET = "your_test_dataset"
DEFAULT_BQ_TABLE = "your_source_table"
DEFAULT_ES_URL = "http://localhost:9200"
# BQからの同期先インデックス名（サンプル用）
DEFAULT_ES_INDEX_BQ = "sample_synced_data"
# ToolのOpenAPIスキーマ生成時、AIにコンテキストを与えるための設定
# [優先度1] AIが分析するElasticsearchのインデックス名。実データがあれば、これを元にスキーマが生成される。
DEFAULT_ES_INDEX_SCHEMA = "sample_product_index"
# [優先度2] 上記インデックスに接続できない/データがない場合の代替データ。
# このサンプルデータを元にスキーマが生成され、Elasticsearch環境なしでもテストが可能になる。
DEFAULT_DOC_EXAMPLES_SCHEMA = [
{"item_id": "AB-123", "productName": "Super Duper Gadget", "price_usd": 199.99, "inStock": True,
"details": {"color": "red", "weight_grams": 250}, "tags": ["electronics", "new", "popular"]},
{"item_id": "XY-789", "productName": "Basic Utility Tool", "price_usd": 25.00, "inStock": False,
"description": "A simple tool for everyday tasks.", "manufacturer_code": "MFR-002"}
]
DEFAULT_GEMINI_MODEL = "gemini-1.5-flash"
# Playbook Instructions生成時、AIにコンテキストを与えるための設定
DEFAULT_ES_INDEX_PLAYBOOK_INSTRUCTIONS = "sample_ticket_index"
DEFAULT_DOC_EXAMPLES_PLAYBOOK_INSTRUCTIONS = [
{"ticket_id": "TICKET-001", "customer_name": "Alice Wonderland", "email": "alice@example.com",
"subject": "Login Issue", "description": "I cannot log in to my account.", "status": "open", "priority": "high",
"created_at": "2024-03-10T10:00:00Z", "agent_assigned": None, "tags": ["login", "account_access"]},
{"ticket_id": "TICKET-002", "customer_name": "Bob The Builder", "subject": "Billing question",
"status": "pending_customer_response", "priority": "medium", "created_at": "2024-03-09T15:30:00Z",
"last_reply_by": "agent_support", "resolution_details": None, "related_issues": ["TICKET-005", "TICKET-008"]}
]
# スタンドアロンPlaybook作成時に使用する既存Tool名の仮定（サンプル用）
DEFAULT_CUSTOM_TOOL_NAME_PLAYBOOK_INSTRUCTIONS = "Sample_Search_Tool"
# --- サンプル用ID (実行前に実際の値に置き換えてください) ---
DEFAULT_DF_PROJECT_ID = "your-gcp-project-id"
DEFAULT_DF_LOCATION_ID = "global"
DEFAULT_DF_AGENT_ID = "your-dialogflow-agent-id"
DEFAULT_PARENT_PLAYBOOK_ID = "your-parent-playbook-id"
# toolの表示名・説明 (サンプル用の汎用的な値)
DEFAULT_GATC_TOOL_DISPLAY_NAME_BASE = "Sample_Tool"
DEFAULT_GATC_TOOL_DESCRIPTION = "AIが生成したスキーマに基づく汎用検索ツール（スクリプト経由）。"
# playbookの表示名・Goal (サンプル用の汎用的な値)
DEFAULT_GACP_PLAYBOOK_DISPLAY_NAME_BASE = "Sample_Playbook"
DEFAULT_GACP_PLAYBOOK_GOAL = "ユーザーの要求に基づいて情報を検索し、応答する。"
DEFAULT_DF_REFERENCED_TOOL_IDS = ["your-referenced-tool-id-1"]
# --- NEW: コンソール出力ヘルパー ---
def print_flow_header(title):
print(f"\n===== {title} =====")
def print_flow_footer(success=True):
status = "完了" if success else "中断"
print(f"===== フロー{status} =====")
def print_step_header(message):
print(f"\n--- {message} ---")
def print_process(message, indent=0):
prefix = " " * indent
print(f"{prefix}⏳ {message}...")
def print_api_result(path, success, indent=0):
prefix = " " * indent
status = "[OK]" if success else "[ERROR]"
print(f"{prefix}--> API通信: {path} {status}")
def print_success(message, indent=0):
prefix = " " * indent
print(f"{prefix}✅ {message}")
def print_warning(message, indent=0):
prefix = " " * indent
print(f"{prefix}⚠️ {message}")
def print_info_block(title, content, indent=0):
prefix = " " * indent
print(f"{prefix}--- {title} ---")
for line in str(content).splitlines():
print(f"{prefix} {line}")
print(f"{prefix}{'-' * (len(title) + 8)}")
def prompt_user(message, default_value="", indent=0):
prefix = " " * indent
prompt_text = f"{prefix}▶ {message}"
if default_value:
prompt_text += f" (デフォルト: {default_value})"
# ユーザー入力を正規化
user_input = unicodedata.normalize('NFC', input(f"{prompt_text}: ").strip())
return user_input or default_value
def prompt_for_action(message, skip_prompt, indent=0):
prefix = " " * indent
full_message = f"{prefix}▶ {message} (yes/no)"
if skip_prompt:
print(f"{full_message} (自動実行 - yes)")
return True
user_input = input(f"{full_message}: ").strip().lower()
return user_input in ["yes", "y"]
# --- ヘルパー関数 (APIリクエスト部分のみ改修) ---
def make_api_request(endpoint_path, payload, method="POST", timeout=240, indent=0):
url = f"{BASE_URL}{endpoint_path}"
response = None
try:
if method.upper() == "POST":
response = requests.post(url, json=payload, timeout=timeout)
else:
print_warning(f"未対応のHTTPメソッドです: {method}", indent)
return None
response.raise_for_status()
print_api_result(endpoint_path, success=True, indent=indent)
return response.json()
except requests.exceptions.HTTPError as http_err:
print_api_result(endpoint_path, success=False, indent=indent)
print_warning(f"HTTPエラー: {http_err}", indent + 1)
if response is not None:
try:
error_body = json.dumps(response.json(), indent=2, ensure_ascii=False)
print_info_block("エラーレスポンス", error_body, indent + 1)
except json.JSONDecodeError:
print_info_block("エラーレスポンス (非JSON)", response.text, indent + 1)
except requests.exceptions.RequestException as req_err:
print_api_result(endpoint_path, success=False, indent=indent)
print_warning(f"リクエストエラー: {req_err}", indent + 1)
except Exception as e:
print_api_result(endpoint_path, success=False, indent=indent)
print_warning(f"予期せぬエラー: {e}", indent + 1)
return None
def check_api_key(key_name, key_type="API"):
key_value = getattr(vars, key_name, None) if vars else None
placeholder_value = "YOUR_KEY_HERE"
key_present = key_value and key_value != placeholder_value and "YOUR_" not in str(key_value)
if not key_present:
print_warning(f"vars.pyの {key_name} が未設定です。この{key_type}関連のテストは失敗する可能性があります。")
return key_present
def check_dialogflow_ids(project_id, agent_id, skip_prompt):
if "your-gcp-project-id" in project_id or "your-dialogflow-agent-id" in agent_id:
print_warning("Dialogflowのproject_idまたはagent_idがデフォルト値です。")
if not prompt_for_action("この設定で続行しますか？", skip_prompt, indent=1):
return False
return True
def check_playbook_tool_ids(tool_ids, skip_prompt):
if not tool_ids or any("your-" in tid for tid in tool_ids):
print_warning("Playbook作成に使用するtool_idsがデフォルト値です。")
if not prompt_for_action("この設定で続行しますか？", skip_prompt, indent=1):
return False
return True
# --- テスト関数 (UIのみ改修) ---
def test_sync_bq_to_es(skip_prompt=False):
print_flow_header("データ同期 (BigQuery → Elasticsearch)")
if not prompt_for_action("このテストを実行しますか？", skip_prompt):
print(" スキップされました。");
print_flow_footer(success=False);
return
# --- インタラクティブな入力部 ---
if skip_prompt:
# スキップ時はデフォルト値を使用
dataset_id = DEFAULT_BQ_DATASET
table_id = DEFAULT_BQ_TABLE
es_url = DEFAULT_ES_URL
es_index = DEFAULT_ES_INDEX_BQ
print_info_block("使用する設定（自動）",
f"BQ Dataset: {dataset_id}\n"
f"BQ Table: {table_id}\n"
f"ES URL: {es_url}\n"
f"ES Index: {es_index}")
else:
# ユーザーに入力を促す
print_step_header("同期設定の入力")
dataset_id = prompt_user("BigQueryデータセットIDを入力してください", default_value=DEFAULT_BQ_DATASET)
table_id = prompt_user("BigQueryテーブルIDを入力してください", default_value=DEFAULT_BQ_TABLE)
es_url = prompt_user("ElasticsearchのURLを入力してください", default_value=DEFAULT_ES_URL)
es_index = prompt_user("Elasticsearchのインデックス名を入力してください", default_value=DEFAULT_ES_INDEX_BQ)
# --- ペイロード作成 ---
payload = {
"dataset_id": dataset_id,
"table_id": table_id,
"elasticsearch_url": es_url,
"elasticsearch_index": es_index,
"elasticsearch_encoded_api_key": getattr(vars, "ELASTIC_CLOUD_ENCODED_API_KEY", None) if vars else None
}
# APIキーに関する警告
if not payload["elasticsearch_encoded_api_key"] and "localhost" not in es_url and "127.0.0.1" not in es_url:
print_warning("Elasticsearch APIキーがvars.pyに未設定で、ES URLもローカルホストではありません。")
if not prompt_for_action("この設定でテストを続行しますか？", skip_prompt, indent=1):
print(" テストを中断しました。")
print_flow_footer(success=False)
return
# --- APIリクエスト実行 ---
print_step_header("同期実行")
print_process("データを同期しています")
response = make_api_request(f"/{APP_PREFIX}/data/sync_bq_to_es", payload, timeout=180)
if response:
print_success("データ同期が完了しました")
print_info_block("APIからのメッセージ", response.get('status', '詳細不明'))
else:
print_warning("データ同期に失敗しました。詳細は上記のエラーを確認してください。")
print_flow_footer(success=bool(response))
def test_generate_schema_and_create_tool_interactive(
es_index_for_schema_gen, df_project_id, df_location_id, df_agent_id,
df_tool_name_base, df_tool_description, skip_prompts,
df_playbook_name_base, df_playbook_goal, parent_playbook_id):
print_flow_header("Tool/Playbook連続作成フロー")
if not prompt_for_action(f"このフローを実行しますか？ (Agent: {df_agent_id})", skip_prompts):
print(" スキップされました。");
print_flow_footer(success=False);
return
if not check_api_key("GEMINI_API_KEY") or not check_dialogflow_ids(df_project_id, df_agent_id, skip_prompts):
print_flow_footer(success=False);
return
# --- スキーマ生成 ---
print_step_header("ステップ1: OpenAPIスキーマ生成")
generated_openapi_schema_obj = None
for i in range(3):
print_process(f"スキーマを生成しています... (試行 {i + 1}/3)")
payload = {"elasticsearch_index_name": es_index_for_schema_gen,
"elasticsearch_document_examples": DEFAULT_DOC_EXAMPLES_SCHEMA,
"gemini_model_name": DEFAULT_GEMINI_MODEL}
response = make_api_request(f"/{APP_PREFIX}/generate/openapi_schema", payload, timeout=180, indent=1)
if response:
generated_openapi_schema_obj = response
print_info_block("生成されたOpenAPIスキーマ", json.dumps(response, indent=2, ensure_ascii=False), indent=1)
if skip_prompts:
user_choice = "yes"
else:
user_choice = prompt_user("このスキーマでToolを作成しますか？ (yes/no/retry)", indent=1).lower()
if user_choice in ["yes", "y"]: break
if user_choice in ["no", "n"]: print(" ユーザーにより中断されました。"); print_flow_footer(
False); return None
else: # response is None
if not prompt_for_action("スキーマ生成を再試行しますか？", skip_prompts, indent=1):
print(" ユーザーにより中断されました。");
print_flow_footer(False);
return None
if not generated_openapi_schema_obj:
print_warning("有効なスキーマが得られませんでした。");
print_flow_footer(False);
return None
# --- Tool作成 ---
print_step_header("ステップ2: Dialogflow Tool作成")
tool_name = prompt_user("Toolの表示名を入力してください", default_value=df_tool_name_base)
tool_desc = prompt_user("Toolの説明を入力してください", default_value=df_tool_description)
clean_es_index_name = es_index_for_schema_gen.replace('-', '_').replace('.', '_')
final_tool_name = tool_name
if es_index_for_schema_gen and not es_index_for_schema_gen in final_tool_name and not clean_es_index_name in final_tool_name:
final_tool_name += f"_{clean_es_index_name}"
print_info_block("最終的なTool表示名", final_tool_name)
print_process("Toolを作成しています")
payload = {"project_id": df_project_id, "location_id": df_location_id, "agent_id": df_agent_id,
"tool_display_name": final_tool_name, "tool_description": tool_desc,
"openapi_schema_string": json.dumps(generated_openapi_schema_obj)}
response = make_api_request(f"/{APP_PREFIX}/dialogflow/create_tool", payload, timeout=120)
if not (response and "tool_id" in response):
print_warning("Toolの作成に失敗しました。");
print_flow_footer(False);
return None
created_tool_id = response['tool_id']
created_tool_display_name = response.get('display_name', final_tool_name)
print_success(f"Tool『{created_tool_display_name}』(ID: {created_tool_id}) を作成しました")
# --- Playbook作成 ---
if prompt_for_action(f"作成したTool『{created_tool_display_name}』を使用してPlaybookを作成しますか？", skip_prompts):
playbook_info = test_generate_instructions_and_create_playbook_interactive(
is_standalone=False, # このフラグでスタンドアロンでないことを示す
es_index_for_instr_gen=es_index_for_schema_gen,
doc_examples_for_instr_gen=DEFAULT_DOC_EXAMPLES_SCHEMA,
gemini_model_for_instr_gen=DEFAULT_GEMINI_MODEL,
tool_name_for_instr_gen=created_tool_display_name,
df_project_id_interactive=df_project_id,
df_location_id_interactive=df_location_id,
df_agent_id_interactive=df_agent_id,
df_playbook_display_name_base_interactive=df_playbook_name_base,
df_playbook_goal_interactive=df_playbook_goal,
df_tool_ids_interactive=[created_tool_id],
skip_prompts_interactive=skip_prompts
)
# --- 親Playbookへ追加 ---
if playbook_info and parent_playbook_id:
if prompt_for_action(
f"作成したPlaybook『{playbook_info['display_name']}』を親Playbook(ID:{parent_playbook_id})に追加しますか？",
skip_prompts):
add_child_playbooks_to_parent_interactive(
parent_playbook_id=parent_playbook_id,
created_child_playbooks=[playbook_info],
df_project_id=df_project_id,
df_location_id=df_location_id,
df_agent_id=df_agent_id,
skip_prompts=skip_prompts
)
print_flow_footer()
return {"tool_id": created_tool_id, "display_name": created_tool_display_name}
def test_generate_instructions_and_create_playbook_interactive(
is_standalone, es_index_for_instr_gen, doc_examples_for_instr_gen, gemini_model_for_instr_gen,
tool_name_for_instr_gen, df_project_id_interactive, df_location_id_interactive, df_agent_id_interactive,
df_playbook_display_name_base_interactive, df_playbook_goal_interactive, df_tool_ids_interactive,
skip_prompts_interactive):
header_text = "ステップ3: 関連Playbook作成" if not is_standalone else "ステップ1: Playbook作成"
print_step_header(header_text)
if is_standalone: # スタンドアロンの場合のみ前提条件チェック
if not check_api_key("GEMINI_API_KEY") or not check_dialogflow_ids(df_project_id_interactive,
df_agent_id_interactive,
skip_prompts_interactive):
return None
if not check_playbook_tool_ids(df_tool_ids_interactive, skip_prompts_interactive):
return None
# Instructions生成
generated_instructions_markdown = None
for i in range(3):
print_process(f"Playbookの指示を生成しています... (試行 {i + 1}/3)", indent=1)
payload = {"elasticsearch_index_name": es_index_for_instr_gen,
"elasticsearch_document_examples": doc_examples_for_instr_gen,
"gemini_model_name": gemini_model_for_instr_gen, "tool_name_in_playbook": tool_name_for_instr_gen}
response = make_api_request(f"/{APP_PREFIX}/generate/playbook_instructions", payload, timeout=240, indent=2)
if response and "playbook_instructions_markdown" in response:
generated_instructions_markdown = response["playbook_instructions_markdown"]
print_info_block("生成されたPlaybook Instructions", generated_instructions_markdown, indent=2)
if skip_prompts_interactive:
user_choice = "yes"
else:
user_choice = prompt_user("この指示でPlaybookを作成しますか？ (yes/no/retry)", indent=2).lower()
if user_choice in ["yes", "y"]: break
if user_choice in ["no", "n"]: print(" ユーザーにより中断されました。"); return None
else:
if not prompt_for_action("指示の生成を再試行しますか？", skip_prompts_interactive, indent=2):
print(" ユーザーにより中断されました。");
return None
if not generated_instructions_markdown:
print_warning("有効な指示が得られませんでした。", indent=1);
return None
# Playbook作成
playbook_name = prompt_user("Playbookの表示名を入力してください",
default_value=df_playbook_display_name_base_interactive, indent=1)
playbook_goal = prompt_user("PlaybookのGoalを入力してください", default_value=df_playbook_goal_interactive,
indent=1)
clean_es_index_name = es_index_for_instr_gen.replace('-', '_').replace('.', '_')
final_playbook_name = playbook_name
if es_index_for_instr_gen and not es_index_for_instr_gen in final_playbook_name and not clean_es_index_name in final_playbook_name:
final_playbook_name += f"_{clean_es_index_name}"
print_info_block("最終的なPlaybook表示名", final_playbook_name, indent=1)
print_process("Playbookを作成しています", indent=1)
payload = {"project_id": df_project_id_interactive, "location_id": df_location_id_interactive,
"agent_id": df_agent_id_interactive, "playbook_display_name": final_playbook_name,
"playbook_goal": playbook_goal, "playbook_instruction_text": generated_instructions_markdown,
"tool_ids": df_tool_ids_interactive}
response = make_api_request(f"/{APP_PREFIX}/dialogflow/create_playbook", payload, timeout=120, indent=1)
if not (response and "playbook_id" in response):
print_warning("Playbookの作成に失敗しました。", indent=1);
return None
created_playbook_id = response['playbook_id']
created_playbook_name = response.get('display_name', final_playbook_name)
print_success(f"Playbook『{created_playbook_name}』(ID: {created_playbook_id}) を作成しました", indent=1)
return {"playbook_id": created_playbook_id, "display_name": created_playbook_name}
def add_child_playbooks_to_parent_interactive(
parent_playbook_id, created_child_playbooks, df_project_id, df_location_id, df_agent_id, skip_prompts):
header_text = "ステップ4: 親Playbookへの追加"
print_step_header(header_text)
print_process("親Playbookの既存の指示を取得しています")
get_instr_payload = {"project_id": df_project_id, "location_id": df_location_id, "agent_id": df_agent_id,
"playbook_id": parent_playbook_id}
parent_instr_response = make_api_request(f"/{APP_PREFIX}/dialogflow/get_playbook_instructions", get_instr_payload,
timeout=120)
if not (parent_instr_response and "instructions_text" in parent_instr_response):
print_warning("親Playbookの指示取得に失敗しました。");
return
existing_parent_instructions = parent_instr_response["instructions_text"]
child_playbook_info = created_child_playbooks[0]
child_name = child_playbook_info["display_name"]
child_desc_prompt = f"子Playbook『{child_name}』の説明を入力してください (親Playbookに追加されます)"
if skip_prompts:
child_description = f"これは '{child_name}' Playbook の自動生成された説明です。"
print(f" ▶ {child_desc_prompt}: (自動入力) {child_description}")
else:
child_description = prompt_user(child_desc_prompt, default_value=f"『{child_name}』を実行する", indent=1)
instruction_to_add = f"\n- - ${{PLAYBOOK:{child_name}}} - {child_description}"
new_parent_instructions = existing_parent_instructions + instruction_to_add
print_info_block("更新後の親Playbookの指示プレビュー", new_parent_instructions, indent=1)
if prompt_for_action("この内容で親Playbookを更新しますか？", skip_prompts, indent=1):
print_process("親Playbookを更新しています", indent=2)
update_instr_payload = {"project_id": df_project_id, "location_id": df_location_id, "agent_id": df_agent_id,
"playbook_id": parent_playbook_id, "new_instructions_text": new_parent_instructions}
update_response = make_api_request(f"/{APP_PREFIX}/dialogflow/update_playbook_instructions",
update_instr_payload, timeout=120, indent=2)
if update_response:
print_success("親Playbookの更新に成功しました。", indent=2)
if "new_instructions_preview" in update_response:
print_info_block("更新後の最終的な指示", update_response['new_instructions_preview'], indent=2)
else:
print_warning("親Playbookの更新に失敗しました。", indent=2)
else:
print(" 更新はキャンセルされました。")
# --- メイン実行部 (元のロジックを維持) ---
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Integration Service APIテストスクリプト")
parser.add_argument(
"--test",
choices=["sync", "generate_and_create_tool", "generate_and_create_playbook", "all"],
nargs="+",
help="実行するテストを選択 (複数指定可)",
default=["all"]
)
parser.add_argument("-y", "--yes", action="store_true",
help="全ての確認プロンプト（表示名入力含む）をスキップし、自動で 'yes' と回答、またはデフォルト値を使用します")
parser.add_argument("--df_project_id", default=DEFAULT_DF_PROJECT_ID,
help=f"Dialogflow Project ID (デフォルト: {DEFAULT_DF_PROJECT_ID})")
parser.add_argument("--df_location_id", default=DEFAULT_DF_LOCATION_ID,
help=f"Dialogflow Agent Location ID (デフォルト: {DEFAULT_DF_LOCATION_ID})")
parser.add_argument("--df_agent_id", default=DEFAULT_DF_AGENT_ID,
help=f"Dialogflow Agent ID (デフォルト: {DEFAULT_DF_AGENT_ID})")
parser.add_argument("--parent_playbook_id", default=DEFAULT_PARENT_PLAYBOOK_ID,
help=f"子Playbookを追加する先の親PlaybookのID (デフォルト: {DEFAULT_PARENT_PLAYBOOK_ID})")
parser.add_argument("--gact_tool_name_base", default=DEFAULT_GATC_TOOL_DISPLAY_NAME_BASE,
help=f"Tool表示名入力時のデフォルトのベース名。(デフォルト: {DEFAULT_GATC_TOOL_DISPLAY_NAME_BASE})")
parser.add_argument("--gact_es_index", default=DEFAULT_ES_INDEX_SCHEMA,
help=f"Tool作成テスト（スキーマ生成元）用ESインデックス名。(デフォルト: {DEFAULT_ES_INDEX_SCHEMA})")
parser.add_argument("--gacp_playbook_name_base", default=DEFAULT_GACP_PLAYBOOK_DISPLAY_NAME_BASE,
help=f"Playbook表示名入力時のデフォルトのベース名。(デフォルト: {DEFAULT_GACP_PLAYBOOK_DISPLAY_NAME_BASE})")
parser.add_argument("--gacp_es_index", default=DEFAULT_ES_INDEX_PLAYBOOK_INSTRUCTIONS,
help=f"スタンドアロンPlaybook作成テスト（Instructions生成元）用ESインデックス名。(デフォルト: {DEFAULT_ES_INDEX_PLAYBOOK_INSTRUCTIONS})")
parser.add_argument("--gacp_playbook_goal", default=DEFAULT_GACP_PLAYBOOK_GOAL,
help=f"Playbook作成テスト用Playbookのゴール。(デフォルト: {DEFAULT_GACP_PLAYBOOK_GOAL})")
parser.add_argument("--gacp_tool_ids", nargs="*", default=DEFAULT_DF_REFERENCED_TOOL_IDS,
help=f"スタンドアロンPlaybook作成テスト用参照ツールIDリスト。(デフォルト: {DEFAULT_DF_REFERENCED_TOOL_IDS})")
args = parser.parse_args()
print(f"テスト対象サービスURL: {BASE_URL}")
print(f"実行対象テスト: {', '.join(args.test)}")
if args.yes:
print("注意: 全ての確認プロンプト（表示名入力含む）はスキップされます。")
tests_to_run_flags = {"sync": False, "generate_and_create_tool": False, "generate_and_create_playbook": False}
if "all" in args.test:
for key in tests_to_run_flags: tests_to_run_flags[key] = True
else:
for test_name in args.test:
if test_name in tests_to_run_flags:
tests_to_run_flags[test_name] = True
if tests_to_run_flags["sync"]:
test_sync_bq_to_es(skip_prompt=args.yes)
if tests_to_run_flags["generate_and_create_tool"]:
test_generate_schema_and_create_tool_interactive(
es_index_for_schema_gen=args.gact_es_index,
df_project_id=args.df_project_id,
df_location_id=args.df_location_id,
df_agent_id=args.df_agent_id,
df_tool_name_base=args.gact_tool_name_base,
df_tool_description=DEFAULT_GATC_TOOL_DESCRIPTION,
skip_prompts=args.yes,
df_playbook_name_base=args.gacp_playbook_name_base,
df_playbook_goal=args.gacp_playbook_goal,
parent_playbook_id=args.parent_playbook_id
)
if tests_to_run_flags["generate_and_create_playbook"]:
print_flow_header("スタンドアロンPlaybook作成フロー")
if prompt_for_action(f"このフローを実行しますか？ (Agent: {args.df_agent_id})", args.yes):
created_playbook_info = test_generate_instructions_and_create_playbook_interactive(
is_standalone=True,
es_index_for_instr_gen=args.gacp_es_index,
doc_examples_for_instr_gen=DEFAULT_DOC_EXAMPLES_PLAYBOOK_INSTRUCTIONS,
gemini_model_for_instr_gen=DEFAULT_GEMINI_MODEL,
tool_name_for_instr_gen=DEFAULT_CUSTOM_TOOL_NAME_PLAYBOOK_INSTRUCTIONS,
df_project_id_interactive=args.df_project_id,
df_location_id_interactive=args.df_location_id,
df_agent_id_interactive=args.df_agent_id,
df_playbook_display_name_base_interactive=args.gacp_playbook_name_base,
df_playbook_goal_interactive=args.gacp_playbook_goal,
df_tool_ids_interactive=args.gacp_tool_ids,
skip_prompts_interactive=args.yes
)
if created_playbook_info and args.parent_playbook_id:
if prompt_for_action(
f"作成したPlaybook『{created_playbook_info['display_name']}』を親Playbookに追加しますか？",
args.yes):
add_child_playbooks_to_parent_interactive(
parent_playbook_id=args.parent_playbook_id,
created_child_playbooks=[created_playbook_info],
df_project_id=args.df_project_id,
df_location_id=args.df_location_id,
df_agent_id=args.df_agent_id,
skip_prompts=args.yes
)
print_flow_footer(success=bool(created_playbook_info))
else:
print(" スキップされました。")
print_flow_footer(success=False)
print("\nすべての選択されたテストの実行試行が完了しました。")
