Strands Agents の GoalLoop プラグインが追加されていたので触ってみた

uv init --bare --name goal-strands
uv add strands-agents

from strands import Agent
from strands.models.bedrock import BedrockModel
from strands.vended_plugins.goal import GoalLoop

model = BedrockModel(
    model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0",
    region_name="us-east-1",
)

concise = GoalLoop(
    goal="3文以内で、小学生にもわかる言葉で説明すること。専門用語は使わない。",
    max_attempts=3,
)

agent = Agent(model=model, plugins=[concise])
result = agent("虹はどうしてできるの？")

goal_result = concise.last_result(agent)
print("\n--- GoalResult ---")
print(f"passed: {goal_result.passed}")
print(f"stop_reason: {goal_result.stop_reason}")
print(f"attempts: {len(goal_result.attempts)}")
for attempt in goal_result.attempts:
    print(f"  #{attempt.attempt}: feedback={attempt.feedback}")

uv run python nl_goal.py

--- GoalResult ---
passed: True
stop_reason: satisfied
attempts: 2
  #1: feedback=ゴールの3つの要件すべてが満たされていません。(1)回答は3文を大幅に超えており、複数のセクション、箇条書き、詳細説明を含んでいます。(2)「屈折」「反射」「波長」「グラデーション」など多数の専門用語が使用されており、小学生には理解困難です。要件を満たすには、3文以内で、「光が雨の水に当たってはね返ったり曲がったりして、色が分かれて見える」のような非常にシンプルな説明に変更してください。
  #2: feedback=None

太陽の光が雨の水滴に当たるとき、光がはね返ったり曲がったりします。
このとき、光に含まれている色がそれぞれ違う角度ではね返るので、
赤・黄・青などの色が分かれて見えるのです。
だから雨の日に太陽を背にして見ると、虹が見えるんですよ。

from strands import Agent
from strands.models.bedrock import BedrockModel
from strands.vended_plugins.goal import GoalLoop

model = BedrockModel(
    model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0",
    region_name="us-east-1",
)

def char_count_validator(response, agent):
    text = "".join(
        block["text"] for block in response["content"] if "text" in block
    )
    chars = len(text)
    if chars <= 100:
        return True
    return {
        "passed": False,
        "feedback": f"回答が長すぎます（{chars}文字）。100文字以内に収めてください。",
    }

plugin = GoalLoop(
    goal=char_count_validator,
    max_attempts=5,
    timeout=60.0,
)

agent = Agent(model=model, plugins=[plugin])
result = agent("生成AIエージェントとは何か説明してください。")

goal_result = plugin.last_result(agent)
print("\n--- GoalResult ---")
print(f"passed: {goal_result.passed}")
print(f"stop_reason: {goal_result.stop_reason}")
print(f"attempts: {len(goal_result.attempts)}")
for attempt in goal_result.attempts:
    print(f"  #{attempt.attempt}: feedback={attempt.feedback}")

uv run python validator_goal.py

--- GoalResult ---
passed: True
stop_reason: satisfied
attempts: 3
  #1: feedback=回答が長すぎます（576文字）。100文字以内に収めてください。
  #2: feedback=回答が長すぎます（131文字）。100文字以内に収めてください。
  #3: feedback=None

生成AIエージェントとは

自律的に複数のツールを組み合わせ、タスクを自動実行するAIシステムです。
ユーザーの指示に対し、自ら判断・計画して目標達成まで動作します。

from strands import Agent
from strands.models.bedrock import BedrockModel
from strands.vended_plugins.goal import GoalLoop

model = BedrockModel(
    model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0",
    region_name="us-east-1",
)

def impossible_validator(response, agent):
    return {
        "passed": False,
        "feedback": "この検証は絶対に合格しません（動作確認用）。",
    }

plugin = GoalLoop(goal=impossible_validator, max_attempts=2)

agent = Agent(model=model, plugins=[plugin])
result = agent("こんにちは！")

goal_result = plugin.last_result(agent)
print("\n--- GoalResult ---")
print(f"passed: {goal_result.passed}")
print(f"stop_reason: {goal_result.stop_reason}")
print(f"attempts: {len(goal_result.attempts)}")

