센서 이벤트에 대한 자동 응답
온도 > 임계값 → 경고 → 작동의 에이전트 기반 IoT 제어 루프를 구축합니다.
센서 이벤트에 대한 자동 응답은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 3번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Agents 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.
이벤트 기반 자동 Agent 응답
센서가 임계값을 넘으면 Agent는 사람의 개입 없이 자동으로 응답해야 합니다. 핵심 과제는 어떤 동작을 수행할지 결정하고, 같은 이벤트가 중복 동작을 발생시키지 않도록 하며, Agent가 작동기에 명령을 쏟아 내지 않도록 쿨다운 기간을 준수하는 것입니다.
동작 정책 정의
동작 정책은 센서 조건을 Agent의 응답에 연결합니다. 로직 코드를 직접 수정하지 않아도 쉽게 읽고 변경할 수 있도록 정책을 선언적으로 정의하십시오. 각 정책에는 조건, 우선순위 및 하나 이상의 동작이 있습니다.
ACTION_POLICIES = [
{
'name': 'HIGH_TEMP_ALERT',
'topic': 'sensors/temperature',
'condition': lambda v: v > 38,
'priority': 'critical',
'actions': ['TURN_ON_COOLING', 'ALERT_MAINTENANCE', 'LOG_EVENT']
},
{
'name': 'HIGH_TEMP_WARNING',
'topic': 'sensors/temperature',
'condition': lambda v: 35 < v <= 38,
'priority': 'warning',
'actions': ['ALERT_MAINTENANCE', 'LOG_EVENT']
},
{
'name': 'LOW_HUMIDITY',
'topic': 'sensors/humidity',
'condition': lambda v: v < 30,
'priority': 'warning',
'actions': ['TURN_ON_HUMIDIFIER', 'LOG_EVENT']
}
]
def match_policies(topic: str, value: float) -> list:
return [
p for p in ACTION_POLICIES
if p['topic'] == topic and p['condition'](value)
]
if __name__ == '__main__':
matches = match_policies('sensors/temperature', 39)
print('Matched policies for temperature=39:')
for p in matches:
print(f" {p['name']} ({p['priority']}): {p['actions']}")
동작 큐
동작 큐는 이벤트 감지와 동작 실행을 분리합니다. 이벤트는 큐에 추가되고 작업자가 꺼내 실행합니다. 이를 통해 MQTT 수신 루프가 차단되는 것을 방지하고 동작이 실패했을 때 재시도할 수 있습니다.
import queue
import threading
from datetime import datetime
action_queue: queue.Queue = queue.Queue(maxsize=500)
def enqueue_action(action_name: str, context: dict, priority: str = 'normal'):
item = {
'action': action_name,
'context': context,
'priority': priority,
'enqueued_at': datetime.utcnow().isoformat()
}
try:
action_queue.put_nowait(item)
print(f'Enqueued: {action_name}')
except queue.Full:
print(f'WARNING: Action queue full, dropping {action_name}')
def action_worker(executor_fn):
"""Run in a background thread, executing actions from the queue."""
while True:
item = action_queue.get()
try:
executor_fn(item['action'], item['context'])
except Exception as e:
print(f'Action failed: {item["action"]} — {e}')
finally:
action_queue.task_done()
# Start worker thread:
# worker_thread = threading.Thread(target=action_worker, args=(execute_action,), daemon=True)
# worker_thread.start()
if __name__ == '__main__':
enqueue_action('TURN_ON_COOLING', {'zone': 'server-room'}, priority='critical')
enqueue_action('LOG_EVENT', {'msg': 'temperature nominal'})
print('Queue size:', action_queue.qsize())
이벤트 중복 제거
중복 제거가 없으면 1초에 1번 판독하는 환경에서 38°C를 10분 동안 초과한 온도가 동일한 이벤트 600개를 생성합니다. 중복 제거를 사용하면 같은 (토픽, 조건, 동작) 조합이 이벤트마다 한 번만 실행되고, 조건이 해제되면 다시 실행할 수 있도록 상태가 초기화됩니다.
class EventDeduplicator:
def __init__(self):
# active_events: (topic, policy_name) -> event_start_time
self._active: dict = {}
def is_new_event(self, topic: str, policy_name: str) -> bool:
key = (topic, policy_name)
return key not in self._active
def mark_active(self, topic: str, policy_name: str):
self._active[(topic, policy_name)] = datetime.utcnow()
def clear_event(self, topic: str, policy_name: str):
key = (topic, policy_name)
if key in self._active:
duration = (datetime.utcnow() - self._active.pop(key)).seconds
print(f'Event cleared: {policy_name} (lasted {duration}s)')
def clear_topic_if_normal(
self, topic: str, value: float, normal_fn
):
if normal_fn(value):
keys = [k for k in self._active if k[0] == topic]
for k in keys:
self.clear_event(k[0], k[1])
dedup = EventDeduplicator()
dedup.mark_active('sensors/temperature', 'HIGH_TEMP_ALERT')
print('New event?', dedup.is_new_event('sensors/temperature', 'HIGH_TEMP_ALERT'))쿨다운 기간
이벤트가 해제된 후 다시 발생하더라도 쿨다운 기간을 두면 빠른 재실행을 방지할 수 있습니다. 중복 제거(조건이 유지되는 동안 한 번만 실행)와 쿨다운(조건이 해제된 후 같은 경보를 다시 허용하기 전에 N분 대기)을 함께 사용하십시오.
from datetime import datetime, timedelta
class CoolDownManager:
def __init__(self, cool_down_minutes: int = 15):
self.cool_down = timedelta(minutes=cool_down_minutes)
self._cleared_at: dict = {} # (topic, policy) -> cleared_datetime
def is_in_cool_down(self, topic: str, policy_name: str) -> bool:
key = (topic, policy_name)
cleared_at = self._cleared_at.get(key)
if cleared_at is None:
return False
return datetime.utcnow() - cleared_at < self.cool_down
def record_clear(self, topic: str, policy_name: str):
self._cleared_at[(topic, policy_name)] = datetime.utcnow()
def time_remaining(self, topic: str, policy_name: str) -> int:
key = (topic, policy_name)
cleared_at = self._cleared_at.get(key)
if cleared_at is None:
return 0
elapsed = datetime.utcnow() - cleared_at
remaining = self.cool_down - elapsed
return max(0, int(remaining.total_seconds()))
cooldown = CoolDownManager(cool_down_minutes=15)
cooldown.record_clear('sensors/temperature', 'HIGH_TEMP_ALERT')
print('In cool-down?', cooldown.is_in_cool_down('sensors/temperature', 'HIGH_TEMP_ALERT'))LLM 지원 동작 결정
동시에 여러 경보가 발생하거나, 정책이 충돌하거나, 판독값이 특이하게 조합되는 복잡한 상황에서는 결정을 LLM에 위임하십시오. LLM은 전체 센서 맥락을 받고 우선순위가 정해진 동작 계획을 추천합니다.
import anthropic
import json
def llm_decide_actions(
sensor_readings: dict,
active_policies: list
) -> list:
client = anthropic.Anthropic(api_key='YOUR_API_KEY')
context = json.dumps({
'readings': sensor_readings,
'triggered_policies': [p['name'] for p in active_policies]
}, indent=2)
prompt = (
f'Current sensor state:\n{context}\n\n'
'Multiple alert policies are active. '
'Recommend an ordered list of actions to take. '
'Consider conflicting effects (e.g., humidifier and cooling may conflict).\n'
'Return JSON: {"recommended_actions": [str], "reasoning": str}'
)
response = client.messages.create(
model='claude-opus-4-5', max_tokens=512,
messages=[{'role': 'user', 'content': prompt}]
)
return json.loads(response.content[0].text)MQTT를 통한 동작 실행
동작은 장치별 MQTT 토픽에 명령 메시지를 발행하여 실행합니다. 명령 페이로드는 표준 스키마를 따릅니다. 동작 이름, 매개변수, 확인 응답을 위한 요청 ID, 그리고 TTL(장치가 너무 오래 오프라인이면 명령이 만료됨)이 포함됩니다.
import json
import uuid
from datetime import datetime, timedelta
ACTION_TOPICS = {
'TURN_ON_COOLING': 'devices/hvac/commands',
'TURN_OFF_COOLING': 'devices/hvac/commands',
'TURN_ON_HUMIDIFIER': 'devices/humidifier/commands',
'ALERT_MAINTENANCE': 'notifications/maintenance',
'LOG_EVENT': 'logs/agent_events'
}
def execute_action(action_name: str, context: dict, mqtt_client) -> str:
topic = ACTION_TOPICS.get(action_name)
if not topic:
print(f'No topic defined for action: {action_name}')
return 'unknown_action'
request_id = str(uuid.uuid4())[:8]
ttl = (datetime.utcnow() + timedelta(minutes=5)).isoformat()
payload = json.dumps({
'action': action_name,
'request_id': request_id,
'context': context,
'ttl': ttl
})
mqtt_client.publish(topic, payload, qos=1)
print(f'Executed {action_name} -> {topic} (req={request_id})')
return request_id
if __name__ == '__main__':
class FakeMQTT:
def publish(self, topic, payload, qos=1):
pass
execute_action('TURN_ON_COOLING', {'zone': 'server-room'}, FakeMQTT())
동작 확인 응답
장치는 확인 응답 토픽에 발행하여 수신한 명령을 확인해야 합니다. Agent는 확인 응답 토픽을 구독하고, 제한 시간 내에 확인 응답이 수신되지 않으면 재시도할 수 있습니다.
import threading
from collections import defaultdict
class AckTracker:
def __init__(self, timeout_seconds: int = 30):
self.timeout = timeout_seconds
self._pending: dict = {} # request_id -> {'action', 'send_time', 'ack_event'}
def register(self, request_id: str, action_name: str):
event = threading.Event()
self._pending[request_id] = {
'action': action_name,
'send_time': datetime.utcnow(),
'ack_event': event
}
# Schedule timeout check
t = threading.Timer(self.timeout, self._on_timeout, args=[request_id])
t.daemon = True
t.start()
def acknowledge(self, request_id: str):
entry = self._pending.pop(request_id, None)
if entry:
entry['ack_event'].set()
print(f'Ack received for {entry["action"]} (req={request_id})')
def _on_timeout(self, request_id: str):
if request_id in self._pending:
action = self._pending.pop(request_id)['action']
print(f'TIMEOUT: No ack for {action} (req={request_id}) — retry?')전체 센서 이벤트 처리 흐름
모든 구성 요소를 결합하면 다음과 같습니다. MQTT 수신 → 정책 일치 → 중복 제거 및 쿨다운 확인 → 동작 큐에 추가 → 작업자가 MQTT 발행을 통해 실행 → 확인 응답 추적. 이 아키텍처는 차단 없이 분당 수천 개의 센서 이벤트를 처리합니다.
class IoTAgentPipeline:
def __init__(self, mqtt_client):
self.mqtt = mqtt_client
self.dedup = EventDeduplicator()
self.cooldown = CoolDownManager(cool_down_minutes=15)
self.ack_tracker = AckTracker(timeout_seconds=30)
def on_sensor_message(self, topic: str, value: float):
policies = match_policies(topic, value)
self.dedup.clear_topic_if_normal(
topic, value,
normal_fn=lambda v: v <= 35 # below warning threshold
)
for policy in policies:
name = policy['name']
if not self.dedup.is_new_event(topic, name):
continue # already active, skip
if self.cooldown.is_in_cool_down(topic, name):
print(f'In cool-down: {name}')
continue
self.dedup.mark_active(topic, name)
ctx = {'topic': topic, 'value': value, 'policy': name}
for action in policy['actions']:
enqueue_action(action, ctx, policy['priority'])에스컬레이션 경로
명령 확인 응답의 반복적인 실패, 지속되는 중요 상태 또는 서로 충돌하는 여러 정책과 같이 사람의 개입이 필요한 상황도 있습니다. 푸시 알림을 보내거나 티켓을 생성하는 에스컬레이션 경로를 정의하십시오.
import requests
def escalate_to_human(
reason: str,
sensor_data: dict,
webhook_url: str = 'https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK'
):
message = {
'text': (
f'*IoT Agent Escalation* \n'
f'Reason: {reason}\n'
f'Sensor data: {sensor_data}\n'
f'Time: {datetime.utcnow().isoformat()}'
)
}
try:
response = requests.post(webhook_url, json=message, timeout=5)
response.raise_for_status()
print(f'Escalation sent: {reason}')
except requests.RequestException as e:
print(f'Escalation failed: {e}')
# Fall back: log to file
with open('escalations.log', 'a') as f:
import json
f.write(json.dumps({'reason': reason, 'data': sensor_data}) + '\n')이벤트 처리 흐름 테스트
이벤트 처리 흐름을 운영 환경에 배포하기 전에 센서 이벤트를 시뮬레이션하고 올바른 동작이 큐에 추가되는지 검증하는 자동화된 테스트를 작성하십시오. 각 정책, 중복 제거 동작 및 쿨다운 만료를 개별적으로 테스트하십시오.
import time
def test_high_temp_policy_fires_once():
dedup = EventDeduplicator()
cooldown = CoolDownManager(cool_down_minutes=0) # disable cooldown for test
pipeline = IoTAgentPipeline(None)
pipeline.dedup = dedup
pipeline.cooldown = cooldown
actions_fired = []
action_queue.queue.clear()
# Fire same event 5 times in a row
for _ in range(5):
pipeline.on_sensor_message('sensors/temperature', 40.0)
# Only 1 set of actions should have been enqueued
actions = list(action_queue.queue)
assert len(actions) > 0, 'At least one action should fire'
print(f'Actions enqueued: {len(actions)} (expected: just 1 event worth)')
return True
result = test_high_temp_policy_fires_once()
print('Test passed:', result)지식 확인
센서 이벤트 처리 흐름에서 이벤트 중복 제거의 주된 목적은 무엇입니까?
복습: 센서 이벤트에 대한 자동 응답
훌륭합니다! 학습한 내용은 다음과 같습니다.
- 동작 정책: 우선순위가 포함된 선언적 조건-동작 매핑
- 동작 큐: 감지와 실행을 분리하며 작업자 스레드가 동작을 처리함
- 중복 제거: 판독값마다가 아니라 이벤트마다 한 번만 실행함
- 쿨다운: 조건이 해제된 직후 다시 실행되는 것을 방지함
- LLM 에스컬레이션: 여러 정책이 관련된 복잡한 상황을 LLM의 추론에 위임함
- 확인 응답 추적: 확인 응답이 없는 명령을 감지하고 재시도함
다음 주제: 라즈베리 파이와 같은 엣지 장치에 경량 Agent 배포하기
자주 묻는 질문
“센서 이벤트에 대한 자동 응답” 강의는 무료인가요?
네 — “센서 이벤트에 대한 자동 응답” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.
“센서 이벤트에 대한 자동 응답”에서 뭘 배우나요?
온도 > 임계값 → 경고 → 작동의 에이전트 기반 IoT 제어 루프를 구축합니다. 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.
AI Agents을(를) 시작하는 데 경험이 필요한가요?
사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 3번째 강의입니다.
“센서 이벤트에 대한 자동 응답” 강의는 얼마나 걸리나요?
대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.
이 AI Agents 강의에서 코드를 작성하고 실행할 수 있나요?
네. 모든 AI Agents 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.
이 강의의 모든 강의
- 에이전트 통합을 위한 MQTT 프로토콜
- 에이전트의 시계열 데이터 처리
- 센서 이벤트에 대한 자동 응답
- 경량 에이전트의 엣지 배포