336 lines
No EOL
18 KiB
Python
336 lines
No EOL
18 KiB
Python
"""Воркер одного LLM-тика (запускается сабпроцессом из isolated_runner).
|
||
|
||
Читает payload из stdin, гоняет create_harness() один раз и умирает —
|
||
каждый тик получает чистый asyncio-рантайм Strands.
|
||
Трасса пишется напрямую в event store (es доступен).
|
||
Результат — в JSON-файл (path в payload['result_path']); stdout — только лог.
|
||
"""
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
|
||
def _rag_tool_for(level: str, trace_id: str):
|
||
"""RAG-инструмент тика (t-260929212316): семантический поиск по базе
|
||
кооператива (вендор /root/rag, fastembed bge-small + sqlite).
|
||
|
||
Ограничения зашиты в сам инструмент (не в промпт):
|
||
1 запрос на тик, k<=5, таймаут 20с, сабпроцесс rag-venv — упавший
|
||
вендор не валит тик. RAG_DISABLED=1 выключает.
|
||
"""
|
||
from strands import tool as strands_tool
|
||
from harness import rag_lookup as _rag
|
||
state = {"calls": 0}
|
||
|
||
@strands_tool
|
||
def rag_lookup(question: str) -> str:
|
||
"""Семантический поиск по базе знаний кооператива: контекст, проекты,
|
||
порталы, встречи, документы. Возвращает релевантные фрагменты.
|
||
Один вызов на тик — формулируй запрос целиком."""
|
||
if state["calls"] >= 1:
|
||
return "лимит: 1 RAG-запрос на тик (использован)"
|
||
state["calls"] += 1
|
||
chunks = _rag.rag_query(question, k=_rag.MAX_K,
|
||
level=level, trace_id=trace_id)
|
||
if not chunks:
|
||
return "(поиск ничего не вернул — отвечай из контекста тика)"
|
||
return _rag.format_chunks(chunks)
|
||
|
||
return rag_lookup
|
||
|
||
|
||
def _portal_mcp_tool_for(level: str, trace_id: str):
|
||
"""Тул тика `portal_mcp` — MCP-шлюз портала кооператива (t-261010093054).
|
||
|
||
Звонки/расписание (booking_slots/booking_create), реестры, казна,
|
||
лента звонков портала — живые данные cil.by прямо из тика. Клиент и
|
||
лимиты — harness/portal_mcp.py (4 вызова на тик, таймаут 15с, трейс
|
||
PortalMcpCalled); тул не поднимается без токена агента. Guardrail:
|
||
деньги-тулы (product_create/product_set_mode) запрещены по умолчанию
|
||
(SIMBIOT_MCP_ALLOW_ALL=1 снимает).
|
||
"""
|
||
from strands import tool as strands_tool
|
||
from harness import portal_mcp as _mcp
|
||
state = {"calls": 0}
|
||
|
||
@strands_tool
|
||
def portal_mcp(tool: str, arguments: str = "") -> str:
|
||
"""Вызов инструмента портала кооператива (MCP cil.by): расписание и
|
||
создание звонков с Алексом (booking_slots → booking_create), реестры
|
||
членов/проектов/агентов, казна, лента звонков и событий, whoami.
|
||
Сначала вызови с tool="booking_slots", arguments='{"type": "call"}',
|
||
слот передавай в booking_create как есть. arguments — JSON-строка.
|
||
Лимит 4 вызова на тик. Ответ — JSON-текст портала; ошибки — "error: …"."""
|
||
if state["calls"] >= _mcp.MAX_CALLS_PER_TICK:
|
||
return (f"лимит: {_mcp.MAX_CALLS_PER_TICK} вызовов portal_mcp "
|
||
f"на тик (использованы)")
|
||
state["calls"] += 1
|
||
return _mcp.call_tool(tool, arguments, level=level, trace_id=trace_id)
|
||
|
||
return portal_mcp
|
||
|
||
|
||
def _wire_tool_budget(agent, max_calls: int = 30):
|
||
"""Хард-кап тул-коллов на тик (e-260930031800: мин-воркер уходил в
|
||
90+ шагов исследования и умирал по таймауту 300с).
|
||
|
||
Механизм (e2e-проверено): BeforeToolCallEvent.cancel_tool отменяет
|
||
каждый вызов сверх лимита; AfterToolsEvent.end_turn останавливает
|
||
цикл агента ПОСЛЕ партии инструментов — модель больше не
|
||
вызывается, финальный текст тика — наш отчёт о недовыполненном.
|
||
Без end_turn модель «настаивает»: в тесте слала 20 шагов подряд,
|
||
получая отказы (дёшево, но цикл не кончается).
|
||
|
||
Партия: 20 параллельных tool-вызовов за один ход. Лимит считается
|
||
по РЕАЛЬНЫМ исполнениям (отменённые не съедают бюджет — модель
|
||
не может заблокировать тик потоком отмен), поэтому первые
|
||
max_calls вызова партии выполняются, хвост отменяется, затем
|
||
цикл останавливается с итоговым текстом.
|
||
"""
|
||
from strands.hooks import AfterToolsEvent, BeforeToolCallEvent
|
||
state = {"executed": 0, "halted": False}
|
||
|
||
def on_before_tool(event):
|
||
if state["halted"] or state["executed"] >= max_calls:
|
||
event.cancel_tool = (
|
||
f"тиковый бюджет шагов исчерпан ({max_calls} tool-вызовов): "
|
||
f"вызов отменён; новые вызовы инструментов отключены до конца тика"
|
||
)
|
||
return
|
||
state["executed"] += 1
|
||
|
||
def on_after_tools(event):
|
||
# партия инструментов закончилась; лимит пробит → стоп-цикл
|
||
# (end_turn: цикл агента не зовёт модель повторно)
|
||
if not state["halted"] and state["executed"] >= max_calls:
|
||
state["halted"] = True
|
||
event.end_turn = (
|
||
f"тик завершён: бюджет шагов исчерпан ({max_calls} tool-вызовов). "
|
||
f"Выше — результат того, что успело выполниться. Не успевшее "
|
||
f"оформи задачей (cli.py task) или продолжи в следующем тике."
|
||
)
|
||
|
||
agent.add_hook(on_before_tool, BeforeToolCallEvent)
|
||
agent.add_hook(on_after_tools, AfterToolsEvent)
|
||
return agent
|
||
|
||
|
||
def build_level_model_from_payload(payload: dict):
|
||
"""Модель для воркера — та же сборка, что и в runtime.build_level_model:
|
||
9router понимает chat/completions (usage там же), Responses API — шим без usage.
|
||
|
||
SIMBIOT_LLM_CALL_TIMEOUT (e-260930161900-6da5c9, e-260930153143-85246b):
|
||
per-call таймаут LLM-вызова в сек. Роутер-лог показал: зависший тик = POST
|
||
в ollama-апстрим принят, но байт стрима так и не пришёл (DONE без TTFT),
|
||
при этом соседние вызовы ходили за 1-3с. Таймаут уровня (300/2880с) убивает
|
||
ВЕСЬ тик из-за одного висящего вызова; теперь вызов режется (до 90с,
|
||
hour 180с), исключение отлавливает event-loop — тик получает результат
|
||
с пометкой «не успел», spend-хвост и задачи не теряются.
|
||
"""
|
||
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"
|
||
# per-call таймаут: один зависший апстрим-вызов не должен убивать весь тик
|
||
call_timeout = int(os.environ.get("SIMBIOT_LLM_CALL_TIMEOUT", "90"))
|
||
return OpenAIModel(
|
||
# timeout открытого клиента (openai SDK) режет каждый chat.completions
|
||
# вызов; при зависании апстрима получаем APITimeoutError за N сек и
|
||
# strands event-loop завершает тик штатно (без потери предыдущих шагов)
|
||
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"),
|
||
"timeout": call_timeout, "max_retries": 1},
|
||
model_id=model_id,
|
||
**({"params": params} if params else {}),
|
||
)
|
||
|
||
|
||
def _emit(payload, obj):
|
||
rp = payload.get("result_path")
|
||
if rp:
|
||
try:
|
||
with open(rp, "w") as f:
|
||
json.dump(obj, f, ensure_ascii=False)
|
||
except OSError:
|
||
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,
|
||
model: str = None):
|
||
"""Записать BudgetSpent за один модельный вызов.
|
||
|
||
Воркер — единственный автор spend-событий, но scheduler может убить
|
||
его по таймауту в любой момент (SIGKILL). Поэтому usage пишем
|
||
ИНКРЕМЕНТАЛЬНО: каждый завершённый модельный вызов — своё событие.
|
||
Хвост недописанного вызова теряется, но весь тик — никогда.
|
||
|
||
model (e-260930062120): запрошенный model_id уровня пишется в каждое
|
||
событие — в кредитах роутера видно, ЧЬИ это траты; "auto" резолвится
|
||
роутером, фактическую модель до usage strands не выносит.
|
||
"""
|
||
if not usage_delta:
|
||
return
|
||
try:
|
||
from harness import es
|
||
es.append_event("harness", "budget", "BudgetSpent", {
|
||
"level": level, "trace_id": trace_id, "source": "worker",
|
||
**({"model": model} if model else {}),
|
||
**usage_delta,
|
||
})
|
||
except Exception:
|
||
pass # трекинг не должен валить тик
|
||
|
||
|
||
def _wire_budget_hooks(agent, level: str, trace_id: str, model: str = None):
|
||
"""После каждого модельного вызова — 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,
|
||
}, model=model)
|
||
|
||
agent.add_hook(on_message_added, MessageAddedEvent)
|
||
return agent
|
||
|
||
|
||
def resolve_tool_budget(level: str) -> int:
|
||
"""Бюджет тул-коллов тика по уровню (e-260930071900, просьба Alex:
|
||
часовому уровню нужен большой цикл — «хоть 300»).
|
||
|
||
Приоритет: SIMBIOT_MAX_TOOL_CALLS_<LEVEL> > SIMBIOT_MAX_TOOL_CALLS
|
||
> per-level дефолт. Минута остаётся компактной (30) — жёсткий кап
|
||
и спасает её от таймаута (e-260930031800: 90+ шагов → 300с → смерть).
|
||
День 10_000 («убрать/сделать супер большим», Alex 2026-10-01) —
|
||
фактически не ограничивает: дневной тик упирается в свой таймаут,
|
||
а не в шаги; инцидент 03:00 — e-261001074527-f0c1b4, фикс закрытия
|
||
чужих задач — отдельная тема той же эскалации.
|
||
"""
|
||
env = (os.environ.get(f"SIMBIOT_MAX_TOOL_CALLS_{level.upper()}")
|
||
or os.environ.get("SIMBIOT_MAX_TOOL_CALLS"))
|
||
if env:
|
||
return int(env)
|
||
return {"min": 30, "hour": 300, "day": 10_000}.get(level, 30)
|
||
|
||
|
||
def main():
|
||
payload = json.loads(sys.stdin.read())
|
||
|
||
from harness import es
|
||
from harness.runtime import setup_provider, _wire_trace_hooks
|
||
|
||
setup_provider()
|
||
from strands_harness import create_harness
|
||
|
||
trace_id = payload.get("trace_prefix") or "tick"
|
||
level = payload.get("level") or "?"
|
||
try:
|
||
# RAG-инструмент тика (t-260929212316): у LLM появляется rag_lookup —
|
||
# поиск по базе кооператива; лимиты (1 запрос/tick, k<=5, таймаут)
|
||
# зашиты в инструмент. Уровень без shell получает только его.
|
||
tools = []
|
||
try:
|
||
tools.append(_rag_tool_for(level, trace_id))
|
||
except Exception:
|
||
pass # RAG — best effort: вендор не должен валить тик
|
||
# MCP портала (t-261010093054): booking_slots/booking_create, реестры,
|
||
# казна, лента звонков — живой портал из тика; без токена тул не поднимется
|
||
try:
|
||
tools.append(_portal_mcp_tool_for(level, trace_id))
|
||
except Exception:
|
||
pass # портал — best effort: шлюз не должен валить тик
|
||
# stdout воркера не читается (subprocess.run с capture_output и
|
||
# длинным выводом переполняет pipe → BrokenPipeError в печати,
|
||
# результат тика терялся, дайджесты 04:11-04:17 ушли пустыми).
|
||
# Результат живёт в JSON-файле (result_path), печать не нужна.
|
||
from strands.handlers.callback_handler import null_callback_handler
|
||
agent = create_harness(
|
||
model=build_level_model_from_payload(payload),
|
||
effort="auto", # effort уже в параметрах модели
|
||
instructions=payload.get("instructions", ""),
|
||
session=None,
|
||
tools=tools,
|
||
callback_handler=null_callback_handler,
|
||
builtin_tools=("read", "write", "edit", "shell") if payload.get("can_shell") else [],
|
||
memory=False,
|
||
skills=False,
|
||
)
|
||
agent = _wire_trace_hooks(agent, es, trace_id, model=payload.get("model"))
|
||
agent = _wire_budget_hooks(agent, level, trace_id, model=payload.get("model"))
|
||
# хард-кап шагов (e-260930031800): cancel_tool + end_turn.
|
||
# ВАЖНО: НЕ используем limits={"turns": N} — нативный стоп SDK
|
||
# обрывает цикл на границе хода, и result.message остаётся
|
||
# toolResult-сообщением → str(result)='' → пустой дайджест
|
||
# (репродукция: stop_reason=limit_turns, text='', тики 04:18+).
|
||
# Наш end_turn (AfterToolsEvent) кладёт в историю ассистентский
|
||
# текст с итогом — дайджест не пустеет.
|
||
agent = _wire_tool_budget(agent, resolve_tool_budget(level))
|
||
result = agent(payload.get("prompt", ""))
|
||
usage = _extract_usage(agent)
|
||
_emit(payload, {"ok": True, "text": str(result), "usage": usage, "trace_id": trace_id})
|
||
except Exception as e:
|
||
import traceback
|
||
_emit(payload, {"ok": False, "error": f"{e}\n{traceback.format_exc()[-800:]}"})
|
||
# орфан-воркер после убийства родителя может умереть на печати — не критично
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |