fix: harden runtime API and simulation handling

This commit is contained in:
Hao OUYANG 2026-07-07 22:46:34 +08:00
parent 60757b3c82
commit 82e24ba135
15 changed files with 118 additions and 43 deletions

View File

@ -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}")

View File

@ -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')

View File

@ -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)

View File

@ -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客户端用于检索丰富上下文

View File

@ -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(

View File

@ -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):

View File

@ -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(

View File

@ -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": "环境已经关闭"

View File

@ -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(

View File

@ -33,6 +33,7 @@ dependencies = [
# 工具库
"python-dotenv>=1.0.0",
"pydantic>=2.0.0",
"psutil>=5.9.0",
]
[project.optional-dependencies]

View File

@ -34,3 +34,4 @@ python-dotenv>=1.0.0
# 数据验证
pydantic>=2.0.0
psutil>=5.9.0

View File

@ -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:

View File

@ -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开始模拟循环...")

View File

@ -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开始模拟循环...")

View File

@ -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" },