Building agentic AI? I co-run a 6-week cohort where you ship a production-ready agent, not another API wrapper.
In my trend digest article I shared a quick tool to keep on top of tech trends, but it's a one-way street: the model gives information, but I still have to decide what to do with it. Let's build the second half: a Telegram bot that shows me each story, guesses a tag, and lets me confirm or overrule it with one tap.
Human-in-the-loop (HITL): the model proposes, you decide #
AI makes suggestions but it can hallucinate, so it's important to have a human in the loop to catch mistakes. The model does the work of categorizing, the human makes the final decision. This is a good example of the control layer above the model and it's where you can make AI more reliable.
This is what we teach in week 4 of our Agentic AI cohort where things come together: expense parsing, AI category suggestion, and the human in the loop to confirm it. This requires the bot to keep state, route responses, and a way to be wrong gracefully. Below is a smaller version so you can get a taste for how this works.
We'll build it in seven steps. Clone the repo to follow along step by step.
Step 1: create the bot and get a token #
Telegram bots are created by another bot. Open Telegram and search for @BotFather (it has a blue checkmark):
- Send
/newbot
. - Give it a display name (anything).
- Give it a username ending in
bot
that is globally unique, e.g.alice_trend_bot
.
BotFather replies with a token like 123456789:ABCdef...
. Treat it like a password. Put it in a .env
file next to your script (or export
in your shell), together with your OpenAI key:
TELEGRAM_BOT_TOKEN=123456789:ABCdef...
OPENAI_API_KEY=sk-...
If the token ever leaks, send /revoke
to BotFather for a fresh one.
Step 2: the dependencies #
The whole thing is one file. I use a PEP 723 header so uv run
resolves everything into its own environment, no virtualenv to manage. Put this at the top of trend_triage_bot.py
:
If you would rather build this inside an existing project, the equivalent is:
uv init && uv add python-telegram-bot openai httpx python-decouple pydantic
Then the imports and a few constants:
import json
import logging
from pathlib import Path
from typing import Literal, Protocol
import httpx
from decouple import config
from openai import AsyncOpenAI
from pydantic import BaseModel
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Message, Update
from telegram.ext import (
Application,
CallbackQueryHandler,
CommandHandler,
ContextTypes,
)
logger = logging.getLogger(__name__)
TAGS = ["read", "lib", "tool", "skip"]
DEFAULT_TOPIC = "rust"
READING_LIST = Path("reading_list.jsonl")
LOBSTERS_FEED = "https://lobste.rs/t/{tag}.json"
Step 3: fetch the stories #
Lobsters has a per-tag JSON feed, no auth required: https://lobste.rs/t/rust.json
returns the latest Rust-tagged stories, .../t/python.json
the Python ones, and so on. It's a tighter, more engineering-focused signal than a broad keyword search, and parameterizing the tag is what lets /digest rust
and /digest python
hit the same code. A Story
is just a title and a URL.
Let's set up the model and fetch the latest five stories for a given tag:
from pydantic import BaseModel, HttpUrl
class Story(BaseModel):
title: str
url: HttpUrl
async def fetch_stories(tag: str, *, limit: int = 5) -> list[Story]:
async with httpx.AsyncClient(
timeout=10, headers={"User-Agent": "trend-triage-bot"}
) as http:
response = await http.get(LOBSTERS_FEED.format(tag=tag))
response.raise_for_status()
return [
Story.model_validate(
{
"title": story["title"],
"url": story["url"] or f"https://lobste.rs/s/{story['short_id']}",
}
)
for story in response.json()[:limit]
if story.get("title")
]
Two small details: Lobsters expects a User-Agent
header, and a text/discussion post has an empty url
, so we fall back to its comments page (/s/{short_id}
), the same pattern you'd use for an HN self-post.
The *
in the function signature makes the limit
keyword-only, so you have to call fetch_stories("rust", limit=10)
, which is a nice safeguard against accidentally changing the default.
Step 4: let the LLM propose a tag #
The model picks one of TAGS
. As the digest topic is variable (/digest rust
, /digest python
), the tags have to be topic-agnostic, so they describe what a story is (read
/ lib
/ tool
), not anything Rust- or Python-specific.
Content-type beats intent here: "is this a tool or a library" is answerable from a headline, whereas "will I read this or build with it" depends on me, not the title. And a tag the model can't infer is a tag you end up correcting every time.
And using structured outputs I get typed values back, not strings I have to parse and second-guess; consistent data types are the foundation of reliable AI.
SYSTEM = (
"Tag this software/tech headline with one of: "
"read (an article, post, or tutorial), "
"lib (a library, framework, or package you import), "
"tool (a CLI, app, or utility you run). "
"Use 'skip' only if it is off-topic or clickbait."
)
class TagChoice(BaseModel):
tag: Literal["read", "lib", "tool", "skip"]
class Classifier(Protocol):
async def tag(self, story: Story) -> str: ...
class OpenAIClassifier:
def __init__(self, api_key: str, model: str = "gpt-4o-mini") -> None:
self._client = AsyncOpenAI(api_key=api_key)
self._model = model
async def tag(self, story: Story) -> str:
completion = await self._client.beta.chat.completions.parse(
model=self._model,
messages=[
{"role": "system", "content": SYSTEM},
{"role": "user", "content": story.title},
],
response_format=TagChoice,
)
choice = completion.choices[0].message.parsed
return choice.tag if choice else "skip"
def _build_classifier() -> Classifier:
return OpenAIClassifier(config("OPENAI_API_KEY"))
_build_classifier()
constructs the client at runtime, not at import; we call it once in main()
and stash the result (more on that in step 7).
This decoupling allows a test to inject a fake tagger without touching OpenAI. It's the same lazy-wiring trick I used demonstrating the repository pattern. The Protocol
means any class with an async tag()
method drops in. Protocols are more flexible here, because they don't require inheritance like ABCs do, so the test double doesn't have to know about the real classifier at all.
Filing a tagged story is a one-liner to a JSONL file. JSONL (or JSON Lines) is a way to store structured data; each line contains a single, valid JSON object.
def save_to_reading_list(story: Story, tag: str) -> None:
with READING_LIST.open("a") as f:
f.write(json.dumps({"tag": tag, **story.model_dump(mode="json")}) + "\n")
Note that model_dump()
hands back a pydantic Url
object (HttpUrl
) that json.dumps
can't serialize; mode="json"
coerces it to a string first.
Step 5: the keyboard that highlights the guess #
The AI's pick is prefixed with >>
, but every other tag is one tap away. The callback_data
stays plain (tag:read
) so the handler never has to strip the decoration:
def triage_keyboard(suggested: str) -> InlineKeyboardMarkup:
buttons = [
InlineKeyboardButton(
f">> {tag}" if tag == suggested else tag,
callback_data=f"tag:{tag}",
)
for tag in TAGS
]
rows = [buttons[i : i + 3] for i in range(0, len(buttons), 3)]
return InlineKeyboardMarkup(rows)
The marker goes in front, and that detail matters: Telegram clips long button labels from the end, so my first attempt, wrapping the tag in >> tool <<
, showed up as >> tool…
with the closing marker eaten. The kind of bug you only catch by testing it on a real phone.
Step 6: two steps, one stashed queue #
A simple bot is stateless: message in, reply out. This one is not. Step one (the /digest
command) fetches and shows the first story; step two fires later, when I tap a button, and needs the queue from step one. context.user_data
is a per-user dict the library keeps between handler calls, so I park the queue there:
async def start_digest(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if update.message is None or context.user_data is None:
return
topic = context.args[0].lower() if context.args else DEFAULT_TOPIC
await update.message.reply_text(f"Fetching today's {topic} stories...")
try:
context.user_data["queue"] = await fetch_stories(topic)
except httpx.HTTPStatusError:
await update.message.reply_text(
f"No feed for '{topic}'. Try a Lobsters tag like rust, python, or go."
)
return
except httpx.RequestError:
await update.message.reply_text(
"Couldn't reach Lobsters right now — try again in a bit."
)
return
await show_next(update.message, context)
async def show_next(message: Message, context: ContextTypes.DEFAULT_TYPE) -> None:
queue: list[Story] = (context.user_data or {}).get("queue", [])
if not queue:
await message.reply_text("Inbox zero. That's all the trends today.")
return
story = queue[0]
suggested = await context.bot_data["classifier"].tag(story)
await message.reply_text(
f"{story.title}\n{story.url}",
reply_markup=triage_keyboard(suggested),
)
context.args
is whatever followed the command: /digest python
gives ["python"]
, a bare /digest
gives []
and falls back to DEFAULT_TOPIC
.
A typo'd topic is a 404 from Lobsters, so I catch HTTPStatusError
and reply with a helpful message, otherwise the user would just stare at a digest that never arrives. Validate at the boundary where untrusted input enters.
The second except
covers the other failure mode: the request never gets an HTTP response at all. HTTPStatusError
only fires once Lobsters answers with a 4xx/5xx — a connect timeout, read timeout, or DNS failure is an httpx.RequestError
, which is a sibling of HTTPStatusError
, not a subclass. Miss it and a flaky network crashes the handler with a traceback instead of a friendly reply. Catching RequestError
covers every transport-level failure (ConnectTimeout
, ReadTimeout
, ConnectError
) in one branch.
Read user_data
with .get(...)
, never [...]
. It lives in memory, so if the bot restarts mid-flow the dict is empty and you want a graceful reply, not a KeyError
.
context.bot_data
is its per-bot sibling: one dict shared across all users. That makes it the right home for the classifier, which holds no per-user state. We build it once in step 7 and read it back here, so every story reuses the same OpenAI client instead of constructing a fresh one each time.
Step 7: the callback, then wire it up #
When I tap a button Telegram sends a callback query, not a message. Three rules keep it sane, numbered in the code:
async def on_tag(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
query = update.callback_query
if query is None or query.data is None:
return
await query.answer() # 1. stop the spinner, first thing
_, tag = query.data.split(":", 1) # 2. "tag:read" -> "read"
queue = (context.user_data or {}).get("queue", [])
if not queue: # stale button after a restart
await query.edit_message_text("Session expired, send /digest again.")
return
story = queue.pop(0)
if tag != "skip":
save_to_reading_list(story, tag) # the human's final say
await query.edit_message_text( # 3. edit, don't reply
f"Filed under {tag}: {story.title}"
if tag != "skip"
else f"Skipped: {story.title}"
)
if isinstance(query.message, Message):
await show_next(query.message, context)
Call await query.answer()
first or the spinner on the button never stops, even when everything else works. Edit the original message instead of replying, or the dead keyboard sits there inviting a second tap on a story you already filed. The same .get(...)
-not-[...]
rule applies here: an old keyboard from before a restart can still send a tap, and you want a "send /digest again" nudge, not a KeyError
.
Routing is by prefix. The pattern="^tag:"
is why a future second keyboard (say setcurrency:EUR
) would not trip this handler:
async def on_error(update: object, context: ContextTypes.DEFAULT_TYPE) -> None:
logger.exception("Handler failed", exc_info=context.error)
def main() -> None:
logging.basicConfig(
format="%(asctime)s %(name)s %(levelname)s %(message)s",
level=logging.INFO,
)
logger.info("Starting trend triage bot, polling for updates")
app = Application.builder().token(config("TELEGRAM_BOT_TOKEN")).build()
app.bot_data["classifier"] = _build_classifier()
app.add_handler(CommandHandler("digest", start_digest))
app.add_handler(CallbackQueryHandler(on_tag, pattern="^tag:"))
app.add_error_handler(on_error)
app.run_polling()
if __name__ == "__main__":
main()
Three things to notice here in main()
:
logging.basicConfig(...)
turns on output:run_polling()
blocks silently otherwise, so without it a freshly started bot looks dead in the terminal even though it's happily polling. -
app.bot_data["classifier"]
builds the OpenAI client once instead of per story. -
add_error_handler
means a network blip or a rate limit gets logged throughon_error
rather than vanishing into the framework.
Run it #
$ export OPENAI_API_KEY=sk-proj-...
$ export TELEGRAM_BOT_TOKEN=...
$ uv run trend_triage_bot.py
2026-06-01 13:19:42,866 __main__ INFO Starting trend triage bot, polling for updates
2026-06-01 13:19:43,190 httpx INFO HTTP Request: POST https://api.telegram.org/bot.../getMe "HTTP/1.1 200 OK"
2026-06-01 13:19:43,242 httpx INFO HTTP Request: POST https://api.telegram.org/bot.../deleteWebhook "HTTP/1.1 200 OK"
2026-06-01 13:19:43,244 telegram.ext.Application INFO Application started
Open your bot in Telegram and send /digest
for the default topic, or use a tag like /digest python
, /digest rust
or any Lobsters tag.
The bot walks you through today's stories one at a time. Tap the highlighted tag to accept the model's guess, or any other tag to overrule it, until you hit inbox zero:
Same bot, any topic: finish the Rust queue, then /digest python
and triage that, no code change:
Filed stories land in reading_list.jsonl
:
{"tag": "read", "title": "One year of Roto, the compiled scripting language for Rust", "url": "https://blog.nlnetlabs.nl/one-year-of-roto-the-compiled-scripting-language-for-rust/"}
{"tag": "lib", "title": "Announcing Rust 1.96.0", "url": "https://blog.rust-lang.org/2026/05/28/Rust-1.96.0/"}
{"tag": "read", "title": "What kache actually caches", "url": "https://kunobi.ninja/blog/what-kache-actually-caches"}
{"tag": "read", "title": "Creusot helps you prove your Rust code is correct", "url": "https://github.com/creusot-rs/creusot/tree/master"}
{"tag": "tool", "title": "uv must be installed to build a standalone Python distribution", "url": "https://github.com/astral-sh/python-build-standalone/commit/c9c40c56eb53136587f0a32382cad9e5cd8d184a"}
{"tag": "tool", "title": "SPy: an interpreter and a compiler for a statically typed variant of Python", "url": "https://github.com/spylang/spy"}
{"tag": "read", "title": "Opaque Types in Python", "url": "https://blog.glyph.im/2026/05/opaque-types-in-python.html"}
{"tag": "read", "title": "uv is fantastic, but its package management UX is a mess", "url": "https://www.loopwerk.io/articles/2026/uv-ux-mess/"}
That file is the actual output; one JSON object per line, ready to feed into whatever reads it next. You can find the repo here.
The interesting question is not whether the model can tag a headline. It's pretty accurate, but it can get it wrong, and that's where you want to have a human in the loop. This has been a simple example to show the flow, but real workflows might involve more interesting things like approving trades, triaging support tickets, or moderating content. The model can do the heavy lifting of making a guess, but the human gets the final say, and that's where the value is.
Keep reading #
Most AI tutorials end at "call the API." This cohort ends with a deployed agent: function calling, structured outputs, three interfaces, Docker, 95%+ test coverage. Six weeks of real engineering, not notebooks. Join the next Agentic AI cohort →