From 82e24ba1351096438e8657ca2b98e0d4efa603a5 Mon Sep 17 00:00:00 2001 From: Hao OUYANG Date: Tue, 7 Jul 2026 22:46:34 +0800 Subject: [PATCH] fix: harden runtime API and simulation handling --- backend/app/api/graph.py | 2 +- backend/app/api/report.py | 10 ++-- backend/app/api/simulation.py | 30 ++++++------ .../app/services/oasis_profile_generator.py | 3 +- .../services/simulation_config_generator.py | 3 +- backend/app/services/simulation_ipc.py | 2 +- backend/app/services/simulation_manager.py | 2 +- backend/app/services/simulation_runner.py | 36 +++++++++++---- backend/app/utils/llm_client.py | 3 +- backend/pyproject.toml | 1 + backend/requirements.txt | 1 + backend/scripts/run_parallel_simulation.py | 46 +++++++++++++++++-- backend/scripts/run_reddit_simulation.py | 6 ++- backend/scripts/run_twitter_simulation.py | 14 +++++- backend/uv.lock | 2 + 15 files changed, 118 insertions(+), 43 deletions(-) diff --git a/backend/app/api/graph.py b/backend/app/api/graph.py index bcd21eba..a4006106 100644 --- a/backend/app/api/graph.py +++ b/backend/app/api/graph.py @@ -494,7 +494,7 @@ def _build_graph_impl(): }), 500 # 解析请求 - data = request.get_json() or {} + data = request.get_json(silent=True) or {} project_id = data.get('project_id') logger.debug(f"请求参数: project_id={project_id}") diff --git a/backend/app/api/report.py b/backend/app/api/report.py index a1e7dbcf..d1ec6aef 100644 --- a/backend/app/api/report.py +++ b/backend/app/api/report.py @@ -55,7 +55,7 @@ def generate_report(): } """ try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} simulation_id = data.get('simulation_id') if not simulation_id: @@ -342,7 +342,7 @@ def get_generate_status(): } """ try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} task_id = data.get('task_id') simulation_id = data.get('simulation_id') @@ -616,7 +616,7 @@ def chat_with_report_agent(): } """ try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} simulation_id = data.get('simulation_id') message = data.get('message') @@ -1064,7 +1064,7 @@ def search_graph_tool(): } """ try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} graph_id = data.get('graph_id') query = data.get('query') @@ -1110,7 +1110,7 @@ def get_graph_statistics_tool(): } """ try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} graph_id = data.get('graph_id') diff --git a/backend/app/api/simulation.py b/backend/app/api/simulation.py index aae76788..0df3f43c 100644 --- a/backend/app/api/simulation.py +++ b/backend/app/api/simulation.py @@ -222,7 +222,7 @@ def create_simulation(): } """ try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} project_id = data.get('project_id') if not project_id: @@ -321,7 +321,7 @@ def _check_simulation_prepared(simulation_id: str) -> tuple: state_file = os.path.join(simulation_dir, "state.json") try: import json - with open(state_file, 'r', encoding='utf-8') as f: + with open(state_file, 'r', encoding='utf-8-sig') as f: state_data = json.load(f) status = state_data.get("status", "") @@ -433,7 +433,7 @@ def prepare_simulation(): from ..config import Config try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} simulation_id = data.get('simulation_id') if not simulation_id: @@ -705,7 +705,7 @@ def get_prepare_status(): from ..models.task import TaskManager try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} task_id = data.get('task_id') simulation_id = data.get('simulation_id') @@ -1141,7 +1141,7 @@ def get_simulation_profiles_realtime(simulation_id: str): state_file = os.path.join(sim_dir, "state.json") if os.path.exists(state_file): try: - with open(state_file, 'r', encoding='utf-8') as f: + with open(state_file, 'r', encoding='utf-8-sig') as f: state_data = json.load(f) status = state_data.get("status", "") is_generating = status == "preparing" @@ -1243,7 +1243,7 @@ def get_simulation_config_realtime(simulation_id: str): state_file = os.path.join(sim_dir, "state.json") if os.path.exists(state_file): try: - with open(state_file, 'r', encoding='utf-8') as f: + with open(state_file, 'r', encoding='utf-8-sig') as f: state_data = json.load(f) status = state_data.get("status", "") error = state_data.get("error") @@ -1438,7 +1438,7 @@ def generate_profiles(): } """ try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} graph_id = data.get('graph_id') if not graph_id: @@ -1540,7 +1540,7 @@ def start_simulation(): } """ try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} simulation_id = data.get('simulation_id') if not simulation_id: @@ -1806,7 +1806,7 @@ def stop_simulation(): } """ try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} simulation_id = data.get('simulation_id') if not simulation_id: @@ -2353,7 +2353,7 @@ def interview_agent(): } """ try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} simulation_id = data.get('simulation_id') agent_id = data.get('agent_id') @@ -2475,7 +2475,7 @@ def interview_agents_batch(): } """ try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} simulation_id = data.get('simulation_id') interviews = data.get('interviews') @@ -2602,7 +2602,7 @@ def interview_all_agents(): } """ try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} simulation_id = data.get('simulation_id') prompt = data.get('prompt') @@ -2706,7 +2706,7 @@ def get_interview_history(): } """ try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} simulation_id = data.get('simulation_id') platform = data.get('platform') # 不指定则返回两个平台的历史 @@ -2768,7 +2768,7 @@ def get_env_status(): } """ try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} simulation_id = data.get('simulation_id') @@ -2835,7 +2835,7 @@ def close_simulation_env(): } """ try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} simulation_id = data.get('simulation_id') timeout = data.get('timeout', 30) diff --git a/backend/app/services/oasis_profile_generator.py b/backend/app/services/oasis_profile_generator.py index de6ce6e3..53ebf668 100644 --- a/backend/app/services/oasis_profile_generator.py +++ b/backend/app/services/oasis_profile_generator.py @@ -257,7 +257,8 @@ class OasisProfileGenerator: self.client = OpenAI( api_key=self.api_key, - base_url=self.base_url + base_url=self.base_url, + default_headers={"User-Agent": "python-requests/2.32.5"} ) # Zep客户端用于检索丰富上下文 diff --git a/backend/app/services/simulation_config_generator.py b/backend/app/services/simulation_config_generator.py index 2c6fbbf1..31a652f0 100644 --- a/backend/app/services/simulation_config_generator.py +++ b/backend/app/services/simulation_config_generator.py @@ -238,7 +238,8 @@ class SimulationConfigGenerator: self.client = OpenAI( api_key=self.api_key, - base_url=self.base_url + base_url=self.base_url, + default_headers={"User-Agent": "python-requests/2.32.5"} ) def generate_config( diff --git a/backend/app/services/simulation_ipc.py b/backend/app/services/simulation_ipc.py index 9d70d0be..67b8f91d 100644 --- a/backend/app/services/simulation_ipc.py +++ b/backend/app/services/simulation_ipc.py @@ -278,7 +278,7 @@ class SimulationIPCClient: return False try: - with open(status_file, 'r', encoding='utf-8') as f: + with open(status_file, 'r', encoding='utf-8-sig') as f: status = json.load(f) return status.get("status") == "alive" except (json.JSONDecodeError, OSError): diff --git a/backend/app/services/simulation_manager.py b/backend/app/services/simulation_manager.py index 73c32e0e..96e8301b 100644 --- a/backend/app/services/simulation_manager.py +++ b/backend/app/services/simulation_manager.py @@ -178,7 +178,7 @@ class SimulationManager: if not os.path.exists(state_file): return None - with open(state_file, 'r', encoding='utf-8') as f: + with open(state_file, 'r', encoding='utf-8-sig') as f: data = json.load(f) state = SimulationState( diff --git a/backend/app/services/simulation_runner.py b/backend/app/services/simulation_runner.py index b1b99897..b2dd1a8e 100644 --- a/backend/app/services/simulation_runner.py +++ b/backend/app/services/simulation_runner.py @@ -12,6 +12,7 @@ import threading import subprocess import signal import atexit +import psutil from typing import Dict, Any, List, Optional, Union from dataclasses import dataclass, field from datetime import datetime @@ -305,7 +306,7 @@ class SimulationRunner: return None try: - with open(state_file, 'r', encoding='utf-8') as f: + with open(state_file, 'r', encoding='utf-8-sig') as f: data = json.load(f) state = SimulationRunState( @@ -1652,7 +1653,23 @@ class SimulationRunner: return False ipc_client = SimulationIPCClient(sim_dir) - return ipc_client.check_env_alive() + if not ipc_client.check_env_alive(): + return False + + process = cls._processes.get(simulation_id) + if process is not None: + return process.poll() is None + + state = cls.get_run_state(simulation_id) + pid = state.process_pid if state else None + if not pid: + return False + + try: + proc = psutil.Process(pid) + return proc.is_running() and proc.status() != psutil.STATUS_ZOMBIE + except psutil.Error: + return False @classmethod def get_env_status_detail(cls, simulation_id: str) -> Dict[str, Any]: @@ -1679,12 +1696,13 @@ class SimulationRunner: return default_status try: - with open(status_file, 'r', encoding='utf-8') as f: + with open(status_file, 'r', encoding='utf-8-sig') as f: status = json.load(f) + env_alive = cls.check_env_alive(simulation_id) return { - "status": status.get("status", "stopped"), - "twitter_available": status.get("twitter_available", False), - "reddit_available": status.get("reddit_available", False), + "status": status.get("status", "stopped") if env_alive else "stopped", + "twitter_available": status.get("twitter_available", False) if env_alive else False, + "reddit_available": status.get("reddit_available", False) if env_alive else False, "timestamp": status.get("timestamp") } except (json.JSONDecodeError, OSError): @@ -1725,7 +1743,7 @@ class SimulationRunner: ipc_client = SimulationIPCClient(sim_dir) - if not ipc_client.check_env_alive(): + if not cls.check_env_alive(simulation_id): raise ValueError(f"模拟环境未运行或已关闭，无法执行Interview: {simulation_id}") logger.info(f"发送Interview命令: simulation_id={simulation_id}, agent_id={agent_id}, platform={platform}") @@ -1787,7 +1805,7 @@ class SimulationRunner: ipc_client = SimulationIPCClient(sim_dir) - if not ipc_client.check_env_alive(): + if not cls.check_env_alive(simulation_id): raise ValueError(f"模拟环境未运行或已关闭，无法执行Interview: {simulation_id}") logger.info(f"发送批量Interview命令: simulation_id={simulation_id}, count={len(interviews)}, platform={platform}") @@ -1897,7 +1915,7 @@ class SimulationRunner: ipc_client = SimulationIPCClient(sim_dir) - if not ipc_client.check_env_alive(): + if not cls.check_env_alive(simulation_id): return { "success": True, "message": "环境已经关闭" diff --git a/backend/app/utils/llm_client.py b/backend/app/utils/llm_client.py index 9d85259b..80c17539 100644 --- a/backend/app/utils/llm_client.py +++ b/backend/app/utils/llm_client.py @@ -106,7 +106,8 @@ class LLMClient: self.client = OpenAI( api_key=self.api_key, - base_url=self.base_url + base_url=self.base_url, + default_headers={"User-Agent": "python-requests/2.32.5"} ) def _create_completion( diff --git a/backend/pyproject.toml b/backend/pyproject.toml index f972e120..50ed9bf4 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -33,6 +33,7 @@ dependencies = [ # 工具库 "python-dotenv>=1.0.0", "pydantic>=2.0.0", + "psutil>=5.9.0", ] [project.optional-dependencies] diff --git a/backend/requirements.txt b/backend/requirements.txt index 8fe1327e..ea39c346 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -34,3 +34,4 @@ python-dotenv>=1.0.0 # 数据验证 pydantic>=2.0.0 +psutil>=5.9.0 diff --git a/backend/scripts/run_parallel_simulation.py b/backend/scripts/run_parallel_simulation.py index 2a627ffd..9ec72f72 100644 --- a/backend/scripts/run_parallel_simulation.py +++ b/backend/scripts/run_parallel_simulation.py @@ -678,8 +678,9 @@ def fetch_new_actions_from_db( if not os.path.exists(db_path): return actions, new_last_rowid + conn = None try: - conn = sqlite3.connect(db_path) + conn = sqlite3.connect(db_path, timeout=5) cursor = conn.cursor() # 使用 rowid 来追踪已处理的记录（rowid 是 SQLite 的内置自增字段） @@ -739,13 +740,35 @@ def fetch_new_actions_from_db( 'action_args': simplified_args, }) - conn.close() except Exception as e: print(f"读取数据库动作失败: {e}") + finally: + if conn is not None: + conn.close() return actions, new_last_rowid +def get_latest_trace_rowid(db_path: str) -> int: + """Return the latest processed trace row id for a platform DB.""" + if not os.path.exists(db_path): + return 0 + + conn = None + try: + conn = sqlite3.connect(db_path, timeout=5) + cursor = conn.cursor() + cursor.execute("SELECT COALESCE(MAX(rowid), 0) FROM trace") + rowid = int(cursor.fetchone()[0] or 0) + return rowid + except Exception as e: + print(f"读取最新动作游标失败: {e}") + return 0 + finally: + if conn is not None: + conn.close() + + def _enrich_action_context( cursor, action_type: str, @@ -1034,6 +1057,9 @@ def create_model(config: Dict[str, Any], use_boost: bool = False): return ModelFactory.create( model_platform=ModelPlatformType.OPENAI, model_type=llm_model, + api_key=llm_api_key or None, + url=llm_base_url or None, + default_headers={"User-Agent": "python-requests/2.32.5"}, ) @@ -1184,10 +1210,16 @@ async def run_twitter_simulation( content = post.get("content", "") try: agent = result.env.agent_graph.get_agent(agent_id) - initial_actions[agent] = ManualAction( + manual_action = ManualAction( action_type=ActionType.CREATE_POST, action_args={"content": content} ) + if agent in initial_actions: + if not isinstance(initial_actions[agent], list): + initial_actions[agent] = [initial_actions[agent]] + initial_actions[agent].append(manual_action) + else: + initial_actions[agent] = manual_action if action_logger: action_logger.log_action( @@ -1204,7 +1236,9 @@ async def run_twitter_simulation( if initial_actions: await result.env.step(initial_actions) - log_info(f"已发布 {len(initial_actions)} 条初始帖子") + last_rowid = get_latest_trace_rowid(db_path) + posted_count = sum(len(action) if isinstance(action, list) else 1 for action in initial_actions.values()) + log_info(f"已发布 {posted_count} 条初始帖子") # 记录 round 0 结束 if action_logger: @@ -1403,7 +1437,9 @@ async def run_reddit_simulation( if initial_actions: await result.env.step(initial_actions) - log_info(f"已发布 {len(initial_actions)} 条初始帖子") + last_rowid = get_latest_trace_rowid(db_path) + posted_count = sum(len(action) if isinstance(action, list) else 1 for action in initial_actions.values()) + log_info(f"已发布 {posted_count} 条初始帖子") # 记录 round 0 结束 if action_logger: diff --git a/backend/scripts/run_reddit_simulation.py b/backend/scripts/run_reddit_simulation.py index 14907cbd..4df88b02 100644 --- a/backend/scripts/run_reddit_simulation.py +++ b/backend/scripts/run_reddit_simulation.py @@ -464,6 +464,9 @@ class RedditSimulationRunner: return ModelFactory.create( model_platform=ModelPlatformType.OPENAI, model_type=llm_model, + api_key=llm_api_key or None, + url=llm_base_url or None, + default_headers={"User-Agent": "python-requests/2.32.5"}, ) def _get_active_agents_for_round( @@ -617,7 +620,8 @@ class RedditSimulationRunner: if initial_actions: await self.env.step(initial_actions) - print(f" 已发布 {len(initial_actions)} 条初始帖子") + posted_count = sum(len(action) if isinstance(action, list) else 1 for action in initial_actions.values()) + print(f" 已发布 {posted_count} 条初始帖子") # 主模拟循环 print("\n开始模拟循环...") diff --git a/backend/scripts/run_twitter_simulation.py b/backend/scripts/run_twitter_simulation.py index caab9e9d..7cb55a87 100644 --- a/backend/scripts/run_twitter_simulation.py +++ b/backend/scripts/run_twitter_simulation.py @@ -457,6 +457,9 @@ class TwitterSimulationRunner: return ModelFactory.create( model_platform=ModelPlatformType.OPENAI, model_type=llm_model, + api_key=llm_api_key or None, + url=llm_base_url or None, + default_headers={"User-Agent": "python-requests/2.32.5"}, ) def _get_active_agents_for_round( @@ -615,16 +618,23 @@ class TwitterSimulationRunner: content = post.get("content", "") try: agent = self.env.agent_graph.get_agent(agent_id) - initial_actions[agent] = ManualAction( + manual_action = ManualAction( action_type=ActionType.CREATE_POST, action_args={"content": content} ) + if agent in initial_actions: + if not isinstance(initial_actions[agent], list): + initial_actions[agent] = [initial_actions[agent]] + initial_actions[agent].append(manual_action) + else: + initial_actions[agent] = manual_action except Exception as e: print(f" 警告: 无法为Agent {agent_id}创建初始帖子: {e}") if initial_actions: await self.env.step(initial_actions) - print(f" 已发布 {len(initial_actions)} 条初始帖子") + posted_count = sum(len(action) if isinstance(action, list) else 1 for action in initial_actions.values()) + print(f" 已发布 {posted_count} 条初始帖子") # 主模拟循环 print("\n开始模拟循环...") diff --git a/backend/uv.lock b/backend/uv.lock index 7f606a13..42ecc6e0 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -996,6 +996,7 @@ dependencies = [ { name = "flask-cors" }, { name = "httpx" }, { name = "openai" }, + { name = "psutil" }, { name = "pydantic" }, { name = "pymupdf" }, { name = "python-dotenv" }, @@ -1026,6 +1027,7 @@ requires-dist = [ { name = "httpx", specifier = ">=0.27.0" }, { name = "openai", specifier = ">=1.0.0" }, { name = "pipreqs", marker = "extra == 'dev'", specifier = ">=0.5.0" }, + { name = "psutil", specifier = ">=5.9.0" }, { name = "pydantic", specifier = ">=2.0.0" }, { name = "pymupdf", specifier = ">=1.24.0" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.0.0" },