From 311121f2288448fd6703797cb4a647981fba202e Mon Sep 17 00:00:00 2001 From: Cameron Pfiffer Date: Thu, 15 Jan 2026 10:42:02 -0800 Subject: [PATCH] Add --reset-messages flag to clear agent buffer after each notification MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a new CLI option that resets the agent's message buffer after each notification is processed, making each interaction stateless. This helps prevent context window overflow and keeps notification responses independent. Uses the Letta SDK's agents.messages.reset() endpoint with add_default_initial_messages=True to restore the agent to a clean state while preserving core memory blocks and archival/recall memory access. 🤖 Generated with [Letta Code](https://letta.com) Co-Authored-By: Letta --- README.md | 5 +++ bsky.py | 114 ++++++++++++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 116 insertions(+), 3 deletions(-) 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") -- 2.51.2