935c75d078
Два корня пустого spend за час: 1) usage-добыча в воркере смотрела m["message"]["usage"] — у strands usage живёт в message["metadata"]["usage"] (event_loop.py), поэтому usage всегда был пуст (тот же баг, что уже чинили в runtime._extract_usage). 2) BudgetSpent писал scheduler ПОСЛЕ успешного возврата воркера — таймаут убивал воркер раньше записи, spend терял весь тик. 3) Модель: strands_harness строит OpenAIResponsesModel, у 9router /responses — шим без usage (проверено: response.completed приходит без usage; non-stream возвращает chat-completion-объект). Перевёл на OpenAIModel (chat/completions, include_usage) — usage приходит и с инструментами, и в reasoning-режиме. Фикс: - воркер пишет BudgetSpent ИНКРЕМЕНТАЛЬНО после каждого модельного вызова (hook на MessageAddedEvent, assistant с metadata.usage, дедуп по объекту) — таймаут теряет максимум хвост текущего вызова, не весь тик; - _extract_usage: корректная структура + суммирование по шагам (model_calls); - scheduler дублирует запись только если воркер не смог разобрать usage; - события помечены source: worker|runner и trace_id тика. Проверено: дельты воркера == итоговый usage тика (39-шаговый тик сходится); убитый по таймауту тик сохраняет все завершённые шаги; spend виден в read model/dashboard.
81 lines
No EOL
4.2 KiB
Python
81 lines
No EOL
4.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")
|
||
# BudgetSpent пишет сам воркер — ИНКРЕМЕНТАЛЬНО, после каждого модельного
|
||
# вызова (см. _llm_worker._wire_budget_hooks). Раньше событие писалось
|
||
# здесь, ПОСЛЕ возврата воркера: таймаут убивал процесс до записи, и
|
||
# spend молча терял весь тик (task t-260929170544-adebea).
|
||
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,
|
||
"level": level_cfg.name, # для записи BudgetSpent самим воркером
|
||
}
|
||
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,
|
||
)
|
||
except subprocess.TimeoutExpired as e:
|
||
# воркер убит по таймауту. Он пишет BudgetSpent сам после каждого
|
||
# модельного вызова, так что spend этого тика уже в сторе —
|
||
# теряется максимум хвост вызова, шедшего на момент убийства.
|
||
# Снаружи дописывать нечего: история воркера недоступна после SIGKILL.
|
||
raise
|
||
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") |