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

> Source: <https://belderbos.dev/blog/human-in-the-loop-telegram-bot-python/>
> Published: 2026-06-01 00:00:00+00:00

# 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)