Agent Behavior-Steering Middleware

Intro

Steering LLM behavior can be challenging. Particularly in dynamic situations, or with long-running ReAct agent loops where the context is constantly growing. The best way to guide behavior is in a well-defined workflow--often a cyclical or acyclical state graph constructed via a framework like LangGraph or PydanticAI...

Sometimes that simply isn't an option.

I was recently on a project and my hands were tied, the situation demanded a dynamic conversational agent. A custom workflow wasn't on the table, and a ReAct agent had to be utilized.

System prompt instructions only get you so far. As the conversation (or requirements) grow, the context degrades and we eventually have a less reliable agent.

Is it possible to dynamically steer and agent with instructions, injecting into the context as needed?

Yes, with "behavior-steering" middleware. It's worth noting that I'll be demonstrating these patterns in LangChain, but they're applicable to any LLM integration framework that provides hooks or middleware to implement.

LangChain Middleware (Crash Course)

I've blogged previously about Using custom Middleware to remove PII and LangChain provides numerous prebuilt middlewares. Middleware hooks enable us to write deterministic code that executes pre or post model execution. This allows us to better engineer the context with customizations. There are several middeware hooks that are provided by LangChain middleware.

LangChain's AgentMiddleware gives you hooks at key points in the agent loop:

• wrap_model_call — intercepts every LLM inference. You receive the ModelRequest (messages, system prompt, tools), call the handler when you're ready, and return the ModelResponse. This is where you inject steering content.
• wrap_tool_call — intercepts every tool execution. You can block, modify, or observe tool calls before and after they run.
• before_model / after_model — lighter hooks for logging or state inspection without wrapping the call.
• before_agent / after_agent — fire once per invoke call, useful for per-invocation setup and teardown.

You can read in-depth about the middleware in langchain here.

Behavior-Steering Middleware

The following sections represent a handful of abstracted behavior-steering middleware types that I've identified. At the time of writing this blog post, I have not identified any other similar projects leveraging this pattern. Do not consider the following exhaustive, certainly more is to be discovered.

The key mechanism for behavior steering is request.override(messages=...). Instead of mutating the system prompt (which should remain static) or appending new messages (which could produce an invalid API call structure), we modify an existing message in the conversation — specifically the last user or tool message — by prepending a \<framework_instruction\> block to its content:

def inject_framework_instruction(
    messages: list[BaseMessage],
    instruction: str,
) -> list[AnyMessage]:
    """Prepend a framework instruction to the last user or tool message."""
    modified: list[AnyMessage] = list(messages)
    tagged = f"<framework_instruction>\n{instruction}\n</framework_instruction>"
 
    for i in range(len(modified) - 1, -1, -1):
        msg = modified[i]
        if isinstance(msg, HumanMessage):
            modified[i] = HumanMessage(
                content=f"{tagged}\n\n{msg.content}",
            )
            logger.debug("Injected framework_instruction into HumanMessage at index %d", i)
            return modified
        if isinstance(msg, ToolMessage):
            modified[i] = ToolMessage(
                content=f"{tagged}\n\n{msg.content}",
                tool_call_id=msg.tool_call_id,
            )
            logger.debug("Injected framework_instruction into ToolMessage at index %d", i)
            return modified
 
    logger.warning("No HumanMessage or ToolMessage found — instruction not injected")
    return modified

Behavior Drift Detect & Correct

The simplest behavior-steering middleware type to explore first is behavior drift detection/correction. While I was developing the OpenPaw framework, I'd encounter situations where my agents wouldn't be yielding intermediary chat-messages during multi-turn exploratory or problem-solving sessions. I had equipped them with a send_message tool, so that they could send updates as they worked instead of waiting for the terminating agent response message for feedback.

Prompting in the AGENT.md or USER.md can only be so reliable, especially in complicated multi-turn scenarios where the context is rapidly growing and the agent is locked onto a target. One solution is to detect the behavior drift by inspecting the message/tool use history with middleware and selectively injecting behavaior steering guidelines into the agent's context.

