PricingLoginStart free
Ejen AI · Pelajaran

Respons Automatik terhadap Peristiwa Sensor

Jika suhu > ambang → amaran → gerakkan: gelung kawalan IoT dipacu ejen.

Pelajaran 3 daripada 413 langkah

Respons Automatik terhadap Peristiwa Sensor ialah pelajaran Ejen AI percuma di CoddyKit. Ini ialah pelajaran 3 daripada 4. Anda boleh membaca keseluruhan pelajaran di bawah secara percuma — kemudian berlatih secara praktikal dalam pelayar menggunakan penyunting kod terbina dalam dan tutor kecerdasan buatan 24/7. Pelajaran ini merupakan sebahagian daripada laluan pembelajaran Ejen AI, dan kemajuan anda disegerakkan merentas web serta aplikasi CoddyKit. Kursus Ejen AI merangkumi sejumlah 4 pelajaran.

Tindak Balas Ejen Dipacu Peristiwa Automatik

Apabila sensor melepasi ambang, ejen mesti bertindak secara automatik tanpa campur tangan manusia. Cabaran utama ialah menentukan tindakan yang perlu diambil, memastikan peristiwa yang sama tidak mencetuskan tindakan pendua dan mematuhi tempoh bertenang supaya ejen tidak membanjiri penggerak dengan arahan.

Menentukan Polisi Tindakan

Polisi tindakan memetakan keadaan sensor kepada tindak balas ejen. Tentukan polisi secara deklaratif supaya polisi mudah dibaca dan diubah suai tanpa menyentuh kod logik. Setiap polisi mempunyai syarat, keutamaan dan satu atau lebih tindakan.

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']}")

Baris Gilir Tindakan

Baris gilir tindakan memisahkan pengesanan peristiwa daripada pelaksanaan tindakan. Peristiwa dimasukkan ke dalam baris gilir; pekerja mengeluarkan dan melaksanakannya. Ini menghalang gelung penerimaan MQTT daripada tersekat dan membolehkan percubaan semula jika tindakan gagal.

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())

Penyahduplikatan Peristiwa

Tanpa penyahduplikatan, suhu yang kekal melebihi 38°C selama 10 minit pada kadar 1 bacaan/saat menghasilkan 600 peristiwa yang serupa. Penyahduplikatan memastikan gabungan (topik, syarat, tindakan) yang sama hanya dicetuskan sekali bagi setiap peristiwa dan ditetapkan semula apabila syarat tidak lagi dipenuhi.

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'))

Tempoh Bertenang

Walaupun selepas sesuatu peristiwa berhenti dan tercetus semula, tempoh bertenang menghalang pencetusan semula yang terlalu pantas. Gabungkan penyahduplikatan (dicetuskan sekali selagi syarat dipenuhi) dengan tempoh bertenang (tunggu N minit selepas syarat tidak lagi dipenuhi sebelum membenarkan amaran yang sama dicetuskan semula).

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'))

Keputusan Tindakan Berbantukan LLM

Untuk situasi kompleks — berbilang amaran serentak, polisi yang bercanggah atau gabungan bacaan yang luar biasa — serahkan keputusan kepada LLM. LLM menerima konteks sensor lengkap dan mencadangkan pelan tindakan mengikut keutamaan.

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)

Melaksanakan Tindakan melalui MQTT

Tindakan dilaksanakan dengan menerbitkan mesej arahan kepada topik MQTT khusus peranti. Muatan arahan mengikut skema standard: nama tindakan, parameter, ID permintaan untuk pengesahan dan TTL (arahan tamat tempoh jika peranti berada di luar talian terlalu lama).

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())

Pengesahan Tindakan

Peranti hendaklah mengesahkan arahan yang diterima dengan menerbitkan mesej kepada topik pengesahan. Ejen melanggan topik pengesahan dan boleh mencuba semula jika tiada pengesahan diterima dalam tempoh tamat masa.

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?')

Saluran Paip Peristiwa Sensor Lengkap

Menggabungkan semua komponen: penerimaan MQTT → padanan polisi → penyahduplikatan + pemeriksaan tempoh bertenang → masukkan tindakan ke baris gilir → pekerja melaksanakan melalui penerbitan MQTT → penjejakan pengesahan. Seni bina ini mengendalikan ribuan peristiwa sensor seminit tanpa tersekat.

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'])

Laluan Eskalasi

Sesetengah situasi memerlukan eskalasi kepada manusia: kegagalan berulang untuk mengesahkan arahan, keadaan kritikal yang berterusan atau berbilang polisi yang bercanggah. Tentukan laluan eskalasi yang menghantar pemberitahuan tolak atau mencipta tiket.

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')

