72 lines
No EOL
3.2 KiB
Python
72 lines
No EOL
3.2 KiB
Python
"""Simbiot Harness — изолированный раннер LLM-тика.
|
||
|
||
Strands держит глобальный asyncio-рантайм: несколько create_harness()
|
||
в одном процессе валит его («cannot schedule new futures after shutdown»).
|
||
Решение — каждый LLM-тик в ОТДЕЛЬНОМ сабпроцессе: чистый asyncio-рантайм
|
||
на тик, полная изоляция, ноль утечек. Демон живёт вечно.
|
||
"""
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
|
||
_HARNESS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
_VENV_PY = os.path.join(_HARNESS_DIR, "venv", "bin", "python")
|
||
_WORKER = os.path.join(_HARNESS_DIR, "harness", "_llm_worker.py")
|
||
MARKER = "@@SIMBIOT_RESULT@@"
|
||
|
||
|
||
def run_llm_tick(level_cfg, prompt: str, session_id=None, trace_prefix=None, timeout=None):
|
||
"""Запускает LLM-тик в сабпроцессе. Возвращает (text, usage, trace_id).
|
||
|
||
События трассы воркер пишет в стор сам (у него есть es).
|
||
Протокол: воркер пишет результат в JSON-ФАЙЛ (путь передан в env);
|
||
stdout/stderr только для лога — библиотечный шум strands не мешает.
|
||
"""
|
||
import tempfile
|
||
result_path = tempfile.mktemp(prefix="simbiot_tick_", suffix=".json", dir="/tmp")
|
||
payload = {
|
||
"model": level_cfg.model,
|
||
"effort": level_cfg.effort,
|
||
"instructions": level_cfg.mandate,
|
||
"prompt": prompt,
|
||
"session_id": None, # сессии пока выключены: .agent/sessions растёт без очистки
|
||
"trace_prefix": trace_prefix,
|
||
"can_shell": level_cfg.can_shell,
|
||
"ctx_tokens": level_cfg.ctx_tokens,
|
||
"result_path": result_path,
|
||
}
|
||
env = dict(os.environ)
|
||
env["SIMBIOT_DB"] = env.get("SIMBIOT_DB", os.path.join(_HARNESS_DIR, "harness", "harness.db"))
|
||
if timeout is None:
|
||
# таймаут уровня: тик может быть длиннее интервала — занятый уровень
|
||
# просто поглощает такты (TickAbsorbed), это норма архитектуры
|
||
timeout = getattr(level_cfg, "tick_timeout", 300)
|
||
timeout = max(timeout, int(level_cfg.every * 0.8))
|
||
try:
|
||
proc = subprocess.run(
|
||
[_VENV_PY, _WORKER],
|
||
input=json.dumps(payload).encode(),
|
||
capture_output=True,
|
||
timeout=timeout,
|
||
cwd=_HARNESS_DIR,
|
||
env=env,
|
||
)
|
||
finally:
|
||
pass
|
||
out = None
|
||
if os.path.exists(result_path):
|
||
try:
|
||
with open(result_path) as f:
|
||
out = json.load(f)
|
||
except ValueError:
|
||
out = None
|
||
try:
|
||
os.unlink(result_path)
|
||
except OSError:
|
||
pass
|
||
if out is None:
|
||
return None, {"error": f"worker result file missing: stdout={(proc.stdout or b'')[:200]!r} stderr={(proc.stderr or b'').decode(errors='replace')[-300:]}"}, None
|
||
if not out.get("ok"):
|
||
return None, {"error": (out.get("error") or "unknown")[:500]}, None
|
||
return out.get("text", ""), out.get("usage", {}), out.get("trace_id") |