""" OASIS dual-platform parallel simulation preset script Run Twitter and Reddit simulations simultaneously with the same configuration file Features: - Dual-platform (Twitter + Reddit) parallel simulation - Keep environment running after simulation completes (enter wait mode) - Support Interview commands via IPC - Support single Agent interview and batch interview - Support remote environment shutdown command Usage: python run_parallel_simulation.py --config simulation_config.json python run_parallel_simulation.py --config simulation_config.json --no-wait # Close immediately after completion python run_parallel_simulation.py --config simulation_config.json --twitter-only python run_parallel_simulation.py --config simulation_config.json --reddit-only Log structure: sim_xxx/ ├── twitter/ │ └── actions.jsonl # Twitter platform action log ├── reddit/ │ └── actions.jsonl # Reddit platform action log ├── simulation.log # Main simulation process log └── run_state.json # Run state (for API queries) """ # ============================================================ # Fix Windows encoding issue: Set UTF-8 encoding before all imports # This is to fix the issue that OASIS third-party library doesn't specify encoding when reading files # ============================================================ import sys import os if sys.platform == 'win32': # Set Python default I/O encoding to UTF-8 # This affects all open() calls without specified encoding os.environ.setdefault('PYTHONUTF8', '1') os.environ.setdefault('PYTHONIOENCODING', 'utf-8') # Reconfigure standard output stream to UTF-8 (fix console encoding issues) if hasattr(sys.stdout, 'reconfigure'): sys.stdout.reconfigure(encoding='utf-8', errors='replace') if hasattr(sys.stderr, 'reconfigure'): sys.stderr.reconfigure(encoding='utf-8', errors='replace') # Force set default encoding (affects default encoding of open() function) # Note: This must be set when Python starts, runtime configuration may not work # So we also need to monkey-patch the built-in open function import builtins _original_open = builtins.open def _utf8_open(file, mode='r', buffering=-1, encoding=None, errors=None, newline=None, closefd=True, opener=None): """ Wrap open() function to use UTF-8 encoding by default for text mode This can fix the issue that third-party libraries (like OASIS) don't specify encoding when reading files """ # Only set default encoding for text mode (non-binary) without specified encoding if encoding is None and 'b' not in mode: encoding = 'utf-8' return _original_open(file, mode, buffering, encoding, errors, newline, closefd, opener) builtins.open = _utf8_open import argparse import asyncio import json import logging import multiprocessing import random import signal import sqlite3 import warnings from datetime import datetime from typing import Dict, Any, List, Optional, Tuple # Global variables: for signal handling _shutdown_event = None _cleanup_done = False # Add backend directory to path # Script is fixed in backend/scripts/ directory _scripts_dir = os.path.dirname(os.path.abspath(__file__)) _backend_dir = os.path.abspath(os.path.join(_scripts_dir, '..')) _project_root = os.path.abspath(os.path.join(_backend_dir, '..')) sys.path.insert(0, _scripts_dir) sys.path.insert(0, _backend_dir) # Load .env file from project root (contains LLM_API_KEY and other configurations) from dotenv import load_dotenv _env_file = os.path.join(_project_root, '.env') if os.path.exists(_env_file): load_dotenv(_env_file) print(f"Loaded environment configuration: {_env_file}") else: # Try to load backend/.env _backend_env = os.path.join(_backend_dir, '.env') if os.path.exists(_backend_env): load_dotenv(_backend_env) print(f"Loaded environment configuration: {_backend_env}") class MaxTokensWarningFilter(logging.Filter): """Filter out camel-ai max_tokens warnings (we intentionally don't set max_tokens to let the model decide)""" def filter(self, record): # Filter out logs containing max_tokens warnings if "max_tokens" in record.getMessage() and "Invalid or missing" in record.getMessage(): return False return True # Add filter immediately when module loads, ensure it takes effect before camel code executes logging.getLogger().addFilter(MaxTokensWarningFilter()) def disable_oasis_logging(): """ Disable verbose logging output from OASIS library OASIS logging is too verbose (logs every agent's observation and action), we use our own action_logger """ # Disable all OASIS loggers oasis_loggers = [ "social.agent", "social.twitter", "social.rec", "oasis.env", "table", ] for logger_name in oasis_loggers: logger = logging.getLogger(logger_name) logger.setLevel(logging.CRITICAL) # Only log critical errors logger.handlers.clear() logger.propagate = False def init_logging_for_simulation(simulation_dir: str): """ Initialize simulation log configuration Args: simulation_dir: Simulation directory path """ # Disable OASIS verbose logging disable_oasis_logging() # Clean up old log directory (if exists) old_log_dir = os.path.join(simulation_dir, "log") if os.path.exists(old_log_dir): import shutil shutil.rmtree(old_log_dir, ignore_errors=True) from action_logger import SimulationLogManager, PlatformActionLogger try: from camel.models import ModelFactory from camel.types import ModelPlatformType import oasis from oasis import ( ActionType, LLMAction, ManualAction, generate_twitter_agent_graph, generate_reddit_agent_graph ) except ImportError as e: print(f"Error: Missing dependency {e}") print("Please install first: pip install oasis-ai camel-ai") sys.exit(1) # Twitter available actions (INTERVIEW not included, INTERVIEW can only be triggered manually via ManualAction) TWITTER_ACTIONS = [ ActionType.CREATE_POST, ActionType.LIKE_POST, ActionType.REPOST, ActionType.FOLLOW, ActionType.DO_NOTHING, ActionType.QUOTE_POST, ] # Reddit available actions (INTERVIEW not included, INTERVIEW can only be triggered manually via ManualAction) REDDIT_ACTIONS = [ ActionType.LIKE_POST, ActionType.DISLIKE_POST, ActionType.CREATE_POST, ActionType.CREATE_COMMENT, ActionType.LIKE_COMMENT, ActionType.DISLIKE_COMMENT, ActionType.SEARCH_POSTS, ActionType.SEARCH_USER, ActionType.TREND, ActionType.REFRESH, ActionType.DO_NOTHING, ActionType.FOLLOW, ActionType.MUTE, ] # IPC-related constants IPC_COMMANDS_DIR = "ipc_commands" IPC_RESPONSES_DIR = "ipc_responses" ENV_STATUS_FILE = "env_status.json" class CommandType: """Command type constants""" INTERVIEW = "interview" BATCH_INTERVIEW = "batch_interview" CLOSE_ENV = "close_env" class ParallelIPCHandler: """ Dual-platform IPC command handler Manage environments of both platforms, handle Interview commands """ def __init__( self, simulation_dir: str, twitter_env=None, twitter_agent_graph=None, reddit_env=None, reddit_agent_graph=None ): self.simulation_dir = simulation_dir self.twitter_env = twitter_env self.twitter_agent_graph = twitter_agent_graph self.reddit_env = reddit_env self.reddit_agent_graph = reddit_agent_graph self.commands_dir = os.path.join(simulation_dir, IPC_COMMANDS_DIR) self.responses_dir = os.path.join(simulation_dir, IPC_RESPONSES_DIR) self.status_file = os.path.join(simulation_dir, ENV_STATUS_FILE) # Ensure directory exists os.makedirs(self.commands_dir, exist_ok=True) os.makedirs(self.responses_dir, exist_ok=True) def update_status(self, status: str): """Update environment status""" with open(self.status_file, 'w', encoding='utf-8') as f: json.dump({ "status": status, "twitter_available": self.twitter_env is not None, "reddit_available": self.reddit_env is not None, "timestamp": datetime.now().isoformat() }, f, ensure_ascii=False, indent=2) def poll_command(self) -> Optional[Dict[str, Any]]: """Poll for pending commands""" if not os.path.exists(self.commands_dir): return None # Get command files (sorted by time) command_files = [] for filename in os.listdir(self.commands_dir): if filename.endswith('.json'): filepath = os.path.join(self.commands_dir, filename) command_files.append((filepath, os.path.getmtime(filepath))) command_files.sort(key=lambda x: x[1]) for filepath, _ in command_files: try: with open(filepath, 'r', encoding='utf-8') as f: return json.load(f) except (json.JSONDecodeError, OSError): continue return None def send_response(self, command_id: str, status: str, result: Dict = None, error: str = None): """Send response""" response = { "command_id": command_id, "status": status, "result": result, "error": error, "timestamp": datetime.now().isoformat() } response_file = os.path.join(self.responses_dir, f"{command_id}.json") with open(response_file, 'w', encoding='utf-8') as f: json.dump(response, f, ensure_ascii=False, indent=2) # Delete command file command_file = os.path.join(self.commands_dir, f"{command_id}.json") try: os.remove(command_file) except OSError: pass def _get_env_and_graph(self, platform: str): """ Get environment and agent_graph for specified platform Args: platform: Platform name ("twitter" or "reddit") Returns: (env, agent_graph, platform_name) or (None, None, None) """ if platform == "twitter" and self.twitter_env: return self.twitter_env, self.twitter_agent_graph, "twitter" elif platform == "reddit" and self.reddit_env: return self.reddit_env, self.reddit_agent_graph, "reddit" else: return None, None, None async def _interview_single_platform(self, agent_id: int, prompt: str, platform: str) -> Dict[str, Any]: """ Execute Interview on a single platform Returns: Dictionary containing result, or dictionary containing error """ env, agent_graph, actual_platform = self._get_env_and_graph(platform) if not env or not agent_graph: return {"platform": platform, "error": f"{platform}platform unavailable"} try: agent = agent_graph.get_agent(agent_id) interview_action = ManualAction( action_type=ActionType.INTERVIEW, action_args={"prompt": prompt} ) actions = {agent: interview_action} await env.step(actions) result = self._get_interview_result(agent_id, actual_platform) result["platform"] = actual_platform return result except Exception as e: return {"platform": platform, "error": str(e)} async def handle_interview(self, command_id: str, agent_id: int, prompt: str, platform: str = None) -> bool: """ Handle single Agent interview command Args: command_id: Command ID agent_id: Agent ID prompt: Interview question platform: Specify platform (optional) - "twitter": Interview only Twitter platform - "reddit": Interview only Reddit platform - None/unspecified: Interview both platforms simultaneously, return integrated result Returns: True means success, False means failure """ # If platform is specified, only interview that platform if platform in ("twitter", "reddit"): result = await self._interview_single_platform(agent_id, prompt, platform) if "error" in result: self.send_response(command_id, "failed", error=result["error"]) print(f" Interview failed: agent_id={agent_id}, platform={platform}, error={result['error']}") return False else: self.send_response(command_id, "completed", result=result) print(f" Interview completed: agent_id={agent_id}, platform={platform}") return True # Platform not specified: interview both platforms simultaneously if not self.twitter_env and not self.reddit_env: self.send_response(command_id, "failed", error="No available simulation environment") return False results = { "agent_id": agent_id, "prompt": prompt, "platforms": {} } success_count = 0 # Interview both platforms in parallel tasks = [] platforms_to_interview = [] if self.twitter_env: tasks.append(self._interview_single_platform(agent_id, prompt, "twitter")) platforms_to_interview.append("twitter") if self.reddit_env: tasks.append(self._interview_single_platform(agent_id, prompt, "reddit")) platforms_to_interview.append("reddit") # Execute in parallel platform_results = await asyncio.gather(*tasks) for platform_name, platform_result in zip(platforms_to_interview, platform_results): results["platforms"][platform_name] = platform_result if "error" not in platform_result: success_count += 1 if success_count > 0: self.send_response(command_id, "completed", result=results) print(f" Interview completed: agent_id={agent_id}, success_platforms={success_count}/{len(platforms_to_interview)}") return True else: errors = [f"{p}: {r.get('error', 'Unknown error')}" for p, r in results["platforms"].items()] self.send_response(command_id, "failed", error="; ".join(errors)) print(f" Interview failed: agent_id={agent_id}, All platforms failed") return False async def handle_batch_interview(self, command_id: str, interviews: List[Dict], platform: str = None) -> bool: """ Handle batch interview command Args: command_id: Command ID interviews: [{"agent_id": int, "prompt": str, "platform": str(optional)}, ...] platform: default platform (can be overridden by each interview item) - "twitter": Interview only Twitter platform - "reddit": Interview only Reddit platform - None/unspecified: Interview both platforms simultaneously for each Agent """ # Group by platform twitter_interviews = [] reddit_interviews = [] both_platforms_interviews = [] # Need to interview both platforms simultaneously for interview in interviews: item_platform = interview.get("platform", platform) if item_platform == "twitter": twitter_interviews.append(interview) elif item_platform == "reddit": reddit_interviews.append(interview) else: # Platform not specified: interview both platforms both_platforms_interviews.append(interview) # Split both_platforms_interviews to two platforms if both_platforms_interviews: if self.twitter_env: twitter_interviews.extend(both_platforms_interviews) if self.reddit_env: reddit_interviews.extend(both_platforms_interviews) results = {} # Handle Twitter platform interview if twitter_interviews and self.twitter_env: try: twitter_actions = {} for interview in twitter_interviews: agent_id = interview.get("agent_id") prompt = interview.get("prompt", "") try: agent = self.twitter_agent_graph.get_agent(agent_id) twitter_actions[agent] = ManualAction( action_type=ActionType.INTERVIEW, action_args={"prompt": prompt} ) except Exception as e: print(f" Warning: Unable to get Twitter Agent {agent_id}: {e}") if twitter_actions: await self.twitter_env.step(twitter_actions) for interview in twitter_interviews: agent_id = interview.get("agent_id") result = self._get_interview_result(agent_id, "twitter") result["platform"] = "twitter" results[f"twitter_{agent_id}"] = result except Exception as e: print(f" Twitter batch Interview failed: {e}") # Handle Reddit platform interview if reddit_interviews and self.reddit_env: try: reddit_actions = {} for interview in reddit_interviews: agent_id = interview.get("agent_id") prompt = interview.get("prompt", "") try: agent = self.reddit_agent_graph.get_agent(agent_id) reddit_actions[agent] = ManualAction( action_type=ActionType.INTERVIEW, action_args={"prompt": prompt} ) except Exception as e: print(f" Warning: Unable to get Reddit Agent {agent_id}: {e}") if reddit_actions: await self.reddit_env.step(reddit_actions) for interview in reddit_interviews: agent_id = interview.get("agent_id") result = self._get_interview_result(agent_id, "reddit") result["platform"] = "reddit" results[f"reddit_{agent_id}"] = result except Exception as e: print(f" Reddit batch Interview failed: {e}") if results: self.send_response(command_id, "completed", result={ "interviews_count": len(results), "results": results }) print(f" Batch Interview completed: {len(results)} Agents") return True else: self.send_response(command_id, "failed", error="No successful interviews") return False def _get_interview_result(self, agent_id: int, platform: str) -> Dict[str, Any]: """Get the latest Interview result from database""" db_path = os.path.join(self.simulation_dir, f"{platform}_simulation.db") result = { "agent_id": agent_id, "response": None, "timestamp": None } if not os.path.exists(db_path): return result try: conn = sqlite3.connect(db_path) cursor = conn.cursor() # Query the latest Interview record cursor.execute(""" SELECT user_id, info, created_at FROM trace WHERE action = ? AND user_id = ? ORDER BY created_at DESC LIMIT 1 """, (ActionType.INTERVIEW.value, agent_id)) row = cursor.fetchone() if row: user_id, info_json, created_at = row try: info = json.loads(info_json) if info_json else {} result["response"] = info.get("response", info) result["timestamp"] = created_at except json.JSONDecodeError: result["response"] = info_json conn.close() except Exception as e: print(f" Failed to read Interview result: {e}") return result async def process_commands(self) -> bool: """ Process all pending commands Returns: True means continue running, False means should exit """ command = self.poll_command() if not command: return True command_id = command.get("command_id") command_type = command.get("command_type") args = command.get("args", {}) print(f"\nReceived IPC command: {command_type}, id={command_id}") if command_type == CommandType.INTERVIEW: await self.handle_interview( command_id, args.get("agent_id", 0), args.get("prompt", ""), args.get("platform") ) return True elif command_type == CommandType.BATCH_INTERVIEW: await self.handle_batch_interview( command_id, args.get("interviews", []), args.get("platform") ) return True elif command_type == CommandType.CLOSE_ENV: print("Received close environment command") self.send_response(command_id, "completed", result={"message": "Environment will close"}) return False else: self.send_response(command_id, "failed", error=f"Unknown command type: {command_type}") return True def load_config(config_path: str) -> Dict[str, Any]: """Load configuration file""" with open(config_path, 'r', encoding='utf-8') as f: return json.load(f) # Non-core action types to be filtered (these actions have low analytical value) FILTERED_ACTIONS = {'refresh', 'sign_up'} # Action type mapping table (Database name -> standard name) ACTION_TYPE_MAP = { 'create_post': 'CREATE_POST', 'like_post': 'LIKE_POST', 'dislike_post': 'DISLIKE_POST', 'repost': 'REPOST', 'quote_post': 'QUOTE_POST', 'follow': 'FOLLOW', 'mute': 'MUTE', 'create_comment': 'CREATE_COMMENT', 'like_comment': 'LIKE_COMMENT', 'dislike_comment': 'DISLIKE_COMMENT', 'search_posts': 'SEARCH_POSTS', 'search_user': 'SEARCH_USER', 'trend': 'TREND', 'do_nothing': 'DO_NOTHING', 'interview': 'INTERVIEW', } def get_agent_names_from_config(config: Dict[str, Any]) -> Dict[int, str]: """ Get mapping of agent_id -> entity_name from simulation_config This allows displaying real entity names in actions.jsonl instead of codes like "Agent_0" Args: config: Content of simulation_config.json Returns: Mapping dictionary of agent_id -> entity_name """ agent_names = {} agent_configs = config.get("agent_configs", []) for agent_config in agent_configs: agent_id = agent_config.get("agent_id") entity_name = agent_config.get("entity_name", f"Agent_{agent_id}") if agent_id is not None: agent_names[agent_id] = entity_name return agent_names def fetch_new_actions_from_db( db_path: str, last_rowid: int, agent_names: Dict[int, str] ) -> Tuple[List[Dict[str, Any]], int]: """ Get new action records from Database and supplement complete context information Args: db_path: Database file path last_rowid: Maximum rowid value from last read (use rowid instead of created_at because different platforms have different created_at formats) agent_names: agent_id -> agent_name mapping Returns: (actions_list, new_last_rowid) - actions_list: List of actions, each element contains agent_id, agent_name, action_type, action_args (including context information) - new_last_rowid: New maximum rowid value """ actions = [] new_last_rowid = last_rowid if not os.path.exists(db_path): return actions, new_last_rowid try: conn = sqlite3.connect(db_path) cursor = conn.cursor() # Use rowid to track processed records (rowid is SQLite's built-in auto-increment field) # This avoids created_at format differences (Twitter uses integers, Reddit uses datetime strings) cursor.execute(""" SELECT rowid, user_id, action, info FROM trace WHERE rowid > ? ORDER BY rowid ASC """, (last_rowid,)) for rowid, user_id, action, info_json in cursor.fetchall(): # Update maximum rowid new_last_rowid = rowid # Filter non-core actions if action in FILTERED_ACTIONS: continue # Parse action arguments try: action_args = json.loads(info_json) if info_json else {} except json.JSONDecodeError: action_args = {} # Simplify action_args, keep only key fields (keep full content, no truncation) simplified_args = {} if 'content' in action_args: simplified_args['content'] = action_args['content'] if 'post_id' in action_args: simplified_args['post_id'] = action_args['post_id'] if 'comment_id' in action_args: simplified_args['comment_id'] = action_args['comment_id'] if 'quoted_id' in action_args: simplified_args['quoted_id'] = action_args['quoted_id'] if 'new_post_id' in action_args: simplified_args['new_post_id'] = action_args['new_post_id'] if 'follow_id' in action_args: simplified_args['follow_id'] = action_args['follow_id'] if 'query' in action_args: simplified_args['query'] = action_args['query'] if 'like_id' in action_args: simplified_args['like_id'] = action_args['like_id'] if 'dislike_id' in action_args: simplified_args['dislike_id'] = action_args['dislike_id'] # Convert action type names action_type = ACTION_TYPE_MAP.get(action, action.upper()) # Supplement context information (post content, usernames, etc.) _enrich_action_context(cursor, action_type, simplified_args, agent_names) actions.append({ 'agent_id': user_id, 'agent_name': agent_names.get(user_id, f'Agent_{user_id}'), 'action_type': action_type, 'action_args': simplified_args, }) conn.close() except Exception as e: print(f"Failed to read Database actions: {e}") return actions, new_last_rowid def _enrich_action_context( cursor, action_type: str, action_args: Dict[str, Any], agent_names: Dict[int, str] ) -> None: """ for actionSupplement context information (post content, usernames, etc.) Args: cursor: Database cursor action_type: Action type action_args: Action arguments (will be modified) agent_names: agent_id -> agent_name mapping """ try: # Like/dislike post: supplement post content and author if action_type in ('LIKE_POST', 'DISLIKE_POST'): post_id = action_args.get('post_id') if post_id: post_info = _get_post_info(cursor, post_id, agent_names) if post_info: action_args['post_content'] = post_info.get('content', '') action_args['post_author_name'] = post_info.get('author_name', '') # Repost: supplement original post content and author elif action_type == 'REPOST': new_post_id = action_args.get('new_post_id') if new_post_id: # Repost's original_post_id points to original post cursor.execute(""" SELECT original_post_id FROM post WHERE post_id = ? """, (new_post_id,)) row = cursor.fetchone() if row and row[0]: original_post_id = row[0] original_info = _get_post_info(cursor, original_post_id, agent_names) if original_info: action_args['original_content'] = original_info.get('content', '') action_args['original_author_name'] = original_info.get('author_name', '') # Quote post: supplement original post content, author, and quote comment elif action_type == 'QUOTE_POST': quoted_id = action_args.get('quoted_id') new_post_id = action_args.get('new_post_id') if quoted_id: original_info = _get_post_info(cursor, quoted_id, agent_names) if original_info: action_args['original_content'] = original_info.get('content', '') action_args['original_author_name'] = original_info.get('author_name', '') # Get quote post comment content (quote_content) if new_post_id: cursor.execute(""" SELECT quote_content FROM post WHERE post_id = ? """, (new_post_id,)) row = cursor.fetchone() if row and row[0]: action_args['quote_content'] = row[0] # Follow user: supplement followed user name elif action_type == 'FOLLOW': follow_id = action_args.get('follow_id') if follow_id: # Get followee_id from follow table cursor.execute(""" SELECT followee_id FROM follow WHERE follow_id = ? """, (follow_id,)) row = cursor.fetchone() if row: followee_id = row[0] target_name = _get_user_name(cursor, followee_id, agent_names) if target_name: action_args['target_user_name'] = target_name # Mute user: supplement muted user name elif action_type == 'MUTE': # Get user_id or target_id from action_args target_id = action_args.get('user_id') or action_args.get('target_id') if target_id: target_name = _get_user_name(cursor, target_id, agent_names) if target_name: action_args['target_user_name'] = target_name # Like/dislike comment: supplement comment content and author elif action_type in ('LIKE_COMMENT', 'DISLIKE_COMMENT'): comment_id = action_args.get('comment_id') if comment_id: comment_info = _get_comment_info(cursor, comment_id, agent_names) if comment_info: action_args['comment_content'] = comment_info.get('content', '') action_args['comment_author_name'] = comment_info.get('author_name', '') # Post comment: supplement commented post information elif action_type == 'CREATE_COMMENT': post_id = action_args.get('post_id') if post_id: post_info = _get_post_info(cursor, post_id, agent_names) if post_info: action_args['post_content'] = post_info.get('content', '') action_args['post_author_name'] = post_info.get('author_name', '') except Exception as e: # Context supplement failure does not affect main process print(f"Failed to supplement action context: {e}") def _get_post_info( cursor, post_id: int, agent_names: Dict[int, str] ) -> Optional[Dict[str, str]]: """ Get post information Args: cursor: Database cursor post_id: Post ID agent_names: agent_id -> agent_name mapping Returns: Dictionary containing content and author_name, or None """ try: cursor.execute(""" SELECT p.content, p.user_id, u.agent_id FROM post p LEFT JOIN user u ON p.user_id = u.user_id WHERE p.post_id = ? """, (post_id,)) row = cursor.fetchone() if row: content = row[0] or '' user_id = row[1] agent_id = row[2] # Preferentially use name from agent_names author_name = '' if agent_id is not None and agent_id in agent_names: author_name = agent_names[agent_id] elif user_id: # Get name from user table cursor.execute("SELECT name, user_name FROM user WHERE user_id = ?", (user_id,)) user_row = cursor.fetchone() if user_row: author_name = user_row[0] or user_row[1] or '' return {'content': content, 'author_name': author_name} except Exception: pass return None def _get_user_name( cursor, user_id: int, agent_names: Dict[int, str] ) -> Optional[str]: """ Get user name Args: cursor: Database cursor user_id: User ID agent_names: agent_id -> agent_name mapping Returns: User name, or None """ try: cursor.execute(""" SELECT agent_id, name, user_name FROM user WHERE user_id = ? """, (user_id,)) row = cursor.fetchone() if row: agent_id = row[0] name = row[1] user_name = row[2] # Preferentially use name from agent_names if agent_id is not None and agent_id in agent_names: return agent_names[agent_id] return name or user_name or '' except Exception: pass return None def _get_comment_info( cursor, comment_id: int, agent_names: Dict[int, str] ) -> Optional[Dict[str, str]]: """ Get comment information Args: cursor: Database cursor comment_id: Comment ID agent_names: agent_id -> agent_name mapping Returns: Dictionary containing content and author_name, or None """ try: cursor.execute(""" SELECT c.content, c.user_id, u.agent_id FROM comment c LEFT JOIN user u ON c.user_id = u.user_id WHERE c.comment_id = ? """, (comment_id,)) row = cursor.fetchone() if row: content = row[0] or '' user_id = row[1] agent_id = row[2] # Preferentially use name from agent_names author_name = '' if agent_id is not None and agent_id in agent_names: author_name = agent_names[agent_id] elif user_id: # Get name from user table cursor.execute("SELECT name, user_name FROM user WHERE user_id = ?", (user_id,)) user_row = cursor.fetchone() if user_row: author_name = user_row[0] or user_row[1] or '' return {'content': content, 'author_name': author_name} except Exception: pass return None def create_model(config: Dict[str, Any], use_boost: bool = False): """ Create LLM model Support dual LLM configuration for acceleration during parallel simulation： - Common configuration：LLM_API_KEY, LLM_BASE_URL, LLM_MODEL_NAME - Acceleration configuration (optional)：LLM_BOOST_API_KEY, LLM_BOOST_BASE_URL, LLM_BOOST_MODEL_NAME If acceleration LLM is configured, different platforms can use different API providers during parallel simulation to improve concurrency. Args: config: Simulation configuration dictionary use_boost: Whether to use acceleration LLM configuration (if available) """ # Check if acceleration configuration exists boost_api_key = os.environ.get("LLM_BOOST_API_KEY", "") boost_base_url = os.environ.get("LLM_BOOST_BASE_URL", "") boost_model = os.environ.get("LLM_BOOST_MODEL_NAME", "") has_boost_config = bool(boost_api_key) # Choose which LLM to use based on parameters and configuration if use_boost and has_boost_config: # Use acceleration configuration llm_api_key = boost_api_key llm_base_url = boost_base_url llm_model = boost_model or os.environ.get("LLM_MODEL_NAME", "") config_label = "[Acceleration LLM]" else: # useCommon configuration llm_api_key = os.environ.get("LLM_API_KEY", "") llm_base_url = os.environ.get("LLM_BASE_URL", "") llm_model = os.environ.get("LLM_MODEL_NAME", "") config_label = "[Common LLM]" # If model name is not in .env, use config as fallback if not llm_model: llm_model = config.get("llm_model", "gpt-4o-mini") # Set environment variables required by camel-ai if llm_api_key: os.environ["OPENAI_API_KEY"] = llm_api_key if not os.environ.get("OPENAI_API_KEY"): raise ValueError("Missing API Key configuration, please set LLM_API_KEY in .env file in project root") if llm_base_url: os.environ["OPENAI_API_BASE_URL"] = llm_base_url print(f"{config_label} model={llm_model}, base_url={llm_base_url[:40] if llm_base_url else 'default'}...") return ModelFactory.create( model_platform=ModelPlatformType.OPENAI, model_type=llm_model, ) def get_active_agents_for_round( env, config: Dict[str, Any], current_hour: int, round_num: int ) -> List: """Decide which Agents to activate this round based on time and configuration""" time_config = config.get("time_config", {}) agent_configs = config.get("agent_configs", []) base_min = time_config.get("agents_per_hour_min", 5) base_max = time_config.get("agents_per_hour_max", 20) peak_hours = time_config.get("peak_hours", [9, 10, 11, 14, 15, 20, 21, 22]) off_peak_hours = time_config.get("off_peak_hours", [0, 1, 2, 3, 4, 5]) if current_hour in peak_hours: multiplier = time_config.get("peak_activity_multiplier", 1.5) elif current_hour in off_peak_hours: multiplier = time_config.get("off_peak_activity_multiplier", 0.3) else: multiplier = 1.0 target_count = int(random.uniform(base_min, base_max) * multiplier) candidates = [] for cfg in agent_configs: agent_id = cfg.get("agent_id", 0) active_hours = cfg.get("active_hours", list(range(8, 23))) activity_level = cfg.get("activity_level", 0.5) if current_hour not in active_hours: continue if random.random() < activity_level: candidates.append(agent_id) selected_ids = random.sample( candidates, min(target_count, len(candidates)) ) if candidates else [] active_agents = [] for agent_id in selected_ids: try: agent = env.agent_graph.get_agent(agent_id) active_agents.append((agent_id, agent)) except Exception: pass return active_agents class PlatformSimulation: """Platform simulation result container""" def __init__(self): self.env = None self.agent_graph = None self.total_actions = 0 async def run_twitter_simulation( config: Dict[str, Any], simulation_dir: str, action_logger: Optional[PlatformActionLogger] = None, main_logger: Optional[SimulationLogManager] = None, max_rounds: Optional[int] = None ) -> PlatformSimulation: """Run Twitter simulation Args: config: Simulation configuration simulation_dir: Simulation directory action_logger: Action logger main_logger: Main logger manager max_rounds: Maximum simulation rounds (optional, used to truncate long simulations) Returns: PlatformSimulation: Result object containing env and agent_graph """ result = PlatformSimulation() def log_info(msg): if main_logger: main_logger.info(f"[Twitter] {msg}") print(f"[Twitter] {msg}") log_info("Initializing...") # Twitter use common LLM configuration model = create_model(config, use_boost=False) # OASIS Twitter uses CSV format profile_path = os.path.join(simulation_dir, "twitter_profiles.csv") if not os.path.exists(profile_path): log_info(f"Error: Profile file does not exist: {profile_path}") return result result.agent_graph = await generate_twitter_agent_graph( profile_path=profile_path, model=model, available_actions=TWITTER_ACTIONS, ) # Get Agent real name mapping from config (use entity_name instead of default Agent_X) agent_names = get_agent_names_from_config(config) # If an agent is not in config, use OASIS default name for agent_id, agent in result.agent_graph.get_agents(): if agent_id not in agent_names: agent_names[agent_id] = getattr(agent, 'name', f'Agent_{agent_id}') db_path = os.path.join(simulation_dir, "twitter_simulation.db") if os.path.exists(db_path): os.remove(db_path) result.env = oasis.make( agent_graph=result.agent_graph, platform=oasis.DefaultPlatformType.TWITTER, database_path=db_path, semaphore=30, # Limit maximum concurrent LLM requests to prevent API overload ) await result.env.reset() log_info("Environment started") if action_logger: action_logger.log_simulation_start(config) total_actions = 0 last_rowid = 0 # Track last processed row in Database (use rowid to avoid created_at format differences) # Execute initial events event_config = config.get("event_config", {}) initial_posts = event_config.get("initial_posts", []) # Log round 0 start (initial event phase) if action_logger: action_logger.log_round_start(0, 0) # round 0, simulated_hour 0 initial_action_count = 0 if initial_posts: initial_actions = {} for post in initial_posts: agent_id = post.get("poster_agent_id", 0) content = post.get("content", "") try: agent = result.env.agent_graph.get_agent(agent_id) initial_actions[agent] = ManualAction( action_type=ActionType.CREATE_POST, action_args={"content": content} ) if action_logger: action_logger.log_action( round_num=0, agent_id=agent_id, agent_name=agent_names.get(agent_id, f"Agent_{agent_id}"), action_type="CREATE_POST", action_args={"content": content} ) total_actions += 1 initial_action_count += 1 except Exception: pass if initial_actions: await result.env.step(initial_actions) log_info(f"Published {len(initial_actions)} initial posts") # Log round 0 end if action_logger: action_logger.log_round_end(0, initial_action_count) # Main simulation loop time_config = config.get("time_config", {}) total_hours = time_config.get("total_simulation_hours", 72) minutes_per_round = time_config.get("minutes_per_round", 30) total_rounds = (total_hours * 60) // minutes_per_round # If maximum rounds specified, truncate if max_rounds is not None and max_rounds > 0: original_rounds = total_rounds total_rounds = min(total_rounds, max_rounds) if total_rounds < original_rounds: log_info(f"Rounds truncated: {original_rounds} -> {total_rounds} (max_rounds={max_rounds})") start_time = datetime.now() for round_num in range(total_rounds): # Check if received exit signal if _shutdown_event and _shutdown_event.is_set(): if main_logger: main_logger.info(f"Received exit signal，at round {round_num + 1} stop simulation") break simulated_minutes = round_num * minutes_per_round simulated_hour = (simulated_minutes // 60) % 24 simulated_day = simulated_minutes // (60 * 24) + 1 active_agents = get_active_agents_for_round( result.env, config, simulated_hour, round_num ) # Log round start regardless of active agents if action_logger: action_logger.log_round_start(round_num + 1, simulated_hour) if not active_agents: # Log round end even without active agents (actions_count=0) if action_logger: action_logger.log_round_end(round_num + 1, 0) continue actions = {agent: LLMAction() for _, agent in active_agents} await result.env.step(actions) # Get actual executed actions from Database and log actual_actions, last_rowid = fetch_new_actions_from_db( db_path, last_rowid, agent_names ) round_action_count = 0 for action_data in actual_actions: if action_logger: action_logger.log_action( round_num=round_num + 1, agent_id=action_data['agent_id'], agent_name=action_data['agent_name'], action_type=action_data['action_type'], action_args=action_data['action_args'] ) total_actions += 1 round_action_count += 1 if action_logger: action_logger.log_round_end(round_num + 1, round_action_count) if (round_num + 1) % 20 == 0: progress = (round_num + 1) / total_rounds * 100 log_info(f"Day {simulated_day}, {simulated_hour:02d}:00 - Round {round_num + 1}/{total_rounds} ({progress:.1f}%)") # Note: Do not close environment, keep for Interview use if action_logger: action_logger.log_simulation_end(total_rounds, total_actions) result.total_actions = total_actions elapsed = (datetime.now() - start_time).total_seconds() log_info(f"Simulation loop completed! Time taken: {elapsed:.1f}seconds, Total actions: {total_actions}") return result async def run_reddit_simulation( config: Dict[str, Any], simulation_dir: str, action_logger: Optional[PlatformActionLogger] = None, main_logger: Optional[SimulationLogManager] = None, max_rounds: Optional[int] = None ) -> PlatformSimulation: """Run Reddit simulation Args: config: Simulation configuration simulation_dir: Simulation directory action_logger: Action logger main_logger: Main logger manager max_rounds: Maximum simulation rounds (optional, used to truncate long simulations) Returns: PlatformSimulation: Result object containing env and agent_graph """ result = PlatformSimulation() def log_info(msg): if main_logger: main_logger.info(f"[Reddit] {msg}") print(f"[Reddit] {msg}") log_info("Initializing...") # Reddit use acceleration LLM configuration(if available，otherwise fallback toCommon configuration） model = create_model(config, use_boost=True) profile_path = os.path.join(simulation_dir, "reddit_profiles.json") if not os.path.exists(profile_path): log_info(f"Error: Profile file does not exist: {profile_path}") return result result.agent_graph = await generate_reddit_agent_graph( profile_path=profile_path, model=model, available_actions=REDDIT_ACTIONS, ) # Get Agent real name mapping from config (use entity_name instead of default Agent_X) agent_names = get_agent_names_from_config(config) # If an agent is not in config, use OASIS default name for agent_id, agent in result.agent_graph.get_agents(): if agent_id not in agent_names: agent_names[agent_id] = getattr(agent, 'name', f'Agent_{agent_id}') db_path = os.path.join(simulation_dir, "reddit_simulation.db") if os.path.exists(db_path): os.remove(db_path) result.env = oasis.make( agent_graph=result.agent_graph, platform=oasis.DefaultPlatformType.REDDIT, database_path=db_path, semaphore=30, # Limit maximum concurrent LLM requests to prevent API overload ) await result.env.reset() log_info("Environment started") if action_logger: action_logger.log_simulation_start(config) total_actions = 0 last_rowid = 0 # Track last processed row in Database (use rowid to avoid created_at format differences) # Execute initial events event_config = config.get("event_config", {}) initial_posts = event_config.get("initial_posts", []) # Log round 0 start (initial event phase) if action_logger: action_logger.log_round_start(0, 0) # round 0, simulated_hour 0 initial_action_count = 0 if initial_posts: initial_actions = {} for post in initial_posts: agent_id = post.get("poster_agent_id", 0) content = post.get("content", "") try: agent = result.env.agent_graph.get_agent(agent_id) if agent in initial_actions: if not isinstance(initial_actions[agent], list): initial_actions[agent] = [initial_actions[agent]] initial_actions[agent].append(ManualAction( action_type=ActionType.CREATE_POST, action_args={"content": content} )) else: initial_actions[agent] = ManualAction( action_type=ActionType.CREATE_POST, action_args={"content": content} ) if action_logger: action_logger.log_action( round_num=0, agent_id=agent_id, agent_name=agent_names.get(agent_id, f"Agent_{agent_id}"), action_type="CREATE_POST", action_args={"content": content} ) total_actions += 1 initial_action_count += 1 except Exception: pass if initial_actions: await result.env.step(initial_actions) log_info(f"Published {len(initial_actions)} initial posts") # Log round 0 end if action_logger: action_logger.log_round_end(0, initial_action_count) # Main simulation loop time_config = config.get("time_config", {}) total_hours = time_config.get("total_simulation_hours", 72) minutes_per_round = time_config.get("minutes_per_round", 30) total_rounds = (total_hours * 60) // minutes_per_round # If maximum rounds specified, truncate if max_rounds is not None and max_rounds > 0: original_rounds = total_rounds total_rounds = min(total_rounds, max_rounds) if total_rounds < original_rounds: log_info(f"Rounds truncated: {original_rounds} -> {total_rounds} (max_rounds={max_rounds})") start_time = datetime.now() for round_num in range(total_rounds): # Check if received exit signal if _shutdown_event and _shutdown_event.is_set(): if main_logger: main_logger.info(f"Received exit signal，at round {round_num + 1} stop simulation") break simulated_minutes = round_num * minutes_per_round simulated_hour = (simulated_minutes // 60) % 24 simulated_day = simulated_minutes // (60 * 24) + 1 active_agents = get_active_agents_for_round( result.env, config, simulated_hour, round_num ) # Log round start regardless of active agents if action_logger: action_logger.log_round_start(round_num + 1, simulated_hour) if not active_agents: # Log round end even without active agents (actions_count=0) if action_logger: action_logger.log_round_end(round_num + 1, 0) continue actions = {agent: LLMAction() for _, agent in active_agents} await result.env.step(actions) # Get actual executed actions from Database and log actual_actions, last_rowid = fetch_new_actions_from_db( db_path, last_rowid, agent_names ) round_action_count = 0 for action_data in actual_actions: if action_logger: action_logger.log_action( round_num=round_num + 1, agent_id=action_data['agent_id'], agent_name=action_data['agent_name'], action_type=action_data['action_type'], action_args=action_data['action_args'] ) total_actions += 1 round_action_count += 1 if action_logger: action_logger.log_round_end(round_num + 1, round_action_count) if (round_num + 1) % 20 == 0: progress = (round_num + 1) / total_rounds * 100 log_info(f"Day {simulated_day}, {simulated_hour:02d}:00 - Round {round_num + 1}/{total_rounds} ({progress:.1f}%)") # Note: Do not close environment, keep for Interview use if action_logger: action_logger.log_simulation_end(total_rounds, total_actions) result.total_actions = total_actions elapsed = (datetime.now() - start_time).total_seconds() log_info(f"Simulation loop completed! Time taken: {elapsed:.1f}seconds, Total actions: {total_actions}") return result async def main(): parser = argparse.ArgumentParser(description='OASIS Dual-Platform Parallel Simulation') parser.add_argument( '--config', type=str, required=True, help='Configuration file path (simulation_config.json)' ) parser.add_argument( '--twitter-only', action='store_true', help='Only run Twitter simulation' ) parser.add_argument( '--reddit-only', action='store_true', help='Only run Reddit simulation' ) parser.add_argument( '--max-rounds', type=int, default=None, help='Maximum simulation rounds (optional, used to truncate long simulations)' ) parser.add_argument( '--no-wait', action='store_true', default=False, help='Close environment immediately after simulation completes, do not enter wait mode' ) args = parser.parse_args() # Create shutdown event at the start of main function to ensure the whole program can respond to exit signal global _shutdown_event _shutdown_event = asyncio.Event() if not os.path.exists(args.config): print(f"Error: Configuration file does not exist: {args.config}") sys.exit(1) config = load_config(args.config) simulation_dir = os.path.dirname(args.config) or "." wait_for_commands = not args.no_wait # Initialize logging configuration (disable OASIS logs, clean up old files) init_logging_for_simulation(simulation_dir) # Create log manager log_manager = SimulationLogManager(simulation_dir) twitter_logger = log_manager.get_twitter_logger() reddit_logger = log_manager.get_reddit_logger() log_manager.info("=" * 60) log_manager.info("OASIS dual-platform parallel simulation") log_manager.info(f"Configuration file: {args.config}") log_manager.info(f"Simulation ID: {config.get('simulation_id', 'unknown')}") log_manager.info(f"Wait mode: {'Enabled' if wait_for_commands else 'Disabled'}") log_manager.info("=" * 60) time_config = config.get("time_config", {}) total_hours = time_config.get('total_simulation_hours', 72) minutes_per_round = time_config.get('minutes_per_round', 30) config_total_rounds = (total_hours * 60) // minutes_per_round log_manager.info(f"Simulation parameters:") log_manager.info(f" - Total simulation duration: {total_hours}hours") log_manager.info(f" - Time per round: {minutes_per_round}minutes") log_manager.info(f" - Configured total rounds: {config_total_rounds}") if args.max_rounds: log_manager.info(f" - Maximum rounds limit: {args.max_rounds}") if args.max_rounds < config_total_rounds: log_manager.info(f" - Actual execution rounds: {args.max_rounds} (Truncated)") log_manager.info(f" - Number of Agents: {len(config.get('agent_configs', []))}") log_manager.info("Log structure:") log_manager.info(f" - Main log: simulation.log") log_manager.info(f" - Twitter actions: twitter/actions.jsonl") log_manager.info(f" - Reddit actions: reddit/actions.jsonl") log_manager.info("=" * 60) start_time = datetime.now() # Store simulation results of both platforms twitter_result: Optional[PlatformSimulation] = None reddit_result: Optional[PlatformSimulation] = None if args.twitter_only: twitter_result = await run_twitter_simulation(config, simulation_dir, twitter_logger, log_manager, args.max_rounds) elif args.reddit_only: reddit_result = await run_reddit_simulation(config, simulation_dir, reddit_logger, log_manager, args.max_rounds) else: # Run in parallel (each platform uses independent logger) results = await asyncio.gather( run_twitter_simulation(config, simulation_dir, twitter_logger, log_manager, args.max_rounds), run_reddit_simulation(config, simulation_dir, reddit_logger, log_manager, args.max_rounds), ) twitter_result, reddit_result = results total_elapsed = (datetime.now() - start_time).total_seconds() log_manager.info("=" * 60) log_manager.info(f"Simulation loop completed! Total time: {total_elapsed:.1f}seconds") # Whether to enter wait mode if wait_for_commands: log_manager.info("") log_manager.info("=" * 60) log_manager.info("Enter wait mode - environment keeps running") log_manager.info("Supported commands: interview, batch_interview, close_env") log_manager.info("=" * 60) # Create IPC handler ipc_handler = ParallelIPCHandler( simulation_dir=simulation_dir, twitter_env=twitter_result.env if twitter_result else None, twitter_agent_graph=twitter_result.agent_graph if twitter_result else None, reddit_env=reddit_result.env if reddit_result else None, reddit_agent_graph=reddit_result.agent_graph if reddit_result else None ) ipc_handler.update_status("alive") # Command wait loop (using global _shutdown_event) try: while not _shutdown_event.is_set(): should_continue = await ipc_handler.process_commands() if not should_continue: break # Use wait_for instead of sleep to respond to shutdown_event try: await asyncio.wait_for(_shutdown_event.wait(), timeout=0.5) break # Received exit signal except asyncio.TimeoutError: pass # Timeout continue loop except KeyboardInterrupt: print("\nReceived interrupt signal") except asyncio.CancelledError: print("\nTask was cancelled") except Exception as e: print(f"\nError processing command: {e}") log_manager.info("\nClose environment...") ipc_handler.update_status("stopped") # Close environment if twitter_result and twitter_result.env: await twitter_result.env.close() log_manager.info("[Twitter] Environment closed") if reddit_result and reddit_result.env: await reddit_result.env.close() log_manager.info("[Reddit] Environment closed") log_manager.info("=" * 60) log_manager.info(f"All completed!") log_manager.info(f"Log files:") log_manager.info(f" - {os.path.join(simulation_dir, 'simulation.log')}") log_manager.info(f" - {os.path.join(simulation_dir, 'twitter', 'actions.jsonl')}") log_manager.info(f" - {os.path.join(simulation_dir, 'reddit', 'actions.jsonl')}") log_manager.info("=" * 60) def setup_signal_handlers(loop=None): """ Set signal handlers to ensure proper exit when receiving SIGTERM/SIGINT Persistent simulation scenario：Simulation completeafter does not exit，Wait for interview command When receiving termination signal, need to： 1. Notify asyncio loop to exit wait 2. Give program a chance to clean up resources properly (close database, environment, etc.) 3. Then exit """ def signal_handler(signum, frame): global _cleanup_done sig_name = "SIGTERM" if signum == signal.SIGTERM else "SIGINT" print(f"\nReceived {sig_name} signal, exiting...") if not _cleanup_done: _cleanup_done = True # Set event to notify asyncio loop to exit (give loop a chance to clean up) if _shutdown_event: _shutdown_event.set() # Don't directly sys.exit(), let asyncio loop exit normally and clean up # Force exit only if signal is received repeatedly else: print("Force exit...") sys.exit(1) signal.signal(signal.SIGTERM, signal_handler) signal.signal(signal.SIGINT, signal_handler) if __name__ == "__main__": setup_signal_handlers() try: asyncio.run(main()) except KeyboardInterrupt: print("\nProgram interrupted") except SystemExit: pass finally: # Clean up multiprocessing resource tracker (prevent warning on exit) try: from multiprocessing import resource_tracker resource_tracker._resource_tracker._stop() except Exception: pass print("Simulation process exited")