أنماط وكلاء المحفز والإجراء

ما الوكيل القائم على المشغّل والإجراء؟

أنواع المشغّلات

مرحلة الاكتشاف

import json

def detect_new_email(current_emails, last_seen_id):
    new_emails = [
        e for e in current_emails
        if e['id'] > last_seen_id
    ]
    return new_emails

# Simulate detection
current = [{'id': 3, 'subject': 'Meeting'}, {'id': 4, 'subject': 'Invoice'}]
new = detect_new_email(current, last_seen_id=2)
print('New emails:', [e['subject'] for e in new])

مرحلة اتخاذ القرار

import openai

client = openai.OpenAI(api_key='sk-...')

def decide_action(event_description):
    response = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[
            {'role': 'system', 'content': 'You are an automation agent. Decide what action to take for the event.'},
            {'role': 'user', 'content': f'Event: {event_description}'}
        ],
        tools=[
            {'type': 'function', 'function': {'name': 'send_reply', 'description': 'Reply to email', 'parameters': {'type': 'object', 'properties': {'message': {'type': 'string'}}, 'required': ['message']}}}
        ]
    )
    return response.choices[0].message

result = decide_action('New email: Invoice for $500 from supplier')
print(result)

مرحلة التنفيذ

import logging
import sys

logging.basicConfig(level=logging.INFO, stream=sys.stdout)
logger = logging.getLogger('agent')

def execute_action(action_name, params):
    try:
        if action_name == 'send_reply':
            # In real code, call Gmail API here
            logger.info(f'Sending reply: {params["message"]}')
            return {'status': 'success'}
        elif action_name == 'create_task':
            logger.info(f'Creating task: {params["title"]}')
            return {'status': 'success'}
        else:
            raise ValueError(f'Unknown action: {action_name}')
    except Exception as e:
        logger.error(f'Action failed: {e}')
        return {'status': 'error', 'message': str(e)}

if __name__ == '__main__':
    result = execute_action('send_reply', {'message': 'Thanks for reaching out!'})
    print('Result:', result)

نموذج آلة الحالات

from enum import Enum

class AgentState(Enum):
    IDLE = 'idle'
    DETECTING = 'detecting'
    DECIDING = 'deciding'
    EXECUTING = 'executing'
    ERROR = 'error'

class AutomationAgent:
    def __init__(self):
        self.state = AgentState.IDLE
    
    def transition(self, new_state):
        print(f'State: {self.state.value} -> {new_state.value}')
        self.state = new_state
    
    def run_cycle(self, event=None):
        self.transition(AgentState.DETECTING)
        if event:
            self.transition(AgentState.DECIDING)
            self.transition(AgentState.EXECUTING)
        self.transition(AgentState.IDLE)

agent = AutomationAgent()
agent.run_cycle(event={'type': 'email', 'subject': 'Test'})

نمط المشغّل عند استلام البريد الإلكتروني

from fastapi import FastAPI, Request
import base64, json

app = FastAPI()

@app.post('/gmail-push')
async def gmail_push(request: Request):
    body = await request.json()
    # Decode Pub/Sub message
    message = body.get('message', {})
    data = base64.b64decode(message.get('data', '')).decode('utf-8')
    notification = json.loads(data)
    
    email_address = notification.get('emailAddress')
    history_id = notification.get('historyId')
    
    print(f'New email for {email_address}, historyId: {history_id}')
    # Fetch email details and run agent here
    return {'status': 'ok'}

نمط المشغّل عند رفع ملف

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time

class UploadHandler(FileSystemEventHandler):
    def on_created(self, event):
        if event.is_directory:
            return
        print(f'New file detected: {event.src_path}')
        self.process_file(event.src_path)
    
    def process_file(self, filepath):
        # Run agent logic on new file
        print(f'Processing: {filepath}')

observer = Observer()
handler = UploadHandler()
observer.schedule(handler, path='/tmp/uploads/', recursive=False)
observer.start()

try:
    time.sleep(30)  # Watch for 30 seconds
finally:
    observer.stop()
    observer.join()

نمط المشغّل القائم على الوقت

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

scheduler = BlockingScheduler()

def daily_report_agent():
    print(f'Daily report running at {datetime.now()}')
    # Fetch data, call LLM, send report
    pass

def hourly_check_agent():
    print(f'Hourly check at {datetime.now()}')
    pass

# Run at 8am every day
scheduler.add_job(daily_report_agent, 'cron', hour=8, minute=0)

# Run every 30 minutes
scheduler.add_job(hourly_check_agent, 'interval', minutes=30)

print('Scheduler started')
scheduler.start()

تنفيذ الإجراءات متساوية الأثر

import hashlib

processed_events = set()  # In production, use Redis or DB

def compute_event_id(event):
    content = f"{event['type']}:{event['source_id']}:{event['timestamp']}"
    return hashlib.sha256(content.encode()).hexdigest()[:16]

def handle_event_idempotent(event):
    event_id = compute_event_id(event)
    
    if event_id in processed_events:
        print(f'Skipping duplicate event: {event_id}')
        return {'status': 'duplicate', 'event_id': event_id}
    
    # Process event
    print(f'Processing event: {event_id}')
    processed_events.add(event_id)
    return {'status': 'processed', 'event_id': event_id}

# Simulate duplicate event
event = {'type': 'email', 'source_id': 'abc123', 'timestamp': '2024-01-01T09:00:00'}
print(handle_event_idempotent(event))
print(handle_event_idempotent(event))  # Duplicate - skipped

حالات الخطأ والاسترداد

import time

def execute_with_retry(action_fn, max_retries=3):
    for attempt in range(max_retries):
        try:
            result = action_fn()
            print(f'Success on attempt {attempt + 1}')
            return result
        except Exception as e:
            wait = 2 ** attempt  # Exponential backoff: 1s, 2s, 4s
            print(f'Attempt {attempt + 1} failed: {e}. Retrying in {wait}s')
            if attempt < max_retries - 1:
                time.sleep(wait)
            else:
                print('All retries exhausted. Moving to dead-letter queue.')
                raise

# Example usage
call_count = [0]

def flaky_action():
    call_count[0] += 1
    if call_count[0] < 3:
        raise ConnectionError('Service unavailable')
    return 'Done'

execute_with_retry(flaky_action)

اختبار المعرفة: أنماط المشغّلات والإجراءات

جمع المكوّنات