--- GoalResult ---
passed: False
stop_reason: max_attempts
attempts: 2

=== preserve_context=True ===
attempts: 3
最終的な会話履歴のメッセージ数: 6
会話履歴の合計文字数: 1393

=== preserve_context=False ===
attempts: 3
最終的な会話履歴のメッセージ数: 3
会話履歴の合計文字数: 402

[0] user: 生成AIエージェントとは何か説明してください。
[1] user: Your previous attempt did not satisfy the goal. ...（直前の試行へのフィードバック）
[2] assistant: （合格した回答）

agent = Agent(
    model=model,
    plugins=[plugin],
    conversation_manager=SlidingWindowConversationManager(window_size=4),
)

最終的な会話履歴のメッセージ数: 4
[0] user: Your previous attempt did not satisfy the goal. ...（1回目へのフィードバック）
[1] assistant: （2回目の失敗した回答）
[2] user: Your previous attempt did not satisfy the goal. ...（2回目へのフィードバック）
[3] assistant: （合格した回答）

async def main():
    async for event in agent.stream_async("生成AIエージェントとは何か説明してください。"):
        if "data" in event:
            print(event["data"], end="", flush=True)

# 生成AIエージェントとは
## 基本的な定義
生成AIエージェントは、生成AI（LLM等）を中核に...
（↑ 1回目の長い回答が最後まで流れてくる）
# 生成AIエージェントとは
生成AI（LLM）を搭載し、ユーザーの指示に対して...
（↑ 区切りなくそのまま2回目の回答が流れてくる）
# 生成AIエージェントとは
生成AIを搭載し、ユーザーの指示に対して自律的に...
（↑ さらに続けて3回目の合格した回答が流れてくる）

init_event_loop    ← 試行1の開始
（data イベントが流れる）
message(assistant) ← 試行1の応答が確定
init_event_loop    ← 試行2の開始
（data イベントが流れる）
message(assistant) ← 試行2の応答が確定
init_event_loop    ← 試行3の開始
（data イベントが流れる）
message(assistant) ← 試行3の応答が確定
result             ← 全体の完了

async def main():
    buffer = ""
    attempt = 0
    async for event in agent.stream_async("生成AIエージェントとは何か説明してください。"):
        if "init_event_loop" in event:
            if buffer:
                print(f"（試行 {attempt} は不合格のため破棄。リトライします）")
            attempt += 1
            buffer = ""
        if "data" in event:
            buffer += event["data"]

    print(f"\n=== 合格した試行 {attempt} の回答だけを表示 ===")
    print(buffer)

（試行 1 は不合格のため破棄。リトライします）
（試行 2 は不合格のため破棄。リトライします）

=== 合格した試行 3 の回答だけを表示 ===
# 生成AIエージェントとは

自律的に判断・行動するAIシステム。ユーザーの指示を受けて、自ら計画を立てて複数のツールを組み合わせ、主体的に目標を達成します。

from strands.vended_plugins.goal import GoalLoop, JudgeConfig

plugin = GoalLoop(
    goal="出典を2つ以上引用すること。",
    max_attempts=3,
    judge=JudgeConfig(
        model=BedrockModel(model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0"),
    ),
)

from strands.vended_plugins.goal import (
    GoalLoop,
    build_judge_prompt,
    JUDGE_SYSTEM_PROMPT,
    JudgeOutcome,
)

async def custom_judge(response, agent):
    from strands import Agent as JudgeAgent
    from strands.models.bedrock import BedrockModel

    judge = JudgeAgent(
        model=BedrockModel(model_id="us.amazon.nova-lite-v1:0"),
        callback_handler=None,
        system_prompt=JUDGE_SYSTEM_PROMPT,
        structured_output_model=JudgeOutcome,
    )
    prompt = build_judge_prompt("Be concise.", agent.messages)
    result = await judge.invoke_async(prompt)
    outcome = result.structured_output
    return {"passed": outcome.passed, "feedback": outcome.feedback}

plugin = GoalLoop(goal=custom_judge, max_attempts=3)