fix(spend): BudgetSpent пишется воркером инкрементально + живой usage из 9router (t-260929170544-adebea)
Два корня пустого 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.
This commit is contained in:
Simbiot Harness
parent
c68335e7fb
commit
935c75d078
4 changed files with 189 additions and 24 deletions
|
|
@ -12,6 +12,26 @@ import sys
|
|||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
|
||||
def build_level_model_from_payload(payload: dict):
|
||||
"""Модель для воркера — та же сборка, что и в runtime.build_level_model:
|
||||
9router понимает chat/completions (usage там же), Responses API — шим без usage."""
|
||||
from strands.models.openai import OpenAIModel
|
||||
import os
|
||||
params = {}
|
||||
effort = payload.get("effort") or "auto"
|
||||
if effort and effort != "auto":
|
||||
params["reasoning_effort"] = effort
|
||||
model_id = payload.get("model") or "auto"
|
||||
if model_id == "none":
|
||||
model_id = "auto"
|
||||
return OpenAIModel(
|
||||
client_args={"api_key": os.environ.get("OPENAI_API_KEY", ""),
|
||||
"base_url": os.environ.get("OPENAI_BASE_URL", "http://127.0.0.1:20128/v1")},
|
||||
model_id=model_id,
|
||||
**({"params": params} if params else {}),
|
||||
)
|
||||
|
||||
|
||||
def _emit(payload, obj):
|
||||
rp = payload.get("result_path")
|
||||
if rp:
|
||||
|
|
@ -22,6 +42,90 @@ def _emit(payload, obj):
|
|||
pass
|
||||
|
||||
|
||||
def _extract_usage(agent) -> dict:
|
||||
"""Usage из истории сообщений агента.
|
||||
|
||||
Strands прикрепляет usage к каждой assistant-сообщении в
|
||||
message["metadata"]["usage"] (event_loop.py) — по одному словарю
|
||||
на модельный вызов (шаг тика). Накапливаем по шагам: тик может
|
||||
состоять из многих вызовов, бюджет интересует суммарный.
|
||||
|
||||
Старый разбор m["message"]["usage"] не находил ничего (структура
|
||||
другая) — usage всегда возвращался пустым, spend не рос.
|
||||
"""
|
||||
usage = {"prompt_tokens": 0, "completion_tokens": 0, "model_calls": 0}
|
||||
try:
|
||||
msgs = getattr(agent, "messages", None) or []
|
||||
for m in msgs:
|
||||
u = ((m or {}).get("metadata") or {}).get("usage") or {}
|
||||
p = u.get("inputTokens") or 0
|
||||
c = u.get("outputTokens") or 0
|
||||
if isinstance(p, (int, float)):
|
||||
usage["prompt_tokens"] += p
|
||||
if isinstance(c, (int, float)):
|
||||
usage["completion_tokens"] += c
|
||||
if p or c:
|
||||
usage["model_calls"] += 1
|
||||
except Exception:
|
||||
pass
|
||||
if not usage["model_calls"]:
|
||||
return {}
|
||||
return usage
|
||||
|
||||
|
||||
def _emit_budget_incremental(level: str, trace_id: str, usage_delta: dict):
|
||||
"""Записать BudgetSpent за один модельный вызов.
|
||||
|
||||
Воркер — единственный автор spend-событий, но scheduler может убить
|
||||
его по таймауту в любой момент (SIGKILL). Поэтому usage пишем
|
||||
ИНКРЕМЕНТАЛЬНО: каждый завершённый модельный вызов — своё событие.
|
||||
Хвост недописанного вызова теряется, но весь тик — никогда.
|
||||
"""
|
||||
if not usage_delta:
|
||||
return
|
||||
try:
|
||||
from harness import es
|
||||
es.append_event("harness", "budget", "BudgetSpent", {
|
||||
"level": level, "trace_id": trace_id, "source": "worker",
|
||||
**usage_delta,
|
||||
})
|
||||
except Exception:
|
||||
pass # трекинг не должен валить тик
|
||||
|
||||
|
||||
def _wire_budget_hooks(agent, level: str, trace_id: str):
|
||||
"""После каждого модельного вызова — usage-дельта в стор.
|
||||
|
||||
Strands вешает usage на assistant-сообщение в
|
||||
message["metadata"]["usage"] (одна запись на модельный вызов).
|
||||
MessageAddedEvent стреляет на добавление КАЖДОЙ записи истории,
|
||||
нас интересуют только assistant с usage.
|
||||
"""
|
||||
from strands.hooks import MessageAddedEvent
|
||||
|
||||
seen = set()
|
||||
|
||||
def on_message_added(event):
|
||||
msg = getattr(event, "message", None) or {}
|
||||
if msg.get("role") != "assistant":
|
||||
return
|
||||
usage = ((msg.get("metadata") or {}).get("usage")) or {}
|
||||
p = usage.get("inputTokens") or 0
|
||||
c = usage.get("outputTokens") or 0
|
||||
if not (p or c):
|
||||
return
|
||||
key = id(msg)
|
||||
if key in seen: # MessageUpdatedEvent может переигрывать — дедуп по объекту
|
||||
return
|
||||
seen.add(key)
|
||||
_emit_budget_incremental(level, trace_id, {
|
||||
"prompt_tokens": p, "completion_tokens": c,
|
||||
})
|
||||
|
||||
agent.add_hook(on_message_added, MessageAddedEvent)
|
||||
return agent
|
||||
|
||||
|
||||
def main():
|
||||
payload = json.loads(sys.stdin.read())
|
||||
|
||||
|
|
@ -32,10 +136,11 @@ def main():
|
|||
from strands_harness import create_harness
|
||||
|
||||
trace_id = payload.get("trace_prefix") or "tick"
|
||||
level = payload.get("level") or "?"
|
||||
try:
|
||||
agent = create_harness(
|
||||
model=f"openai/{payload['model']}",
|
||||
effort=payload.get("effort", "auto"),
|
||||
model=build_level_model_from_payload(payload),
|
||||
effort="auto", # effort уже в параметрах модели
|
||||
instructions=payload.get("instructions", ""),
|
||||
session=None,
|
||||
builtin_tools=("read", "write", "edit", "shell") if payload.get("can_shell") else [],
|
||||
|
|
@ -43,20 +148,9 @@ def main():
|
|||
skills=False,
|
||||
)
|
||||
agent = _wire_trace_hooks(agent, es, trace_id)
|
||||
agent = _wire_budget_hooks(agent, level, trace_id)
|
||||
result = agent(payload.get("prompt", ""))
|
||||
usage = {}
|
||||
try:
|
||||
msgs = getattr(agent, "messages", None) or []
|
||||
for m in reversed(msgs):
|
||||
u = (m or {}).get("message", {}).get("usage")
|
||||
if u:
|
||||
usage = {
|
||||
"prompt_tokens": u.get("inputTokens") or u.get("prompt_tokens"),
|
||||
"completion_tokens": u.get("outputTokens") or u.get("completion_tokens"),
|
||||
}
|
||||
break
|
||||
except Exception:
|
||||
pass
|
||||
usage = _extract_usage(agent)
|
||||
_emit(payload, {"ok": True, "text": str(result), "usage": usage, "trace_id": trace_id})
|
||||
except Exception as e:
|
||||
import traceback
|
||||
|
|
|
|||
|
|
@ -25,6 +25,10 @@ def run_llm_tick(level_cfg, prompt: str, session_id=None, trace_prefix=None, tim
|
|||
"""
|
||||
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,
|
||||
|
|
@ -35,6 +39,7 @@ def run_llm_tick(level_cfg, prompt: str, session_id=None, trace_prefix=None, tim
|
|||
"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"))
|
||||
|
|
@ -52,8 +57,12 @@ def run_llm_tick(level_cfg, prompt: str, session_id=None, trace_prefix=None, tim
|
|||
cwd=_HARNESS_DIR,
|
||||
env=env,
|
||||
)
|
||||
finally:
|
||||
pass
|
||||
except subprocess.TimeoutExpired as e:
|
||||
# воркер убит по таймауту. Он пишет BudgetSpent сам после каждого
|
||||
# модельного вызова, так что spend этого тика уже в сторе —
|
||||
# теряется максимум хвост вызова, шедшего на момент убийства.
|
||||
# Снаружи дописывать нечего: история воркера недоступна после SIGKILL.
|
||||
raise
|
||||
out = None
|
||||
if os.path.exists(result_path):
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -23,13 +23,42 @@ def _router_key():
|
|||
return env
|
||||
|
||||
|
||||
def _router_base_url():
|
||||
return os.environ.get("ROUTER_BASE_URL", "http://127.0.0.1:20128/v1")
|
||||
|
||||
|
||||
def build_level_model(level_cfg):
|
||||
"""Модель уровня для 9router: CHAT COMPLETIONS, не Responses API.
|
||||
|
||||
strands_harness для провайдера openai строит OpenAIResponsesModel,
|
||||
но у 9router /responses — шим: usage в response.completed отсутствует
|
||||
(проверено: usage:{} и в стриме, и в non-stream). Токены видны только
|
||||
в /chat/completions (OpenAIModel, stream_options include_usage) —
|
||||
а без них BudgetSpent бесполезен. Проверено: OpenAIModel + 9router
|
||||
отдаёт usage и с инструментами, и в reasoning-режиме.
|
||||
|
||||
Сессии выключены, поэтому server-side state Responses API не нужен;
|
||||
chat/completions stateless и полностью покрывает потребности тика.
|
||||
"""
|
||||
from strands.models.openai import OpenAIModel
|
||||
params = {}
|
||||
if level_cfg.effort and level_cfg.effort != "auto":
|
||||
params["reasoning_effort"] = level_cfg.effort
|
||||
return OpenAIModel(
|
||||
client_args={"api_key": os.environ.get("OPENAI_API_KEY", ""),
|
||||
"base_url": os.environ.get("OPENAI_BASE_URL", "http://127.0.0.1:20128/v1")},
|
||||
model_id=level_cfg.model if level_cfg.model != "none" else "auto",
|
||||
**({"params": params} if params else {}),
|
||||
)
|
||||
|
||||
|
||||
def setup_provider():
|
||||
"""9router как openai-совместимый провайдер strands."""
|
||||
key = _router_key()
|
||||
if not key:
|
||||
raise RuntimeError("ROUTER_KEY не найден (.env)")
|
||||
os.environ["OPENAI_API_KEY"] = key
|
||||
os.environ["OPENAI_BASE_URL"] = os.environ.get("ROUTER_BASE_URL", "http://127.0.0.1:20128/v1")
|
||||
os.environ["OPENAI_BASE_URL"] = _router_base_url()
|
||||
|
||||
|
||||
def _wire_trace_hooks(agent, es, trace_id: str):
|
||||
|
|
@ -79,8 +108,8 @@ def build_agent(level_cfg, session_id=None, extra_instructions=""):
|
|||
|
||||
instructions = level_cfg.mandate + ("\n\n" + extra_instructions if extra_instructions else "")
|
||||
agent = create_harness(
|
||||
model=f"openai/{level_cfg.model}",
|
||||
effort=level_cfg.effort,
|
||||
model=build_level_model(level_cfg),
|
||||
effort="auto", # effort уже в параметрах модели (build_level_model)
|
||||
instructions=instructions,
|
||||
session={"id": session_id} if session_id else None,
|
||||
builtin_tools=("read", "write", "edit", "shell") if level_cfg.can_shell else [],
|
||||
|
|
@ -98,8 +127,8 @@ def run_cycle(level_cfg, es, prompt: str, session_id=None, trace_prefix=None):
|
|||
trace_id = trace_prefix or uuid.uuid4().hex[:12]
|
||||
instructions = level_cfg.mandate
|
||||
agent = create_harness(
|
||||
model=f"openai/{level_cfg.model}",
|
||||
effort=level_cfg.effort,
|
||||
model=build_level_model(level_cfg),
|
||||
effort="auto", # effort уже в параметрах модели (build_level_model)
|
||||
instructions=instructions,
|
||||
session={"id": session_id} if session_id else None,
|
||||
builtin_tools=("read", "write", "edit", "shell") if level_cfg.can_shell else [],
|
||||
|
|
|
|||
|
|
@ -26,6 +26,8 @@ ACTIVE = ["sec", "min", "hour", "day"]
|
|||
|
||||
# состояние исполнителей уровня: level → {"busy": bool, "task": id|None}
|
||||
_runners = {}
|
||||
# когда уровень последний раз слал TG-уведомление о занятости (анти-спам)
|
||||
_NOTIFY_TS = {}
|
||||
|
||||
|
||||
def _now():
|
||||
|
|
@ -140,6 +142,25 @@ def build_context(level_cfg, model: dict, task=None):
|
|||
return "\n\n".join(parts)
|
||||
|
||||
|
||||
def _notify_busy(level_cfg, task):
|
||||
"""TG-visibility: уровни сами сообщают о занятости.
|
||||
|
||||
Анти-спам: по одному уведомлению на уровень не чаще раза в 15 минут —
|
||||
повторные взятия задач внутри часа не заспамят чат.
|
||||
"""
|
||||
global _NOTIFY_TS
|
||||
now = time.time()
|
||||
if now - _NOTIFY_TS.get(level_cfg.name, 0) < 900: # 15 минут
|
||||
return
|
||||
_NOTIFY_TS[level_cfg.name] = now
|
||||
from . import tg_inbox
|
||||
tg_inbox.notify_chat(
|
||||
f"🔧 Уровень {level_cfg.name} взял задачу [{task.get('size', '?')}] "
|
||||
f"«{task.get('title', '')[:70]}»\n"
|
||||
f"Срез лестницы по запросу: /лестница"
|
||||
)
|
||||
|
||||
|
||||
def run_level_tick(level_cfg, seq: int, model: dict):
|
||||
"""Один тик уровня с LLM: дайджест + (опц.) задача. Возвращает текст.
|
||||
|
||||
|
|
@ -155,6 +176,12 @@ def run_level_tick(level_cfg, seq: int, model: dict):
|
|||
es.append_event("task", task_id, "TaskClaimed", {
|
||||
"level": level_cfg.name, "session_id": f"s-{level_cfg.name}-{seq}",
|
||||
})
|
||||
# TG-visibility: взяли задачу — сообщаем в чат (анти-спам: не чаще 15м на уровень)
|
||||
try:
|
||||
from . import tg_inbox
|
||||
_notify_busy(level_cfg, task)
|
||||
except Exception:
|
||||
pass # уведомление — best effort, не валить тик
|
||||
prompt = build_context(level_cfg, model, task)
|
||||
text, usage, trace_id = isolated_runner.run_llm_tick(
|
||||
level_cfg, prompt,
|
||||
|
|
@ -174,9 +201,15 @@ def run_level_tick(level_cfg, seq: int, model: dict):
|
|||
"level": level_cfg.name, "period": period.isoformat(),
|
||||
"summary_ref": trace_id, "text": (text or "")[:2000],
|
||||
})
|
||||
if usage:
|
||||
# BudgetSpent воркер уже записал сам — инкрементально, после каждого
|
||||
# модельного вызова (устраняет потерю spend при таймауте воркера).
|
||||
# Если usage в результате пуст (воркер не смог разобрать историю —
|
||||
# старый воркер / нетекстовая структура), дублируем итог, чтобы тик
|
||||
# не терялся в spend вовсе: дедуп по trace_id в read model.
|
||||
if usage and not usage.get("model_calls"):
|
||||
es.append_event("harness", "budget", "BudgetSpent", {
|
||||
"level": level_cfg.name, **{k: v for k, v in usage.items() if k != "error"},
|
||||
"level": level_cfg.name, "trace_id": trace_id, "source": "runner",
|
||||
**{k: v for k, v in usage.items() if k != "error"},
|
||||
})
|
||||
if task:
|
||||
es.append_event("task", task_id, "TaskCompleted", {
|
||||
|
|
|
|||
Loading…
Reference in a new issue