4
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

「2026 Japan AWS Jr. Champions 真夏のQiitaリレー」
の4日目の投稿です!
過去の投稿は以下の(リンク集)からご覧ください!
https://qiita.com/ys-yoshida/private/6f7c7f85155a993e2c86

はじめに

⚠️ この記事はAI(Claude)と一緒に作成しました。

6月末にあったAWS SummitでAWSの方がフィジカルAIによる出展を行っていました。
ハードの知識0の自分でもその技術の凄さだけは理解できたので、
自分もこれやってみてえな!って事で0→1の活動として、
なんちゃってフィジカルAIエージェントのロボットを自作しました。

「右手だけ少し上げて」→ロボット「👋」
「ダンスして」→ロボット「🕺🕺🕺」

を全部Amazon BedrockのTool Useだけで動かしています。実際の動作動画はこちらです
(英語で話しかけている理由は後述します)。

イメージはLLMにロボットを操作するツールを与えて、命令通りになるようLLMがエージェンティックにツールの操作を行う事でロボットを動かしています。

image.png

操作方法は2種類あり、簡単な2軸ロボットをAWS Bedrock(Claude)のTool Useから自然言語で操作するテキスト版と、

image.png

Amazon Nova Sonicを使ってリアルタイム音声対話で動かせる音声版があります。

image.png

この記事は「ロボットを作りました」の紹介というより、
AIエージェントに現実世界(サーボモーター)を操作させてみて分かった
「AIに任せるべきこと・任せないこと」の記録
です。

(需要がありそうならハンズオン記事書いてみます...!)

ハードウェア・ソフトウェア構成(クリックで開く)

ハードウェア構成

  • ESP32(Dev Module)
  • PCA9685(I2C接続のPWMドライバ、16チャンネル)
  • SG90サーボ ×2(右手=チャンネル4、左手=チャンネル3)

配線はシンプルです。

PCA9685 ESP32
SDA GPIO21
SCL GPIO22
VCC 3.3V
GND GND
V+ 外部5V(SG90用)

PCA9685は安価なボードだとOE(Output Enable)ピンがフローティングになっていることがあり、その場合は出力が無効化されてサーボが全く動きません。OEをGNDに落とす必要があります。

ソフトウェア構成

arm_control.py         … サーボ制御の共通ロジック(ツール定義・ディスパッチ・システムプロンプト)
├── bedrock_servo.py       … テキストチャット版(Bedrock Converse API)
└── bedrock_servo_voice.py … リアルタイム音声版(Amazon Nova Sonic)

テキスト版と音声版でAIとの対話方式は全く異なりますが、「AIが選んだツール名と引数を受け取って、実際にサーボを動かす」というロジックは共通なのでarm_control.pyに集約し、両方から呼び出す構成にしています。

ESP32側の処理

ESP32の役割は「JSON受信 → 角度をクランプ → PWM出力」のシンプルな3ステップです。

#define ARM_CENTER 90
#define ARM_RANGE  60
#define ARM_MIN (ARM_CENTER - ARM_RANGE)
#define ARM_MAX (ARM_CENTER + ARM_RANGE)

#define CH_REVERSED  3  // 右腕は取り付けが逆向きなので反転補正

void setServoAngle(uint8_t channel, int angle) {
  angle = constrain(angle, ARM_MIN, ARM_MAX);

  if (channel == CH_REVERSED) {
    angle = ARM_CENTER - (angle - ARM_CENTER);
  }

  uint16_t pulse = map(angle, 0, 180, SERVO_MIN, SERVO_MAX);
  pca.setPWM(channel, 0, pulse);
}

void loop() {
  if (!Serial.available()) return;

  String line = Serial.readStringUntil('\n');
  line.trim();
  if (line.isEmpty()) return;

  JsonDocument doc;
  DeserializationError err = deserializeJson(doc, line);
  if (err) {
    Serial.println("{\"status\":\"error\",\"message\":\"invalid json\"}");
    return;
  }

  uint8_t channel = doc["channel"];
  int angle       = doc["angle"];
  int clamped     = constrain(angle, ARM_MIN, ARM_MAX);
  setServoAngle(channel, clamped);

  Serial.printf("{\"status\":\"ok\",\"channel\":%d,\"angle\":%d}\n", channel, clamped);
}

ポイントは2つです。

  1. 可動域を基準90度±60度(30〜150度)にクランプして、無理な動きでサーボやアーム構造を壊さないようにする
  2. 右腕(チャンネル3)は物理的に取り付けが逆向きだったため、CH_REVERSEDで基準角度を軸に角度を反転させて補正する

このおかげでPython側やAI側は「チャンネルごとの向きの違い」を意識せず、「角度が大きいほど上がる」という単純な規則だけを扱えます。

AIに操作させてみたハマったこと

ロボットの土台ができた後、実際にAIに操作させてみて色々ハマりました。
共通するのは「AIに任せていい範囲」を見誤ると簡単に崩れる、という話です。

1. AIが用意したダンス振り付けを全部踊りはじめる

