Middleware that forces a direct, tool-free answer when the model-call budget is nearly used up.

class langgraph_agent_toolkit.agents.components.middlewares.immediate_generation.ImmediateGenerationMiddleware(model_call_limit=None, instruction=None)

Bases: AgentMiddleware

Force a final, tool-free answer on the last allowed model call of a run.

Mirrors the custom create_react_agent’s immediate_generation router: rather than running out of steps mid-tool-loop (returning “need more steps” or hitting the recursion limit), the model is asked to synthesize a direct answer from what it already has once it reaches the budget. The tool results are always kept — only tools and the system message are changed — so the model answers from the data it already collected.

The budget is the number of model calls already made in the current run (AI messages since the latest human message), so the middleware needs no extra state. It should fire only as a graceful fallback near the recursion limit, not cut off legitimate multi-step work, so model_call_limit defaults to about half the recursion limit (a model→tools cycle is ~2 steps): with the default recursion limit of 64 that is 32 model calls.

Parameters:

• model_call_limit (int | None)

• instruction (str | None)

__init__(model_call_limit=None, instruction=None)

Parameters:

• model_call_limit (int | None)

• instruction (str | None)

Return type:

None

wrap_model_call(request, handler)

Intercept and control model execution via handler callback.

Async version is awrap_model_call

The handler callback executes the model request and returns a ModelResponse. Middleware can call the handler multiple times for retry logic, skip calling it to short-circuit, or modify the request/response. Multiple middleware compose with first in list as outermost layer.

Parameters:

• request (ModelRequest) – Model request to execute (includes state and runtime).

• handler (Callable[[ModelRequest], ModelResponse]) –

  Callback that executes the model request and returns ModelResponse.

  Call this to execute the model.

  Can be called multiple times for retry logic.

  Can skip calling it to short-circuit.

Returns:

The model call result.

Return type:

ModelResponse

Examples

!!! example “Retry on error”

```python def wrap_model_call(self, request, handler):

for attempt in range(3):

try:

return handler(request)

except Exception:

if attempt == 2:

raise

```

!!! example “Rewrite response”

```python def wrap_model_call(self, request, handler):

response = handler(request) ai_msg = response.result[0] return ModelResponse(

result=[AIMessage(content=f”[{ai_msg.content}]”)], structured_response=response.structured_response,

)

```

!!! example “Error to fallback”

```python def wrap_model_call(self, request, handler):

try:

return handler(request)

except Exception:

return ModelResponse(result=[AIMessage(content=”Service unavailable”)])

```

!!! example “Cache/short-circuit”

```python def wrap_model_call(self, request, handler):

if cached := get_cache(request):

return cached # Short-circuit with cached result

response = handler(request) save_cache(request, response) return response

```

!!! example “Simple AIMessage return (converted automatically)”

```python def wrap_model_call(self, request, handler):

response = handler(request) # Can return AIMessage directly for simple cases return AIMessage(content=”Simplified response”)

```

async awrap_model_call(request, handler)

Intercept and control async model execution via handler callback.

The handler callback executes the model request and returns a ModelResponse.

Middleware can call the handler multiple times for retry logic, skip calling it to short-circuit, or modify the request/response. Multiple middleware compose with first in list as outermost layer.

Parameters:

• request (ModelRequest) – Model request to execute (includes state and runtime).

• handler (Callable[[ModelRequest], Awaitable[ModelResponse]]) –

  Async callback that executes the model request and returns ModelResponse.

  Call this to execute the model.

  Can be called multiple times for retry logic.

  Can skip calling it to short-circuit.

Returns:

The model call result.

Return type:

ModelResponse

Examples

!!! example “Retry on error”

```python async def awrap_model_call(self, request, handler):

for attempt in range(3):

try:

return await handler(request)

except Exception:

if attempt == 2:

raise

```

async aafter_agent(state, runtime)

Async logic to run after the agent execution completes.

Parameters:

• state (StateT) – The current agent state.

• runtime (Runtime[ContextT]) – The runtime context.

Returns:

Agent state updates to apply after agent execution.

Return type:

dict[str, Any] | None

async aafter_model(state, runtime)

Async logic to run after the model is called.

Parameters:

• state (StateT) – The current agent state.

• runtime (Runtime[ContextT]) – The runtime context.

Returns:

Agent state updates to apply after model call.

Return type:

dict[str, Any] | None

async abefore_agent(state, runtime)

Async logic to run before the agent execution starts.

Parameters:

• state (StateT) – The current agent state.

• runtime (Runtime[ContextT]) – The runtime context.

Returns:

Agent state updates to apply before agent execution.

Return type:

dict[str, Any] | None

async abefore_model(state, runtime)

