TypeScript для Python ML-инженеров: создаём агентный сервис

Viacheslav Dubrov

2026-08-15 · Обновлено 2026-10-02 · 23 мин · Инструменты разработчика · ИИ-агенты

Автоматический перевод

Эта статья была автоматически переведена с оригинальной английской версии.

Это руководство для опытных Python ML- и бэкенд-инженеров, которым нужно построить сервис AI на TypeScript и Node. Оно отталкивается от понятий Python-сервисов, а не от основ программирования или браузерного кода. К концу вы сможете сопоставить стек Python-сервиса с TypeScript, узнавать привычки Python, из-за которых появляются баги в JavaScript, а также запускать и менять стриминговый агентный сервис.

Пример сервиса проводит запрос через валидацию, выполнение инструментов, HTTP-стриминг и сохранение данных. Он отвечает на три вопроса: что происходит, когда аргументы инструмента приходят фрагментами, когда стрим падает после HTTP 200 и когда воркер забирает одну и ту же задачу дважды? Мы рассматриваем только бэкенд и AI: без React и без браузерного бандлера.

Код лежит в slavadubrov/typescript-agent-service. В нём две реализации агентного цикла, история запусков в Postgres, воркер и MCP-сервер. Офлайн-демо запускается без API-ключа, а затем сопоставление «Python → TypeScript» ниже поможет читать и менять сервис. Ресурсоёмкие численные расчёты остаются в Python; здесь речь об оркестрации, HTTP и стриминге.

Сначала запустите сопутствующий репозиторий

Установите Node 24 и pnpm по официальной инструкции pnpm. Затем клонируйте репозиторий и выполните:

pnpm install
pnpm demo
pnpm check

pnpm demo прогоняет HTTP-обработчик, агентный цикл и SSE-стрим с моделью с заранее заданными ответами. Он также вызывает вычисление runSweep из воркера, но не запускает процесс воркера и его очередь в базе данных. Демо не требует API-ключа, базы данных или Docker. pnpm check запускает проверку типов, линтер, проверку форматирования и тесты.

Демо печатает SSE-фреймы одного чат-запроса, а затем sweep воркера. В сокращённом виде:

event: tool_call
data: {"type":"tool_call","callId":"call_1","toolName":"estimate_kv_cache","args":{"model":"llama-3.1-8b","seqLen":32768,"batchSize":16}}

event: tool_result
data: {"type":"tool_result","callId":"call_1","toolName":"estimate_kv_cache","ok":true,"result":{"model":"llama-3.1-8b","seqLen":32768,"batchSize":16,"kvCacheGiB":64},"durationMs":0}

event: done
data: {"type":"done","stopReason":"stop","steps":2,"text":"Llama-3.1-8B at 32k context and batch 16 needs 64 GiB of KV cache."}

--- worker: capacity sweep ---
batch=  1  kv_cache=4.0 GiB
batch= 16  kv_cache=64.0 GiB

Каждый блок event:/data: — это один SSE-фрейм: модель просит инструмент, инструмент отвечает, и запуск заканчивается событием done. Последняя часть — sweep воркера по размерам батча.

Чтобы запустить настоящие API и воркер:

cp .env.example .env          # add OPENAI_API_KEY or an OpenAI-compatible URL
pnpm db:up                    # start Postgres in Docker
pnpm db:push                  # create the database schema
pnpm dev:api                  # API at http://localhost:8080
pnpm dev:worker               # run this in a second terminal

Запустите стрим агента. Ответ — серия SSE-фреймов, как в выводе демо выше:

curl -N -X POST localhost:8080/v1/chat \
  -H 'content-type: application/json' \
  -d '{"message":"How much KV cache for llama-3.1-70b at 32k context, batch 8?"}'

Поставьте в очередь долгую задачу и опросите её. Первый вызов возвращает 202 с {"id":"...","status":"queued"}. Второй возвращает сохранённую строку, а после завершения работы воркера в ней появляется результат:

curl -X POST localhost:8080/v1/sweeps \
  -H 'content-type: application/json' \
  -d '{"model":"llama-3.1-70b","seqLen":32768,"batchSizes":[1,4,16]}'

curl localhost:8080/v1/runs/<id>

Скрипты pnpm выше — это команды, которые я использую в остальной части статьи. Репозиторий прячет низкоуровневые команды Node и TypeScript за именованными скриптами pnpm, примерно как Python-проект может прятать команды uv run за целями make. Не смешивайте npm install с этим pnpm-репозиторием. Используйте pnpm install, чтобы pnpm-lock.yaml оставался единственным lock-файлом.

Что делают Node, npm, pnpm, TypeScript и tsx

За похожими названиями скрываются разные задачи:

• JavaScript — язык.
• Node.js — рантайм, примерно как CPython для JavaScript. Для этого проекта установите Node 24.
• Реестр npm — индекс пакетов, примерно как PyPI. Команда npm поставляется с Node и умеет устанавливать пакеты из этого реестра.
• pnpm — менеджер пакетов, который выбрал этот репозиторий. Он ставит пакеты из реестра npm, управляет рабочим пространством монорепозитория и запускает команды, объявленные в package.json.
• package.json — манифест проекта, ближайший аналог pyproject.toml. Его раздел scripts даёт имена вроде demo, check и dev:api более длинным командам.
• TypeScript — это JavaScript со статическими типами. Команда tsc проверяет эти типы. Репозиторий запускает её через pnpm check или pnpm typecheck.
• tsx запускает файлы .ts без отдельной сборки. Скрипты разработки используют его режим watch, чтобы перезапускать API или воркер после изменения исходников. В этом руководстве вы не вызываете его напрямую.