Menguji Saluran Paip Peristiwa

Sebelum menggunakan saluran paip peristiwa Anda dalam persekitaran pengeluaran, tulis ujian automatik yang mensimulasikan peristiwa sensor dan mengesahkan bahawa tindakan yang betul dimasukkan ke baris gilir. Uji setiap polisi secara berasingan, tingkah laku penyahduplikatan dan tamat tempoh bertenang.

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)

Semakan Pengetahuan

Apakah tujuan utama penyahduplikatan peristiwa dalam saluran paip peristiwa sensor?

Jawab dalam pelajaran interaktif

Imbas Kembali: Tindak Balas Automatik terhadap Peristiwa Sensor

Cemerlang! Inilah perkara yang Anda pelajari:

  • Polisi tindakan: pemetaan deklaratif syarat kepada tindakan dengan keutamaan
  • Baris gilir tindakan: memisahkan pengesanan daripada pelaksanaan; utas pekerja memproses tindakan
  • Penyahduplikatan: dicetuskan sekali bagi setiap peristiwa, bukan sekali bagi setiap bacaan
  • Tempoh bertenang: menghalang pencetusan semula serta-merta selepas syarat tidak lagi dipenuhi
  • Eskalasi LLM: situasi kompleks dengan berbilang polisi diserahkan kepada penaakulan LLM
  • Penjejakan pengesahan: mengesan dan mencuba semula arahan yang tidak disahkan

Seterusnya: menggunakan ejen ringan pada peranti pinggir seperti Raspberry Pi.

Percuma untuk bermula

Pelajari Ejen AI dengan tutor kecerdasan buatan — percuma

Tulis dan jalankan kod sebenar dalam pelayar anda, dapatkan bantuan segera daripada tutor kecerdasan buatan yang tersedia 24/7, dan sambung semula dari tempat anda berhenti di web atau dalam aplikasi.

Kursus
60
Pelajaran
239
Mula belajar secara percumaLihat kursus penuh

Soalan Lazim

Adakah pelajaran “Respons Automatik terhadap Peristiwa Sensor” percuma?

Ya — teks penuh “Respons Automatik terhadap Peristiwa Sensor” boleh dibaca secara percuma di web ini. Untuk berlatih secara interaktif menggunakan penyunting kod terbina dalam dan tutor kecerdasan buatan 24/7, serta membuka kunci baki kursus Ejen AI, tingkat taraf kepada CoddyKit PRO. Kursus Ejen AI merangkumi sejumlah 4 pelajaran.

Apakah yang akan saya pelajari dalam “Respons Automatik terhadap Peristiwa Sensor”?

Jika suhu > ambang → amaran → gerakkan: gelung kawalan IoT dipacu ejen. Anda berlatih Ejen AI menggunakan kod praktikal yang dijalankan terus dalam pelayar, manakala tutor kecerdasan buatan 24/7 menjawab soalan anda semasa anda mengikuti pelajaran.

Adakah saya memerlukan pengalaman untuk memulakan Ejen AI?

Tiada pengalaman terdahulu diperlukan. Pembelajaran Ejen AI di CoddyKit disusun untuk pelajar daripada peringkat pemula hingga lanjutan, jadi anda boleh bermula di sini atau dari awal dan belajar mengikut kadar anda sendiri. Ini ialah pelajaran 3 daripada 4.

Berapa lamakah pelajaran “Respons Automatik terhadap Peristiwa Sensor” diambil?

Kebanyakan pelajaran CoddyKit mengambil masa kira-kira 5–10 minit. Setiap pelajaran ringkas dan interaktif, jadi anda boleh membuat kemajuan secara berterusan dan menyambung tepat dari tempat anda berhenti di web atau aplikasi.

Bolehkah saya menulis dan menjalankan kod dalam pelajaran Ejen AI ini?

Ya. Setiap pelajaran Ejen AI menyertakan penyunting kod terbina dalam, jadi anda boleh menulis dan menjalankan kod sebenar terus dalam pelayar serta menerima maklum balas kecerdasan buatan serta-merta — tanpa memerlukan persediaan setempat.

Semua pelajaran dalam kursus ini

  1. Protokol MQTT untuk Integrasi Ejen
  2. Pemprosesan Data Siri Masa dalam Ejen
  3. Respons Automatik terhadap Peristiwa Sensor
  4. Pelaksanaan Ejen Ringan di Pinggir
← Kembali ke Ejen AI