From b7df4880c8334c9e23683ce947c62dd5c04a54d0 Mon Sep 17 00:00:00 2001 From: liyizhouAI Date: Thu, 16 Apr 2026 22:44:40 +0800 Subject: [PATCH] fix: v0.3.4 - production stability fixes for simulation pipeline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - LLM JSON truncation: add _repair_truncated_json() to handle max_tokens cutoff - Ontology generation: increase max_tokens 4096→8192 to prevent truncation - Config generation: parallel batch processing (3 threads × 30 agents/batch) for ~3x speedup - Rate limit: reduce semaphore 30→8 to avoid GLM API 429 storms - HuggingFace offline: set HF_HUB_OFFLINE=1 to skip unreachable hf-mirror.com - Simulation recovery: auto-reset stuck "preparing" states on Flask restart - Prepare endpoint: handle concurrent prepare requests gracefully - Frontend: handle page-refresh-during-prepare edge case - API: return 429 instead of 500 for LLM rate limit errors --- backend/app/__init__.py | 7 ++ backend/app/api/graph.py | 13 +- backend/app/api/simulation.py | 41 ++++++- backend/app/services/ontology_generator.py | 8 +- .../services/simulation_config_generator.py | 75 ++++++++---- backend/app/services/simulation_manager.py | 39 ++++++ backend/app/utils/llm_client.py | 112 +++++++++++++++++- backend/scripts/run_parallel_simulation.py | 4 + backend/scripts/run_reddit_simulation.py | 2 +- backend/scripts/run_twitter_simulation.py | 2 +- frontend/src/components/Step2EnvSetup.vue | 13 +- locales/en.json | 1 + locales/zh.json | 1 + 13 files changed, 284 insertions(+), 34 deletions(-) diff --git a/backend/app/__init__.py b/backend/app/__init__.py index 96aebdec..443f4a49 100644 --- a/backend/app/__init__.py +++ b/backend/app/__init__.py @@ -47,6 +47,13 @@ def create_app(config_class=Config): SimulationRunner.register_cleanup() if should_log_startup: logger.info("已注册模拟进程清理函数") + + # 启动时恢复卡在 "preparing" 状态的模拟 + from .services.simulation_manager import SimulationManager, SimulationStatus + try: + SimulationManager.recover_stuck_preparations() + except Exception as e: + logger.warning(f"恢复卡住的准备任务失败（非致命）: {e}") # 请求日志中间件 @app.before_request diff --git a/backend/app/api/graph.py b/backend/app/api/graph.py index f55780b1..8028da23 100644 --- a/backend/app/api/graph.py +++ b/backend/app/api/graph.py @@ -250,13 +250,20 @@ def generate_ontology(): "total_text_length": project.total_text_length } }) - + except Exception as e: + # 区分 429 限流和其他错误 + err_str = str(e) + is_rate_limit = any(kw in err_str.lower() for kw in ['429', 'rate limit', '速率限制', '1302', 'too many']) + status_code = 429 if is_rate_limit else 500 + + logger.error(f"本体生成失败: {type(e).__name__}: {err_str[:300]}") return jsonify({ "success": False, - "error": str(e), + "error": "LLM 限流，请等待 1-2 分钟后重试" if is_rate_limit else str(e), + "error_type": "rate_limit" if is_rate_limit else "server_error", "traceback": traceback.format_exc() - }), 500 + }), status_code # ============== 接口2：构建图谱 ============== diff --git a/backend/app/api/simulation.py b/backend/app/api/simulation.py index cff6f832..4de212c8 100644 --- a/backend/app/api/simulation.py +++ b/backend/app/api/simulation.py @@ -430,7 +430,7 @@ def prepare_simulation(): # 检查是否强制重新生成 force_regenerate = data.get('force_regenerate', False) logger.info(f"开始处理 /prepare 请求: simulation_id={simulation_id}, force_regenerate={force_regenerate}") - + # 检查是否已经准备完成（避免重复生成） if not force_regenerate: logger.debug(f"检查模拟 {simulation_id} 是否已准备完成...") @@ -448,8 +448,23 @@ def prepare_simulation(): "prepare_info": prepare_info } }) - else: - logger.info(f"模拟 {simulation_id} 未准备完成，将启动准备任务") + + # 检查是否正在准备中（避免并发准备） + if state.status == SimulationStatus.PREPARING: + logger.info(f"模拟 {simulation_id} 正在准备中，返回进行中状态") + return jsonify({ + "success": True, + "data": { + "simulation_id": simulation_id, + "status": "preparing", + "message": t('api.preparing'), + "already_prepared": False, + "profiles_count": state.profiles_count, + "entities_count": state.entities_count + } + }) + + logger.info(f"模拟 {simulation_id} 未准备完成，将启动准备任务") # 从项目获取必要信息 project = ProjectManager.get_project(state.project_id) @@ -744,10 +759,26 @@ def get_prepare_status(): } }) - # 如果没有task_id，返回错误 + # 如果没有task_id，检查模拟状态 if not task_id: if simulation_id: - # 有simulation_id但未准备完成 + # 检查模拟当前状态 + sim_state = SimulationManager().get_simulation(simulation_id) + if sim_state and sim_state.status == SimulationStatus.PREPARING: + # 模拟正在准备中（可能是页面刷新丢失了task_id） + return jsonify({ + "success": True, + "data": { + "simulation_id": simulation_id, + "status": "preparing", + "progress": -1, # -1 表示进度未知 + "message": t('api.preparing'), + "already_prepared": False, + "profiles_count": sim_state.profiles_count, + "entities_count": sim_state.entities_count + } + }) + # 未准备完成且不在准备中 return jsonify({ "success": True, "data": { diff --git a/backend/app/services/ontology_generator.py b/backend/app/services/ontology_generator.py index b69dbc34..949a1b82 100644 --- a/backend/app/services/ontology_generator.py +++ b/backend/app/services/ontology_generator.py @@ -225,12 +225,16 @@ class OntologyGenerator: {"role": "user", "content": user_message} ] + # max_tokens: 本体 JSON（10 实体 + 10 关系 + 属性）需要较大输出空间 + # Qwen 32B 尤其需要更大余量，避免截断 + ontology_max_tokens = 8192 + # 调用 LLM（主 LLM 重试 5 次全失败后，自动降级到 fallback） try: result = self.llm_client.chat_json( messages=messages, temperature=0.3, - max_tokens=4096 + max_tokens=ontology_max_tokens ) except Exception as primary_err: if not self._fallback_llm: @@ -242,7 +246,7 @@ class OntologyGenerator: result = self._fallback_llm.chat_json( messages=messages, temperature=0.3, - max_tokens=4096 + max_tokens=ontology_max_tokens ) # 验证和后处理 diff --git a/backend/app/services/simulation_config_generator.py b/backend/app/services/simulation_config_generator.py index d9fce5ea..d49b2c01 100644 --- a/backend/app/services/simulation_config_generator.py +++ b/backend/app/services/simulation_config_generator.py @@ -212,8 +212,10 @@ class SimulationConfigGenerator: # 上下文最大字符数 MAX_CONTEXT_LENGTH = 50000 - # 每批生成的Agent数量 - AGENTS_PER_BATCH = 15 + # 每批生成的Agent数量（增大以减少LLM调用次数） + AGENTS_PER_BATCH = 30 + # 并行生成Agent配置的线程数（加速配置生成） + MAX_PARALLEL_BATCHES = 3 # 各步骤的上下文截断长度（字符数） TIME_CONFIG_CONTEXT_LENGTH = 10000 # 时间配置 @@ -305,25 +307,58 @@ class SimulationConfigGenerator: event_config = self._parse_event_config(event_config_result) reasoning_parts.append(f"{t('progress.eventConfigLabel')}: {event_config_result.get('reasoning', t('common.success'))}") - # ========== 步骤3-N: 分批生成Agent配置 ========== + # ========== 步骤3-N: 分批生成Agent配置（并行加速） ========== all_agent_configs = [] - for batch_idx in range(num_batches): - start_idx = batch_idx * self.AGENTS_PER_BATCH - end_idx = min(start_idx + self.AGENTS_PER_BATCH, len(entities)) - batch_entities = entities[start_idx:end_idx] - - report_progress( - 3 + batch_idx, - t('progress.generatingAgentConfig', start=start_idx + 1, end=end_idx, total=len(entities)) - ) - - batch_configs = self._generate_agent_configs_batch( - context=context, - entities=batch_entities, - start_idx=start_idx, - simulation_requirement=simulation_requirement - ) - all_agent_configs.extend(batch_configs) + + if num_batches <= 1 or len(entities) <= self.AGENTS_PER_BATCH: + # 单批次，直接串行 + for batch_idx in range(num_batches): + start_idx = batch_idx * self.AGENTS_PER_BATCH + end_idx = min(start_idx + self.AGENTS_PER_BATCH, len(entities)) + batch_entities = entities[start_idx:end_idx] + report_progress( + 3 + batch_idx, + t('progress.generatingAgentConfig', start=start_idx + 1, end=end_idx, total=len(entities)) + ) + batch_configs = self._generate_agent_configs_batch( + context=context, + entities=batch_entities, + start_idx=start_idx, + simulation_requirement=simulation_requirement + ) + all_agent_configs.extend(batch_configs) + else: + # 多批次并行生成，加速配置生成 + import concurrent.futures + + def _gen_batch(batch_idx): + s = batch_idx * self.AGENTS_PER_BATCH + e = min(s + self.AGENTS_PER_BATCH, len(entities)) + be = entities[s:e] + report_progress( + 3 + batch_idx, + t('progress.generatingAgentConfig', start=s + 1, end=e, total=len(entities)) + ) + return s, self._generate_agent_configs_batch( + context=context, + entities=be, + start_idx=s, + simulation_requirement=simulation_requirement + ) + + # 分组并行：每 MAX_PARALLEL_BATCHES 个一批 + batch_results = {} + for group_start in range(0, num_batches, self.MAX_PARALLEL_BATCHES): + group_end = min(group_start + self.MAX_PARALLEL_BATCHES, num_batches) + with concurrent.futures.ThreadPoolExecutor(max_workers=self.MAX_PARALLEL_BATCHES) as executor: + futures = {executor.submit(_gen_batch, bi): bi for bi in range(group_start, group_end)} + for future in concurrent.futures.as_completed(futures): + idx, configs = future.result() + batch_results[idx] = configs + + # 按 batch 顺序组装结果 + for idx in sorted(batch_results.keys()): + all_agent_configs.extend(batch_results[idx]) reasoning_parts.append(t('progress.agentConfigResult', count=len(all_agent_configs))) diff --git a/backend/app/services/simulation_manager.py b/backend/app/services/simulation_manager.py index c1c2285a..7f27545a 100644 --- a/backend/app/services/simulation_manager.py +++ b/backend/app/services/simulation_manager.py @@ -144,6 +144,45 @@ class SimulationManager: def clear_accelerate(cls, simulation_id: str) -> None: cls._accelerate_flags.pop(simulation_id, None) + @classmethod + def recover_stuck_preparations(cls) -> None: + """启动时恢复卡在 'preparing' 状态的模拟（Flask 重启导致后台线程丢失）""" + logger.info("检查卡住的准备任务...") + data_dir = os.path.join( + os.path.dirname(__file__), + '../../uploads/simulations' + ) + if not os.path.exists(data_dir): + return + + recovered = 0 + for sim_dir_name in os.listdir(data_dir): + sim_dir = os.path.join(data_dir, sim_dir_name) + state_file = os.path.join(sim_dir, "state.json") + if not os.path.isdir(sim_dir) or not os.path.exists(state_file): + continue + + try: + with open(state_file, 'r', encoding='utf-8') as f: + data = json.load(f) + + if data.get("status") == "preparing": + # 卡在 preparing 状态，重置为 created + data["status"] = "created" + data["error"] = "准备任务因服务重启而中断，请重新准备" + data["updated_at"] = datetime.now().isoformat() + with open(state_file, 'w', encoding='utf-8') as f: + json.dump(data, f, ensure_ascii=False, indent=2) + logger.info(f"已重置卡住的准备任务: {sim_dir_name} -> created") + recovered += 1 + except Exception as e: + logger.warning(f"恢复准备任务失败 {sim_dir_name}: {e}") + + if recovered > 0: + logger.info(f"共恢复 {recovered} 个卡住的准备任务") + else: + logger.info("没有卡住的准备任务") + def __init__(self): # 确保目录存在 os.makedirs(self.SIMULATION_DATA_DIR, exist_ok=True) diff --git a/backend/app/utils/llm_client.py b/backend/app/utils/llm_client.py index 9477cbf6..af3ee611 100644 --- a/backend/app/utils/llm_client.py +++ b/backend/app/utils/llm_client.py @@ -163,5 +163,115 @@ class LLMClient: try: return json.loads(cleaned_response) except json.JSONDecodeError: - raise ValueError(f"LLM返回的JSON格式无效: {cleaned_response}") + # 尝试修复截断的 JSON（LLM 输出被 max_tokens 截断时常见） + repaired = self._repair_truncated_json(cleaned_response) + if repaired is not None: + logger.warning("LLM JSON 解析失败，但修复截断后成功") + return repaired + raise ValueError(f"LLM返回的JSON格式无效: {cleaned_response[:500]}") + + @staticmethod + def _repair_truncated_json(text: str) -> Optional[Dict[str, Any]]: + """ + 尝试修复被截断的 JSON： + 1. 找到最内层的完整对象/数组，闭合外层括号 + 2. 移除末尾不完整的键值对/元素 + """ + # 移除尾部不完整的字符串（引号未闭合） + # 找到最后一个完整的值结束位置 + repaired = text.rstrip() + + # 策略：逐步回退到最近的有效 JSON 结构 + # 1. 移除末尾的逗号和空白 + repaired = re.sub(r'[,\s]+$', '', repaired) + + # 2. 如果有未闭合的字符串（奇数引号），截断到最后一个完整值 + # 向前找最后一个完整 key:value 或 数组元素 的结束位置 + for trim_pattern in [ + r',\s*"[^"]*$', # 末尾是不完整的 key + r',\s*"[^"]*":\s*$', # 末尾是 key: 但无值 + r',\s*"[^"]*":\s*"[^"]*$', # 末尾是不完整的字符串值 + r',\s*\{$', # 末尾是空对象开始 + r',\s*\[$', # 末尾是空数组开始 + r',\s*"[^"]*":\s*\[[^\]]*$', # 末尾是不完整的数组 + ]: + new_repaired = re.sub(trim_pattern + r'\s*', '', repaired) + if new_repaired != repaired: + repaired = new_repaired + + # 3. 闭合所有未闭合的括号 + stack = [] + i = 0 + in_string = False + escape_next = False + while i < len(repaired): + ch = repaired[i] + if escape_next: + escape_next = False + elif ch == '\\' and in_string: + escape_next = True + elif ch == '"' and not escape_next: + in_string = not in_string + elif not in_string: + if ch in '{[': + stack.append(ch) + elif ch == '}': + if stack and stack[-1] == '{': + stack.pop() + elif ch == ']': + if stack and stack[-1] == '[': + stack.pop() + i += 1 + + # 闭合剩余的括号（逆序） + for bracket in reversed(stack): + if bracket == '{': + repaired += '}' + elif bracket == '[': + repaired += ']' + + try: + result = json.loads(repaired) + # 只接受 dict 结果（本体生成期望 dict） + if isinstance(result, dict): + return result + except json.JSONDecodeError: + pass + + # 最后尝试：找到第一个完整的顶层 JSON 对象 + depth = 0 + best_end = -1 + in_str = False + esc = False + for idx, ch in enumerate(text): + if esc: + esc = False + continue + if ch == '\\' and in_str: + esc = True + continue + if ch == '"' and not esc: + in_str = not in_str + continue + if in_str: + continue + if ch == '{': + if depth == 0: + best_end = -1 # reset + depth += 1 + elif ch == '}': + depth -= 1 + if depth == 0: + best_end = idx + break # 第一个完整顶层对象 + + if best_end > 0: + try: + candidate = json.loads(text[:best_end + 1]) + if isinstance(candidate, dict): + return candidate + except json.JSONDecodeError: + pass + + return None diff --git a/backend/scripts/run_parallel_simulation.py b/backend/scripts/run_parallel_simulation.py index ef99677f..e816c53c 100644 --- a/backend/scripts/run_parallel_simulation.py +++ b/backend/scripts/run_parallel_simulation.py @@ -32,6 +32,10 @@ OASIS 双平台并行模拟预设脚本 import sys import os +# Force HuggingFace offline mode - model already cached locally +os.environ['HF_HUB_OFFLINE'] = '1' +os.environ['TRANSFORMERS_OFFLINE'] = '1' + if sys.platform == 'win32': # 设置 Python 默认 I/O 编码为 UTF-8 # 这会影响所有未指定编码的 open() 调用 diff --git a/backend/scripts/run_reddit_simulation.py b/backend/scripts/run_reddit_simulation.py index 14907cbd..0f721397 100644 --- a/backend/scripts/run_reddit_simulation.py +++ b/backend/scripts/run_reddit_simulation.py @@ -578,7 +578,7 @@ class RedditSimulationRunner: agent_graph=self.agent_graph, platform=oasis.DefaultPlatformType.REDDIT, database_path=db_path, - semaphore=30, # 限制最大并发 LLM 请求数，防止 API 过载 + semaphore=8, # GLM RPM 有限，8 并发避免 429 疯狂触发 ) await self.env.reset() diff --git a/backend/scripts/run_twitter_simulation.py b/backend/scripts/run_twitter_simulation.py index caab9e9d..bf86b146 100644 --- a/backend/scripts/run_twitter_simulation.py +++ b/backend/scripts/run_twitter_simulation.py @@ -593,7 +593,7 @@ class TwitterSimulationRunner: agent_graph=self.agent_graph, platform=oasis.DefaultPlatformType.TWITTER, database_path=db_path, - semaphore=30, # 限制最大并发 LLM 请求数，防止 API 过载 + semaphore=8, # GLM RPM 有限，8 并发避免 429 疯狂触发 ) await self.env.reset() diff --git a/frontend/src/components/Step2EnvSetup.vue b/frontend/src/components/Step2EnvSetup.vue index adf324cc..8df68caa 100644 --- a/frontend/src/components/Step2EnvSetup.vue +++ b/frontend/src/components/Step2EnvSetup.vue @@ -807,7 +807,18 @@ const startPrepareSimulation = async () => { await loadPreparedData() return } - + + // 模拟正在准备中（页面刷新或并发请求） + if (res.data.status === 'preparing' && !res.data.task_id) { + addLog(t('log.preparingInprogress') || '模拟环境正在准备中，开始监控进度...') + if (res.data.entities_count) { + expectedTotal.value = res.data.entities_count + } + startPolling() + startProfilesPolling() + return + } + taskId.value = res.data.task_id addLog(t('log.prepareTaskStarted')) addLog(t('log.prepareTaskId', { taskId: res.data.task_id })) diff --git a/locales/en.json b/locales/en.json index ee74c4fb..2ac631e6 100644 --- a/locales/en.json +++ b/locales/en.json @@ -349,6 +349,7 @@ "prepareStarted": "Preparation task started. Query progress via /api/simulation/prepare/status.", "alreadyPrepared": "Preparation already complete. No need to regenerate.", "notStartedPrepare": "Preparation not started. Please call /api/simulation/prepare.", + "preparing": "Simulation environment is being prepared, please wait...", "taskCompletedPrepared": "Task completed (preparation already exists)", "requireTaskOrSimId": "Please provide task_id or simulation_id", "configNotFound": "Simulation config not found. Please call /prepare first.", diff --git a/locales/zh.json b/locales/zh.json index 8c5ddc26..9cf04283 100644 --- a/locales/zh.json +++ b/locales/zh.json @@ -349,6 +349,7 @@ "prepareStarted": "准备任务已启动，请通过 /api/simulation/prepare/status 查询进度", "alreadyPrepared": "已有完成的准备工作，无需重复生成", "notStartedPrepare": "尚未开始准备，请调用 /api/simulation/prepare 开始", + "preparing": "模拟环境准备中，请稍候...", "taskCompletedPrepared": "任务已完成（准备工作已存在）", "requireTaskOrSimId": "请提供 task_id 或 simulation_id", "configNotFound": "模拟配置不存在，请先调用 /prepare 接口",