danceは最初、3種類の振り付け(dance1/dance2/dance3)を全部AIに見せて、
「ランダムに1つ選んで踊って」とプロンプトで指示する作りでした。

# 最初の設計(うまくいかなかった)
TOOLS = [
    {"toolSpec": {"name": "dance1", ...}},
    {"toolSpec": {"name": "dance2", ...}},
    {"toolSpec": {"name": "dance3", ...}},
]
# システムプロンプト: "dance1・dance2・dance3の中からランダムに1つを選んで呼んでください"

結果、「ダンスして」の1回の指示で3つ全部連続で実行されるという挙動になりました。
さらに、2回目に「もう一回ダンスして」と言うと、もう踊ったのでできませんと拒否される
1回目は暴走、2回目は逆に頑固、というかなり気分屋な感じになってしまいました。

原因は、「選択」と「もう実行したかどうかの状態管理」をAIに丸ごと委ねていたことです。
AIは会話の文脈から都度もっともらしい解釈を作るので、
こちらが期待する「1回の指示につき1回、毎回ランダムに」という
一貫したルールを守ってくれるとは限りません。

対処はシンプルで、AIに見せるツールをdance1個に統一し、
どれ実行するかはPython側のrandom.choice()で決定的に選ぶように変更しました。

DANCE_VARIANTS = [dance1, dance2, dance3]

def dance(ser: serial.Serial) -> str:
    variant = random.choice(DANCE_VARIANTS)
    logger.info("dance: picked %s", variant.__name__)
    return variant(ser)

AIから見えるツールがdanceの1つだけになったので、
「1回の指示で複数回呼ばれる」余地自体がなくなります。

学び: 「選択」「乱数」「状態管理」のような決定的にやりたい処理はAIに委ねず、コード側に持たせる。AIの役割は「自然言語の意図をツール呼び出しに変換する」ところまでに絞ると安定する。

2. 腕上げろって言ってるのに下げる

「right arm up」と言っても腕が下がる、というバグもありました。
原因はシステムプロンプトに「角度の大小がどちらの方向を意味するか」を
書いていなかったことです。人間ならupなら上がるでしょと当然のように解釈しますが、
AIは文脈だけで逆方向に解釈することがあります。

def build_system_prompt() -> str:
    return (
        "あなたはロボットの腕を操作するアシスタントです。"
        ...
        "角度が大きいほど腕は上がり、角度が小さいほど腕は下がります。"
        "「上げて」と言われた場合は現在角度に20〜30度を加算し、"
        "「下げて」と言われた場合は現在角度から20〜30度を減算した絶対角度を計算して"
        "move_servoツールを呼んでください。"
        "可動域を超える場合は上限・下限に留めてください。"
        ...
    )

現在角度(current_angles)もシステムプロンプトに毎回埋め込んでいるので、
「もう少し上げて」のような相対指示も、AIが現在角度を起点に計算できます。

current_angles = {3: ARM_CENTER, 4: ARM_CENTER}

f"現在の角度: 右手(チャンネル4)={current_angles[4]}度, 左手(チャンネル3)={current_angles[3]}度。"

当たり前に思える前提でも、AIにとっては明示しないと存在しない情報です。
ツール設計は結局プロンプト設計に帰着する、というのがここでの学びです。

3. Nova Sonicでハマったこと

音声版のNova Sonicでも、変わった仕様にいくつも遭遇しました。

日本語で話しかけても英語だと認識される
調べてみたら、Nova Sonicは英語・仏・独・伊・西などのみ対応で、
日本語は非対応でした。動画で英語を使っているのはこれが理由です。
発音が下手すぎて認識が悪いのは別問題です。

image.png

toolUseイベントが来ても、まだツールを実行してはいけない
toolUseイベントはツール名の通知のみで、実行のトリガーではありません。後続のcontentEndイベント(type: "TOOL")が来て初めて実行する、という2段階の仕様でした。

elif "toolUse" in ev:
    tool_use = ev["toolUse"]
    self._pending_tool_name    = tool_use["toolName"]
    self._pending_tool_use_id  = tool_use["toolUseId"]
    self._pending_tool_input   = json.loads(tool_use["content"])
    # ここではまだ実行しない

elif "contentEnd" in ev and ev["contentEnd"].get("type") == "TOOL":
    tool_name   = self._pending_tool_name
    tool_use_id = self._pending_tool_use_id
    tool_input  = self._pending_tool_input

    result = await asyncio.to_thread(dispatch_tool, self.ser, tool_name, tool_input)
    await self.send_tool_result(tool_use_id, str(uuid.uuid4()), result)

ダンスを実行するとマイクが詰まってクラッシュする
dance内のtime.sleep()を素朴に呼んだら、
asyncioのイベントループ全体が数秒ブロックされ、
その間マイク音声の送信が滞留してValidationExceptionでクラッシュしました。
原因は同期処理がイベントループを止めていたことなので、
asyncio.to_thread()で別スレッドに逃がして解決しています。

result = await asyncio.to_thread(dispatch_tool, self.ser, tool_name, tool_input)

AIにロボットを動かさせる仕組み(Tool Use)

上記の3つの罠にどう対処したかの土台として、Tool Useの仕組みを説明します。

