Real-time progress bars with LangChain streaming

LangChain’s streaming callbacks let you surface token-by-token output to users, but the library doesn’t ship a progress bar out of the box. This guide shows how to build a langchain streaming progress bar callback that tracks tokens, handles async iteration, and survives tool calls without blocking the event loop.

Step 1: understand the callback interface

LangChain callbacks fire at three points that matter for progress: on_llm_start, on_llm_new_token, and on_llm_end. The BaseCallbackHandler class defines these hooks. Your handler receives a token: str on each on_llm_new_token call — that’s your atomic unit of progress.

from langchain.callbacks.base import BaseCallbackHandler
from typing import Any, Dict, List, Optional
from uuid import UUID

class TokenCountingHandler(BaseCallbackHandler):
    def __init__(self) -> None:
        self.token_count: int = 0
        self.start_time: Optional[float] = None

    def on_llm_start(
        self,
        serialized: Dict[str, Any],
        prompts: List[str],
        *,
        run_id: UUID,
        parent_run_id: Optional[UUID] = None,
        tags: Optional[List[str]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> None:
        import time
        self.start_time = time.perf_counter()
        self.token_count = 0

    def on_llm_new_token(
        self,
        token: str,
        *,
        run_id: UUID,
        parent_run_id: Optional[UUID] = None,
        tags: Optional[List[str]] = None,
        **kwargs: Any,
    ) -> None:
        self.token_count += 1

    def on_llm_end(
        self,
        response: Any,
        *,
        run_id: UUID,
        parent_run_id: Optional[UUID] = None,
        tags: Optional[List[str]] = None,
        **kwargs: Any,
    ) -> None:
        pass  # handled by the progress bar

This handler is deliberately minimal. It counts tokens and records a start timestamp. The progress bar itself lives outside the callback — separation of concerns keeps the handler testable and the UI portable.

Step 2: wire a synchronous progress bar

For CLI tools and scripts, tqdm is the lowest-friction choice. It renders a clean bar, handles terminal resizing, and computes ETA automatically. Wrap the handler in a context manager that owns the tqdm instance.

from contextlib import contextmanager
from tqdm import tqdm
import time

@contextmanager
def streaming_progress_bar(
    unit: str = "tok",
) -> TokenCountingHandler:
    handler = TokenCountingHandler()
    pbar: Optional[tqdm] = None

    original_on_start = handler.on_llm_start
    original_on_token = handler.on_llm_new_token
    original_on_end = handler.on_llm_end

    def on_llm_start(*args: Any, **kwargs: Any) -> None:
        nonlocal pbar
        original_on_start(*args, **kwargs)
        pbar = tqdm(
            total=0,  # unknown total; we'll update dynamically
            desc=description,
            unit=unit,
            dynamic_ncols=True,
            leave=True,
        )

    def on_llm_new_token(*args: Any, **kwargs: Any) -> None:
        original_on_token(*args, **kwargs)
        if pbar:
            pbar.update(1)
            # update postfix with throughput
            elapsed = time.perf_counter() - handler.start_time
            if elapsed > 0:
                pbar.set_postfix({"tok/s": f"{handler.token_count / elapsed:.1f}"})

    def on_llm_end(*args: Any, **kwargs: Any) -> None:
        original_on_end(*args, **kwargs)
        if pbar:
            pbar.close()

    handler.on_llm_start = on_llm_start  # type: ignore[method-assign]
    handler.on_llm_new_token = on_llm_new_token  # type: ignore[method-assign]
    handler.on_llm_end = on_llm_end  # type: ignore[method-assign]

    try:
        yield handler
    finally:
        if pbar:
            pbar.close()

Usage is a one-liner:

from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage

llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)

with streaming_progress_bar("Writing summary") as handler:
    response = llm([HumanMessage(content="Summarize the plot of Dune in 200 words.")], callbacks=[handler])

print(f"\nDone. {handler.token_count} tokens in {time.perf_counter() - handler.start_time:.2f}s")

Run it. You’ll see a bar that increments per token, a live tokens-per-second rate, and a clean close when the stream ends.

Step 3: support async streaming

Production services use astream or ainvoke to avoid blocking threads. The callback interface is identical, but your progress bar must not call blocking tqdm methods from the event loop. Run the bar in a thread or use an async-friendly library like rich.progress.

Here’s a rich-based async handler. rich renders via a background thread internally, so update calls from the event loop are safe.

from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TaskProgressColumn, TimeElapsedColumn
from rich.console import Console
import asyncio

class AsyncProgressHandler(BaseCallbackHandler):
    def __init__(self, description: str = "Streaming") -> None:
        self.description = description
        self.token_count = 0
        self._progress: Optional[Progress] = None
        self._task_id: Optional[int] = None
        self._console = Console()
        self._start_time: Optional[float] = None

    def on_llm_start(self, *args: Any, **kwargs: Any) -> None:
        import time
        self._start_time = time.perf_counter()
        self.token_count = 0
        self._progress = Progress(
            SpinnerColumn(),
            TextColumn("[bold blue]{task.description}"),
            BarColumn(),
            TaskProgressColumn(),
            TimeElapsedColumn(),
            console=self._console,
            transient=False,
        )
        self._progress.start()
        self._task_id = self._progress.add_task(self.description, total=None)

    def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
        self.token_count += 1
        if self._progress and self._task_id is not None:
            self._progress.update(self._task_id, advance=1)
            elapsed = time.perf_counter() - self._start_time
            if elapsed > 0:
                rate = self.token_count / elapsed
                self._progress.update(self._task_id, description=f"{self.description} ({rate:.1f} tok/s)")

    def on_llm_end(self, *args: Any, **kwargs: Any) -> None:
        if self._progress:
            self._progress.stop()

Async usage:

from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage

async def main() -> None:
    llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)
    handler = AsyncProgressHandler("Generating response")

    async for chunk in llm.astream([HumanMessage(content="Explain quantum entanglement simply.")], callbacks=[handler]):
        pass  # chunks arrive via callback; iterator yields final message

    print(f"\nTotal tokens: {handler.token_count}")

