fix: v0.3.4 - production stability fixes for simulation pipeline
- 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
This commit is contained in:
liyizhouAI
parent
3bdaeaf5e7
commit
b7df4880c8
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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：构建图谱 ==============
|
||||
|
|
|
|||
|
|
@ -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": {
|
||||
|
|
|
|||
|
|
@ -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
|
||||
)
|
||||
|
||||
# 验证和后处理
|
||||
|
|
|
|||
|
|
@ -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)))
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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() 调用
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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 }))
|
||||
|
|
|
|||
|
|
@ -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.",
|
||||
|
|
|
|||
|
|
@ -349,6 +349,7 @@
|
|||
"prepareStarted": "准备任务已启动，请通过 /api/simulation/prepare/status 查询进度",
|
||||
"alreadyPrepared": "已有完成的准备工作，无需重复生成",
|
||||
"notStartedPrepare": "尚未开始准备，请调用 /api/simulation/prepare 开始",
|
||||
"preparing": "模拟环境准备中，请稍候...",
|
||||
"taskCompletedPrepared": "任务已完成（准备工作已存在）",
|
||||
"requireTaskOrSimId": "请提供 task_id 或 simulation_id",
|
||||
"configNotFound": "模拟配置不存在，请先调用 /prepare 接口",
|
||||
|
|
|
|||
Loading…
Reference in New Issue