diff --git a/README.md b/README.md index 2dd816c..5155ff2 100644 --- a/README.md +++ b/README.md @@ -199,10 +199,15 @@ python bsky.py --reasoning # Use simplified log format (void - LEVEL - message) python bsky.py --simple-logs + +# Reset agent message buffer after each notification (stateless mode) +python bsky.py --reset-messages ``` **Note**: The default config path is `configs/config.yaml`. +The `--reset-messages` flag resets the agent's conversation history after each notification is processed, making each interaction stateless. This can help prevent context window overflow and keeps each notification response independent. + ### Running Multiple Bots (`run_bots.py`) Run all configured bots simultaneously with aggregated, color-coded logs: diff --git a/bsky.py b/bsky.py index 31a6f39..d5e8eb9 100644 --- a/bsky.py +++ b/bsky.py @@ -1,7 +1,7 @@ # Rich imports removed - using simple text formatting from time import sleep from letta_client import Letta -from bsky_utils import thread_to_yaml_string, count_thread_posts +from bsky_utils import thread_to_yaml_string, count_thread_posts, extract_images_from_thread import os import logging import json @@ -12,6 +12,7 @@ from datetime import datetime, timedelta from collections import defaultdict import time import argparse +import random from utils import ( upsert_block, @@ -19,8 +20,79 @@ from utils import ( ) from config_loader import get_letta_config, get_config, get_queue_config +# Vision support (optional - requires pillow) +VISION_ENABLED = False +try: + from integrate_vision import create_message_with_vision + VISION_ENABLED = True +except ImportError as e: + pass # Vision not available - pillow not installed + import bsky_utils from datetime import date + +# Downrank configuration +BSKY_DOWNRANK_FILE = Path("bsky_downrank_handles.txt") +DEFAULT_DOWNRANK_RATE = 0.1 # 10% response rate for downranked users + + +def load_downrank_handles() -> dict: + """Load handles that should be downranked (responded to less frequently). + + File format (one per line): + handle.bsky.social # Uses default rate (10%) + handle.bsky.social:0.05 # Custom rate (5%) + # Comments start with # + + Returns: + Dict mapping handle -> response rate (0.0 to 1.0) + """ + try: + if not BSKY_DOWNRANK_FILE.exists(): + return {} + + downrank_handles = {} + with open(BSKY_DOWNRANK_FILE, 'r') as f: + for line in f: + line = line.strip() + if not line or line.startswith('#'): + continue + + # Check for custom rate + if ':' in line: + handle, rate_str = line.split(':', 1) + try: + rate = float(rate_str) + except ValueError: + rate = DEFAULT_DOWNRANK_RATE + else: + handle = line + rate = DEFAULT_DOWNRANK_RATE + + downrank_handles[handle.lower()] = rate + + if downrank_handles: + logger.info(f"Loaded {len(downrank_handles)} downrank handles") + return downrank_handles + except Exception as e: + logger.error(f"Error loading downrank handles: {e}") + return {} + + +def should_respond_to_handle(handle: str, downrank_handles: dict) -> bool: + """Check if we should respond to this handle. + + Returns True 100% of the time for non-downranked users. + Returns True at the configured rate for downranked users. + """ + handle_lower = handle.lower() + if handle_lower not in downrank_handles: + return True + + rate = downrank_handles[handle_lower] + should_respond = random.random() < rate + logger.info(f"Downranked handle @{handle}: {'responding' if should_respond else 'skipping'} ({rate*100:.0f}% chance)") + return should_respond from notification_db import NotificationDB def extract_handles_from_data(data): @@ -98,6 +170,7 @@ start_time = time.time() # Testing mode flag TESTING_MODE = False +RESET_MESSAGES_AFTER_NOTIFICATION = False # Skip git operations flag SKIP_GIT = False @@ -243,6 +316,12 @@ def process_mention(void_agent, atproto_client, notification_data, queue_filepat logger.info(f"[{correlation_id}] Skipping mention from @{author_handle} (not in allowed_handles)") return True # Remove from queue + # Check if handle is downranked (reduced response rate) + downrank_handles = load_downrank_handles() + if not should_respond_to_handle(author_handle, downrank_handles): + logger.info(f"[{correlation_id}] Skipping mention from @{author_handle} (downranked, not selected)") + return True # Remove from queue + # Retrieve the entire thread associated with the mention try: thread = atproto_client.app.bsky.feed.get_post_thread({ @@ -411,10 +490,24 @@ If you choose to reply, use the add_post_to_bluesky_reply_thread tool. Each call logger.error(f"Error attaching user blocks: {e}") try: + # Extract images from thread for vision support + thread_images = extract_images_from_thread(thread, max_images=4) + + # Build message (with or without images) + if VISION_ENABLED and thread_images: + logger.info(f"Thread contains {len(thread_images)} images, downloading for vision...") + message = create_message_with_vision(prompt, thread_images, max_images=4) + if isinstance(message.get('content'), list): + logger.info(f"Vision message created with {len([c for c in message['content'] if c.get('type') == 'image'])} images") + else: + if thread_images and not VISION_ENABLED: + logger.debug(f"Thread has {len(thread_images)} images but vision not enabled (install pillow)") + message = {"role": "user", "content": prompt} + # Use streaming to avoid 524 timeout errors message_stream = CLIENT.agents.messages.stream( agent_id=void_agent.id, - messages=[{"role": "user", "content": prompt}], + messages=[message], stream_tokens=False, # Step streaming only (faster than token streaming) max_steps=100 ) @@ -1325,6 +1418,17 @@ def load_and_process_queued_notifications(void_agent, atproto_client, testing_mo processed_uris = load_processed_notifications() processed_uris.add(notif_data['uri']) save_processed_notifications(processed_uris) + + # Reset agent message buffer if enabled + if RESET_MESSAGES_AFTER_NOTIFICATION: + try: + CLIENT.agents.messages.reset( + agent_id=void_agent.id, + add_default_initial_messages=True + ) + logger.info(f"Reset agent message buffer after processing notification") + except Exception as e: + logger.warning(f"Failed to reset agent messages: {e}") elif success is None: # Special case for moving to error directory error_path = QUEUE_ERROR_DIR / filepath.name @@ -2024,6 +2128,7 @@ def main(): parser.add_argument('--synthesis-interval', type=int, default=600, help='Send synthesis message every N seconds (default: 600 = 10 minutes, 0 to disable)') parser.add_argument('--synthesis-only', action='store_true', help='Run in synthesis-only mode (only send synthesis messages, no notification processing)') parser.add_argument('--debug', action='store_true', help='Enable debug logging') + parser.add_argument('--reset-messages', action='store_true', help='Reset agent message buffer after each notification is processed') args = parser.parse_args() # Initialize configuration with custom path @@ -2122,7 +2227,7 @@ def main(): # Create Rich console for pretty printing # Console no longer used - simple text formatting - global TESTING_MODE, SKIP_GIT, SHOW_REASONING + global TESTING_MODE, SKIP_GIT, SHOW_REASONING, RESET_MESSAGES_AFTER_NOTIFICATION TESTING_MODE = args.test # Store no-git flag globally for use in export_agent_state calls @@ -2134,6 +2239,9 @@ def main(): # Store reasoning flag globally SHOW_REASONING = args.reasoning + # Store reset-messages flag globally + RESET_MESSAGES_AFTER_NOTIFICATION = args.reset_messages + if TESTING_MODE: logger.info("=== RUNNING IN TESTING MODE ===") logger.info(" - No messages will be sent to Bluesky")