yumoqing/pipeline-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 全局并发agent数限制(max_concurrent_agents params表)+机构工作目录workspace_base动态化(_is_safe_workdir读params表)

Browse Source
This commit is contained in:
ymq 2026-08-15 10:00:03 +08:00
parent 2cc259bc7d
commit 82b0944424
3 changed files with 80 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
pipeline_service/agent_loop.py
View File

@ -120,7 +120,7 @@ def _resolve_workspace(workspace_dir):
# ── Agent 工具函数 ──
def _is_safe_workdir(workdir):
"""检查目录是否在允许范围内。"""
"""检查目录是否在允许范围内（同步版，仅用模块级白名单兜底）。"""
wd = os.path.abspath(workdir)
for allowed in _ALLOWED_WORKDIRS:
awd = os.path.abspath(os.path.expanduser(allowed))
@ -129,10 +129,47 @@ def _is_safe_workdir(workdir):
return False
_allowed_workdirs_cache = None
_allowed_workdirs_cache_time = 0
async def _get_allowed_workdirs():
"""动态读 workspace_base 参数（params 表），合并到允许目录列表（60s 缓存）。"""
global _allowed_workdirs_cache, _allowed_workdirs_cache_time
import time as _t
now = _t.time()
if _allowed_workdirs_cache is not None and (now - _allowed_workdirs_cache_time) < 60:
return _allowed_workdirs_cache
allowed = list(_ALLOWED_WORKDIRS)
try:
db = _get_db()
async with db.sqlorContext("pipeline") as sor:
from .workspace import get_workspace_base
base = await get_workspace_base(sor)
if base and base not in allowed:
allowed.append(base)
except Exception:
pass
_allowed_workdirs_cache = allowed
_allowed_workdirs_cache_time = now
return allowed
async def _is_safe_workdir_async(workdir):
"""检查目录是否在允许范围内（含 params 表动态 workspace_base）。"""
wd = os.path.abspath(workdir)
allowed = await _get_allowed_workdirs()
for a in allowed:
awd = os.path.abspath(os.path.expanduser(a))
if wd.startswith(awd):
return True
return False
async def _run_shell(command, workdir, timeout=120):
"""安全执行 shell 命令。返回 {"rc": int, "stdout": str, "stderr": str}"""
cwd = os.path.abspath(workdir) if workdir else os.path.expanduser('~/pipeline_ws')
if not _is_safe_workdir(cwd):
if not await _is_safe_workdir_async(cwd):
return {"rc": -1, "stdout": "", "stderr": f"安全限制：目录 {cwd} 不在允许范围"}
if not os.path.isdir(cwd):
return {"rc": -1, "stdout": "", "stderr": f"目录不存在: {cwd}"}
@ -590,7 +627,7 @@ async def _exec_agent_tool(tool, params, workspace_dir):
path = p.get('path', '')
if not path: return 'FAIL: 需要文件路径'
full = os.path.join(workspace_dir, path)
if not _is_safe_workdir(full): return 'FAIL: 路径不在允许范围'
if not await _is_safe_workdir_async(full): return 'FAIL: 路径不在允许范围'
if not os.path.isfile(full): return f'FAIL: 文件不存在 {path}'
with open(full, encoding='utf-8') as f:
return f.read()[:30000]
@ -599,7 +636,7 @@ async def _exec_agent_tool(tool, params, workspace_dir):
content = p.get('content', '')
if not path: return 'FAIL: 需要文件路径'
full = os.path.join(workspace_dir, path)
if not _is_safe_workdir(full): return 'FAIL: 路径不在允许范围'
if not await _is_safe_workdir_async(full): return 'FAIL: 路径不在允许范围'
os.makedirs(os.path.dirname(full), exist_ok=True)
with open(full, 'w', encoding='utf-8') as f:
f.write(content)
@ -607,7 +644,7 @@ async def _exec_agent_tool(tool, params, workspace_dir):
elif tool == 'list_files':
path = p.get('path', '') or '.'
full = os.path.join(workspace_dir, path)
if not _is_safe_workdir(full): return 'FAIL: 路径不在允许范围'
if not await _is_safe_workdir_async(full): return 'FAIL: 路径不在允许范围'
if not os.path.isdir(full): return f'FAIL: 目录不存在 {path}'
items = os.listdir(full)[:50]
lines = []

18
pipeline_service/init.py
View File

@ -555,12 +555,26 @@ def load_pipeline_service():
"UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL, updated_at=NOW() "
"WHERE state='running' AND pipeline_id='role_task' "
"AND updated_at < (NOW() - INTERVAL 20 MINUTE)", {})
# 全局并发 agent 数限制（max_concurrent_agents，appbase params 表，默认 3）。
# 以 DB state='running' 计数为准（权威），超出上限时本轮回合不再派发。
from .workspace import get_max_concurrent_agents
max_n = await get_max_concurrent_agents(sor)
running_rows = await sor.sqlExe(
"SELECT COUNT(*) as c FROM pipeline_tasks "
"WHERE state='running' AND pipeline_id='role_task'", {})
cur = getattr(running_rows[0], 'c', 0) if running_rows else 0
avail = max(0, max_n - cur)
if avail <= 0:
# 达到并发上限，本轮不派发（退出 context 后末尾 sleep 再 poll）
continue
recs = await sor.sqlExe(
"SELECT id, tenant_id, role FROM pipeline_tasks "
"WHERE state='submitted' AND pipeline_id='role_task' "
"AND claimed_by IS NULL ORDER BY created_at ASC LIMIT 5",
"AND claimed_by IS NULL ORDER BY created_at ASC LIMIT 20",
{})
for rec in (recs or []):
for rec in (recs or [])[:avail]:
tid = getattr(rec, 'id', '')
pid = getattr(rec, 'tenant_id', '')
r = getattr(rec, 'role', '')

30
pipeline_service/workspace.py
View File

@ -33,17 +33,31 @@ def _get_file_icon(ext):
return "📁"
async def get_workspace_base(sor):
"""读 workspace_base 参数（appbase params 表，表不存在时兜底 WORKSPACE_BASE）。"""
workspace_base = WORKSPACE_BASE
async def get_param(sor, name, default=""):
"""读 appbase params 表配置（params_name → params_value），表不存在/无值时兑底 default。"""
try:
_wbr = await sor.sqlExe(
"SELECT params_value FROM params WHERE params_name='workspace_base' LIMIT 1", {})
if _wbr and getattr(_wbr[0], 'params_value', ''):
workspace_base = getattr(_wbr[0], 'params_value', '')
recs = await sor.sqlExe(
"SELECT params_value FROM params WHERE params_name=${n}$ LIMIT 1", {"n": name})
if recs and getattr(recs[0], 'params_value', '') not in (None, ''):
return getattr(recs[0], 'params_value', '')
except Exception:
pass
return workspace_base
return default
async def get_workspace_base(sor):
"""读 workspace_base 参数（appbase params 表，表不存在时兜底 WORKSPACE_BASE）。"""
return await get_param(sor, 'workspace_base', WORKSPACE_BASE)
async def get_max_concurrent_agents(sor):
"""读全局并发 agent 数限制（appbase params 表 max_concurrent_agents，默认 3）。"""
val = await get_param(sor, 'max_concurrent_agents', '3')
try:
n = int(float(str(val)))
return max(1, n)
except (ValueError, TypeError):
return 3
async def get_workspace_dir(sor, uid):