언제 윙 포일링(Wing Foiling)을 하러 갈지 알려주는 AI 에이전트를 만들었습니다

문제점: 오늘 포일링을 하러 가야 할까?

저는 윙 포일링(wing foiling)에 중독되어 있습니다. 이게 무엇인지 모르신다면 — 공기 주입식 윙을 잡고 하이드로포일(hydrofoil) 보드 위에 서서 말 그대로 물 위를 날아다니는 모습을 상상해 보세요. 정말 놀랍습니다.

하지만 매일 아침 저는 windy.com, 바람을 확인하기 위한 OpenWeatherMap, 조석표(tide charts), 스웰(swells), 그리고 일반적인 날씨를 확인하기 위해 3~4개의 서로 다른 앱을 확인해야 했습니다. 그러고 나서 어떤 장비를 가져갈지 머릿속으로 계산해야 했죠. 서핑을 한다면 보드, 웨트슈트(wetsuit), 리쉬(leash) 등이 다를 수 있지만, 윙 포일링은 다뤄야 할 장비가 훨씬 더 많습니다: 프론트 포일(front foil), 마스트(mast) 크기, 윙(wing) 크기, 보드, 웨트슈트 등등 말이죠! 그래서 생각했습니다. 이 모든 것을 대신 해주고 저에게 그냥 "가라(GO)" 또는 "가지 마라(NO-GO)"라고 말해주는 AI 에이전트를 만들면 어떨까?

그래서 직접 만들었습니다. Strands Agents, MQTT, 그리고 DynamoDB를 사용하여 말이죠. 그 과정을 안내해 드리겠습니다.

아키텍처 (The Architecture)

MQTT 요청(Request) → mqtt_bridge.py → Strands Agent → 도구(Tools) (조석, 바람, 장비, 위험) → MQTT 응답(Response)

아이디어는 간단합니다. 저는 "샌프란시스코의 크리시 필드(Crissy Field)에서 윙 포일링을 해도 될까? 내 몸무게는 85kg이야." 와 같은 메시지를 MQTT 토픽에 발행(publish)하면, 에이전트가 바람 데이터, 조석 예측, 장비 추천, 그리고 위험 평가를 포함한 전체 분석 내용을 응답합니다.

왜 MQTT인가요? 궁극적으로 이것을 IoT 기기에서 실행하고 싶기 때문입니다. 여러분이 자주 가는 해변 명소에 실시간으로 가부(go/no-go) 조건을 보여주는 작은 화면이 있다고 상상해 보세요. MQTT는 가볍고, 발행/구독(pub/sub) 방식이며, 이에 완벽하게 부합합니다.

Strands Agent 설정하기

Strands Agents는 도구를 사용하는 AI 에이전트를 만드는 과정을 매우 단순하게 만들어주는 AWS의 오픈 소스 SDK입니다. @tool 데코레이터를 사용하여 Python 함수로 도구를 정의하고, 에이전트에 시스템 프롬프트(system prompt)를 제공하면, 에이전트가 언제 어떻게 도구를 호출할지 스스로 판단합니다.

제 에이전트 설정은 다음과 같습니다:

from strands import Agent
from tools import get_tide_info, get_wind_and_weather, recommend_gear, assess_risks

...

그게 전부입니다. 이 에이전트는 Amazon Bedrock를 통해 Claude를 사용하며, Strands가 모든 도구 오케스트레이션 (Tool Orchestration)을 처리합니다. 저는 "만약 바람이 X보다 크면 Y를 호출하라"와 같은 로직을 작성할 필요가 없습니다. LLM (Large Language Model)이 시스템 프롬프트 (System Prompt)와 도구 설명 (Tool Descriptions)을 바탕으로 워크플로우 (Workflow)를 스스로 파악하기 때문입니다.

도구 (The Tools)

저는 4개의 도구를 구축했으며, Strands가 사용 방법을 알 수 있도록 각각 @tool 데코레이터 (Decorator)를 적용했습니다.

1. get_wind_and_weather — OpenWeatherMap API

@tool
def get_wind_and_weather(latitude: float, longitude: float) -> str:
    """특정 위치의 현재 풍속, 풍향, 돌풍 및 날씨 정보를 가져옵니다."""
...

2. get_tide_info — Storm Glass API

@tool
def get_tide_info(latitude: float, longitude: float) -> str:
    """특정 위치의 현재 조석 상태 및 예측 정보를 가져옵니다."""
...

3. recommend_gear — 순수 로직 (Pure Logic)

