Рантайм для ИИ-агентов на Agent API Вайб-Маркетолога и разбор самого API.
Тестовое задание было свободным: «предложите полезную функцию, идею или решение». Я прочитал документацию, нашёл два места, где она молча наказывает разработчика агента, и закрыл оба — кодом там, где это возможно на стороне клиента, и предложением там, где нужно менять сервер.
- PROPOSALS.md — шесть предложений по API. Главное: подпись вебхука не содержит метки времени, из-за чего перехваченный вызов остаётся валидным бессрочно.
- Библиотека в
src/— рантайм Inbox, который не платит авто-ответчику платформы и не отвечает дважды на одно сообщение.
В разделе «Входящие сообщения» есть строчка:
если агент не забрал сообщение за ~4 секунды, платформа отвечает сама; списание с баланса владельца — Claude Opus 4.8 10 ₽/ответ, Sonnet 4.6 8 ₽/ответ
Ключевое слово — «не забрал». Четыре секунды отсчитываются до момента, когда агент забрал сообщение из очереди, а не до момента ответа. Поэтому цикл, который пишется первым и выглядит совершенно естественно, — платный:
while True:
messages = await poll() # ← здесь мы ждём, всё хорошо
for message in messages:
answer = await llm(message) # ← а здесь мы слепы
await reply(message.id, answer)
# и только теперь снова poll()Пока идёт поход в модель, ни один long-poll не припаркован. Всё, что пришло в это время, пролежит дольше четырёх секунд, и за него ответит платформа — своей моделью, своим тоном, за ваши деньги. Ответ агента, пришедший позже, уже никому не нужен.
Ошибка не в невнимательности: цикл корректен с точки зрения логики и падает только на тайминге, которого не видно в коде.
examples/demo_savings.py прогоняет один и тот же поток сообщений через
наивный цикл и через рантайм. Ни сети, ни ключей не нужно — используется
фейк платформы с её таймингами:
python examples/demo_savings.pyСообщений в канал: 12, время ответа агента ≈ 2× окна авто-ответчика
ответил агент ответила платформа списано, ₽
-------------------------------------------------------------------
Наивный цикл 7 5 50
InboxRuntime 12 0 0
Наивный цикл отдал платформе 42% диалогов — это 50 ₽.
Пересчёт на реальную нагрузку: при 1000 диалогов в сутки и той же доле
пропусков это 4167 ₽ в день на ровном месте.
Дело не только в деньгах: 42 % диалогов ведёт не тот агент, которого настраивал клиент.
Опрос отделён от обработки. Несколько long-poll висят постоянно; вернувшийся паркуется заново немедленно, ещё до того как сообщение уйдёт в работу. Обработкой занимаются отдельные воркеры. Слепого окна не возникает в принципе.
Риск измеряется, а не предполагается. Метрика uncovered_seconds —
суммарное время, когда не припаркован ни один опрос. Это прямая оценка
денежного риска, её можно отдать в Prometheus и повесить алерт: пока она
около нуля, платформа не имеет повода отвечать за вас.
Ответ уходит до дедлайна канала. Бюджет времени (Deadline) передаётся
обработчику и уменьшается по мере работы. Если основной обработчик не
укладывается или падает, ответ отдаёт быстрый запасной. Молчание дороже
неидеального ответа — за молчание платит владелец.
Повторная доставка не пересчитывается. Сообщение, забранное но не
отвеченное за 60 секунд, платформа выдаёт снова. Рантайм отдаёт сохранённый
ответ вместо повторного похода в модель: это и лишние деньги, и — если в
ответе были действия crm.* — риск выполнить их дважды.
Вебхуки проверяются жёстко. Сравнение подписи за постоянное время, поддержка старой и новой схем секрета одновременно (ротация без простоя), отклонение повторов.
Нужен Python 3.11+.
python -m venv .venv && .venv/Scripts/python -m pip install -e ".[dev]".venv/Scripts/python -m pytest -q.venv/Scripts/python examples/demo_savings.pyimport asyncio
from vibe_runtime import Deadline, HttpInboxTransport, InboxMessage, InboxRuntime
async def answer(message: InboxMessage, deadline: Deadline) -> str:
"""Основной обработчик. Бюджет времени — не декорация: на него надо смотреть."""
return await my_llm.complete(message.text, timeout=deadline.remaining)
async def quick(message: InboxMessage) -> str:
"""Запасной ответ. Обязан быть мгновенным: шаблон или кэш, не модель."""
return "Уточняю детали, вернусь с ответом через минуту."
async def main() -> None:
async with HttpInboxTransport(token="oc_...") as transport:
runtime = InboxRuntime(transport, answer, fast_handler=quick)
await runtime.run()
asyncio.run(main())Ответ может быть строкой или объектом Reply с действиями Bitrix24:
return Reply(
text="Создал сделку, менеджер свяжется сегодня.",
actions=({"method": "crm.deal.add", "params": {"fields": {"TITLE": "Заявка"}}},),
)src/vibe_runtime/
├── inbox.py ядро: припаркованный опрос, воркеры, дедлайны, дедупликация
├── deadlines.py бюджет времени, передаваемый вниз по стеку вызовов
├── webhooks.py проверка подписи и защита от повторов
├── transport.py протокол доступа к API + реализация на httpx
├── metrics.py счётчики, включая оценку денежного риска
└── models.py входящие сообщения и ответы
tests/
├── fake_platform.py фейк платформы с её таймингами и биллингом
└── test_*.py 29 тестов
Тесты гоняются против фейка, воспроизводящего окна платформы (4 с / 22 с / 60 с) в масштабе 1:100 — настоящий asyncio, настоящая конкуренция, весь набор проходит за три секунды.
Ключевой тест — test_naive_loop_pays_platform_but_runtime_does_not:
он на одном сценарии показывает, что наивный цикл отдаёт деньги, а рантайм
нет. Если рантайм сломается, тест это заметит; если сломается сам тест —
он падает на проверке «наивный цикл обязан был пропустить сообщения»,
а не молча зеленеет.
Регистрацию я не проходил и живые ключи не использовал: код проверен против
фейка платформы, построенного строго по документации. Если какие-то тайминги
или формат полей на практике отличаются от описанных — это исправляется
константами в RuntimeConfig и разбором в models.py, логика не меняется.
Защита от повторов вебхуков — лечение симптома. Настоящее решение требует метки времени в подписи на стороне платформы, и это предложение №1 в PROPOSALS.md.