Add synthesis timing functionality to X bot · cameron.stream/void@acdb622

+52 -80

1 changed file

expand all

x.py

+52 -80

x.py

··· 599 599 return yaml.dump(simplified_thread, default_flow_style=False, sort_keys=False) 600 600 601 601 602 - def ensure_x_user_blocks_attached(thread_data: Dict, agent_id: Optional[str] = None) -> None: 603 - """ 604 - Ensure all users in the thread have their X user blocks attached. 605 - Creates blocks with initial content including their handle if they don't exist. 606 - 607 - Args: 608 - thread_data: Dict with 'tweets' and 'users' keys from get_thread_context() 609 - agent_id: The Letta agent ID to attach blocks to (defaults to config agent_id) 610 - """ 611 - if not thread_data or "users" not in thread_data: 612 - return 613 - 614 - try: 615 - from tools.blocks import attach_x_user_blocks, x_user_note_set 616 - from config_loader import get_letta_config 617 - from letta_client import Letta 618 - 619 - # Get Letta client and agent_id from config 620 - config = get_letta_config() 621 - client = Letta(token=config['api_key'], timeout=config['timeout']) 622 - 623 - # Use provided agent_id or get from config 624 - if agent_id is None: 625 - agent_id = config['agent_id'] 626 - 627 - # Get agent info to create a mock agent_state for the functions 628 - class MockAgentState: 629 - def __init__(self, agent_id): 630 - self.id = agent_id 631 - 632 - agent_state = MockAgentState(agent_id) 633 - 634 - users_data = thread_data["users"] 635 - user_ids = list(users_data.keys()) 636 - 637 - if not user_ids: 638 - return 639 - 640 - logger.info(f"Ensuring X user blocks for {len(user_ids)} users: {user_ids}") 641 - 642 - # Get current blocks to check which users already have blocks with content 643 - current_blocks = client.agents.blocks.list(agent_id=agent_id) 644 - existing_user_blocks = {} 645 - 646 - for block in current_blocks: 647 - if block.label.startswith("x_user_"): 648 - user_id = block.label.replace("x_user_", "") 649 - existing_user_blocks[user_id] = block 650 - 651 - # Attach all user blocks (this will create missing ones with basic content) 652 - attach_result = attach_x_user_blocks(user_ids, agent_state) 653 - logger.info(f"X user block attachment result: {attach_result}") 654 - 655 - # For newly created blocks, update with user handle information 656 - for user_id in user_ids: 657 - if user_id not in existing_user_blocks: 658 - user_info = users_data[user_id] 659 - username = user_info.get('username', 'unknown') 660 - name = user_info.get('name', 'Unknown') 661 - 662 - # Set initial content with handle information 663 - initial_content = f"# X User: {user_id}\n\n**Handle:** @{username}\n**Name:** {name}\n\nNo additional information about this user yet." 664 - 665 - try: 666 - x_user_note_set(user_id, initial_content, agent_state) 667 - logger.info(f"Set initial content for X user {user_id} (@{username})") 668 - except Exception as e: 669 - logger.error(f"Failed to set initial content for X user {user_id}: {e}") 670 - 671 - except Exception as e: 672 - logger.error(f"Error ensuring X user blocks: {e}") 673 - 674 602 675 603 # X Caching and Queue System Functions 676 604 ··· 1554 1482 logger.info("Found #voidstop, skipping this mention") 1555 1483 return True 1556 1484 1557 - # Ensure X user blocks are attached 1558 - try: 1559 - ensure_x_user_blocks_attached(thread_data, void_agent.id) 1560 - except Exception as e: 1561 - logger.warning(f"Failed to ensure X user blocks: {e}") 1562 - # Continue without user blocks rather than failing completely 1485 + # Note: X user block attachment removed - no longer using user-specific memory blocks 1563 1486 1564 1487 # Create prompt for Letta agent 1565 1488 # First try to use cached author info from queued mention ··· 2131 2054 2132 2055 return void_agent 2133 2056 2134 - def x_main_loop(testing_mode=False, cleanup_interval=10): 2057 + def x_main_loop(testing_mode=False, cleanup_interval=10, synthesis_interval=600, synthesis_only=False): 2135 2058 """ 2136 2059 Main X bot loop that continuously monitors for mentions and processes them. 2137 2060 Similar to bsky.py main() but for X/Twitter. ··· 2139 2062 Args: 2140 2063 testing_mode: If True, don't actually post to X 2141 2064 cleanup_interval: Run user block cleanup every N cycles (0 to disable) 2065 + synthesis_interval: Send synthesis message every N seconds (0 to disable) 2066 + synthesis_only: If True, only send synthesis messages (no notification processing) 2142 2067 """ 2143 2068 import time 2144 2069 from time import sleep 2145 2070 from config_loader import get_letta_config 2146 2071 from letta_client import Letta 2072 + from bsky import send_synthesis_message 2147 2073 2148 2074 logger.info("=== STARTING X VOID BOT ===") 2149 2075 ··· 2173 2099 else: 2174 2100 logger.info("User block cleanup disabled") 2175 2101 2102 + if synthesis_interval > 0: 2103 + logger.info(f"Synthesis messages enabled every {synthesis_interval} seconds ({synthesis_interval/60:.1f} minutes)") 2104 + else: 2105 + logger.info("Synthesis messages disabled") 2106 + 2107 + # Synthesis-only mode 2108 + if synthesis_only: 2109 + if synthesis_interval <= 0: 2110 + logger.error("Synthesis-only mode requires --synthesis-interval > 0") 2111 + return 2112 + 2113 + logger.info(f"Starting X synthesis-only mode, interval: {synthesis_interval} seconds ({synthesis_interval/60:.1f} minutes)") 2114 + 2115 + while True: 2116 + try: 2117 + # Send synthesis message immediately on first run 2118 + logger.info("🧠 Sending X synthesis message") 2119 + # No atproto_client for X bot, just pass None 2120 + send_synthesis_message(letta_client, void_agent.id, atproto_client=None) 2121 + 2122 + # Wait for next interval 2123 + logger.info(f"Waiting {synthesis_interval} seconds until next synthesis...") 2124 + sleep(synthesis_interval) 2125 + 2126 + except KeyboardInterrupt: 2127 + logger.info("=== X SYNTHESIS MODE STOPPED BY USER ===") 2128 + break 2129 + except Exception as e: 2130 + logger.error(f"Error in X synthesis loop: {e}") 2131 + logger.info(f"Sleeping for {synthesis_interval} seconds due to error...") 2132 + sleep(synthesis_interval) 2133 + 2176 2134 cycle_count = 0 2177 2135 start_time = time.time() 2136 + last_synthesis_time = time.time() 2178 2137 2179 2138 while True: 2180 2139 try: ··· 2188 2147 if cleanup_interval > 0 and cycle_count % cleanup_interval == 0: 2189 2148 logger.debug(f"Running periodic user block cleanup (cycle {cycle_count})") 2190 2149 periodic_user_block_cleanup(letta_client, void_agent.id) 2150 + 2151 + # Check if synthesis interval has passed 2152 + if synthesis_interval > 0: 2153 + current_time = time.time() 2154 + if current_time - last_synthesis_time >= synthesis_interval: 2155 + logger.info(f"⏰ {synthesis_interval/60:.1f} minutes have passed, triggering X synthesis") 2156 + send_synthesis_message(letta_client, void_agent.id, atproto_client=None) 2157 + last_synthesis_time = current_time 2191 2158 2192 2159 # Log cycle completion 2193 2160 elapsed_time = time.time() - start_time ··· 2260 2227 parser.add_argument('--test', action='store_true', help='Run in testing mode (no actual posts)') 2261 2228 parser.add_argument('--cleanup-interval', type=int, default=10, 2262 2229 help='Run user block cleanup every N cycles (default: 10, 0 to disable)') 2230 + parser.add_argument('--synthesis-interval', type=int, default=600, 2231 + help='Send synthesis message every N seconds (default: 600 = 10 minutes, 0 to disable)') 2232 + parser.add_argument('--synthesis-only', action='store_true', 2233 + help='Run in synthesis-only mode (only send synthesis messages, no notification processing)') 2263 2234 args = parser.parse_args() 2264 - x_main_loop(testing_mode=args.test, cleanup_interval=args.cleanup_interval) 2235 + x_main_loop(testing_mode=args.test, cleanup_interval=args.cleanup_interval, 2236 + synthesis_interval=args.synthesis_interval, synthesis_only=args.synthesis_only) 2265 2237 elif sys.argv[1] == "loop": 2266 2238 x_notification_loop() 2267 2239 elif sys.argv[1] == "reply":

Configure Feed

Configure Feed