@yaebal/ai

llm superpowers for telegram bots. ctx.ai.replyStream() streams a model's answer straight into the chat — with telegram's native draft animation ("Thinking…", live-updating preview) in private chats and throttled edits with a typing cursor everywhere else. conversation memory, per-user limits, and a provider-agnostic model contract are built in.

install

pnpm add @yaebal/ai

quick start

install ai() once — it adds ctx.ai via derive, no other plugin required. this is the full streamed chat bot from examples/ai-chat: per-user memory on by default, polite rate limits, streamed answers.

import { AiLimitError, ai, openaiCompatible } from "@yaebal/ai";
import { createBot } from "yaebal";

const bot = createBot(process.env.BOT_TOKEN!).install(
  ai({
    model: openaiCompatible({ model: "gpt-4o-mini", apiKey: process.env.OPENAI_API_KEY }),
    system: "you are a concise, friendly assistant.",
    limits: { perUser: "20/h" },
  }),
);

bot.command("reset", async (ctx) => {
  await ctx.ai.reset();
  await ctx.send("clean slate. who are you again?");
});

bot.on("message:text", async (ctx) => {
  try {
    // private chats: telegram's native draft animation ("Thinking…", live preview).
    // groups: a message growing through throttled edits with a ▍ cursor.
    await ctx.ai.replyStream(ctx.text);
  } catch (error) {
    if (error instanceof AiLimitError) {
      const minutes = Math.ceil(error.retryAfterMs / 60_000);
      await ctx.reply(`easy! you hit the hourly limit — try again in ~${minutes}m.`);
      return;
    }
    throw error;
  }
});

await bot.start();

models

any provider works through one of four doors:

import { ai, aiSdk, anthropicModel, customModel, openaiCompatible } from "@yaebal/ai";

// 1. any /chat/completions provider — openai, ollama, openrouter, groq, mistral, deepseek…
openaiCompatible({ model: "llama3.2", baseUrl: "http://localhost:11434/v1" });

// 2. anthropic, natively
anthropicModel({ model: "claude-sonnet-5", apiKey: process.env.ANTHROPIC_API_KEY! });

// 3. the whole vercel ai sdk ecosystem (needs `pnpm add ai` + a provider package)
import { anthropic } from "@ai-sdk/anthropic";
ai({ model: anthropic("claude-sonnet-5") }); // LanguageModel is auto-detected

// 4. anything that yields strings — the escape hatch
customModel(async function* ({ messages }) {
  yield "hello ";
  yield "world";
});

openaiCompatible and anthropicModel are zero-dependency fetch adapters. the ai sdk bridge lazy-loads the ai package only when used, so bots on the built-in adapters never pay for it. a provider http failure throws AiProviderError with the status and response body.

what lands on ctx

ai() is a Plugin<Context, AiControl> — after .install(ai(...)) every handler sees ctx.ai:

prompt is a plain string or an AiMessage[]. every call accepts per-call overrides (system, memory, temperature, maxTokens, signal), and the reply variants take onPart to observe every telegram message as it lands.

// full text, nothing sent — for pipelines and custom rendering
const { text, usage } = await ctx.ai.generate("summarize this: " + ctx.text);

// raw token stream — render it however you like
for await (const piece of ctx.ai.stream(ctx.text)) {
  process.stdout.write(piece);
}

// inspect / wipe this conversation's memory
const turns = await ctx.ai.history();
await ctx.ai.reset();

options

streaming behavior

replyStream() picks the best rendering mechanism telegram offers for the chat:

• private chats — the answer renders through sendMessageDraft: users see telegram's own native "Thinking…" placeholder before the first token, then an animated draft, finalized into a real message via sendMessage.
• groups and channels — drafts don't exist there, so the answer grows by throttled editMessageText calls with a ▍ cursor appended while in flight. a typing action covers the silence before the first token.
• long answers — output past 4096 characters is split at word boundaries (via @yaebal/split) and finalized message-by-message while the stream keeps going.
• formatting — parseMode applies to finalized messages only; in-flight previews stay plain, so a half-open **bold can never 400 the stream. if telegram rejects the finished entities, the text is resent plain instead of being lost.
• resilience — preview ticks are cosmetic: a flood limit or race on a tick is swallowed and the next tick catches up; only finalization errors propagate.

bot.install(
  ai({
    model,
    parseMode: "MarkdownV2",           // applied at finalization only
    streaming: {
      intervalMs: 800,                 // min gap between preview updates
      drafts: true,                    // sendMessageDraft in private chats
      cursor: "▍",                     // in-flight cursor for edit mode, "" disables
    },
  }),
);

the result reports how it went: { text, messages, mode, ticks, aborted } — mode is "draft" or "edit", and aborted is true when the call's signal fired mid-stream. the engine itself is exported as streamToChat(target, source, options) for use outside ctx.ai.

memory

conversation memory is on by default: keyed per user per chat (each group member gets their own thread with the bot), windowed to the last 32 turns, stored in-process. pass any @yaebal/sklad StorageAdapter to persist it across restarts, or memory: false for stateless calls.

import { redisStorage } from "@yaebal/sklad";

bot.install(
  ai({
    model,
    memory: {
      storage: redisStorage({ client: redis }), // any sklad StorageAdapter
      window: 32,                               // stored turns, user + assistant combined
      // key: default is per user per chat — each group member gets their own thread
    },
  }),
);

// or stateless: every call sees only its own prompt
bot.install(ai({ model, memory: false }));

per call, generate() defaults to no memory while reply(), replyStream() and stream() read it; pass { memory: true | false } in the call options to override either way.

limits

limits: { perUser: "20/h" } gates every model call with an in-process sliding window. an exhausted call throws AiLimitError before the provider is hit — retryAfterMs says when the window frees up, so you can answer politely instead of burning tokens.

import { AiLimitError } from "@yaebal/ai";

bot.install(ai({ model, limits: { perUser: "20/h" } })); // also "5/m", "100/d", "1/s"

bot.on("message:text", async (ctx) => {
  try {
    await ctx.ai.replyStream(ctx.text);
  } catch (error) {
    if (error instanceof AiLimitError) {
      await ctx.reply(`try again in ${Math.ceil(error.retryAfterMs / 1000)}s`);
      return;
    }
    throw error;
  }
});

testing

customModel() makes tests deterministic — a generator that yields scripted pieces stands in for the provider, so nothing touches the network. drive the bot with @yaebal/test and assert on the recorded sendMessageDraft / sendMessage / editMessageText calls. the streaming engine is also injectable: streaming.now replaces the throttle clock.

import { ai, customModel } from "@yaebal/ai";
import { createTestEnv } from "@yaebal/test";

// a deterministic model — no network, no api key
const bot = createBot("test-token").install(
  ai({
    model: customModel(async function* () {
      yield "hello ";
      yield "world";
    }),
  }),
);
bot.on("message:text", (ctx) => ctx.ai.replyStream(ctx.text));

const env = createTestEnv(bot);
await env.createUser().sendMessage("hi");
// assert on the recorded sendMessageDraft / sendMessage calls

related

• ai tooling — the mcp server and agent installer shipped by this same package
• examples/ai-chat — the runnable bot this page's quick start is based on
• @yaebal/split — the word-boundary splitter used for >4096 answers
• @yaebal/typing — standalone typing keep-alive for non-ai long calls
• @yaebal/sklad — storage adapters for persistent memory