이 도구는 API를 호출하지 않습니다. 윙 포일링 (Wing Foiling) 지식을 규칙으로 인코딩한 것입니다.

@tool
def recommend_gear(wind_knots: float, rider_weight_kg: float = 80.0) -> str:
    """바람과 라이더의 체중에 기반하여 윙(Wing)과 포일(Foil) 사이즈를 추천합니다."""
...

4. assess_risks — 안전 우선 (Safety First)

@tool
def assess_risks(wind_knots: float, gust_knots: float, conditions: str, tide_state: str) -> str:
    """현재 조건에 기반하여 윙 포일링 위험 요소를 평가합니다."""
...

Strands의 묘미는 제가 좋은 독스트링 (Docstring)과 타입 힌트 (Type Hint)를 사용하여 이 도구들을 정의하기만 하면, 에이전트가 도구를 "언제" 호출해야 하는지, 그리고 도구들을 "어떻게" 서로 연결해야 하는지를 스스로 안다는 점입니다. 에이전트는 먼저 바람 정보를 호출하고, 그다음 조석 정보를 호출한 뒤, 풍속 데이터를 장비 추천 도구로 전달하고, 마지막으로 이 모든 정보를 위험 평가 도구에 입력합니다. 이 모든 과정이 LLM에 의해 오케스트레이션됩니다.

MQTT 추가: IoT 브릿지 (The IoT Bridge)

저는 이것이 단순한 CLI (Command Line Interface) 챗봇이 아니라 이벤트 드리븐 (Event-driven) 방식으로 동작하기를 원했습니다. MQTT는 자연스러운 선택이었습니다.

import paho.mqtt.client as mqtt
from agent import create_agent

...

이제 저는 스마트폰 앱, 해변에 있는 Raspberry Pi, Home Assistant 자동화 등 MQTT를 지원하는 무엇이든 사용하여 어디서나 에이전트를 트리거 (Trigger)할 수 있습니다. 에이전트는 메시지를 듣고, 생각하고, 다시 발행 (Publish)합니다.

사용 방법은 다음과 같습니다:

# 에이전트에게 질문하기
mosquitto_pub -t wingfoil/request -m "Can I wing foil at Tarifa, Spain (36.0143, -5.6044)? 75kg rider."

...

DynamoDB 추가하기: API 호출 캐싱 (Caching API Calls)

여기서 제가 빠르게 맞닥뜨린 문제가 있습니다. Storm Glass API는 무료 티어(하루 10회 요청)가 매우 제한적이며, OpenWeatherMap 또한 무제한이 아닙니다. 만약 하루 동안 같은 장소에 대해 여러 번 질문한다면, 데이터가 변하지 않았음에도 API 호출을 낭비하게 됩니다.

해결책: 간단한 "오늘 가져왔는가?(fetched today?)" 확인 절차를 포함한 DynamoDB 캐시(Cache)를 사용하는 것입니다.

import boto3
import time
from datetime import date, datetime
...

캐시 키(Cache key)는 source#lat#lng 형식입니다 (인근 쿼리가 동일한 캐시에 접근할 수 있도록 소수점 둘째 자리까지 반올림합니다). 만약 데이터가 오늘 가져온 것이라면 이를 사용합니다. 그렇지 않다면 API를 새로 호출합니다. 또한 오래된 항목이 자동으로 삭제되도록 테이블에 TTL(Time To Live)을 활성화했습니다.

DynamoDB 테이블은 요청당 비용이 발생하는 온디맨드(On-demand) 방식이므로, 제 사용량 정도로는 비용이 거의 들지 않습니다. 그리고 간단한 설정 스크립트를 만들었습니다:

