LangChain Reliability: Streaming, Batching, Retries, Fallbacks, Middleware

Up to now the helpdesk agent has had a deliberately narrow kind of freedom. It may read service ownership and recent incidents, but before writing a ticket note it must pause, show the arguments it intends to use and wait for a human. That arrangement holds up as long as every model and tool call returns promptly. In practice they don't.

One model request might take fifteen seconds. A provider might turn traffic away for a whole minute on a rate limit. A single ticket in a batch might have broken input. A fallback model could answer in a different way than the primary one, and a user might shut the browser while an approved tool is mid-run. Every one of those cases deserves a named behavior with a test behind it.

This step adds five boundaries around the agent. For a single request:

one request
  -> stream honest progress
  -> retry a small transient failure
  -> use a labeled fallback if attempts are exhausted
  -> stop at time and call limits

For a pile of unrelated requests:

many independent requests
  -> run a bounded batch
  -> keep one result or error beside each input

The rule throughout: don't hide a failure behind a cheerful answer.

The ticket we keep testing with

The test incident doesn't change: EU checkout requests break after the 14:05 deployment, with "database connection refused" in the checkout-api logs. The user asks the agent to look up the owner and recent incidents, suggest a next step and add a note to INC-2048 after they sign off.

On the happy path the agent reads the owner and recent incidents, writes a recommendation, proposes a note, waits for approval, writes once and returns the result. Make it harder: submit three more tickets at once, make one provider call fail twice, make one ticket invalid and drop the client connection partway through the first request. Everything below is built to cope with exactly that.

Give failures names first

"The call failed" doesn't tell you what to do. Sort failures into categories that lead to different actions.

• Temporary provider failure. A reset connection, a request timeout, or a provider reply that clearly asks you to come back later. Retry a few times with a pause in between.
• Rate limit. The provider is guarding a quota or its capacity. Follow any retry information it sends, slow the caller down and lower concurrency. Repeating immediately only makes it worse.
• Invalid input. The ticket lacks an ID, the schema doesn't match, or the text exceeds an application limit. Answer with a clear client error, and don't resend the same malformed request.
• Authorization failure. The caller isn't allowed to read a runbook or write to the ticket. Halt there; no switching credentials and no asking another model to work around the refusal.
• Tool business error. The ticket is already closed, the transition isn't allowed, or the same note exists. Turn the tool's answer into a meaningful application error, because retrying could well be wrong.
• Model output validation failure. The provider returned data that breaks the output contract. For a read-only result, one repair attempt can be acceptable, but there still has to be a final failure path.
• Unknown failure. The application couldn't classify the error. Fail closed, record the error type and investigate, and never treat every exception as temporary.

Together these categories determine which failures get retried, which may fall back and which go straight to the caller.

Project layout and setup

The snapshot for this step contains:

langchain-helpdesk/
├── app.py
├── reliability.py
├── batch_runner.py
├── stream_view.py
├── middleware_config.py
└── tests/
    ├── test_reliability.py
    └── test_stream_view.py

Install the packages:

python -m pip install -U \
  langchain \
  langchain-core \
  langgraph \
  pytest

The tests never call a provider.

Streaming delivers; it doesn't verify

A blank screen while tools run and the model thinks is a poor experience. Streaming fixes that by emitting events as the work progresses. LangChain's current agent streaming gives you three views:

• updates: completed agent steps
• messages: model tokens plus metadata
• custom: progress data that your application defines

They carry different kinds of information, so don't show all of them as if they were one answer.

A basic stream loop

for chunk in agent.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": request_text,
            }
        ]
    },
    config,
    stream_mode=["updates", "custom"],
    version="v2",
):
    send_to_client(chunk)

Taking the call apart: agent.stream(...) kicks off the same run that invoke(...) would, yet yields pieces while it executes. The first argument is the state update, which here is one user message. config holds settings scoped to the request, like the thread ID, and secrets don't belong in it. stream_mode=["updates", "custom"] requests finished-step data plus your own progress messages, so no token-by-token text comes back in this example. version="v2" picks the current chunk format, with a type, a namespace and a data payload on each chunk. send_to_client(chunk) stands for your own code, which might emit Server-Sent Events, a WebSocket message or a terminal line.

Translate raw chunks for the UI

The snapshot converts chunks before they reach a user interface:

@dataclass(frozen=True)
class ProgressEvent:
    kind: str
    label: str
    payload: Any

def readable_events(chunks):
    for chunk in chunks:
        chunk_type = chunk.get("type")
        data = chunk.get("data")        if chunk_type == "updates" and isinstance(data, dict):
            for step_name, step_data in data.items():
                yield ProgressEvent(
                    kind="step",
                    label=step_name,
                    payload=step_data,
                )

That adapter gives the UI a small, stable contract. It can display lines like these:

Checking service ownership...
Reading recent incidents...
Waiting for ticket-note approval...
Ticket note approved and written.

What it must not show, before the run has actually finished, is this:

Done.

Partial text isn't final text

Streaming tokens makes an answer feel quicker, but it adds nothing to safety or accuracy. The model can open a sentence and then head elsewhere, a tool can fail after tokens have already gone out, and a policy check can veto the proposed action.

Use explicit event states: running, waiting_for_approval, completed, failed and cancelled. Only completed carries a final result.

Decide what a disconnect means

A closed browser doesn't prove the server stopped. Settle the policy in advance:

• Cancel read-only work when cancellation exists and no one is waiting for the answer.
• Let an approved durable job finish if the user plans to return for it.
• Never assume a write didn't happen just because its response got lost.
• Look the operation up by idempotency key before attempting the write again.

Streaming and cancellation are related but they aren't one feature.

Batching independent inputs

LangChain runnables expose batch and abatch. The default batch runs independent calls in a thread pool, which suits many network-bound operations, and a provider-specific runnable can replace that with a native batch API. The helpdesk worker accepts several unrelated tickets:

outputs = runnable.batch(
    tickets,
    config={"max_concurrency": 4},
    return_exceptions=True,
)

tickets is the ordered list of inputs. max_concurrency limits how many calls run side by side, and four is just an illustration; derive your number from provider quotas, tool capacity, cost and measured latency. With return_exceptions=True, one item failing no longer throws away the results of the others. Afterward the application matches each output to its input:

for ticket, output in zip(tickets, outputs, strict=True):
    if isinstance(output, BaseException):
        save_failure(ticket.ticket_id, type(output).__name__)
    else:
        save_recommendation(ticket.ticket_id, output)

That keeps one result per ticket.

Skip the parallel batch in these cases: item two depends on item one; every item updates the same record without concurrency control; order of execution is itself a business rule; the provider has a bulk transaction with different guarantees; or one user can submit enough items to use up shared capacity. Batching moves load around and doesn't make it vanish.

Retry the smallest safe boundary

Retrying means doing the work again, so be careful what you repeat. Re-running a full workflow can redo retrieval, redo model calls and redo a ticket write. A safer approach retries only the read step known to fail transiently. The snapshot declares a dedicated error for that:

class TemporaryProviderError(RuntimeError):
    """A short-lived error that may succeed when tried again."""

and wraps a single runnable:

primary_call = RunnableLambda(primary).with_retry(
    retry_if_exception_type=(TemporaryProviderError,),
    wait_exponential_jitter=False,
    stop_after_attempt=3,
)

RunnableLambda(primary) turns a plain Python function into a runnable. retry_if_exception_type acts as an allowlist, so an invalid ticket doesn't match and goes directly back to the caller. Setting wait_exponential_jitter=False keeps the offline test quick and deterministic. stop_after_attempt=3 counts every try, meaning three attempts altogether and not three retries on top of the first. Against a real network service, exponential backoff with jitter is generally the better choice, since it keeps all workers from hitting the service at the same moment.

Writes aren't safe to retry blindly

Imagine the ticket service adds the note, then the connection breaks before the success response arrives. The caller sees a timeout, yet the write may exist already, and a naive retry would add the note twice. Use an idempotency key:

operation_id = "ticket-note:INC-2048:approval-7331"

The ticket service saves that key alongside the first successful write. If the same key shows up again, it returns the stored result and doesn't apply the change a second time. When the external system can't do idempotency, check its current state before retrying, and accept that some failure windows can't be fully closed.

A fallback is a different path

Once the allowed attempts are used up, a fallback runnable can take over:

reliable = primary_call.with_fallbacks(
    [RunnableLambda(conservative_fallback)],
    exceptions_to_handle=(TemporaryProviderError,),
)

It runs only for the named temporary error, and what it returns is this:

Recommendation(
    ticket_id=ticket.ticket_id,
    text=(
        "The recommendation service is temporarily unavailable. "
        "Keep the ticket open and compare the failed deployment "
        "with the last known-good release."
    ),
    source="fallback",
)

The reply admits what happened. It doesn't claim that runbooks were searched or that the ticket was updated.

Switching to another model also isn't the same as restarting the first one. The second model may behave differently with tool calls, have a smaller context limit, support structured output differently, apply other safety behavior, differ in latency and price, or fall under another provider data policy. Treat the fallback route as a configuration of its own and test it separately. The best fallback is sometimes no model at all, just a fixed, conservative message or a queue for human review.

Timeouts live in several layers

One timeout value isn't enough. Give separate budgets to connection setup, a single model request, a single tool request, the whole agent run, the approval wait and the client connection. The full-run budget has to exceed any individual-call budget, but it still needs a limit.

