🙌
Strands+Galileo(Agent Control)を試す
Galileo Agent Control について
Galileo が2026年3月に出した OSS のガードレール基盤。
"AI agent governance" と呼ばれる、新規市場の役者の一人。
この市場、まだ軸が定まっていない。
ガードレール(個別呼び出しを止める)、オーケストレーション(エージェント自体を管理する)、observability(事後に観測する) の3つが混ざっていて、
Galileo Agent Control に関していうと「ガードレール軸の中で OSS + フレームワーク非依存 + 中央集権ポリシー」という比較的明確なポジション。
「中央集権的なポリシープレーン」という思想自体が一つの選択であって、NeMo Guardrails のような分散ガード、対話フロー全体を記述する DSL アプローチとは設計思想が異なる。
特徴
- (1) コードとポリシーの分離が徹底している — @control() デコレータと外部ポリシー定義の組み合わせで、エージェント側コードを変えずにruntime挙動を変えられる。これは他にも謳う製品はあるけど、Agent Control は「policies once, deploy anywhere」という portable policies を一次的な売りに置いている
- (2) フレームワーク非依存 — Strands、CrewAI、LangChain、Google ADK のどれでも同じポリシーが効く。これは LangChain 系のように「自社フレームワークに統合された guardrail」とは思想が逆。
- (3) Apache-2.0 オープンソース + vendor-neutral evaluator — regex、list、Luna-2 (Galileo自社モデル)、custom evaluator まで差し替え可能。lock-in を避ける設計。
1. このプロジェクト（Galileoためすプロジェクト）の位置づけ
これは Agent Control(Galileo OSS / Apache-2.0)の挙動を手で触って理解するための学習プロジェクト。
つくるもの
Agent Control 入門サンドボックス(旅程プランナー)
なぜ旅程プランナーなのか
学習用の題材に必要だった条件:
- Agent Control の3つの action(deny / warn / steer)が自然に全部出る
- pre / post の非対称性が意味を持つ
- multi-agent 的要素はあるが、題材複雑性に溺れない
- 半日〜1日で動く
- staged trust delegation(tool単位の段階的権限委譲)の実地写像になる
旅程プランナーは、tool 呼び出しごとに「予算ガード」「整合性ガード」「トーン調整」が自然に必要になるので、学習素材として密度が高い。
このプロジェクトで得たいもの
-
@control()デコレータ、policy 記述、evaluator、decision(allow/warn/steer/deny)の実地感覚 - pre と post のどちらで介入するかという設計判断の肌感
- 中央集権的ポリシープレーンという思想に対する自分なりの評価軸
- dashboard に記録される事象の語彙から、Galileo の設計者が何を可観測にしたかを読む
得なくていいもの:
- 旅程プランナーとしての完成度
- Strands の高度な multi-agent 機能の習熟
- 本番運用を想定した設計
2. 技術スタック
| 要素 | 選択 | 理由 |
|---|---|---|
| エージェントフレームワーク | Strands Agents (Python) | 私が既に習熟しているため。あと今のところベンダーロックインがないため。 |
| LLM | OpenAI gpt-5.4-mini
|
低コスト・低レイテンシ、学習用途に適う |
| ガバナンス層 | Agent Control (Docker Compose) | このプロジェクトの主役 |
| 言語 | Python 3.12+ | Agent Control SDK の前提 |
| パッケージ管理 | uv | 推奨、軽い |
| 環境変数 |
.env + python-dotenv |
OPENAI_API_KEY を gitignore 配下で管理 |
依存パッケージ
strands-agents[openai]
agent-control-sdk
python-dotenv
3. エージェント構成
単一エージェント + 3 tool で構成する。multi-agent hand-off は使わない。
理由: @control() の介入点と観察対象の対応をシンプルに保つため。複雑性は tool の数で稼ぐ。
エージェント
-
planner_agent- system prompt: 旅程プランナーとして振る舞う。予算・日程・制約を守って旅程を組み立てる
- 利用可能 tool:
search_flights,search_hotels,assemble_itinerary
Tool 定義(すべてパラメトリックモック、外部API呼ばない)
| tool | 入力 | 出力 |
|---|---|---|
search_flights |
origin, destination, date, budget_max
|
固定ルールで3件のダミー結果を生成 |
search_hotels |
city, checkin, checkout, budget_per_night
|
固定ルールで3件のダミー結果を生成 |
assemble_itinerary |
flights, hotels, activities, user_constraints
|
組み立てた旅程を dict で返す |
モックの実装方針:
- 引数に応じて結果を動的生成(例: budget_max 以下の価格帯を返す)
- ランダム要素は入れない(失敗の再現性を確保)
- LLM が「矛盾した引数を渡す」「制約を忘れる」といった failure mode を起こせる余地を残す
4. Agent Control による介入設計
3つの control を異なる思想で配置する。これが本プロジェクトの中核実験。
Control A: 予防型 (pre / deny)
-
配置:
search_flightsの呼び出し前 -
トリガ:
budget_maxがユーザー指定の総予算の 80% を超えている - action: deny
- 思想: 「信用しない」。委譲しない判断。強制的に止める
- staged trust delegation での位置: 権限を与えない
Control B: ステアリング型 (pre / steer)
-
配置:
assemble_itineraryの呼び出し前 - トリガ: ユーザー制約(例: 「飛行機なし指定」)と引数内容に矛盾がある
- action: steer(引数を補正して続行させる)
- 思想: 「部分的に信用」。LLM の見落としを構造的に補正
- staged trust delegation での位置: 引数ガード付きで委譲
Control C: 事後検閲型 (post / deny)
-
配置:
assemble_itineraryの出力後 - トリガ: 最終行程の合計金額がユーザー総予算を超えている
- action: deny
- 思想: 「自由にやらせて結果だけ見る」。局所判断は通すが全体整合だけ検閲
- staged trust delegation での位置: 結果検閲付きで委譲
観察するメトリクス
- 各 control の発火回数
- pre で止まった場合と post で止まった場合の、LLM 再試行挙動の違い
- steer で書き換えた後の LLM の振る舞い(素直に従うか、元の意図に戻ろうとするか)
- Agent Control dashboard(localhost:8000)に記録される事象の語彙
5. 実装フェーズ（計画）
Phase 0: 環境構築
- ディレクトリ作成、
pyproject.toml、.env.example、.gitignore - Docker Compose で Agent Control server + UI + Postgres 起動
-
uv venv、依存インストール -
localhost:8000/healthで疎通確認
Phase 1: controlなしの最小構成
- 3 tool のパラメトリックモック実装
- planner_agent を Strands
OpenAIModel(model_id="gpt-5.4-mini")で構築 - 動作確認スクリプト: 「東京→札幌、予算5万、2泊3日」で旅程が組めるか
- この時点では Agent Control は起動しているだけ、
@control()は付けない
Phase 2: Control A(予防型 / deny)追加
-
search_flightsに@control()を付ける - Agent Control UI か setup スクリプトで control を登録
- 「予算内だが flight budget だけ超過する」ケースで deny 発火を確認
- dashboard で記録を確認
Phase 3: Control B(ステアリング型 / steer)追加
-
assemble_itineraryに@control()を追加 - 「飛行機なし」指定で国際線が混入するケースで steer 発火を確認
- steer 後の LLM 挙動を観察
Phase 4: Control C(事後検閲型 / post / deny)追加
-
assemble_itineraryの post stage に control 追加 - 合計金額超過ケースで deny 発火を確認
- 3つの control の相互作用を観察
Phase 5: 振り返り
- 発火パターンの記録
- L3/L4 の観点で「この設計が何を可観測にしているか」をメモ
- 中央集権プレーンという思想への評価メモ
基本コード
agent.py
from __future__ import annotations
import os
import agent_control
from dotenv import load_dotenv
from .tools import (
assemble_itinerary,
assemble_itinerary_plain,
search_flights,
search_flights_plain,
search_hotels,
)
SYSTEM_PROMPT = """
You are a travel planner.
Respect user budget, dates, and constraints.
Use the available tools to search flights, search hotels, and assemble an itinerary.
If the user constraint conflicts with a proposed option, explain the conflict clearly.
""".strip()
AGENT_NAME = "agent-control-travel-sandbox"
AGENT_DESCRIPTION = "Travel planner sandbox for Agent Control learning"
def build_planner_agent(model_id: str = "gpt-5.4-mini", with_control: bool = True):
"""Build the Strands agent for the travel sandbox."""
load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
raise RuntimeError("OPENAI_API_KEY is required to build the planner agent.")
try:
from strands import Agent
from strands.models.openai import OpenAIModel
except ImportError as exc: # pragma: no cover - import guard
raise RuntimeError(
"Install strands-agents[openai] before running the planner agent."
) from exc
if with_control:
server_url = (
os.getenv("AGENT_CONTROL_SERVER_URL")
or os.getenv("AGENT_CONTROL_URL")
or "http://localhost:8000"
)
agent_control.init(
agent_name=AGENT_NAME,
agent_description=AGENT_DESCRIPTION,
server_url=server_url,
policy_refresh_interval_seconds=0,
)
model = OpenAIModel(
client_args={"api_key": api_key},
model_id=model_id,
params={
"max_completion_tokens": 1200,
"temperature": 0.2,
},
)
tools = (
[search_flights, search_hotels, assemble_itinerary]
if with_control
else [search_flights_plain, search_hotels, assemble_itinerary_plain]
)
return Agent(model=model, tools=tools, system_prompt=SYSTEM_PROMPT)
tools.py
from __future__ import annotations
from dataclasses import asdict, dataclass
from datetime import date
from typing import Any, Iterable
from agent_control import control
from strands import tool
@dataclass(frozen=True)
class FlightOption:
airline: str
origin: str
destination: str
date: str
departure: str
arrival: str
price: int
baggage_included: bool
@dataclass(frozen=True)
class HotelOption:
name: str
city: str
checkin: str
checkout: str
nightly_rate: int
nights: int
total_price: int
breakfast_included: bool
def _clean_budget(value: Any) -> int:
try:
return max(0, int(value))
except (TypeError, ValueError):
return 0
def _count_nights(checkin: str, checkout: str) -> int:
try:
start = date.fromisoformat(checkin)
end = date.fromisoformat(checkout)
except ValueError:
return 1
nights = (end - start).days
return nights if nights > 0 else 1
def _search_flights_impl(
origin: str,
destination: str,
date: str,
budget_max: int,
) -> list[dict[str, Any]]:
"""Return deterministic flight options that stay within the budget."""
budget_max = _clean_budget(budget_max)
if budget_max <= 0:
return []
base_price = max(6_000, budget_max // 3)
price_steps = (
base_price,
min(budget_max, base_price + max(1_000, budget_max // 10)),
min(budget_max, base_price + max(2_000, budget_max // 6)),
)
airlines = ("Aurora Air", "Northwind", "Skyline Express")
departures = ("07:10", "11:20", "18:40")
arrivals = ("09:05", "13:15", "20:35")
results: list[dict[str, Any]] = []
for index, price in enumerate(price_steps):
results.append(
asdict(
FlightOption(
airline=airlines[index],
origin=origin,
destination=destination,
date=date,
departure=departures[index],
arrival=arrivals[index],
price=min(price, budget_max),
baggage_included=index != 2,
)
)
)
return results
@control(policy="flight-budget-guard", step_name="search_flights")
@tool
def _search_flights_guarded(
origin: str,
destination: str,
date: str,
budget_max: int,
) -> list[dict[str, Any]]:
"""Guarded flight search used to exercise Agent Control."""
return _search_flights_impl(
origin=origin,
destination=destination,
date=date,
budget_max=budget_max,
)
@tool
def search_flights(
origin: str,
destination: str,
date: str,
budget_max: int,
) -> list[dict[str, Any]]:
"""Return deterministic flight options that stay within the budget."""
return _search_flights_guarded(
origin=origin,
destination=destination,
date=date,
budget_max=budget_max,
)
@tool
def search_flights_plain(
origin: str,
destination: str,
date: str,
budget_max: int,
) -> list[dict[str, Any]]:
"""Return deterministic flight options without Agent Control."""
return _search_flights_impl(
origin=origin,
destination=destination,
date=date,
budget_max=budget_max,
)
@tool
def search_hotels(
city: str,
checkin: str,
checkout: str,
budget_per_night: int,
) -> list[dict[str, Any]]:
"""Return deterministic hotel options that stay within the nightly budget."""
budget_per_night = _clean_budget(budget_per_night)
if budget_per_night <= 0:
return []
nights = _count_nights(checkin, checkout)
base_rate = max(5_000, budget_per_night // 2)
rate_steps = (
base_rate,
min(budget_per_night, base_rate + max(750, budget_per_night // 8)),
min(budget_per_night, base_rate + max(1_500, budget_per_night // 5)),
)
names = ("Harbor Stay", "City Nest", "Quiet Lodge")
results: list[dict[str, Any]] = []
for index, nightly_rate in enumerate(rate_steps):
nightly_rate = min(nightly_rate, budget_per_night)
results.append(
asdict(
HotelOption(
name=names[index],
city=city,
checkin=checkin,
checkout=checkout,
nightly_rate=nightly_rate,
nights=nights,
total_price=nightly_rate * nights,
breakfast_included=index != 0,
)
)
)
return results
def _activity_cost(activity: dict[str, Any]) -> int:
raw_cost = activity.get("estimated_cost", activity.get("budget", 0))
return _clean_budget(raw_cost)
def _assemble_itinerary_impl(
flights: Iterable[dict[str, Any]],
hotels: Iterable[dict[str, Any]],
activities: Iterable[dict[str, Any]],
user_constraints: dict[str, Any],
) -> dict[str, Any]:
"""Assemble a deterministic trip plan from already generated options."""
flight_list = [dict(item) for item in flights]
hotel_list = [dict(item) for item in hotels]
activity_list = [dict(item) for item in activities]
selected_flight = flight_list[0] if flight_list else None
selected_hotel = hotel_list[0] if hotel_list else None
total_budget = _clean_budget(user_constraints.get("total_budget", 0))
activity_total = sum(_activity_cost(item) for item in activity_list)
flight_total = _clean_budget(selected_flight["price"]) if selected_flight else 0
hotel_total = _clean_budget(selected_hotel["total_price"]) if selected_hotel else 0
total_cost = flight_total + hotel_total + activity_total
warnings: list[str] = []
if user_constraints.get("flight_allowed") is False and flight_list:
warnings.append("Flight options are present even though flight travel was disallowed.")
if total_budget and total_cost > total_budget:
warnings.append(f"Total cost {total_cost} exceeds budget {total_budget}.")
return {
"selected_flight": selected_flight,
"selected_hotel": selected_hotel,
"activities": activity_list,
"budget": {
"total_budget": total_budget,
"flight_total": flight_total,
"hotel_total": hotel_total,
"activity_total": activity_total,
"total_cost": total_cost,
"within_budget": not total_budget or total_cost <= total_budget,
},
"user_constraints": dict(user_constraints),
"warnings": warnings,
}
@tool
def assemble_itinerary(
flights: Iterable[dict[str, Any]],
hotels: Iterable[dict[str, Any]],
activities: Iterable[dict[str, Any]],
user_constraints: dict[str, Any],
) -> dict[str, Any]:
"""Assemble a deterministic trip plan from already generated options."""
return _assemble_itinerary_guarded(
flights=flights,
hotels=hotels,
activities=activities,
user_constraints=user_constraints,
)
@tool
def assemble_itinerary_plain(
flights: Iterable[dict[str, Any]],
hotels: Iterable[dict[str, Any]],
activities: Iterable[dict[str, Any]],
user_constraints: dict[str, Any],
) -> dict[str, Any]:
"""Assemble a deterministic trip plan without Agent Control."""
return _assemble_itinerary_impl(
flights=flights,
hotels=hotels,
activities=activities,
user_constraints=user_constraints,
)
@control(policy="itinerary-steer", step_name="assemble_itinerary")
@tool(name="assemble_itinerary_guarded")
def _assemble_itinerary_guarded(
flights: Iterable[dict[str, Any]],
hotels: Iterable[dict[str, Any]],
activities: Iterable[dict[str, Any]],
user_constraints: dict[str, Any],
) -> dict[str, Any]:
"""Guarded itinerary assembly used to exercise pre-steer controls."""
return _assemble_itinerary_impl(
flights=flights,
hotels=hotels,
activities=activities,
user_constraints=user_constraints,
)
@control(policy="itinerary-post-deny", step_name="assemble_itinerary")
@tool(name="assemble_itinerary_post_guarded")
def _assemble_itinerary_post_guarded(
flights: Iterable[dict[str, Any]],
hotels: Iterable[dict[str, Any]],
activities: Iterable[dict[str, Any]],
user_constraints: dict[str, Any],
) -> dict[str, Any]:
"""Guarded itinerary assembly used to exercise post-deny controls."""
return _assemble_itinerary_impl(
flights=flights,
hotels=hotels,
activities=activities,
user_constraints=user_constraints,
)
main.py
from __future__ import annotations
import argparse
import json
from typing import Any
from .agent import build_planner_agent
from .tools import assemble_itinerary, search_flights, search_hotels
def _demo_payload() -> dict[str, Any]:
flights = search_flights("Tokyo", "Sapporo", "2026-05-01", 50_000)
hotels = search_hotels("Sapporo", "2026-05-01", "2026-05-03", 15_000)
activities = [
{"name": "Otaru walk", "estimated_cost": 3_000},
{"name": "Seafood market lunch", "estimated_cost": 2_500},
]
constraints = {
"total_budget": 50_000,
"flight_allowed": True,
"note": "Two nights in Hokkaido",
}
return assemble_itinerary(flights, hotels, activities, constraints)
def run_demo() -> None:
print(json.dumps(_demo_payload(), ensure_ascii=False, indent=2))
def run_agent() -> None:
agent = build_planner_agent()
prompt = (
"Plan a two-night trip from Tokyo to Sapporo with a total budget of 50000 "
"JPY. Include one or two simple activities and keep the output concise."
)
response = agent(prompt)
print(response)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Travel sandbox CLI")
parser.add_argument(
"--mode",
choices=("demo", "agent"),
default="demo",
help="Run the deterministic demo or the Strands agent.",
)
return parser.parse_args()
def main() -> None:
args = parse_args()
if args.mode == "agent":
run_agent()
return
run_demo()
if __name__ == "__main__":
main()
Agent Control サーバーに control を登録するためのスクリプト
setup_controls.py
from __future__ import annotations
import asyncio
import os
import sys
from typing import Any
from pathlib import Path
import agent_control
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from src.agent import AGENT_DESCRIPTION, AGENT_NAME
import src.tools # noqa: F401 # Registers @control-decorated steps.
CONTROL_NAME = "flight-budget-guard"
STEER_CONTROL_NAME = "itinerary-flight-conflict-steer"
POST_DENY_CONTROL_NAME = "itinerary-total-budget-deny"
def build_control_data() -> dict[str, Any]:
return {
"description": (
"Block flight searches when the requested budget is at or above "
"the 80% budget threshold used in the travel sandbox."
),
"enabled": True,
"execution": "server",
"scope": {
"step_types": ["tool"],
"step_names": ["search_flights"],
"stages": ["pre"],
},
"condition": {
"selector": {"path": "input.budget_max"},
"evaluator": {
"name": "regex",
"config": {
"pattern": r"^(?:[4-9][0-9]{4}|[1-9][0-9]{5,})$",
},
},
},
"action": {"decision": "deny"},
"tags": ["phase-2", "flight", "budget", "deny"],
}
def build_itinerary_control_data() -> dict[str, Any]:
return {
"description": (
"Steer itinerary assembly when flights are present but the user "
"explicitly disallows flying."
),
"enabled": True,
"execution": "server",
"scope": {
"step_types": ["tool"],
"step_names": ["assemble_itinerary"],
"stages": ["pre"],
},
"condition": {
"and": [
{
"selector": {"path": "input"},
"evaluator": {
"name": "regex",
"config": {
"pattern": r"flight_allowed': False",
},
},
},
{
"selector": {"path": "input"},
"evaluator": {
"name": "regex",
"config": {
"pattern": r"(Aurora Air|Northwind|Skyline Express)",
},
},
},
]
},
"action": {
"decision": "steer",
"steering_context": {
"message": (
"User explicitly disallowed flights, but flight options are "
"present. Remove flights from the itinerary and retry with "
"ground-only or no-flight alternatives."
)
},
},
"tags": ["phase-3", "itinerary", "flight", "steer"],
}
def build_post_deny_control_data() -> dict[str, Any]:
return {
"description": "Deny itinerary assembly when the final trip exceeds the total budget.",
"enabled": True,
"execution": "server",
"scope": {
"step_types": ["tool"],
"step_names": ["assemble_itinerary"],
"stages": ["post"],
},
"condition": {
"selector": {"path": "output"},
"evaluator": {
"name": "regex",
"config": {
"pattern": r"'within_budget': False",
},
},
},
"action": {"decision": "deny"},
"tags": ["phase-4", "itinerary", "budget", "deny"],
}
def _control_id(record: dict[str, Any]) -> int | None:
value = record.get("control_id", record.get("id"))
return int(value) if value is not None else None
async def ensure_control(server_url: str, api_key: str | None) -> dict[str, Any]:
existing = await agent_control.list_controls(
server_url=server_url,
api_key=api_key,
name=CONTROL_NAME,
limit=20,
)
for control in existing.get("controls", []):
if control.get("name") == CONTROL_NAME:
control_id = _control_id(control)
if control_id is None:
raise RuntimeError("Existing control missing control_id.")
await agent_control.add_agent_control(
agent_name=AGENT_NAME,
control_id=control_id,
server_url=server_url,
api_key=api_key,
)
return {"control_id": control_id, "configured": True, "reused": True}
created = await agent_control.create_control(
name=CONTROL_NAME,
data=build_control_data(),
server_url=server_url,
api_key=api_key,
)
control_id = _control_id(created)
if control_id is None:
raise RuntimeError("Created control missing control_id.")
await agent_control.add_agent_control(
agent_name=AGENT_NAME,
control_id=control_id,
server_url=server_url,
api_key=api_key,
)
return {"control_id": control_id, "configured": True, "reused": False}
async def ensure_itinerary_control(server_url: str, api_key: str | None) -> dict[str, Any]:
existing = await agent_control.list_controls(
server_url=server_url,
api_key=api_key,
name=STEER_CONTROL_NAME,
limit=20,
)
for control in existing.get("controls", []):
if control.get("name") == STEER_CONTROL_NAME:
control_id = _control_id(control)
if control_id is None:
raise RuntimeError("Existing control missing control_id.")
await agent_control.delete_control(
control_id=control_id,
force=True,
server_url=server_url,
api_key=api_key,
)
break
created = await agent_control.create_control(
name=STEER_CONTROL_NAME,
data=build_itinerary_control_data(),
server_url=server_url,
api_key=api_key,
)
control_id = _control_id(created)
if control_id is None:
raise RuntimeError("Created control missing control_id.")
await agent_control.add_agent_control(
agent_name=AGENT_NAME,
control_id=control_id,
server_url=server_url,
api_key=api_key,
)
return {"control_id": control_id, "configured": True, "reused": False}
async def ensure_post_deny_control(server_url: str, api_key: str | None) -> dict[str, Any]:
existing = await agent_control.list_controls(
server_url=server_url,
api_key=api_key,
name=POST_DENY_CONTROL_NAME,
limit=20,
)
for control in existing.get("controls", []):
if control.get("name") == POST_DENY_CONTROL_NAME:
control_id = _control_id(control)
if control_id is None:
raise RuntimeError("Existing control missing control_id.")
await agent_control.delete_control(
control_id=control_id,
force=True,
server_url=server_url,
api_key=api_key,
)
break
created = await agent_control.create_control(
name=POST_DENY_CONTROL_NAME,
data=build_post_deny_control_data(),
server_url=server_url,
api_key=api_key,
)
control_id = _control_id(created)
if control_id is None:
raise RuntimeError("Created control missing control_id.")
await agent_control.add_agent_control(
agent_name=AGENT_NAME,
control_id=control_id,
server_url=server_url,
api_key=api_key,
)
return {"control_id": control_id, "configured": True, "reused": False}
def main() -> None:
server_url = (
os.getenv("AGENT_CONTROL_SERVER_URL")
or os.getenv("AGENT_CONTROL_URL")
or "http://localhost:8000"
)
api_key = os.getenv("AGENT_CONTROL_API_KEY")
agent_control.init(
agent_name=AGENT_NAME,
agent_description=AGENT_DESCRIPTION,
server_url=server_url,
policy_refresh_interval_seconds=0,
)
result = asyncio.run(ensure_control(server_url=server_url, api_key=api_key))
print(result)
if __name__ == "__main__":
main()
実装：Phase 0
-
uv venvでセットアップを開始した。 -
uv pip install -e .による editable install を完了した。 - docker-compose でGalileo Agent Control を起動
services:
postgres:
image: postgres:16-alpine
container_name: agent_control_postgres
ports:
- "${AGENT_CONTROL_DB_HOST_PORT:-5432}:5432"
environment:
POSTGRES_DB: agent_control
POSTGRES_USER: agent_control
POSTGRES_PASSWORD: "${AGENT_CONTROL_POSTGRES_PASSWORD:-agent_control}"
volumes:
- pgdata:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U agent_control -d agent_control"]
interval: 5s
timeout: 5s
retries: 10
restart: unless-stopped
server:
platform: linux/amd64
image: galileoai/agent-control-server:latest
container_name: agent_control_server
ports:
- "${AGENT_CONTROL_SERVER_HOST_PORT:-8000}:8000"
environment:
AGENT_CONTROL_DB_URL: "postgresql+psycopg://agent_control:${AGENT_CONTROL_POSTGRES_PASSWORD:-agent_control}@postgres:5432/agent_control"
AGENT_CONTROL_HOST: 0.0.0.0
AGENT_CONTROL_PORT: 8000
AGENT_CONTROL_API_KEY_ENABLED: ${AGENT_CONTROL_API_KEY_ENABLED:-false}
AGENT_CONTROL_API_KEYS: ${AGENT_CONTROL_API_KEYS:-}
AGENT_CONTROL_ADMIN_API_KEYS: ${AGENT_CONTROL_ADMIN_API_KEYS:-}
AGENT_CONTROL_SESSION_SECRET: ${AGENT_CONTROL_SESSION_SECRET:-}
AGENT_CONTROL_CORS_ORIGINS: ${AGENT_CONTROL_CORS_ORIGINS:-http://localhost:4000}
depends_on:
postgres:
condition: service_healthy
restart: unless-stopped
volumes:
pgdata:
- 再現性のある実験のために deterministic な mock tool を用意した。
- 同じ入力を入れたら、毎回ほぼ同じ結果が返るようにした
- 外部APIや乱数に頼らず、動きがぶれないようにした
- あとで control の違いを比べやすくするための土台を作った
実装：Phase 1
☞「まずは制御を入れずに、旅程生成の基本動作を再現可能な形で確認する」
- build_planner_agent(with_control=False) で control なし
- search_flights_plain / assemble_itinerary_plain のような 素の deterministic mock tool を使う
- その結果、同じ入力なら毎回だいたい同じ旅程が出るかを確認する
run_phase1.py
from __future__ import annotations
import json
import sys
from typing import Any
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from src.agent import build_planner_agent
def build_prompt() -> str:
return (
"Plan a two-night trip from Tokyo to Sapporo with a total budget of 50000 JPY. "
"Include one or two simple activities and keep the output concise."
)
def run() -> dict[str, Any]:
agent = build_planner_agent(with_control=False)
response = agent(build_prompt())
print(response)
return {"prompt": build_prompt(), "response": str(response)}
if __name__ == "__main__":
result = run()
print(json.dumps(result, ensure_ascii=False, indent=2))
実行結果
$ uv run python scripts/run_phase1.py
Tool #1: search_flights_plain
Tool #2: search_hotels
Tool #3: assemble_itinerary_plain
Here’s a concise 2-night Sapporo trip within your 50,000 JPY budget:
**Trip dates:** Apr 17–19, 2026
**Flight:** Aurora Air Tokyo → Sapporo
- 07:10–09:05
- **16,666 JPY**
- Baggage included
**Hotel:** Harbor Stay
- 2 nights
- **25,000 JPY total**
- 12,500 JPY/night
**Simple activities:**
- Walk around **Odori Park**
- Visit **Sapporo Clock Tower**
- Optional: quick stop at **Nijo Market**
**Estimated total:** **41,666 JPY**
**Budget remaining:** **8,334 JPY**
If you want, I can also turn this into a day-by-day mini itinerary.Here’s a concise 2-night Sapporo trip within your 50,000 JPY budget:
**Trip dates:** Apr 17–19, 2026
**Flight:** Aurora Air Tokyo → Sapporo
- 07:10–09:05
- **16,666 JPY**
- Baggage included
**Hotel:** Harbor Stay
- 2 nights
- **25,000 JPY total**
- 12,500 JPY/night
**Simple activities:**
- Walk around **Odori Park**
- Visit **Sapporo Clock Tower**
- Optional: quick stop at **Nijo Market**
**Estimated total:** **41,666 JPY**
**Budget remaining:** **8,334 JPY**
If you want, I can also turn this into a day-by-day mini itinerary.
{
"prompt": "Plan a two-night trip from Tokyo to Sapporo with a total budget of 50000 JPY. Include one or two simple activities and keep the output concise.",
"response": "Here’s a concise 2-night Sapporo trip within your 50,000 JPY budget:\n\n**Trip dates:** Apr 17–19, 2026 \n**Flight:** Aurora Air Tokyo → Sapporo \n- 07:10–09:05 \n- **16,666 JPY** \n- Baggage included\n\n**Hotel:** Harbor Stay \n- 2 nights \n- **25,000 JPY total** \n- 12,500 JPY/night\n\n**Simple activities:**\n- Walk around **Odori Park**\n- Visit **Sapporo Clock Tower** \n- Optional: quick stop at **Nijo Market**\n\n**Estimated total:** **41,666 JPY** \n**Budget remaining:** **8,334 JPY**\n\nIf you want, I can also turn this into a day-by-day mini itinerary.\n"
}
phase1 の評価
- Agent Control なしで planner を実行する。
- Tokyo → Sapporo の最初の成功例を記録する。
- 結果: deterministic な mock tool で、41,666 円の妥当な旅程が生成された
実装: phase2
☞ 「飛行機検索に入る前に、予算的に危ないなら止める」
- search_flights を呼ぶ
- その時点で budget_max=50,000 が flight-budget-guard の条件に引っかかる
- なので flight 候補を返す前に deny する
run_phase2.py
from __future__ import annotations
import asyncio
import json
import os
import sys
from typing import Any
from pathlib import Path
import agent_control
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from policies.setup_controls import ensure_control
from src.agent import AGENT_DESCRIPTION, AGENT_NAME, build_planner_agent
from src.tools import search_flights
def build_prompt() -> str:
return (
"Plan a two-night trip from Tokyo to Sapporo with a total budget of 50000 JPY. "
"Include one or two simple activities and keep the output concise."
)
async def _prepare_controls() -> dict[str, Any]:
server_url = (
os.getenv("AGENT_CONTROL_SERVER_URL")
or os.getenv("AGENT_CONTROL_URL")
or "http://localhost:8000"
)
api_key = os.getenv("AGENT_CONTROL_API_KEY")
agent_control.init(
agent_name=AGENT_NAME,
agent_description=AGENT_DESCRIPTION,
server_url=server_url,
policy_refresh_interval_seconds=0,
)
return await ensure_control(server_url=server_url, api_key=api_key)
def _verify_flight_guard() -> dict[str, Any]:
try:
search_flights(
origin="Tokyo",
destination="Sapporo",
date="2026-05-01",
budget_max=50_000,
)
except agent_control.ControlViolationError as exc:
return {
"blocked": True,
"control_name": exc.control_name,
"message": str(exc),
}
return {"blocked": False}
def run() -> dict[str, Any]:
control_state = asyncio.run(_prepare_controls())
guard_check = _verify_flight_guard()
agent = build_planner_agent()
prompt = build_prompt()
try:
response = agent(prompt)
print(response)
return {
"prompt": prompt,
"response": str(response),
"control_state": control_state,
"guard_check": guard_check,
"expected": "deny",
}
except agent_control.ControlViolationError as exc:
print(f"ControlViolationError: {exc}")
return {
"prompt": prompt,
"error": str(exc),
"control_name": exc.control_name,
"control_state": control_state,
"guard_check": guard_check,
"expected": "deny",
}
if __name__ == "__main__":
result = run()
print(json.dumps(result, ensure_ascii=False, indent=2))
実行結果
$ uv run python scripts/run_phase2.py
Tool #1: search_flights
Tool #2: search_hotels
Tool #3: search_flights
I can’t complete the flight search because the flight tool blocks budgets in the 40,000–99,999 JPY range, and your total trip budget is 50,000 JPY.
What I could confirm:
- Hotel options in Sapporo for 2 nights fit within budget:
- Harbor Stay: 25,000 JPY total
- City Nest: 31,250 JPY total
- Quiet Lodge: 35,000 JPY total
That leaves too little for a Tokyo–Sapporo flight under your total budget, so the trip isn’t feasible as requested.
If you want, I can still help by:
- raising the total budget,
- switching to a cheaper destination,
- or planning a very low-cost Sapporo trip with a different transport assumption.I can’t complete the flight search because the flight tool blocks budgets in the 40,000–99,999 JPY range, and your total trip budget is 50,000 JPY.
What I could confirm:
- Hotel options in Sapporo for 2 nights fit within budget:
- Harbor Stay: 25,000 JPY total
- City Nest: 31,250 JPY total
- Quiet Lodge: 35,000 JPY total
That leaves too little for a Tokyo–Sapporo flight under your total budget, so the trip isn’t feasible as requested.
If you want, I can still help by:
- raising the total budget,
- switching to a cheaper destination,
- or planning a very low-cost Sapporo trip with a different transport assumption.
{
"prompt": "Plan a two-night trip from Tokyo to Sapporo with a total budget of 50000 JPY. Include one or two simple activities and keep the output concise.",
"response": "I can’t complete the flight search because the flight tool blocks budgets in the 40,000–99,999 JPY range, and your total trip budget is 50,000 JPY.\n\nWhat I could confirm:\n- Hotel options in Sapporo for 2 nights fit within budget:\n - Harbor Stay: 25,000 JPY total\n - City Nest: 31,250 JPY total\n - Quiet Lodge: 35,000 JPY total\n\nThat leaves too little for a Tokyo–Sapporo flight under your total budget, so the trip isn’t feasible as requested.\n\nIf you want, I can still help by:\n- raising the total budget,\n- switching to a cheaper destination,\n- or planning a very low-cost Sapporo trip with a different transport assumption.\n",
"control_state": {
"control_id": 1,
"configured": true,
"reused": true
},
"guard_check": {
"blocked": true,
"control_name": "flight-budget-guard",
"message": "Control violation [flight-budget-guard]: Pattern '^(?:[4-9][0-9]{4}|[1-9][0-9]{5,})$' found"
},
"expected": "deny"
}
ri@rwin-gal:~/work/20260418/gallileo/agent-control-travel-sandbox$
phase2 の評価
- flight の予算チェック用に pre-deny control を追加する。
- 結果:
flight-budget-guardはbudget_maxが 50,000 円のケースでsearch_flightsを止めた。 - その結果、agent は flight 選択を完了せず、予算に関する説明へフォールバックした。
実装: phase3
飛行機なしなのに flight が混ざった旅程を組もうとすると、assemble_itinerary を止める代わりに、flight を外してやり直すよう steer する
run_phase3.py
from __future__ import annotations
import asyncio
import json
import os
import sys
from pathlib import Path
from typing import Any
import agent_control
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from policies.setup_controls import ensure_itinerary_control
from src.agent import AGENT_DESCRIPTION, AGENT_NAME, build_planner_agent
from src.tools import assemble_itinerary, search_hotels
def build_prompt() -> str:
return (
"Plan a short trip from Tokyo to Sapporo on a 30000 JPY budget, but do "
"not use flights. Keep the output concise and recommend ground-only "
"alternatives if needed."
)
async def _prepare_controls() -> dict[str, Any]:
server_url = (
os.getenv("AGENT_CONTROL_SERVER_URL")
or os.getenv("AGENT_CONTROL_URL")
or "http://localhost:8000"
)
api_key = os.getenv("AGENT_CONTROL_API_KEY")
agent_control.init(
agent_name=AGENT_NAME,
agent_description=AGENT_DESCRIPTION,
server_url=server_url,
policy_refresh_interval_seconds=0,
)
return await ensure_itinerary_control(server_url=server_url, api_key=api_key)
def _verify_itinerary_steer() -> dict[str, Any]:
flights = [
{
"airline": "Aurora Air",
"origin": "Tokyo",
"destination": "Sapporo",
"date": "2026-05-01",
"departure": "07:10",
"arrival": "09:05",
"price": 16_666,
"baggage_included": True,
}
]
hotels = search_hotels("Sapporo", "2026-05-01", "2026-05-03", 10_000)
activities = [
{"name": "Odori Park stroll", "estimated_cost": 1_500},
{"name": "Sapporo Clock Tower visit", "estimated_cost": 1_000},
]
constraints = {
"total_budget": 30_000,
"flight_allowed": False,
"note": "Ground-only trip requested",
}
try:
assemble_itinerary(
flights=flights,
hotels=hotels,
activities=activities,
user_constraints=constraints,
)
except agent_control.ControlSteerError as exc:
return {
"steered": True,
"control_name": exc.control_name,
"message": str(exc),
}
return {"steered": False}
def run() -> dict[str, Any]:
control_state = asyncio.run(_prepare_controls())
steer_check = _verify_itinerary_steer()
agent = build_planner_agent()
prompt = build_prompt()
try:
response = agent(prompt)
print(response)
return {
"prompt": prompt,
"response": str(response),
"control_state": control_state,
"steer_check": steer_check,
"expected": "steer",
}
except agent_control.ControlSteerError as exc:
print(f"ControlSteerError: {exc}")
return {
"prompt": prompt,
"error": str(exc),
"control_name": exc.control_name,
"control_state": control_state,
"steer_check": steer_check,
"expected": "steer",
}
if __name__ == "__main__":
result = run()
print(json.dumps(result, ensure_ascii=False, indent=2))
実行結果
$ uv run python scripts/run_phase3.py
Tool #1: search_hotels
Tool #2: assemble_itinerary
Short ground-only trip from Tokyo to Sapporo, within 30,000 JPY:
- Transport: No flights. Best ground-only option is the JR route via Shinkansen + limited express, but it will likely exceed your total budget by itself.
- Hotel options in Sapporo for 2 nights:
- Harbor Stay — 15,000 JPY total
- City Nest — 18,750 JPY total
- Quiet Lodge — 21,000 JPY total
Budget note:
- Since ground transport from Tokyo to Sapporo is typically far above 30,000 JPY round trip, a true Tokyo–Sapporo trip is not feasible on this budget without flights.
- If you want to stay within 30,000 JPY, consider a ground-only alternative closer to Tokyo, or increase the budget substantially.
Recommended alternative:
- Swap Sapporo for a nearby ground-access destination like Sendai, Nagano, or Niigata for a short 2-day trip.Short ground-only trip from Tokyo to Sapporo, within 30,000 JPY:
- Transport: No flights. Best ground-only option is the JR route via Shinkansen + limited express, but it will likely exceed your total budget by itself.
- Hotel options in Sapporo for 2 nights:
- Harbor Stay — 15,000 JPY total
- City Nest — 18,750 JPY total
- Quiet Lodge — 21,000 JPY total
Budget note:
- Since ground transport from Tokyo to Sapporo is typically far above 30,000 JPY round trip, a true Tokyo–Sapporo trip is not feasible on this budget without flights.
- If you want to stay within 30,000 JPY, consider a ground-only alternative closer to Tokyo, or increase the budget substantially.
Recommended alternative:
- Swap Sapporo for a nearby ground-access destination like Sendai, Nagano, or Niigata for a short 2-day trip.
{
"prompt": "Plan a short trip from Tokyo to Sapporo on a 30000 JPY budget, but do not use flights. Keep the output concise and recommend ground-only alternatives if needed.",
"response": "Short ground-only trip from Tokyo to Sapporo, within 30,000 JPY:\n\n- Transport: No flights. Best ground-only option is the JR route via Shinkansen + limited express, but it will likely exceed your total budget by itself.\n- Hotel options in Sapporo for 2 nights:\n - Harbor Stay — 15,000 JPY total\n - City Nest — 18,750 JPY total\n - Quiet Lodge — 21,000 JPY total\n\nBudget note:\n- Since ground transport from Tokyo to Sapporo is typically far above 30,000 JPY round trip, a true Tokyo–Sapporo trip is not feasible on this budget without flights.\n- If you want to stay within 30,000 JPY, consider a ground-only alternative closer to Tokyo, or increase the budget substantially.\n\nRecommended alternative:\n- Swap Sapporo for a nearby ground-access destination like Sendai, Nagano, or Niigata for a short 2-day trip.\n",
"control_state": {
"control_id": 10,
"configured": true,
"reused": false
},
"steer_check": {
"steered": true,
"control_name": "itinerary-flight-conflict-steer",
"message": "Control steering [itinerary-flight-conflict-steer]: Condition tree matched\nSteering context: User explicitly disallowed flights, but flight options are present. Remove flights from the itinerary and retry with ground-only or no-flight alternatives."
},
"expected": "steer"
}
Phase 3 の評価
- 制約修正用に pre-steer control を追加する。
- 結果:
itinerary-flight-conflict-steerはflight_allowed=Falseなのに flight があるときにassemble_itineraryを steer した。 - 直接の guard check では、ground-only alternatives を促す
ControlSteerErrorが返った。
実装: phase4
☞「最後に出来上がった旅程を見て、within_budget=False なら止める」
- assemble_itinerary が旅程を作る
- その結果が予算を超えていたら
- post-execution で deny する
run_phase4.py
from __future__ import annotations
import asyncio
import json
import os
import sys
from pathlib import Path
from typing import Any
import agent_control
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from policies.setup_controls import ensure_post_deny_control
from src.agent import AGENT_DESCRIPTION, AGENT_NAME, build_planner_agent
from src.tools import _assemble_itinerary_post_guarded, search_flights, search_hotels
def build_prompt() -> str:
return (
"Plan a trip from Tokyo to Sapporo with a total budget of 20000 JPY. "
"Flights are allowed, but keep the final plan concise."
)
async def _prepare_controls() -> dict[str, Any]:
server_url = (
os.getenv("AGENT_CONTROL_SERVER_URL")
or os.getenv("AGENT_CONTROL_URL")
or "http://localhost:8000"
)
api_key = os.getenv("AGENT_CONTROL_API_KEY")
agent_control.init(
agent_name=AGENT_NAME,
agent_description=AGENT_DESCRIPTION,
server_url=server_url,
policy_refresh_interval_seconds=0,
)
control_state = await ensure_post_deny_control(server_url=server_url, api_key=api_key)
agent_control.init(
agent_name=AGENT_NAME,
agent_description=AGENT_DESCRIPTION,
server_url=server_url,
policy_refresh_interval_seconds=0,
)
return control_state
def _verify_post_deny() -> dict[str, Any]:
flights = search_flights("Tokyo", "Sapporo", "2026-05-01", 20_000)
hotels = search_hotels("Sapporo", "2026-05-01", "2026-05-03", 10_000)
activities = [
{"name": "Odori Park stroll", "estimated_cost": 5_000},
{"name": "Sapporo Clock Tower visit", "estimated_cost": 4_500},
]
constraints = {
"total_budget": 20_000,
"flight_allowed": True,
"note": "Flights allowed; test total budget enforcement",
}
try:
_assemble_itinerary_post_guarded(
flights=flights,
hotels=hotels,
activities=activities,
user_constraints=constraints,
)
except agent_control.ControlViolationError as exc:
return {
"blocked": True,
"control_name": exc.control_name,
"message": str(exc),
}
return {"blocked": False}
def run() -> dict[str, Any]:
control_state = asyncio.run(_prepare_controls())
deny_check = _verify_post_deny()
agent = build_planner_agent()
prompt = build_prompt()
try:
response = agent(prompt)
print(response)
return {
"prompt": prompt,
"response": str(response),
"control_state": control_state,
"deny_check": deny_check,
"expected": "deny",
}
except agent_control.ControlViolationError as exc:
print(f"ControlViolationError: {exc}")
return {
"prompt": prompt,
"error": str(exc),
"control_name": exc.control_name,
"control_state": control_state,
"deny_check": deny_check,
"expected": "deny",
}
if __name__ == "__main__":
result = run()
print(json.dumps(result, ensure_ascii=False, indent=2))
実行結果
$ uv run python scripts/run_phase4.py
Tool #1: search_flights
Tool #2: search_hotels
Here’s a concise budget plan for Tokyo → Sapporo on **2026-04-17**:
- **Flight:** Aurora Air — **6,666 JPY**
- Depart **07:10** from Tokyo, arrive **09:05** in Sapporo
- Baggage included
- **Hotel:** Harbor Stay — **6,667 JPY/night**
- **1 night**: 2026-04-17 to 2026-04-18
**Total:** **13,333 JPY**
**Budget remaining:** **6,667 JPY**
If you want, I can also suggest a slightly more comfortable option while staying under budget.Here’s a concise budget plan for Tokyo → Sapporo on **2026-04-17**:
- **Flight:** Aurora Air — **6,666 JPY**
- Depart **07:10** from Tokyo, arrive **09:05** in Sapporo
- Baggage included
- **Hotel:** Harbor Stay — **6,667 JPY/night**
- **1 night**: 2026-04-17 to 2026-04-18
**Total:** **13,333 JPY**
**Budget remaining:** **6,667 JPY**
If you want, I can also suggest a slightly more comfortable option while staying under budget.
{
"prompt": "Plan a trip from Tokyo to Sapporo with a total budget of 20000 JPY. Flights are allowed, but keep the final plan concise.",
"response": "Here’s a concise budget plan for Tokyo → Sapporo on **2026-04-17**:\n\n- **Flight:** Aurora Air — **6,666 JPY** \n - Depart **07:10** from Tokyo, arrive **09:05** in Sapporo \n - Baggage included\n\n- **Hotel:** Harbor Stay — **6,667 JPY/night** \n - **1 night**: 2026-04-17 to 2026-04-18\n\n**Total:** **13,333 JPY** \n**Budget remaining:** **6,667 JPY**\n\nIf you want, I can also suggest a slightly more comfortable option while staying under budget.\n",
"control_state": {
"control_id": 11,
"configured": true,
"reused": false
},
"deny_check": {
"blocked": true,
"control_name": "itinerary-total-budget-deny",
"message": "Control violation [itinerary-total-budget-deny]: Pattern ''within_budget': False' found"
},
"expected": "deny"
}
Phase 4 の評価
- 総予算チェック用に post-deny control を追加する。
- 結果:
itinerary-total-budget-denyは、組み立てた旅程が総予算を超えたときに post-execution で止めた。 - 直接の guard check では、
within_budgetが false であることを示すControlViolationErrorが返った。
Phase 5 の評価
- control モデルについての振り返りを記録する。
- pre-deny は、条件に合わない tool 呼び出しを 実行前に止める ための仕組みだった。
そのため、処理が business logic に入る前に止まり、権限の境界をかなり強く切る使い方だと分かった。 - pre-steer は、修正して続ければよいと伝えられるときに有効だった。
今回は、flight を外す、あるいは別の条件に直すという、実行可能な指示として表現できたときにうまく機能した。 - post-deny は、まず局所的な生成を許し、最後に全体として正しいかを確認して止めるやり方だった。
そのため、個々の判断は通しつつ、最終結果だけで全体制約を守らせたい場合に向いていると感じた。 - centralized policy plane は、どの control がいつ発火したかを一か所で見られるので観測しやすかった。 一方で、判断と正しさがその中心に集まるため、そこが信頼できることが前提になる構造でもあると分かった。
- 今回は検証を優先して、control の条件判定を regex ベースの簡易実装に寄せた。構造化データに対する意味的な判定ではなく、入力や出力の文字列表現に依存しているため、本来のポリシープレーンの理想形とは距離がある。ただし、その分、発火条件を再現しやすく、deny / steer / post-deny の違いを観察するには十分だった。
- また一方で、実行ログからは control の効き方にも差があることが見えた。
- Phase 2 では、
search_flightsが deny されたあとに LLM が再試行しており、pre-deny は単に止めるだけでなく再試行を誘発することが分かった。 - Phase 3 では、steering_context が実質的に LLM への修正文として働き、flight を外した ground-only の回答に切り替わった。
- Phase 4 では、post-deny は
assemble_itineraryを通る経路では有効だったが、LLM が tool を使わずに自力で答える経路は別に存在していた。つまり、post-deny は tool 経路の検閲としては強いが、LLM の直答経路までは捕まえない。
- Phase 2 では、
ダッシュボードUIもある
ダッシュボードからもコントロール作成できるみたい
可視化もある
感想
Agent Control は、エージェントのどこをそのまま任せず（@control が付いている関数＝介入を明示）、いつ介入するかを deny（入力前に止める） / steer （途中で直させる）/ post-deny（出力後に検閲する） で具体的に扱える点がよかった。
あとコーディングもpythonのデコレータ（@control）とかでシンプルに表現できたり
しっかりみてないけど、ダッシュボードもあるから直感的につかえそうな気がしている。
ただしそのコンパクトさゆえに？状態管理ももたず、スコープ的には個別呼び出しの pre/post 評価のみなので守備範囲は狭い。。複雑でも致命的でもない何らかの処理に、手軽に最低限のガードレール設置は置いときたいってときに使えそうだが具体的なケースは？まだ思いついてない。
Splunk との統合の線もみえており、結果 policy に「どの team のエージェントか」「どの role が書いた policy か」「どの compliance framework に属するか」といった所属や権限ベース照合みたいなのまで統合された、Splunk エンタープライズガバナンス製品になる可能性もある。
現状、先行き不透明な製品との認識で、先述のように守備範囲も狭いため、今後ガバナンスはこれを使っていこう！とおもえるような段階ではないと思う。
ひとまず今回の検証を通して、制御点（control の発火点）と委譲境界の設計を、コードと実行結果の両方から確認できた。
そのことで、「その設計判断が実際にどう見えるか」を試せたことは体験価値だったと思う。
とりあえずはエージェントガードレールにライトに入門するための勉強用と割り切って Strands＋ Galileo Agent Controlを試してみるのも良いのではないだろうか？
Discussion