The following is an example copy/pasted from a LangSmith trace demonstrating the behavior guidelines being injected into a tool's output response:

"<framework_instruction>You have not sent a status update in 2 turns. Please use the `send_message` tool to report your progress.</framework_instruction>

Report section written: Grid-Scale Battery Technology:
Grid-scale battery ..."

The full flow can be followed from the implemented middleware class itself in my code-exmaples,

class CheckInMiddleware(AgentMiddleware):
    """Middleware that monitors a specific tool and reminds the agent to use it.
 
    Steering content is injected as a ``<framework_instruction>`` block
    prepended to the last user or tool message — the system prompt is not
    modified and no new messages are appended. The agent's system prompt
    should mention that ``<framework_instruction>`` tags will appear with
    guidance.
 
    Example:
        >>> config = CheckInConfig(target_tool="send_message", max_silent_turns=2)
        >>> middleware = CheckInMiddleware(config)
        >>> agent = create_agent(
        ...     model=ChatOpenAI(model="gpt-4.1-mini"),
        ...     tools=[send_message, do_research],
        ...     middleware=[middleware],
        ...     state_schema=CheckInAgentState,
        ...     system_prompt=(
        ...         "You are a research agent. Send regular status updates. "
        ...         "You will receive <framework_instruction> tags with "
        ...         "guidance — follow them."
        ...     ),
        ... )
    """
 
    state_schema = CheckInAgentState
 
    def __init__(self, config: CheckInConfig) -> None:
        """Initialise the middleware.
 
        Args:
            config: Check-in monitoring configuration.
        """
        super().__init__()
        self._config = config
        self._detector = CheckInDetector(config)
 
    @property
    def detector(self) -> CheckInDetector:
        """Expose the underlying detector for observability and testing."""
        return self._detector
 
    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse | ExtendedModelResponse:
        """Intercept the model call to inject check-in reminders when needed.
 
        Steps:
        1. If the detector says to intervene, inject a framework_instruction
           into the last user/tool message via ``request.override``.
        2. Call the handler (the real model).
        3. Record the turn so counters stay accurate.
        4. Return an ExtendedModelResponse that writes check-in fields back
           into the agent state.
 
        Args:
            request: Incoming model request with messages and metadata.
            handler: Callable that executes the actual LLM call.
 
        Returns:
            ExtendedModelResponse with updated check-in state fields.
        """
        if self._detector.should_intervene():
            reminder = self._detector.build_reminder()
            modified_messages = inject_framework_instruction(
                list(request.messages), reminder,
            )
            request = request.override(messages=modified_messages)
            self._detector.record_intervention()
            logger.warning(
                "Check-in intervention: agent silent for %d turns without %s",
                self._detector.turns_since_target,
                self._config.target_tool,
            )
 
        response = handler(request)
 
        if response.result:
            self._detector.record_turn(response.result)
 
        return ExtendedModelResponse(
            model_response=response,
            command=Command(
                update={
                    "turns_since_check_in": self._detector.turns_since_target,
                    "intervention_count": self._detector.intervention_count,
                }
            ),
        )
 
    def reset(self) -> None:
        """Reset the middleware for a new conversation."""
        self._detector = CheckInDetector(self._config)
        logger.debug("CheckInMiddleware reset")

In my code-examples, we have a utility function for the drift detector that records turns and inspects if target behavior (determined through tool use in this scenario) is occurring.

    def record_turn(self, messages: list[BaseMessage]) -> None:
        """Update counters after the model responds.
 
        Scans AIMessage tool_calls for the target tool. If found, the silent
        turn counter resets to zero. Otherwise it increments. The cooldown
        counter always increments.
 
        Args:
            messages: Response messages returned by the model this turn.
        """
        target_used = False
 
        for msg in messages:
            if isinstance(msg, AIMessage) and msg.tool_calls:
                for tc in msg.tool_calls:
                    if tc.get("name") == self._config.target_tool:
                        target_used = True
                        logger.debug("Check-in detected: %s called", self._config.target_tool)
                        break
            if target_used:
                break
 
        if target_used:
            self.turns_since_target = 0
        else:
            self.turns_since_target += 1
 
        self.turns_since_last_intervention += 1
 
        logger.debug(
            "record_turn complete — turns_since_target=%d, cooldown=%d",
            self.turns_since_target,
            self.turns_since_last_intervention,
        )