В этом репозитории запускайте скрипты pnpm. Node — это рантайм внутри этих скриптов, а продакшеном занимается готовый Dockerfile.

Что запускает каждый скрипт:

• pnpm demo запускает tsx scripts/demo.ts.
• pnpm dev:api запускает tsx watch.
• pnpm check запускает tsc, Biome (линтер) и Vitest (средство запуска тестов).
• Docker CMD запускает node --import tsx/esm.

Всё это работает на Node 24. В pnpm check tsc и Biome только читают файлы; Vitest исполняет тесты на Node.

Стек, сопоставленный с Python

Zod, Hono, Drizzle, Vitest и Biome покрывают примерно ту же область, что pydantic, FastAPI, SQLAlchemy, pytest и Ruff.

Большинство пар — прямые замены. Три — нет:

В сопутствующем репозитории есть четыре библиотеки, которые стоит объяснить.

Hono для HTTP-слоя

Express и Fastify — альтернативы, ориентированные на Node. Hono использует стандартные веб-API Request и Response и предоставляет адаптеры для Node и serverless-рантаймов. Для этого небольшого стримингового API такая переносимость полезна, поэтому я выбрал Hono. Выбирайте Fastify, если сервис работает только на Node и вам нужны его набор плагинов и сериализация ответов на основе схем.

Drizzle для SQL

Drizzle хранит схему в TypeScript и не требует шага генерации клиента. Он также оставляет доступным чистый SQL, когда конструктор запросов не может аккуратно выразить конструкцию Postgres. Я бы выбрал Prisma, если его сгенерированный клиент и окружающие инструменты лучше подходят команде.

Biome для линтинга и форматирования

Biome выполняет линтинг, форматирование и сортировку импортов одним бинарником и одним файлом конфигурации. Оставьте ESLint, если проект зависит от собственных правил, которых в Biome нет.

Vitest для тестов

Vitest запускает .ts-тесты сопутствующего репозитория без отдельной настройки трансформации. Оставьте Jest в репозитории, где уже есть конфигурация Jest и собственные матчеры.

Читаем синтаксис TypeScript, который встретится ниже

Держите эту таблицу рядом с примерами сервиса как справочник.

Используйте const, если связывание не должно меняться. Используйте let для счётчика, аккумулятора или другого связывания, которое вы будете переназначать.

Короткий перевод enum

Для строковых состояний этот репозиторий использует объект и выводимый из него тип объединения строк:

const Status = { Queued: "queued", Running: "running" } as const;
type Status = (typeof Status)[keyof typeof Status]; // "queued" | "running"

Объект даёт Status.Queued во время работы программы. Строка type во время проверки типов разрешает только "queued" или "running". Вместе они выполняют две роли этого объявления на Python:

from enum import Enum

class Status(str, Enum):
    QUEUED = "queued"
    RUNNING = "running"

Достаточно узнавать этот шаблон. as const оставляет значения объекта точными строками, а не расширяет их до произвольного string.

Семь семантических различий, которые стоят времени

Таблица синтаксиса проведёт вас через примеры. Именно в этих семантических различиях привычки Python приводят к багам.

1. Пустые массивы и объекты истинны

Правило Python «пустой контейнер ложен» переносится хуже всех привычек. if (results) истинно и для пустого массива. Пишите if (results.length).

2. null и undefined — разные вещи

null обычно обозначает намеренное отсутствие. undefined обычно значит, что значение отсутствует или не присвоено, хотя код может присвоить его и явно. Код библиотек постоянно возвращает undefined. Разница проявляется, когда вы пишете значение по умолчанию. || подставляет правую часть всякий раз, когда левая ложна. Это относится к 0, "" и false. ?? подставляет её только для null и undefined. Поэтому 0 || 10 равно 10, а 0 ?? 10 равно 0. Так размер батча, равный нулю, незаметно превращается в десять.

3. Блок catch получает unknown

Конструкции except ValueError: нет. Один блок catch получает всё. Поскольку JavaScript позволяет бросить строку, число или null, TypeScript типизирует пойманное значение как unknown, то есть «может быть буквально чем угодно» при strict. Сопутствующий репозиторий включает strict, и новым проектам обычно стоит делать так же. Чтобы изучить ошибку, сначала сузьте тип значения:

try {
    await risky();
} catch (error) {
    // `error` is `unknown` until you prove otherwise. This line is the
    // TypeScript equivalent of `except Exception as e:` and it is not
    // optional.
    const message = error instanceof Error ? error.message : String(error);
}

4. Промисы стартуют сразу

Вызов async-функции начинает выполнять её тело и возвращает промис. Объект-корутина Python ничего не делает, пока вы не сделаете await или не запланируете его. Promise.all близок к asyncio.gather. Promise.allSettled близок к gather(..., return_exceptions=True), только каждый результат оборачивается в { status, value } или { status, reason }.

Планированием в рантайме занимается Node. Он держит процесс живым, пока существуют активные дескрипторы или запросы, такие как таймеры и сокеты. Обычный ожидающий промис сам по себе Node живым не удерживает. Оборачивать программу в asyncio.run не нужно. В ES-модуле можно использовать await на верхнем уровне, когда запуск должен дождаться асинхронной операции.

5. В JavaScript один обычный числовой тип

Тип number в JavaScript хранит значения как 64-битные числа с плавающей точкой, примерно как float в Python. Технический стандарт этого формата называется IEEE 754. Десятичные значения приближённые, поэтому 0.1 + 0.2 не равно в точности 0.3, а целые числа остаются точными только до 2**53 - 1, то есть 9,007,199,254,740,991.

