How to detect and handle truncated LLM responses

Truncated outputs waste tokens, break downstream parsing, and silently corrupt results. This guide shows you how to detect truncated llm response conditions programmatically, implement automatic recovery, and add observability so you catch regressions before users do.

Step 1: Know the truncation signals every provider returns

Every OpenAI-compatible endpoint includes a finish_reason field in the response object. The value tells you why generation stopped. You care about three values:

The usage object gives you the token accounting: prompt_tokens, completion_tokens, total_tokens. When finish_reason == "length", completion_tokens will equal your requested max_tokens (or the provider’s ceiling).

A minimal response snippet:

{
  "id": "chatcmpl-abc123",
  "object": "chat.completion",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "The quick brown fox jumps over the lazy dog. The fox then..."
      },
      "finish_reason": "length",
      "logprobs": null
    }
  ],
  "usage": {
    "prompt_tokens": 42,
    "completion_tokens": 256,
    "total_tokens": 298
  }
}

If you see "finish_reason": "length" and completion_tokens == max_tokens, the response was truncated. Some providers also return "finish_reason": "max_tokens" — treat it the same way.

Step 2: Build a detection helper you can reuse

Wrap the check in a tiny function so every call site stays clean. This example uses the official OpenAI Python SDK, but the logic is identical for raw HTTP.

# truncation.py
from dataclasses import dataclass
from openai.types.chat import ChatCompletion

@dataclass
class TruncationInfo:
    truncated: bool
    reason: str
    completion_tokens: int
    max_tokens_requested: int | None

def check_truncation(response: ChatCompletion, max_tokens_requested: int | None) -> TruncationInfo:
    """
    Returns TruncationInfo for the first choice.
    Call this on every non-streaming completion.
    """
    choice = response.choices[0]
    finish_reason = choice.finish_reason
    usage = response.usage
    completion_tokens = usage.completion_tokens if usage else 0

    truncated = False
    reason = "completed"

    if finish_reason == "length":
        truncated = True
        reason = "max_tokens_reached"
    elif finish_reason == "max_tokens":  # some providers
        truncated = True
        reason = "provider_max_tokens"
    elif finish_reason == "content_filter":
        truncated = True
        reason = "content_filter"

    return TruncationInfo(
        truncated=truncated,
        reason=reason,
        completion_tokens=completion_tokens,
        max_tokens_requested=max_tokens_requested,
    )

Usage:

from openai import OpenAI
from truncation import check_truncation

client = OpenAI()

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Write a 500-word essay on..."}],
    max_tokens=256,  # deliberately low to demonstrate
)

info = check_truncation(resp, max_tokens_requested=256)
if info.truncated:
    print(f"Truncated: {info.reason}, got {info.completion_tokens} tokens")

Step 3: Handle streaming responses

Streaming chunks don’t include finish_reason until the final chunk. Accumulate content and inspect the last chunk’s finish_reason.

# streaming_truncation.py
from openai import OpenAI
from truncation import TruncationInfo

def stream_with_truncation_check(
    client: OpenAI,
    *,
    model: str,
    messages: list[dict],
    max_tokens: int | None,
) -> tuple[str, TruncationInfo]:
    """
    Returns (full_text, TruncationInfo).
    Raises on API error; truncation is reported via TruncationInfo.
    """
    stream = client.chat.completions.create(
        model=model,
        messages=messages,
        max_tokens=max_tokens,
        stream=True,
    )

    collected: list[str] = []
    finish_reason: str | None = None
    completion_tokens = 0

    for chunk in stream:
        if chunk.choices:
            delta = chunk.choices[0].delta
            if delta.content:
                collected.append(delta.content)
            if chunk.choices[0].finish_reason is not None:
                finish_reason = chunk.choices[0].finish_reason

        # usage only appears on the final chunk (OpenAI) or not at all (some providers)
        if chunk.usage:
            completion_tokens = chunk.usage.completion_tokens

    full_text = "".join(collected)

    # Synthesize a TruncationInfo similar to non-streaming path
    truncated = finish_reason in ("length", "max_tokens", "content_filter")
    reason = finish_reason or "unknown"

    return full_text, TruncationInfo(
        truncated=truncated,
        reason=reason,
        completion_tokens=completion_tokens,
        max_tokens_requested=max_tokens,
    )

Step 4: Implement automatic recovery strategies

Once you detect truncation, choose a recovery strategy based on your use case.

Strategy A: Retry with higher max_tokens

Simplest fix when you control the request. Bump the limit and re-send the same prompt.

# retry_strategy.py
from openai import OpenAI
from truncation import check_truncation, TruncationInfo

MAX_RETRIES = 2
TOKEN_MULTIPLIER = 1.5  # 256 -> 384 -> 576