asyncio.run(main())

The total=None tells rich this is an indeterminate bar — it animates until on_llm_end stops it. If your provider returns usage metadata with completion_tokens, you can switch to a determinate bar by setting total in on_llm_start after peeking at the first chunk’s metadata. Most providers don’t expose that early, so indeterminate is the robust default.

Step 4: handle tool calls and intermediate steps

Agents and chains emit on_tool_start, on_tool_end, and on_agent_action callbacks. A naive progress bar treats every token equally, which misleads users when the model spends 30 seconds in a tool call emitting zero tokens. Extend the handler to show phase-aware progress.

from enum import Enum
from dataclasses import dataclass, field
from typing import Optional

class Phase(Enum):
    IDLE = "idle"
    THINKING = "thinking"
    TOOL_CALL = "tool_call"
    STREAMING = "streaming"

@dataclass
class PhaseAwareHandler(BaseCallbackHandler):
    phase: Phase = Phase.IDLE
    token_count: int = 0
    tool_name: Optional[str] = None
    _progress: Optional[Progress] = None
    _task_id: Optional[int] = None
    _console: Console = field(default_factory=Console)
    _start_time: float = 0.0

    def _ensure_progress(self) -> None:
        if self._progress is None:
            self._progress = Progress(
                SpinnerColumn(),
                TextColumn("[bold]{task.description}"),
                BarColumn(),
                TaskProgressColumn(),
                TimeElapsedColumn(),
                console=self._console,
            )
            self._progress.start()
            self._task_id = self._progress.add_task("Starting…", total=None)

    def on_llm_start(self, *args: Any, **kwargs: Any) -> None:
        import time
        self._start_time = time.perf_counter()
        self.token_count = 0
        self.phase = Phase.THINKING
        self._ensure_progress()
        if self._task_id is not None:
            self._progress.update(self._task_id, description="Thinking…")

    def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
        if self.phase != Phase.STREAMING:
            self.phase = Phase.STREAMING
            if self._task_id is not None:
                self._progress.update(self._task_id, description="Streaming…")
        self.token_count += 1
        self._update_rate()

    def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs: Any) -> None:
        self.phase = Phase.TOOL_CALL
        self.tool_name = serialized.get("name", "tool")
        if self._task_id is not None:
            self._progress.update(self._task_id, description=f"Running {self.tool_name}…")

    def on_tool_end(self, output: str, **kwargs: Any) -> None:
        self.phase = Phase.THINKING
        if self._task_id is not None:
            self._progress.update(self._task_id, description="Processing result…")

    def on_llm_end(self, *args: Any, **kwargs: Any) -> None:
        if self._progress:
            self._progress.stop()

    def _update_rate(self) -> None:
        import time
        if self._task_id is not None and self._start_time:
            elapsed = time.perf_counter() - self._start_time
            if elapsed > 0:
                rate = self.token_count / elapsed
                desc = f"Streaming ({rate:.1f} tok/s)"
                self._progress.update(self._task_id, description=desc, advance=1)

This handler shifts the bar’s label between “Thinking…”, “Running search…”, and “Streaming (42.3 tok/s)” so users understand why the bar paused. The spinner keeps animating during tool calls, signaling liveness.