Храните 64-битные ID как строки на границах сервиса. Преобразование bigint из Postgres в number JavaScript может его округлить. Для более крупных точных целых JavaScript предоставляет отдельный тип BigInt, который не смешивается с обычными числами.

6. Используйте Map, когда нужен словарь в стиле Python

В JavaScript {} создаёт объект. Объекты обычно представляют записи с именованными полями:

const request = { model: "gpt-5", seqLen: 4096 };
console.log(request.model);

Объект — это не чистая таблица «ключ-значение», как dict в Python. Он наследует некоторые имена от самого JavaScript. Это может дать неожиданный результат:

const tools: Record<string, unknown> = {};

tools["constructor"]; // a built-in function, not a missing value
Object.hasOwn(tools, "constructor"); // false

Если внешняя строка выбирает поле объекта, вызовите Object.hasOwn перед чтением. Если нужен универсальный словарь, используйте Map. Map ближе к dict в Python: ключ существует только тогда, когда его добавил ваш код.

const tools = new Map<string, unknown>();
tools.get("constructor"); // undefined

7. Указывайте расширение в относительных импортах

Исходный файл JavaScript, который делится кодом с другими файлами, называется модулем. Этот проект использует современный формат модулей, ES-модули, обычно сокращаемый до ESM. ES расшифровывается как ECMAScript — формальное название языка JavaScript. На практике ESM — это синтаксис import и export, который используется во всём проекте.

Для относительного импорта Node требует точное имя файла. Он не станет гадать, означает ли ./env файл ./env.ts или ./env.js:

import { loadEnv } from "./env.ts";

Импорты из установленных пакетов или пакетов рабочего пространства по-прежнему используют имя пакета без расширения файла:

import { z } from "zod";
import { runAgent } from "@agent/core";

Zod: объявляем схему, выводим тип

В pydantic вы объявляете класс и получаете валидатор. В Zod вы объявляете валидатор и выводите из него тип. Тот же единый источник истины, направление обратное.

На схеме одна и та же настройка PORT показана в обоих вариантах и то, что остаётся от каждого объявления во время работы кода.

Из packages/schemas/src/env.ts:

import { z } from "zod"; // `z` is Zod's whole API, the way `pd` is pandas

const EnvSchema = z.object({
    PORT: z.coerce.number().int().positive().default(8080),
    // See the note below: a bare z.url() would accept "localhost:8000".
    OPENAI_BASE_URL: z
        .url({ protocol: /^https?$/ })
        .default("https://api.openai.com/v1"),
    DATABASE_URL: z.string().optional(),
});

export type Env = z.infer<typeof EnvSchema>;

z.infer<typeof EnvSchema> считывает статический тип из схемы рантайма. z.coerce.number() учитывает, что каждое определённое значение в process.env (аналог os.environ в Node) — строка. Он играет ту же роль, что приведение числовых настроек в pydantic, хотя наборы принимаемых строк различаются. Это повторяет шаблон pydantic-settings и выполняется один раз при старте. Невалидное окружение тогда даёт читаемую ошибку при запуске, а не TypeError внутри обработчика.

Обычный z.url() принимает localhost:8000. Стандарт URL считает схемой всё до первого двоеточия. Поэтому он читает localhost: как протокол с названием «localhost» и принимает строку. Затем значение попадает в HTTP-клиент и ломается с меньшим количеством контекста. Валидация схемой переносит сбои раньше, но если вы написали разрешающую схему, она будет её соблюдать.

Zod 4 также поставляется с z.toJSONSchema, поэтому этому проекту не нужна зависимость zod-to-json-schema, распространённая в старых туториалах. Это важно, когда одну схему должны использовать три потребителя, о чём говорится в разделе «Один инструмент, три потребителя» ниже.

Сервис

Верхний ряд — общий код в packages/. Ниже каждое приложение в apps/ — отдельный процесс. К Postgres обращаются только API и воркер, а сервер MCP общается со своим клиентом через stdin и stdout.

Демо-сервис подбирает размеры развёртываний LLM. Один инструмент ищет архитектурные константы модели. Другой оценивает её объём KV cache: память GPU, которая хранит ключи и значения аттеншна для запросов в обработке. Оба инструмента намеренно сделаны скучной арифметикой. Им не нужна сеть, и они каждый раз дают один и тот же ответ. Благодаря этому сервис можно тестировать без API-ключа. Оценщик KV cache также опубликован через Model Context Protocol (MCP), чтобы его могли вызывать другие клиенты AI.

typescript-agent-service/
├── apps/
│   ├── api/           @agent/api: Hono API, streams Server-Sent Events (below)
│   ├── worker/        polls Postgres for long-running jobs
│   └── mcp/           MCP server: exposes one tool to outside AI clients
├── packages/
│   ├── schemas/       package name @agent/schemas: env, API, and tool schemas
│   ├── agent-core/    package name @agent/core: the loop (twice), tools, storage
│   └── observability/ Pino logging, OpenTelemetry tracing
├── pnpm-workspace.yaml
└── package.json

pnpm-workspace.yaml — файл, который объявляет рабочее пространство. Внутренние пакеты получают имя с областью, например @agent/core, где префикс @agent/ — соглашение об именовании, а не возможность языка. Каждый пакет объявляет свою публичную точку входа в package.json. Эта граница пакета не зависит от того, какая команда запускает приложение.