def complete_with_auto_retry(
    client: OpenAI,
    *,
    model: str,
    messages: list[dict],
    max_tokens: int,
    max_retries: int = MAX_RETRIES,
) -> tuple[str, TruncationInfo]:
    current_max = max_tokens
    last_info: TruncationInfo | None = None

    for attempt in range(max_retries + 1):
        resp = client.chat.completions.create(
            model=model,
            messages=messages,
            max_tokens=current_max,
        )
        info = check_truncation(resp, max_tokens_requested=current_max)
        last_info = info

        if not info.truncated:
            return resp.choices[0].message.content or "", info

        # Truncated — prepare next attempt
        current_max = int(current_max * TOKEN_MULTIPLIER)
        # Respect provider ceiling (e.g., 4096 for gpt-4o-mini output)
        # In production, fetch model limits from a config or /models endpoint
        current_max = min(current_max, 4096)

    # Exhausted retries — return what we have plus truncation info
    return resp.choices[0].message.content or "", last_info

Strategy B: Continue generation from the cutoff

When you can’t increase max_tokens (provider hard limit) or want to avoid re-prompting, ask the model to continue. Append the partial assistant message and a continuation prompt.

# continuation_strategy.py
from openai import OpenAI
from truncation import TruncationInfo

CONTINUATION_PROMPT = "Continue exactly where you left off. Do not repeat."

def complete_with_continuation(
    client: OpenAI,
    *,
    model: str,
    messages: list[dict],
    max_tokens: int,
    max_continuations: int = 3,
) -> tuple[str, TruncationInfo]:
    """
    Returns concatenated full text and final TruncationInfo.
    """
    working_messages = messages.copy()
    full_parts: list[str] = []
    final_info: TruncationInfo | None = None

    for _ in range(max_continuations + 1):
        resp = client.chat.completions.create(
            model=model,
            messages=working_messages,
            max_tokens=max_tokens,
        )
        info = check_truncation(resp, max_tokens_requested=max_tokens)
        final_info = info

        content = resp.choices[0].message.content or ""
        full_parts.append(content)
        working_messages.append({"role": "assistant", "content": content})

        if not info.truncated:
            break

        # Truncated — ask for continuation
        working_messages.append({"role": "user", "content": CONTINUATION_PROMPT})

    return "".join(full_parts), final_info

Caveat: Continuation consumes extra prompt tokens (the prior output feeds back in). For very long outputs, Strategy A is cheaper.

Strategy C: Structured output repair

If you expect JSON or code and truncation breaks parsing, combine continuation with a repair pass.

# json_repair.py
import json
from openai import OpenAI

REPAIR_PROMPT = (
    "The following JSON was truncated. Output ONLY the missing closing braces, "
    "brackets, and any trailing commas needed to make it valid JSON. "
    "No explanation, no markdown."
)

def complete_json_with_repair(
    client: OpenAI,
    *,
    model: str,
    messages: list[dict],
    max_tokens: int,
) -> dict:
    text, info = complete_with_continuation(
        client, model=model, messages=messages, max_tokens=max_tokens
    )

    try:
        return json.loads(text)
    except json.JSONDecodeError:
        # One-shot repair attempt
        repair_resp = client.chat.completions.create(
            model=model,
            messages=[
                *messages,
                {"role": "assistant", "content": text},
                {"role": "user", "content": REPAIR_PROMPT},
            ],
            max_tokens=128,
        )
        repair_text = repair_resp.choices[0].message.content or ""
        fixed = text + repair_text
        return json.loads(fixed)

Step 5: Add observability so truncation doesn’t go unnoticed

Detection and recovery are useless if you don’t know how often they fire. Emit metrics on every completion.

# metrics.py
from dataclasses import dataclass
from truncation import TruncationInfo
import time

@dataclass
class CompletionMetrics:
    model: str
    latency_ms: int
    prompt_tokens: int
    completion_tokens: int
    truncated: bool
    truncation_reason: str
    retry_count: int

def record_completion(metrics: CompletionMetrics) -> None:
    """
    Send to your metrics backend (Prometheus, Datadog, CloudWatch, etc.).
    Example uses a hypothetical `metrics_client`.
    """
    # metrics_client.increment("llm.completions.total", tags={"model": metrics.model})
    # metrics_client.histogram("llm.completion.latency_ms", metrics.latency_ms, tags={"model": metrics.model})
    # metrics_client.increment("llm.completion.truncated", tags={
    #     "model": metrics.model,
    #     "reason": metrics.truncation_reason,
    # })
    # metrics_client.histogram("llm.completion.retry_count", metrics.retry_count, tags={"model": metrics.model})
    pass  # replace with real implementation

Wrap your completion call:

# instrumented_completion.py
from openai import OpenAI
from truncation import check_truncation, TruncationInfo
from metrics import CompletionMetrics, record_completion
import time