Step 5: integrate with LangChain’s Runnable interface

Modern LangChain code uses Runnable pipelines (| operator). Callbacks attach via with_config or the config argument at invoke time. The pattern is identical, but you must pass the handler through the chain.

from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser

prompt = ChatPromptTemplate.from_template("Write a haiku about {topic}.")
llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)
chain = prompt | llm | StrOutputParser()

handler = AsyncProgressHandler("Composing haiku")

# config propagates callbacks to every runnable in the chain
result = await chain.ainvoke({"topic": "rust"}, config={"callbacks": [handler]})
print(result)

If you use astream on the chain, each chunk triggers on_llm_new_token for the final model. Intermediate runnables (prompt formatting, parsers) don’t emit token callbacks — only the model does. That’s the behavior you want.

Step 6: production hardening

Three issues bite teams moving from notebook to service:

1. Callback leakage across requests.
Handlers hold state (token_count, start_time). Reusing a handler instance across concurrent requests mixes counts. Instantiate a fresh handler per request, or reset state in on_llm_start.

# Bad: module-level singleton
handler = AsyncProgressHandler()

# Good: per-request factory
def make_handler() -> AsyncProgressHandler:
    return AsyncProgressHandler("Generating")

2. Backpressure on slow clients.
If you forward tokens to a WebSocket or SSE stream slower than the model produces them, the callback queue backs up. Buffer tokens in the handler and drain asynchronously, or apply asyncio.Queue with a max size and drop/backpressure policy.

class BufferedHandler(BaseCallbackHandler):
    def __init__(self, max_queue: int = 1000) -> None:
        self.queue: asyncio.Queue[str] = asyncio.Queue(maxsize=max_queue)
        self.token_count = 0

    def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
        self.token_count += 1
        try:
            self.queue.put_nowait(token)
        except asyncio.QueueFull:
            # policy: drop oldest, or raise, or block with put()
            _ = self.queue.get_nowait()  # drop one
            self.queue.put_nowait(token)

3. Provider fallback mid-stream.
Some gateways (including n4n.ai) automatically fail over to another provider when the primary hits rate limits or errors. The stream may switch models mid-request, resetting token counters or changing tokenizer behavior. Treat on_llm_start as potentially firing more than once per logical request. Reset token_count each time, but preserve a cumulative total if you need end-to-end metrics.

def on_llm_start(self, *args: Any, **kwargs: Any) -> None:
    # called on each provider attempt
    self._start_time = time.perf_counter()
    self.token_count = 0  # reset for this attempt
    self._cumulative_tokens += getattr(self, "token_count", 0)  # preserve prior
    # ... progress bar init

Verification checklist

Run through these to confirm the implementation works end to end:

1. Basic streaming — invoke a streaming model with the synchronous handler. Bar appears, increments per token, closes cleanly, final token count matches response.usage_metadata["output_tokens"] (if provider returns it).
2. Async streaming — astream with AsyncProgressHandler. Bar animates in terminal, no RuntimeError: Event loop is closed or blocking warnings.
3. Tool call phase — run an agent with a slow tool (e.g., HTTP fetch). Bar shows “Running tool…” with spinner, then resumes “Streaming…” when model resumes.
4. Concurrent requests — fire 5 ainvoke calls simultaneously with separate handlers. Each gets its own bar (or logs), no cross-talk in token counts.
5. Cancellation — Ctrl-C during streaming. Handler’s finally block or on_llm_end closes the progress bar without leaving terminal artifacts.
6. Empty response — model returns zero tokens (rare, but possible with strict stop sequences). Bar starts and stops without error; token_count == 0.

What to avoid

• Don’t print inside callbacks. print calls interleave with tqdm/rich output and corrupt the bar. Use the progress library’s API exclusively.
• Don’t assume token == word. A token is ~0.75 English words. If you need word count, decode tokens with the model’s tokenizer (e.g., tiktoken) — but that adds latency. Token count is the honest metric.
• Don’t hardcode model names in the handler. The handler should be model-agnostic. Pass description strings from the caller.
• Don’t forget streaming=True. Without it, on_llm_new_token never fires. The model returns one giant chunk in on_llm_end.

Next steps

You now have a langchain streaming progress bar callback that works synchronously, asynchronously, through tool calls, and under concurrent load. From here:

• Wrap the handler in a FastAPI dependency that streams tokens via Server-Sent Events to a browser progress bar.
• Add a cost_estimate field using per-token pricing from your provider registry.
• Emit structured logs (JSON) alongside the visual bar for observability pipelines.

The callback system is the right abstraction layer — it keeps streaming concerns out of your business logic and lets you swap UIs (CLI, web, TUI) without touching the chain.