Это приватное рабочее пространство указывает точки входа на исходники .ts, потому что все потребители — часть одного репозитория. Публичные npm-пакеты обычно публикуют JavaScript и декларации типов .d.ts, чтобы обычным потребителям Node не требовались средство запуска TypeScript или настройка сборки автора пакета.

Напишите цикл инструментов вручную один раз

Левая колонка — один шаг модели. Правая колонка — то, что выдаёт цикл: кроме исключения, которое выходит из цикла, эти события — всё, что видят HTTP-маршрут и тесты.

Фреймворки для агентов с tool calls оборачивают один и тот же базовый цикл:

1. Вызвать модель с определениями инструментов.
2. Провалидировать и запустить запрошенные инструменты.
3. Добавить результаты к сообщениям.
4. Снова вызвать модель.

Если написать цикл один раз, скрытые пути отказа становятся видимыми.

Цикл — это async function*, асинхронный генератор, та же форма, что у async def с yield в Python. HTTP-маршрут итерирует этот генератор, превращает каждое событие в фрейм Server-Sent Events и накапливает текст. После окончания стрима маршрут один раз вызывает storage.createRun с итоговым текстом. Тесты вызывают цикл отдельно и собирают его события в массив. SSE-фрейм — это один фрагмент долгоживущего HTTP-ответа. Формат объясняется в разделе ниже.

У цикла два нормальных выхода. Ответ без tool calls отдаёт done с stopReason: "stop". Если модель продолжает запрашивать инструменты вплоть до maxSteps, генератор всё равно завершается и отдаёт done с stopReason: "max_steps". Считайте такой результат завершённым, но усечённым: накопленный текст может быть неполным.

Каждое значение, которое отдаёт цикл, — это AgentEvent, размеченное объединение. Сокращённо из packages/schemas/src/events.ts:

export const AgentEventSchema = z.discriminatedUnion("type", [
    z.object({ type: z.literal("text"), delta: z.string() }),
    z.object({
        type: z.literal("tool_call"),
        callId: z.string(),
        toolName: z.string(),
        args: z.unknown(),
    }),
    z.object({
        type: z.literal("tool_result"),
        callId: z.string(),
        ok: z.boolean(),
        result: z.unknown() /* ... */,
    }),
    z.object({
        type: z.literal("done"),
        stopReason: z.enum(["stop", "max_steps"]),
        steps: z.number(),
        text: z.string(),
    }),
    z.object({ type: z.literal("error"), message: z.string() }),
]);

export type AgentEvent = z.infer<typeof AgentEventSchema>;

z.discriminatedUnion выбирает вариант по полю type, как match по датаклассам с меткой. Внутри if (event.type === "text") компилятор знает, что event.delta существует. Добавление варианта не приводит автоматически к ошибке в неполном switch. Чтобы компилятор сообщил о пропущенном варианте, проверьте оставшееся значение на соответствие never в ветке default. Правило линтера для проверки полноты также может сообщить о пропущенных случаях. И HTTP-маршрут, и версия на AI SDK ниже используют суженные типы событий.

Тело цикла стримит события text и собирает tool calls из фрагментов. Из packages/agent-core/src/loop.ts:

let text = ""; // the assistant text produced during this model step
// Keyed by the `index` field, because a streamed response interleaves
// fragments of several parallel tool calls and only `index` is present
// on every fragment. `id` and `name` arrive once, `arguments` arrives
// in pieces. `a?.b` below reads `b` only if `a` exists, and gives back
// `undefined` instead of throwing if it doesn't.
const partial = new Map<number, PartialToolCall>();

for await (const chunk of stream) {
    const choice = chunk.choices[0];
    if (!choice) continue;

    if (choice.delta.content) {
        text += choice.delta.content;
        yield { type: "text", delta: choice.delta.content };
    }

    for (const fragment of choice.delta.tool_calls ?? []) {
        const slot = partial.get(fragment.index) ?? { id: "", name: "", args: "" };
        if (fragment.id) slot.id = fragment.id;
        if (fragment.function?.name) slot.name = fragment.function.name;
        if (fragment.function?.arguments) slot.args += fragment.function.arguments;
        partial.set(fragment.index, slot);
    }
}

Локальная для цикла переменная text хранит один шаг модели. Цикл использует её в сообщении ассистента для следующего шага или в итоговом событии done. Это не аккумулятор уровня маршрута, который потом сохраняется.

Карта partial — то, что скрывают фреймворки. SDK отдаёт строковые фрагменты аргументов функции, и они могут разрезать сериализованный JSON в произвольных местах. Несколько параллельных вызовов тоже могут перемежаться. В репозитории есть тест, который разрезает {"model":"llama-3.1-8b",...} на четыре чанка.

На схеме эти четыре чанка попадают в запись карты для индекса 0.

Вторая вещь, которую стоит написать самому, — что происходит при провале валидации:

// `tool.parse` is this repo's wrapper, not Zod's. Zod's own `.parse` throws and
// `.safeParse` returns `{ success, data, error }`; this returns
// `{ ok: true, value }` on success and `{ ok: false, error }` on failure, so no
// caller has to catch.
const parsed = tool.parse(raw);
if (!parsed.ok) {
    return {
        ok: false,
        value: { error: `Invalid arguments: ${parsed.error}` },
        parsedArgs: raw,
    };
}