This specific example use case is one of many. Keep in mind, you are not limited to using deterministic evaluations of the target AgentState, you could dispatch another LLM within the middleware with it's own prompting/evaluation criteria and feedback.

View the full drift detection example

Turn-Based Urgency

Next up is "Turn-Based Urgency". The concept is simple, your agent must do something within an allocated turn budget. With a defined turn budget, we could have specific prompts with increasing urgency that are injected into the agent context.

An easy example to envision is a customer service chat bot that is required to make a recommendation in N number of turns. If we have a turn-budget we can then inject behavior guidelines into the agent loop as certain thresholds are crossed.

Each level has a corresponding prompt modifier that gets injected into the last user message via \<framework_instruction\> tags.

Here's a snippet for how urgency could be calculated:

    def _calculate(self, turns_used: int) -> UrgencyState:
        """Calculate urgency state from turn count.
 
        Args:
            turns_used: Number of turns already consumed.
 
        Returns:
            UrgencyState with calculated level and metrics.
        """
        max_turns = self._config.max_turns
        turns_remaining = max(0, max_turns - turns_used)
        remaining_ratio = turns_remaining / max_turns if max_turns > 0 else 0.0
 
        level = self._determine_level(remaining_ratio)
 
        return UrgencyState(
            current_level=level,
            turns_used=turns_used,
            turns_remaining=turns_remaining,
            max_turns=max_turns,
            remaining_ratio=remaining_ratio,
            is_terminal=level == UrgencyLevel.TERMINAL,
        )
 
    def _determine_level(self, remaining_ratio: float) -> UrgencyLevel:
        """Map remaining ratio to urgency level.
 
        Args:
            remaining_ratio: Ratio of turns remaining (0.0 to 1.0).
 
        Returns:
            UrgencyLevel based on configured thresholds.
        """
        thresholds = self._config.thresholds
 
        if remaining_ratio > thresholds.balanced_max:
            return UrgencyLevel.EXPLORATION
        elif remaining_ratio > thresholds.efficiency_max:
            return UrgencyLevel.BALANCED
        elif remaining_ratio > 0:
            return UrgencyLevel.EFFICIENCY
        else:
            return UrgencyLevel.TERMINAL

The beauty of this pattern is that the agent gets prompt guidelines progressively. Resulting in a smoother transitioned over the course of a conversation. The shopping assistant naturally transitions from asking things like "What's your budget?" to delivering "I recommend the AudioMaxx 1000 for your commute."

Here is an example of the modified request when the middleware fires:

<framework_instruction>
URGENCY: EFFICIENCY (turn 4/5)
Time to work toward a recommendation. If you have enough information, use the `recommend_product` tool to make your suggestion.
</framework_instruction>
 
That sounds good. My budget is around $150. Does it fit?

The framework injects framework instructions as urgency thresholds are crossed, leaving the original input text at the end.

Note: This pattern of including framework instructions in a system prompt and then injecting framework_instruction tags into a Human/User message is valuable far beyond behavior-steering middleware. It's an enormously powerful pattern when building agent harnesses.

The full turn urgency example

Phase-Steering

Some tasks have natural phases — gather information, analyze it, make a recommendation. In a structured workflow you'd model these as explicit graph nodes, but in a ReAct agent loop there are no nodes to anchor to. The agent decides what to do next on every iteration, and without guardrails it will skip phases, stall in one indefinitely, or use tools meant for a later stage.