Bedrockには「Tool Use」という機能があり、AIに「使えるツール一覧」を渡しておくと、
会話の内容に応じて「このツールをこの引数で呼んでほしい」と
自分で判断して返してくれます。今回は次の5つのツールを用意しました。

TOOLS = [
    {"toolSpec": {"name": "move_servo", ...}},        # 基本: 絶対角度を指定
    {"toolSpec": {"name": "wave_hand", ...}},          # 手を振る(3往復・自動で元角度に戻す)
    {"toolSpec": {"name": "raise_hands", ...}},        # 両腕を限界まで上げる
    {"toolSpec": {"name": "stand_at_attention", ...}}, # 両腕を限界まで下げる
    {"toolSpec": {"name": "dance", ...}},              # ダンス
]

AIが選んだツール名と引数は次のディスパッチ関数で実際のサーボ操作にマッピングします

def dispatch_tool(ser: serial.Serial, tool_name: str, tool_input: dict) -> str:
    if tool_name == "move_servo":
        return move_servo(ser, tool_input["channel"], tool_input["angle"])
    if tool_name == "wave_hand":
        return wave_hand(ser, tool_input["channel"])
    if tool_name == "raise_hands":
        return raise_hands(ser)
    if tool_name == "stand_at_attention":
        return stand_at_attention(ser)
    if tool_name == "dance":
        return dance(ser)
    return "unknown tool"

最初はmove_servo(channel, angle)だけで十分では、と考えていましたが、
wave_hand(手を振る)のような複数ステップの動きをAI自身に
「angle=120, 60, 120, 60...を順番に呼んでください」と組み立てさせるのは不安定でした。
ステップ数や間隔がブレたり、途中で止まったりします。
そこで「意味のある動作」はツール側で完結させ、AIには「どの動作を呼ぶか」だけを
判断させる設計にしました。AIが呼ぶのは「ツール名+最小限の引数」だけで、
実際のシーケンス(何回、何ミリ秒間隔で動くか)はPython側の関数に閉じ込めています。

テキスト版の実装(bedrock_servo.py)

BedrockのConverse APIは「リクエスト→レスポンス」の単発やりとりです。
stopReasontool_useならツールを実行して結果を返し、再度呼び出す、というループになります。

def chat(client, ser, user_input: str):
    messages = [{"role": "user", "content": [{"text": user_input}]}]

    while True:
        resp = client.converse(
            modelId="us.anthropic.claude-sonnet-5",
            system=[{"text": build_system_prompt()}],
            messages=messages,
            toolConfig={"tools": TOOLS},
        )

        output      = resp["output"]["message"]
        stop_reason = resp["stopReason"]
        messages.append(output)

        if stop_reason == "tool_use":
            tool_results = []
            for block in output["content"]:
                if "toolUse" not in block:
                    continue
                tool_name   = block["toolUse"]["name"]
                tool_input  = block["toolUse"]["input"]
                tool_use_id = block["toolUse"]["toolUseId"]

                result = dispatch_tool(ser, tool_name, tool_input)

                tool_results.append({
                    "toolResult": {
                        "toolUseId": tool_use_id,
                        "content": [{"text": result}]
                    }
                })

            messages.append({"role": "user", "content": tool_results})

        else:
            # 通常のテキスト応答が返ってきたらループ終了
            break

toolResultmessagesに積んでもう一度converseを呼ぶことで、
「ツールの実行結果を踏まえてAIが最終応答を作る」流れになります。
実装としてはこれだけで、Tool Use自体の構造は素直です。

音声版の実装(Amazon Nova Sonic)

Nova SonicはInvokeModelWithBidirectionalStreamという双方向streaming APIです。
通常のリクエスト・レスポンスとは全く違うイベント駆動の構造で、
以下のようなイベントを両方向でやり取りします。

マイク音声をずっとaudioInputとして送り続けながら、
スピーカー出力(audioOutput)・文字起こし(textOutput)・ツール呼び出し(toolUse
が非同期に返ってくる、という構造です。

環境構築で地味に苦労したところ

開発機がWSL2だったため、ここも実装以上に苦労しました。
WindowsからはESP32が見えているのに、WSL2側からはUSBシリアルが見えません。
usbipd-winでパススルーする必要があり、これをUSBを挿し直すたびにやり直す運用になっています。

usbipd bind --busid 2-2
usbipd attach --wsl --busid 2-2

また音声版ではマイク・スピーカーも必要ですが、
WSL2は素の状態ではオーディオデバイスがありません。
WSLg搭載のWSLならPulseAudioソケットが使えるので、
pulseaudio-utilsを入れて解決しました。

まとめ

一連のハマりどころを振り返ると、共通する学びは1つです。

AIに任せていいのは「意図の解釈」までで、実行の詳細・選択・状態管理はコード側の責任にする

これを外すとAIはもっともらしく振る舞いながらも簡単に暴走したり頑固になったりします

Tool Useは「AIに何をやらせるか」より「AIに何をやらせないか」を決める設計だと思って触ると、うまくいきやすいです。

4
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
4
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?