Перед выполнением поиск может не найти инструмент, JSON.parse может отклонить аргументы, а Zod может отклонить их форму. Каждый такой сбой становится сообщением, которое читает модель. Во время выполнения ожидаемый ToolError тоже становится результатом инструмента, чтобы модель могла исправить свой вызов. Неожиданное исключение уходит на путь обработки HTTP-ошибок и не выдаётся за сбой предметной области. z.prettifyError превращает дерево проблем Zod в сообщение, с которым модель может что-то сделать, вместо стека вызовов.

strict: true в определении функции OpenAI просит провайдера ограничить декодирование схемой. К флагу strict в tsconfig TypeScript он отношения не имеет. Это похоже на guided decoding в vLLM, хотя поддерживаемые схемы и детали принудительного применения различаются. Флаг устраняет один вид сбоя, но самостоятельно развёрнутый эндпоинт может его игнорировать. Аргументы также должны проходить JSON.parse.

Цикл вызывает /chat/completions, потому что сопутствующий репозиторий ориентирован на OpenAI-совместимые серверы. vLLM, SGLang и Ollama документируют этот эндпоинт, поэтому OPENAI_BASE_URL может направить тот же клиент на любой из них. Покрытие ими Responses API различается и меняется от релиза к релизу. Если вы контролируете обе стороны, перед выбором между двумя API проверьте актуальную страницу совместимости сервера.

Затем переключитесь на AI SDK и поймите, что вы обменяли

В сопутствующем репозитории тот же агент реализован и на Vercel AI SDK, чтобы можно было сравнить обе версии. Обе выдают значения AgentEvent, но сейчас по-разному обрабатывают ошибки, завершающие выполнение. Необязательный адаптер SDK выдаёт событие error со значением String(part.error) и завершается без done. Текущий HTTP-маршрут пересылает эти подробности и сохраняет запуск как succeeded, поскольку генератор завершился без исключения. Прежде чем заменить стандартный цикл этим адаптером, считайте error или закрытие потока без done неудачным запуском, удаляйте чувствительные подробности из сообщений об ошибках для клиента и проверяйте сохранённый статус вместе с завершающим событием.

Сопутствующий репозиторий использует AI SDK 7. Реализация на фреймворке находится в packages/agent-core/src/loop-ai-sdk.ts:

const result = streamText({
    model: provider.chatModel(options.model),
    prompt: options.message,
    tools: aiSdkTools,          // Zod schemas passed straight through
    stopWhen: stepCountIs(options.maxSteps ?? 6),
});

for await (const part of result.fullStream) {
    switch (part.type) {
        case "text-delta": yield { type: "text", delta: part.text }; break;
        case "tool-call":  yield { type: "tool_call", callId: part.toolCallId, /* ... */ }; break;
        // ...
    }
}

SDK убирает пять фрагментов прикладного кода:

• аккумулятор фрагментов
• JSON.parse и путь его ошибки
• вызов, который прогоняет Zod по разобранным аргументам
• сборку сообщений под конкретного провайдера
• счётчик шагов

stopWhen принимает несколько условий, включая лимит шагов или конкретный tool call. Цикл for await не меняется, когда меняется политика остановки.

Вы теряете прямой контроль над сбоем валидации. Рукописный цикл решает, что увидит модель после отклонённого вызова. Сопутствующий сервис не задаёт параметр SDK repairToolCall, поэтому действует поведение по умолчанию: если разрешено больше одного шага, AI SDK 7 отправляет неудавшийся вызов обратно модели на следующем шаге. Передайте колбэк repairToolCall, чтобы исправлять вызов самому, например вторым вызовом модели.

Обмен работает и в обратную сторону. В версии с SDK смена провайдера локализуется в адаптере провайдера. Она всё равно требует соответствующего пакета провайдера, учётных данных, конфигурации и интеграционных тестов. В «сыром» цикле специфичную для провайдера обработку запроса и стрима вам нужно менять самому.

Для нового сервиса начинайте с SDK, если вам не нужно контролировать эти пути отказа. Рукописная версия даёт небольшую реализацию, которую можно изучить и сравнить с этим выбором.

Стриминг по HTTP: Hono и SSE

Маршруты Hono читаются как маршруты FastAPI. Единственное добавление — zValidator, который делает то, что FastAPI получает бесплатно из аннотаций типов в сигнатуре обработчика. c в обработчике ниже — это контекст запроса Hono, объект, который FastAPI разносит по вашим параметрам. deps — набор зависимостей, которые передаются приложению при создании, а не импортируются напрямую. runAgent — одна из них, и раздел о тестировании показывает, что это даёт.

Сокращённо из apps/api/src/app.ts:

app.post("/v1/chat", zValidator("json", ChatRequestSchema), (c) => {
    const body = c.req.valid("json");
    const log = deps.logger.child({ route: "chat" });

    return streamSSE(c, async (stream) => {
        let text = "";
        let done: Extract<AgentEvent, { type: "done" }> | undefined;
        try {
            // The full route wraps this loop in a `withSpan` tracing span.
            for await (const event of deps.runAgent({
                message: body.message,
                maxSteps: body.maxSteps,
            })) {
                if (event.type === "text") text += event.delta;
                if (event.type === "done") done = event;

                await stream.writeSSE({
                    event: event.type,
                    data: JSON.stringify(event),
                });
            }
        } catch (error) {
            log.error({ err: error }, "agent run failed");
            await stream.writeSSE({
                event: "error",
                data: JSON.stringify({
                    type: "error",
                    message: "Agent run failed",
                }),
            });
            return;
        }

        await deps.storage.createRun({
            kind: "chat",
            status:
                done?.stopReason === "max_steps" ? "truncated" : "succeeded",
            input: { message: body.message },
            output: { text },
        });
    });
});

