AI Human-in-the-loop: News Digest Triage Telegram Bot

A developer built a Telegram bot that uses AI to categorize tech news stories and lets users confirm or override the tags with a single tap, implementing a human-in-the-loop system to catch AI hallucinations. The bot fetches stories from Lobste.rs, suggests tags via OpenAI, and stores confirmed entries in a JSONL file. The project demonstrates how to add a control layer above AI models for more reliable decision-making.

AI Human-in-the-loop: News Digest Triage Telegram Bot 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 /blog/python-ai-trend-digest-asyncio-protocol/ 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 /blog/control-layer-is-the-product/ and it's where you can make AI more reliable. This is what we teach in week 4 of our Agentic AI cohort https://pythonagenticai.com 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 https://github.com/bbelderbos/telegram-hitl 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 /blog/python-ai-trend-digest-asyncio-protocol/ so uv run resolves everything into its own environment, no virtualenv to manage. Put this at the top of trend triage bot.py : /// script requires-python = " =3.12" dependencies = "python-telegram-bot =21", "openai =1.40", "httpx", "python-decouple", "pydantic", /// 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: python 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 https://lobste.rs 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: python 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 /blog/structured-outputs-pydantic-openai/ 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 /blog/repository-pattern-swappable-data-sources/ . 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. php 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: php 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: php 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: php 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 loading 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: php 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 through on error rather than vanishing into the framework. Run it bash $ 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 https://lobste.rs/tags . 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 https://github.com/bbelderbos/telegram-hitl . 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 → https://pythonagenticai.com