Put timeout behavior next to the operation it limits: provider timeouts on the provider client, tool timeouts around the tool client, and the web request deadline at the service boundary. When one fires, record which layer ended the work. A bare "timed out" with no boundary named is hard to fix.

Middleware wraps agent calls

The earlier steps mostly used fixed chains, and the approval step introduced agent middleware. Middleware is also a good home for rules that should cover many model or tool calls: retrying selected model failures, capping model calls, capping tool calls, filtering tools for the current user, adding request metadata for logs and traces, redacting selected personal data and halting when an application budget runs out. The snapshot uses the built-in middleware:

def reliability_middleware():
    return [
        ModelRetryMiddleware(
            max_retries=2,
            retry_on=(
                TemporaryProviderError,
                TimeoutError,
                ConnectionError,
            ),
            on_failure="error",
            initial_delay=0.5,
            backoff_factor=2.0,
            max_delay=4.0,
            jitter=True,
        ),
        ModelCallLimitMiddleware(
            run_limit=8,
            exit_behavior="error",
        ),
        ToolCallLimitMiddleware(
            run_limit=6,
            exit_behavior="end",
        ),
    ]

ModelRetryMiddleware surrounds model calls, and an initial call plus two retries gives at most three attempts. ModelCallLimitMiddleware keeps a model loop from running forever. ToolCallLimitMiddleware stops excessive tool use within a run. These controls do separate jobs: a call limit doesn't replace a timeout, a timeout doesn't replace approval and approval doesn't replace authorization.

Order matters

Hooks that run before a call go in list order, those that run after go in reverse, and wrapping hooks nest the way function calls do. When middleware A wraps B, a retry in A can redo B's work, so sketch the order whenever two components could both retry, change models or alter state. Approval for a write belongs outside any retry that might repeat the approved side effect.

Custom hooks are plain code

LangChain provides node-style hooks (before_model, after_model) and wrap-style hooks (wrap_model_call, wrap_tool_call). Node-style hooks suit checks and state updates at fixed points. Wrap-style hooks are for code that must surround one call, or that decides whether the handler runs zero, one or many times. Not every application rule should become middleware, though. Ticket state transitions, authorization decisions and data ownership rules are easier to follow as named application services.

Testing the failure paths offline

The snapshot has five tests, none needing a provider.

A temporary error succeeds on the second attempt:

result = runnable.invoke(ticket())

assert calls == 2
assert result.source == "primary"

Exhausted attempts use the labeled fallback:

assert calls == 3
assert result.source == "fallback"
assert "temporarily unavailable" in result.text

Invalid input is neither retried nor hidden:

with pytest.raises(InvalidTicketError):
    runnable.invoke(ticket())

assert calls == 1

One batch failure doesn't erase another success:

assert results[0].status == "completed"
assert results[1].status == "failed"
assert results[1].error_type == "RuntimeError"

Progress stays separate from model tokens:

assert [(event.kind, event.label) for event in events] == [
    ("step", "model"),
    ("progress", "custom"),
    ("token", "model"),
]

Run them:

pytest -q

Expected result:

5 passed

These tests prove your local control rules. They don't prove that a provider's errors are mapped correctly, which takes a small integration test with the real provider SDK and a controlled test account.

Write the failure matrix down

Put a table like this in code or documentation before release:

failure                         retry?   fallback?   client result
connection reset               yes      maybe       503 if exhausted
provider rate limit            yes      maybe       503 or queued
invalid ticket                 no       no          422
authorization denied           no       no          403
structured output invalid      limited  maybe       502 if exhausted
ticket already closed          no       no          409
unknown exception              no       no          500 + incident ID

Exact status codes depend on your API. What matters is that every class of failure has one deliberate policy.

Still missing

The runtime behavior is clearer now, but this is still a command-line teaching snapshot. It has no authenticated HTTP API, no startup and readiness checks, no secret loading, no structured request logs, no durable job storage, no deployment configuration, no real provider integration tests, no evaluation set checked before release and no production trace review. A service shape that adds those comes next.

Where the helpdesk flow stands

The flow now has visible states and bounded recovery:

request
  -> progress events
  -> small retry boundary
  -> labeled fallback when allowed
  -> model and tool call limits
  -> approval before write
  -> final result or named failure

batch
  -> bounded concurrency
  -> one outcome per ticket

No control eliminates uncertainty. The aim is to keep it visible, bounded and testable. Next comes the whole helpdesk workflow behind an API, with config, secrets, request IDs, timeouts, logs and tracing hooks, then deterministic tests, evaluation cases and deployment checks, plus a list of what a real company has to supply before the service touches live tickets. All of the above was checked against LangChain's Python documentation on streaming, runnable retry, fallbacks, batch, middleware and testing on August 31, 2026, and package APIs and provider error types may change.