Маршрут сохраняет итоговое событие done. Цикл, остановившийся на max_steps, завершается нормально, поэтому маршрут сохраняет его как truncated, а не succeeded.

zValidator валидирует тело запроса и придаёт c.req.valid("json") тип, который производит схема. Если его пропустить, тело получает тип any — лазейку TypeScript, где любое обращение к свойству компилируется и ничего не проверяется. Это сводит на нет выигрыш схемы в типобезопасности.

Этот маршрут использует Server-Sent Events вместо WebSocket. Сервер держит HTTP-ответ открытым, пока пишет фреймы event: <name> и data: <json>, а после последнего события закрывает его. Трафик идёт от сервера к клиенту, что соответствует этому агентному стриму. WebSocket добавил бы двусторонний обмен сообщениями и смену протокола, которые этому маршруту не нужны.

В стандартном цикле исключение во время стриминга меняет способ сообщения об ошибках HTTP. После отправки первого фрейма со статусом 200 сервер не может заменить ответ на 500. Блок catch записывает перехваченную ошибку в лог, отправляет клиенту событие ошибки с постоянным сообщением и завершается. Этот возврат пропускает storage.createRun, поэтому исключение из генератора не оставляет строки. Этот catch не выполняется, когда необязательный адаптер SDK выдаёт событие error и завершается. Если нужны неудачные запуски для отладки или эвалов, запишите в блоке catch строку failed с частичным текстом.

Схема прослеживает один запрос до трёх его исходов. Код статуса во всех трёх случаях один и тот же, поэтому их различает только последний фрейм.

Этот путь покрыт тестом. Генератор отдаёт одну текстовую дельту и затем бросает исключение. Ответ остаётся 200, а его последний фрейм — событие error с константным сообщением Agent run failed. Логгер сохраняет пойманную ошибку для диагностики на стороне сервера. Любой клиент, который проверяет только код статуса, сообщит об успехе для неудавшегося запуска.

В app.ts есть ещё два небольших решения, которые стоит объяснить. Он рассматривает /healthz как эндпоинт liveness, поэтому этот маршрут намеренно не обращается к Postgres. Сбой liveness во время недоступности базы данных мог бы перезапустить все реплики, не починив зависимость. Добавьте отдельную проверку readiness, когда оркестратор должен перестать направлять трафик на экземпляр, который не может достучаться до Postgres. Пути ошибок логируют пойманную ошибку, но возвращают константную строку. Отдавать error.message в теле ответа — так строки подключения оказываются в чужом браузере.

Часть про Celery — без Celery

Долгим задачам не место в обработчике запроса. POST /v1/sweeps валидирует модель, длину последовательности и список размеров батча, вставляет строку queued и возвращает 202 с id запуска. Воркер забирает строку, считает размер KV cache для каждого размера батча и сохраняет результат. GET /v1/runs/:id читает его.

Верхняя половина показывает состояния, через которые проходит строка sweep. Временная шкала показывает повторный захват устаревшей блокировки, который объясняется ниже: через пять минут второй воркер забирает задачу, которая всё ещё выполняется.

Здесь нет ни Redis, ни BullMQ. PostgreSQL документирует SKIP LOCKED для нескольких потребителей таблицы, похожей на очередь. Это выражение даёт небольшому сервису очередь at-least-once в одной таблице. Она транзакционна вместе с остальными вашими записями, и это на один сервис меньше в docker-compose.yml.

Запрос захвата из packages/agent-core/src/db/storage.ts выглядит так:

const [candidate] = await tx
    .select({ id: runs.id })
    .from(runs)
    // The real query also picks up rows whose lock went stale; trimmed here.
    .where(and(eq(runs.kind, kind), eq(runs.status, "queued")))
    .orderBy(runs.createdAt)
    .limit(1)
    // `.for()` adds a locking clause; `{ skipLocked: true }` adds SKIP LOCKED.
    .for("update", { skipLocked: true });

Строка блокируется на время транзакции, а любой параллельный воркер, выполняющий тот же запрос, пропускает её, а не ждёт. Два одновременных захвата поэтому не получают одну и ту же неустаревшую строку. Интеграционный тест запускает два захвата одновременно через Promise.all и проверяет, что они возвращают разные строки. Наивная версия, SELECT ... LIMIT 1 с последующим UPDATE, этот тест не проходит: обе транзакции читают одну и ту же строку до того, как любая из них запишет, поэтому обе запускают одну и ту же задачу.

Верхняя панель — гонка, которую ловит тест; нижняя панель — приведённый выше запрос захвата.

Это выполнение at-least-once: задача может выполниться больше одного раза. Полный запрос также заново забирает строку running, когда её блокировке больше пяти минут, и ничто не продлевает эту аренду. Выполняющаяся задача, которая работает дольше пяти минут, поэтому может быть захвачена дважды. Делайте задачи идемпотентными. Для долгой работы добавьте heartbeat аренды или задайте порог устаревшей блокировки выше максимального времени выполнения.

Добавьте BullMQ, когда вам нужны отложенные задачи, повторяющиеся расписания, приоритеты, ограничения скорости или дашборд. Эквивалентный шаг в Python — переход от таблицы в базе данных к Celery. До этого Redis — ещё один сервис, который нужно запускать и мониторить.

Воркер заново валидирует то, что читает из jsonb:

// The row was validated on the way in, but it has been through a database.
// A stored row can outlive the schema version that accepted it.
// The job is a batch-size sweep. Zod's own `.parse` throws; the poll loop
// catches that and marks the run failed.
const input = SweepRequestSchema.parse(run.input);