def instrumented_complete(
    client: OpenAI,
    *,
    model: str,
    messages: list[dict],
    max_tokens: int,
) -> tuple[str, TruncationInfo]:
    start = time.perf_counter()
    resp = client.chat.completions.create(
        model=model,
        messages=messages,
        max_tokens=max_tokens,
    )
    latency_ms = int((time.perf_counter() - start) * 1000)

    info = check_truncation(resp, max_tokens_requested=max_tokens)
    usage = resp.usage

    record_completion(CompletionMetrics(
        model=model,
        latency_ms=latency_ms,
        prompt_tokens=usage.prompt_tokens if usage else 0,
        completion_tokens=usage.completion_tokens if usage else 0,
        truncated=info.truncated,
        truncation_reason=info.reason,
        retry_count=0,  # increment if you wrap retry logic
    ))

    return resp.choices[0].message.content or "", info

Alerting rule example (PromQL):

# Alert if >5% of completions truncated over 5m window
sum(rate(llm_completion_truncated_total[5m])) by (model)
/
sum(rate(llm_completions_total[5m])) by (model)
> 0.05

Step 6: Test your detection and recovery

Write tests that simulate each finish_reason. Mock the SDK response object.

# test_truncation.py
import pytest
from openai.types.chat import ChatCompletion, ChatCompletionMessage, Choice, CompletionUsage
from truncation import check_truncation, TruncationInfo

def make_completion(finish_reason: str, completion_tokens: int, max_tokens: int) -> ChatCompletion:
    return ChatCompletion(
        id="test",
        object="chat.completion",
        created=1234567890,
        model="gpt-4o-mini",
        choices=[
            Choice(
                index=0,
                message=ChatCompletionMessage(role="assistant", content="x" * completion_tokens),
                finish_reason=finish_reason,
                logprobs=None,
            )
        ],
        usage=CompletionUsage(prompt_tokens=10, completion_tokens=completion_tokens, total_tokens=10 + completion_tokens),
    )

def test_detects_length_truncation():
    resp = make_completion("length", completion_tokens=256, max_tokens=256)
    info = check_truncation(resp, max_tokens_requested=256)
    assert info.truncated is True
    assert info.reason == "max_tokens_reached"

def test_detects_provider_max_tokens():
    resp = make_completion("max_tokens", completion_tokens=4096, max_tokens=8192)
    info = check_truncation(resp, max_tokens_requested=8192)
    assert info.truncated is True
    assert info.reason == "provider_max_tokens"

def test_normal_completion_not_truncated():
    resp = make_completion("stop", completion_tokens=100, max_tokens=256)
    info = check_truncation(resp, max_tokens_requested=256)
    assert info.truncated is False
    assert info.reason == "completed"

def test_content_filter_flagged():
    resp = make_completion("content_filter", completion_tokens=50, max_tokens=256)
    info = check_truncation(resp, max_tokens_requested=256)
    assert info.truncated is True
    assert info.reason == "content_filter"

Run with pytest -q test_truncation.py. All four should pass.

Step 7: Handle provider-specific quirks

Not every provider follows the spec perfectly. Common deviations:

If you route through a gateway that normalizes 240+ models to a single OpenAI-compatible contract — such as n4n.ai — you get consistent finish_reason and usage across providers, so the detection logic above works without per-provider branches.

Step 8: Verify end-to-end in staging

Before shipping, run a staged rollout:

1. Shadow mode: Deploy detection + metrics only. No retries. Watch the truncation rate dashboard for 24h.
2. Canary with retry: Enable Strategy A (retry with higher max_tokens) for 5% of traffic. Compare error rates and latency vs control.
3. Full rollout: Once canary shows zero regressions, enable for all traffic. Keep the alert from Step 5 active.

Success criteria:

• Truncation rate drops to near zero (only content_filter remains)
• P99 latency increase < 200ms (one extra round-trip on retry)
• No increase in 429/5xx errors from retry storms

Quick reference: Decision flowchart

completion received
       │
       ▼
finish_reason == "stop" ──► Success, record metrics
       │
       ├─► "length" or "max_tokens"
       │       │
       │       ▼
       │  Can increase max_tokens? ──Yes──► Retry with higher limit (Strategy A)
       │       │
       │       No
       │       │
       │       ▼
       │  Need single contiguous output? ──Yes──► Continuation (Strategy B)
       │       │
       │       No
       │       │
       │       ▼
       │  Structured output (JSON/code)? ──Yes──► Continuation + Repair (Strategy C)
       │       │
       │       No
       │       │
       ▼       ▼
   Log truncation, return partial, alert on rate

What to avoid

• Silently truncating — always surface TruncationInfo to callers so they can decide.
• Infinite retry loops — cap retries, back off exponentially, respect Retry-After headers.
• Assuming max_tokens is the only limit — provider output ceilings (4096, 8192, 16384) are hard stops. Query /models or maintain a config map.
• Ignoring content_filter — it’s a truncation signal. Log it separately; it may indicate prompt problems.

You now have detection, three recovery strategies, observability, tests, and a rollout plan. Pick the strategy that fits your latency budget and output requirements, instrument it, and stop losing completions to silent truncation.