Async logic to run before the model is called.

Parameters:

• state (StateT) – The agent state.

• runtime (Runtime[ContextT]) – The runtime context.

Returns:

Agent state updates to apply before model call.

Return type:

dict[str, Any] | None

after_agent(state, runtime)

Logic to run after the agent execution completes.

Parameters:

• state (StateT) – The current agent state.

• runtime (Runtime[ContextT]) – The runtime context.

Returns:

Agent state updates to apply after agent execution.

Return type:

dict[str, Any] | None

after_model(state, runtime)

Logic to run after the model is called.

Parameters:

• state (StateT) – The current agent state.

• runtime (Runtime[ContextT]) – The runtime context.

Returns:

Agent state updates to apply after model call.

Return type:

dict[str, Any] | None

async awrap_tool_call(request, handler)

Intercept and control async tool execution via handler callback.

The handler callback executes the tool call and returns a ToolMessage or Command. Middleware can call the handler multiple times for retry logic, skip calling it to short-circuit, or modify the request/response. Multiple middleware compose with first in list as outermost layer.

Parameters:

• request (ToolCallRequest) –

  Tool call request with call dict, BaseTool, state, and runtime.

  Access state via request.state and runtime via request.runtime.

• handler (Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]]) –

  Async callable to execute the tool and returns ToolMessage or Command.

  Call this to execute the tool.

  Can be called multiple times for retry logic.

  Can skip calling it to short-circuit.

Returns:

ToolMessage or Command (the final result).

Return type:

ToolMessage | Command[Any]

The handler Callable can be invoked multiple times for retry logic.

Each call to handler is independent and stateless.

Examples

!!! example “Async retry on error”

```python async def awrap_tool_call(self, request, handler):

for attempt in range(3):

try:

result = await handler(request) if is_valid(result):

return result

except Exception:

if attempt == 2:

raise

return result

```

```python async def awrap_tool_call(self, request, handler):

if cached := await get_cache_async(request):

return ToolMessage(content=cached, tool_call_id=request.tool_call[“id”])

result = await handler(request) await save_cache_async(request, result) return result

```

before_agent(state, runtime)

Logic to run before the agent execution starts.

Parameters:

• state (StateT) – The current agent state.

• runtime (Runtime[ContextT]) – The runtime context.

Returns:

Agent state updates to apply before agent execution.

Return type:

dict[str, Any] | None

before_model(state, runtime)

Logic to run before the model is called.

Parameters:

• state (StateT) – The current agent state.

• runtime (Runtime[ContextT]) – The runtime context.

Returns:

Agent state updates to apply before model call.

Return type:

dict[str, Any] | None

property name: str

The name of the middleware instance.

Defaults to the class name, but can be overridden for custom naming.

state_schema

alias of _DefaultAgentState

transformers = ()

Stream transformer factories registered by the middleware.

Each entry is a scope-aware factory invoked as factory(scope) so every invocation receives a fresh instance. Factories are merged with the transformers argument of [create_agent][langchain.agents.create_agent] at graph compile time, after the ToolCallTransformer and before any user-supplied entries.

wrap_tool_call(request, handler)

Intercept tool execution for retries, monitoring, or modification.

Async version is awrap_tool_call

Multiple middleware compose automatically (first defined = outermost).

Exceptions propagate unless handle_tool_errors is configured on ToolNode.

Parameters:

• request (ToolCallRequest) –

  Tool call request with call dict, BaseTool, state, and runtime.

  Access state via request.state and runtime via request.runtime.

• handler (Callable[[ToolCallRequest], ToolMessage | Command[Any]]) – Callable to execute the tool (can be called multiple times).

Returns:

ToolMessage or Command (the final result).

Return type:

ToolMessage | Command[Any]

The handler Callable can be invoked multiple times for retry logic.

Each call to handler is independent and stateless.

Examples

!!! example “Modify request before execution”

```python def wrap_tool_call(self, request, handler):

modified_call = {

**request.tool_call, “args”: {

**request.tool_call[“args”], “value”: request.tool_call[“args”][“value”] * 2,

},

} request = request.override(tool_call=modified_call) return handler(request)

```

!!! example “Retry on error (call handler multiple times)”

```python def wrap_tool_call(self, request, handler):

for attempt in range(3):

try:

result = handler(request) if is_valid(result):

return result

except Exception:

if attempt == 2:

raise

return result

```

!!! example “Conditional retry based on response”

```python def wrap_tool_call(self, request, handler):

for attempt in range(3):

result = handler(request) if isinstance(result, ToolMessage) and result.status != “error”:

return result

if attempt < 2:

continue

return result

```

tools

Additional tools registered by the middleware.