Набор тестов также передаёт воркеру строку sweep, у которой seqLen равно строке "lots". Воркер помечает запуск как неудавшийся и продолжает опрос, а не падает и не повторяет вечно ту же некорректную строку.

Работа на CPU показывает ещё одно ограничение Node. Синхронный колбэк выполняется в потоке цикла событий и не вытесняется. Цикл for с арифметикой, работающий две секунды, блокирует на эти две секунды каждый запрос, таймер и проверку liveness в этом процессе. Плотный цикл внутри async def блокирует asyncio точно так же. Оба рантайма требуют явно выносить CPU-работу.

await setTimeout(0) из node:timers/promises (префикс node: означает стандартную библиотеку, так что node:timers относится к Node как os к Python) — это await asyncio.sleep(0). Sweep уступает управление после каждого размера батча, чтобы процесс воркера мог обслуживать таймеры и другие колбэки. Уступка управления не делает CPU-работу параллельной. Node worker_threads могут выполнять JavaScript параллельно. Для чистой CPU-работы на Python в обычной сборке CPython с GIL используйте пул процессов, а не пул потоков. Этот сервис не использует ни то, ни другое. Держите тяжёлые вычисления в Python, где уже есть нужные библиотеки, и выносите их из цикла событий API.

Один инструмент, три потребителя

У EstimateKvCacheInput три потребителя:

• Рукописный цикл преобразует её через z.toJSONSchema.
• AI SDK получает её без изменений.
• MCP-сервер публикует её форму.

Ради такого повторного использования и существует packages/schemas.

Из apps/mcp/src/index.ts:

server.registerTool(
    "estimate_kv_cache",
    {
        description: "Estimate KV-cache VRAM in GiB for a served model...",
        // A Zod object schema keeps the map of fields you passed in on
        // `.shape`. This SDK wants that map, not the schema wrapped around it.
        inputSchema: EstimateKvCacheInput.shape,
    },
    async ({ model, seqLen, batchSize }) => {
        /* ... */
    },
);

await server.connect(new StdioServerTransport());

Перед подключением клиента важны две детали. Во-первых, сервер, запущенный таким образом, использует собственные stdin и stdout для связи с клиентом. Каждая строка — это сообщение JSON-RPC. Случайный console.log, JavaScript-аналог print, тогда портит сообщение. Клиент отключается с ошибкой разбора, в которой не назван ни один файл. Направляйте всю диагностику в stderr.

Во-вторых, сбой предметной области должен возвращать isError: true с сообщением. Вызывающая модель тогда может исправить вызов, как и после невалидных аргументов инструмента в агентном цикле.

Сервер можно запустить через printf и конвейер, и это стоит сделать один раз, прежде чем подключать настоящий клиент:

printf '%s\n' \
  '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"probe","version":"1"}}}' \
  '{"jsonrpc":"2.0","method":"notifications/initialized"}' \
  '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"estimate_kv_cache","arguments":{"model":"llama-3.1-70b","seqLen":8192,"batchSize":4}}}' \
  | pnpm -s mcp   # -s suppresses pnpm's own output so only JSON-RPC comes back

Проба использует ту же версию протокола, что и README сопутствующего репозитория. Для настоящего клиента используйте SDK, а не поддерживайте сообщения JSON-RPC вручную.

Тестируем агента без API-ключа

Vitest выполняет роль pytest, но структура другая. describe группирует связанные тесты. it и test определяют по одному тест-кейсу. test.each близок к parametrize, beforeEach задаёт подготовку для каждого теста, vi.fn() создаёт мок-функцию, а describe.skipIf условно пропускает группу.

Тесты агента опираются на одно решение: runAgent принимает клиент OpenAI как параметр, а не создаёт его сам. Тестовый клиент — объект с методом chat.completions.create, возвращающим скриптованный асинхронный итерируемый объект:

function fakeClient(scripts: Chunk[][]): OpenAI {
    let call = 0;
    return {
        chat: {
            completions: {
                create: async () => {
                    const script = scripts[call++] ?? [];
                    // Defines an async generator and calls it on the same
                    // line, so `create` hands back something you can
                    // `for await` over, which is the shape a real streaming
                    // response has.
                    return (async function* () {
                        for (const chunk of script) yield chunk;
                    })();
                },
            },
        },
        // TypeScript refuses a direct cast between unrelated shapes, so you
        // cast through `unknown` first. This disables type checking for that
        // value, so keep it to test code.
    } as unknown as OpenAI;
}

Тесты разрезают одну строку JSON-аргументов на чанки и обрабатывают две операции tool call в одном ответе. Они также покрывают невалидный batchSize, некорректный JSON, неизвестные имена инструментов и модель, которая продолжает вызывать инструменты, пока её не остановит maxSteps. Файл с тестами выполняется заметно быстрее секунды, без сети и без ключа.

Интеграционные тесты с Postgres используют describe.skipIf(!process.env.DATABASE_URL), поэтому pnpm test работает на свежем клоне без запущенного Postgres, а CI включает их, задавая эту переменную. В репозитории 42 теста. Тридцать восемь из них работают без Postgres и API-ключа.

Логируйте структурированные события с Pino

Pino выполняет ту же роль, что structlog: один JSON-объект на строку, дочерние логгеры с привязанными полями и явное скрытие данных. Сопутствующий репозиторий настраивает его в packages/observability/src/logger.ts:

const log = pino({
    redact: {
        paths: [
            "req.headers.authorization",
            "apiKey",
            "OPENAI_API_KEY",
            "*.apiKey",
        ],
        censor: "[redacted]",
    },
});