Phase-steering middleware solves this by binding behavioral guidance and tool access to named phases. Each phase defines what the agent should focus on and which tools it's allowed to use. The middleware injects the current phase's guidance into the conversation via framework_instruction tags and enforces tool access in wrap_tool_call — if the agent tries to call a tool that belongs to a later phase, the call is blocked with an explanation.

The remaining question is: what triggers a phase transition? There are several options — state content analysis, iteration counts, explicit user commands — but the simplest and most deterministic approach is tool-triggered transitions. Certain tools naturally mark phase boundaries. When a customer-facing agent calls note_preferences, that's an organic signal that discovery is complete. When it calls shortlist_items, curation is done. For this example, we don't need heuristics -- the tool calls are the phase boundaries.

In the companion repository, this is demonstrated with a personalization shopping assistant that progresses through DISCOVERY → CURATION → RECOMMENDATION → COMPLETE. The configuration binds phases, tools, and transitions together:

  PhaseConfig(
      phases={
          Phase.DISCOVERY: PhaseBehavior(
              system_prompt_addition="Ask questions to understand needs...",
              allowed_tools=["browse_catalog", "note_preferences"],
          ),
          Phase.CURATION: PhaseBehavior(
              system_prompt_addition="Filter and compare products...",
              allowed_tools=["filter_products", "compare_options", "shortlist_items"],
          ),
          Phase.RECOMMENDATION: PhaseBehavior(
              system_prompt_addition="Present your recommendation...",
              allowed_tools=["present_recommendation"],
          ),
      },
      transitions=[
          PhaseTransitionRule(
              from_phase=Phase.DISCOVERY,
              to_phase=Phase.CURATION,
              trigger_tool="note_preferences",
          ),
          PhaseTransitionRule(
              from_phase=Phase.CURATION,
              to_phase=Phase.RECOMMENDATION,
              trigger_tool="shortlist_items",
          ),
      ],
  )

The middleware uses two hooks to make this work. wrap_model_call injects the current phase's guidance — what to focus on and which tools are available — as a <framework_instruction> block before each LLM call. wrap_tool_call does double duty: it enforces the per-phase tool allow-list and checks whether the tool that just executed is a transition trigger.

A practical challenge with ReAct agents is that they're eager. Once a transition fires mid-loop, the agent sees the next phase's tools are available and immediately chains forward — potentially blowing through every phase in a single turn. To prevent this, the middleware sets a transition gate after each phase change. While the gate is active, all further tool calls are blocked with a message telling the agent to report back to the user. The gate clears on the next invoke call (via the before_agent hook), giving the user a chance to review progress and drive the next phase forward.

This creates a natural conversation rhythm: the user asks a question, the agent works within its current phase, a transition fires, the agent reports back, and the user sends the next message to begin the new phase. The flat ReAct loop behaves like a structured workflow — but the structure lives in middleware configuration, not in graph topology.

The full implementation — including the PhaseSteeringMiddleware, ToolTriggeredDetector, and a runnable demo is available in my code-examples repository.

Conclusion

Behavior-steering middleware isn't a replacement for well-designed agent workflows. When you can build a proper state graph with explicit nodes and edges, you should. But when you're constrained to a ReAct agent be it client requirements, framework limitations, or the inherently dynamic nature of the problem-solving these patterns give you a practical way to shape agent behavior without modifying the model or the agent architecture.

The three patterns address different failure modes:

• Turn-based urgency prevents open-ended conversations from running indefinitely without delivering a result.
• Behavior Drift catches agents that silently stop using the tools they were given.
• Phase steering imposes workflow structure on an otherwise unstructured agent loop.

They share a common mechanism: intercept the model call, evaluate a runtime condition, and inject <framework_instruction> tagged guidance into an existing message or tool result. The agent sees instructions woven into the conversation that happen to arrive at exactly the right moment.

Remember, these examples all used tools to determine/evaluate behavioral guidelines. It's absolutely possible to steer with an additional LLM in the middleware layer.

All example implementations are available in my code-examples repository.

Happy spec-writing/coding!