ddb.create_table(
    TableName="agent-external-data-cache",
    KeySchema=[{"AttributeName": "cache_key", "KeyType": "HASH"}],
...

모든 과정의 통합 (How It All Comes Together)

제가 다음과 같은 메시지를 보낼 때:

"Should I go wing foiling at Crissy Field, San Francisco (37.8035, -122.4654)? I weigh 85kg."

다음과 같은 일이 일어납니다:

1. **MQTT 브리지 (MQTT bridge)**가 메시지를 수신합니다.
2. **Strands 에이전트 (Strands agent)**가 메시지를 파싱(Parse)하여 위치와 몸무게를 추출합니다.
3. 에이전트가 **get_wind_and_weather(37.8035, -122.4654)**를 호출합니다 $\rightarrow$ DynamoDB 캐시를 확인합니다 $\rightarrow$ 필요 시 OpenWeatherMap을 호출합니다.
4. 에이전트가 **get_tide_info(37.8035, -122.4654)**를 호출합니다 $\rightarrow$ 캐시를 확인합니다 $\rightarrow$ 필요 시 Storm Glass를 호출합니다.
5. 에이전트가 실제로 얻은 풍속과 함께 **recommend_gear(18, 85)**를 호출합니다.
6. 에이전트가 모든 데이터를 사용하여 **assess_risks(18, 24, "partly cloudy", "incoming")**를 호출합니다.
7. 에이전트가 모든 정보를 종합하여 **Go/No-Go 권장 사항 (Go/No-Go recommendation)**을 도출합니다.
8. 응답이 다시 MQTT로 발행(Publish)됩니다.

전체 과정은 약 5~10초가 소요되며, 다음과 같은 응답을 받게 됩니다:

GO ✅ — 남서풍(SW)이 18노트(knots)로 일정하게 불고 있으며, 중조(mid-tide)가 들어오고 있어 완벽한 조건입니다. 5m 윙(wing), 미디엄 포일(medium foil, 1400cm²), 5'2" 보드를 사용하세요. 다리 근처에서 24노트까지 치솟는 돌풍(gusts)을 주의하세요. 최적의 시간대: 만조(tide peaks) 전 향후 3시간 이내.

내가 배운 점

Strands는 도구 사용(tool-use) 에이전트 구현을 매우 쉽게 만듭니다. 진심으로, 가장 어려운 부분은 에이전트 오케스트레이션(orchestration)이 아니라 실제 도구 로직(tool logic)을 작성하는 것이었습니다. @tool 데코레이터와 잘 작성된 독스트링(docstrings)만 있으면 충분합니다.

MQTT + AI 에이전트는 강력한 조합입니다. 이는 "두뇌"와 "인터페이스"를 분리(decouple)해 줍니다. 모바일 앱, e-ink 디스플레이, 음성 비서 등 어떤 프론트엔드(frontend)든 구축할 수 있으며, 이들은 모두 동일한 토픽(topics)에 발행/구독(pub/sub)하기만 하면 됩니다.

캐시(cache)로 DynamoDB를 사용하는 것은 과하지만 편리합니다. Redis나 로컬 JSON 파일을 사용할 수도 있었습니다. 하지만 DynamoDB는 서버리스(serverless)이고 항상 켜져 있으며, 저는 이미 Bedrock과 함께 AWS 생태계 안에 있습니다. 게다가 TTL(Time To Live)이 무료로 데이터 정리(cleanup)를 처리해 줍니다.

모의 데이터(Mock data) 폴백(fallbacks)은 필수적입니다. 두 API 모두 속도 제한(rate limits)이 있습니다. 모의 데이터를 갖추고 있으면 API 호출을 소모하지 않고도 개발 및 데모 중에 에이전트가 계속 작동할 수 있습니다.

다음 단계

• 다중 지점 비교 (Multi-spot comparison) — "오늘 어디로 포일링하러 갈까: Crissy Field, 3rd Ave, 아니면 Alameda?"
• 예보 모드 (Forecast mode) — 향후 3일간의 조건을 확인하고 최적의 시간대를 제안
• Raspberry Pi 배포 (Raspberry Pi deployment) — e-ink 디스플레이를 장착하여 내가 자주 가는 장소에 전용 장치 설치
• 알림 시스템 (Alert system) — "내가 좋아하는 장소의 조건이 좋아지면 알려줘"

직접 시도해 보세요

이 코드는 날씨 조건에 의존하는 카이트보딩(kiteboarding), 서핑(surfing), 세일링(sailing), 패러글라이딩(paragliding) 등 어떤 야외 스포츠에도 쉽게 맞춤 조정할 수 있습니다. 도구(tools)와 시스템 프롬프트(system prompt)만 교체하면 됩니다.

필요한 사항:

• Python 3.10+
• AWS 자격 증명 (Bedrock + DynamoDB용)
• MQTT 브로커 (로컬에서는 Mosquitto가 잘 작동합니다)
• OpenWeatherMap 및 Storm Glass API 키 (선택 사항 — 모의 데이터가 있으면 없이도 작동합니다)

pip install strands-agents paho-mqtt boto3 requests

만약 당신이 윙 포일링 (Wing Foiling)을 좋아한다면 (혹은 좋아하고 싶다면), 저에게 연락해 주세요. 그리고 당신의 스포츠를 위해 이와 유사한 무언가를 만든다면, 꼭 보고 싶습니다! 🤙