Без скрытия log.info({ req }, "...") может скопировать заголовок Authorization в бэкенд логов.

Трейсите прикладную работу вручную созданными спанами

Сопутствующий репозиторий использует OpenTelemetry для трёх прикладных спанов: agent.run, agent.tool и worker.sweep. Автоматическую инструментацию HTTP или Postgres он не устанавливает. startTracing() в packages/observability/src/tracing.ts создаёт NodeSDK с экспортёром трейсов OTLP. Если OTEL_EXPORTER_OTLP_ENDPOINT не задан, трейсинг остаётся выключенным.

Саму работу оборачивает withSpan() из того же файла:

return tracer.startActiveSpan(name, { attributes }, async (span) => {
    try {
        return await fn(span);
    } finally {
        span.end();
    }
});

В JavaScript нет синтаксиса контекстных менеджеров в стиле Python. Здесь колбэк — это блок, который окружил бы контекстный менеджер Python. Полный хелпер также записывает исключения и задаёт статус спана перед повторным броском.

Автоматические спаны HTTP и базы данных — отдельная возможность. Им нужны соответствующие пакеты инструментации и инициализация до загрузки инструментируемых модулей. Добавляйте это, только когда такие спаны полезны, и следуйте настройке OpenTelemetry Node SDK для точных версий пакетов, которые вы деплоите.

Упакуйте монорепозиторий в Docker

Используйте готовый Dockerfile. Контейнер запускает API с загрузчиком tsx. При деплое вам не нужно выбирать или вызывать средство запуска TypeScript.

Сборка использует pnpm fetch, чтобы загрузки зависимостей оставались в кэше, пока не изменится lock-файл. Затем pnpm deploy копирует API и его продакшен-зависимости в самодостаточный каталог. Стадия рантайма работает от пользователя node без прав root, а её CMD в exec-форме позволяет API получать SIGTERM напрямую для корректного завершения.

Путь на три недели

Опытным Python-инженерам можно пропустить материалы, которые учат переменным и циклам. Эта последовательность сосредоточена на том, что отличается от Python. Выполняйте задание из столбца «Постройте» в каждой строке; чтение лишь поддерживает его.

Начните с бесплатных туториалов Total TypeScript. Платите за продвинутые материалы, только когда работаете с обобщениями и условными типами уровня библиотек. Пропускайте любые курсы «введение в JavaScript» и всё, что связано с React, если этого не требует продукт.

Для обзора продакшен-практик goldbergyoni/nodebestpractices — обширный чек-лист, который ведёт сообщество. Проверяйте советы, влияющие на поведение в рантайме или на безопасность, по актуальной документации Node.

Прежде чем адаптировать сервис

Офлайн-тесты проверяют поток управления сервиса на заранее заданных ответах модели. Они не показывают, насколько надёжно модель в реальном сервисе выполнит ваши задачи. Помните об этом различии, когда замените демо-инструменты настоящими.

Валидируйте каждое значение, которое приходит извне: окружение, тело HTTP-запроса, аргументы tool calls от модели и строки, прочитанные из jsonb.

Затем задайте политику завершения: решите, принимает ли ваше приложение запуски truncated, повторяет их или отправляет на проверку, и протестируйте событие и сохранённый статус вместе. Если задачи могут работать дольше порога устаревшей блокировки, добавьте heartbeat аренды, прежде чем добавлять побочные эффекты.

Когда эти поведения станут явными, добавьте строку failed, описанную в разделе о стриминге, и используйте неудавшиеся и усечённые запуски для построения отдельного набора эвалов агента. Именно там проверяется выполнение задач; офлайн-набор остаётся более быстрой проверкой сломанного стриминга, валидации и сохранения данных.

Источники

Демо-репозиторий

• slavadubrov/typescript-agent-service - монорепозиторий, который используется во всей статье: API на Hono с SSE, две реализации агентного цикла, хранилище на Drizzle, воркер, MCP-сервер и 42 теста

Рантайм и язык

• Запуск TypeScript напрямую в Node.js - ограниченная поддержка TypeScript в Node и её ограничение для node_modules
• Параметры компилятора TypeScript - strict и другие проверки, настроенные в сопутствующем репозитории
• Руководство MDN по JavaScript - справочник по языку, который стоит держать открытым
• javascript.info - современный учебник по JavaScript. Прочитайте главы о модулях и промисах.

Инструменты

• pnpm и установка pnpm - менеджер пакетов, рабочие пространства и настройка
• Biome - линтинг, форматирование и сортировка импортов в одном бинарнике
• Vitest - средство запуска тестов, которому не нужна настройка трансформации
• Total TypeScript - бесплатные туториалы и платный курс по продвинутым типам

Библиотеки

• Zod - валидация схем и вывод типов. Версия 4 включает z.toJSONSchema.
• Hono - HTTP-фреймворк на веб-стандартах
• Drizzle ORM - ORM для TypeScript с упором на SQL и миграциями drizzle-kit
• Документация PostgreSQL по SELECT - выражение блокировки FOR UPDATE ... SKIP LOCKED
• BullMQ - очередь на Redis на случай, когда таблицы в базе данных недостаточно

AI и агенты

• Vercel AI SDK - streamText, tool, stopWhen и адаптеры провайдеров
• openai/openai-node - официальный клиент для TypeScript
• MCP TypeScript SDK и спецификация MCP - создание серверов и клиентов

Соглашения

• goldbergyoni/nodebestpractices - чек-лист продакшен-практик, который ведёт сообщество