feat(steering): add experimental steering for modular prompting (#1280)

---------

Co-authored-by: John Tristan <[email protected]>

Author: dbschmigelski

src/strands/agent/state.py

@@ -1,97 +1,6 @@
 """Agent state management."""
 
-import copy
-import json
-from typing import Any, Dict, Optional
+from ..types.json_dict import JSONSerializableDict
 
-
-class AgentState:
-    """Represents an Agent's stateful information outside of context provided to a model.
-
-    Provides a key-value store for agent state with JSON serialization validation and persistence support.
-    Key features:
-    - JSON serialization validation on assignment
-    - Get/set/delete operations
-    """
-
-    def __init__(self, initial_state: Optional[Dict[str, Any]] = None):
-        """Initialize AgentState."""
-        self._state: Dict[str, Dict[str, Any]]
-        if initial_state:
-            self._validate_json_serializable(initial_state)
-            self._state = copy.deepcopy(initial_state)
-        else:
-            self._state = {}
-
-    def set(self, key: str, value: Any) -> None:
-        """Set a value in the state.
-
-        Args:
-            key: The key to store the value under
-            value: The value to store (must be JSON serializable)
-
-        Raises:
-            ValueError: If key is invalid, or if value is not JSON serializable
-        """
-        self._validate_key(key)
-        self._validate_json_serializable(value)
-
-        self._state[key] = copy.deepcopy(value)
-
-    def get(self, key: Optional[str] = None) -> Any:
-        """Get a value or entire state.
-
-        Args:
-            key: The key to retrieve (if None, returns entire state object)
-
-        Returns:
-            The stored value, entire state dict, or None if not found
-        """
-        if key is None:
-            return copy.deepcopy(self._state)
-        else:
-            # Return specific key
-            return copy.deepcopy(self._state.get(key))
-
-    def delete(self, key: str) -> None:
-        """Delete a specific key from the state.
-
-        Args:
-            key: The key to delete
-        """
-        self._validate_key(key)
-
-        self._state.pop(key, None)
-
-    def _validate_key(self, key: str) -> None:
-        """Validate that a key is valid.
-
-        Args:
-            key: The key to validate
-
-        Raises:
-            ValueError: If key is invalid
-        """
-        if key is None:
-            raise ValueError("Key cannot be None")
-        if not isinstance(key, str):
-            raise ValueError("Key must be a string")
-        if not key.strip():
-            raise ValueError("Key cannot be empty")
-
-    def _validate_json_serializable(self, value: Any) -> None:
-        """Validate that a value is JSON serializable.
-
-        Args:
-            value: The value to validate
-
-        Raises:
-            ValueError: If value is not JSON serializable
-        """
-        try:
-            json.dumps(value)
-        except (TypeError, ValueError) as e:
-            raise ValueError(
-                f"Value is not JSON serializable: {type(value).__name__}. "
-                f"Only JSON-compatible types (str, int, float, bool, list, dict, None) are allowed."
-            ) from e
+# Type alias for agent state
+AgentState = JSONSerializableDict

src/strands/experimental/__init__.py

@@ -3,7 +3,7 @@
 This module implements experimental features that are subject to change in future revisions without notice.
 """
 
-from . import tools
+from . import steering, tools
 from .agent_config import config_to_agent
 
-__all__ = ["config_to_agent", "tools"]
+__all__ = ["config_to_agent", "tools", "steering"]

src/strands/experimental/steering/__init__.py

@@ -0,0 +1,46 @@
+"""Steering system for Strands agents.
+
+Provides contextual guidance for agents through modular prompting with progressive disclosure.
+Instead of front-loading all instructions, steering handlers provide just-in-time feedback
+based on local context data populated by context callbacks.
+
+Core components:
+
+- SteeringHandler: Base class for guidance logic with local context
+- SteeringContextCallback: Protocol for context update functions
+- SteeringContextProvider: Protocol for multi-event context providers
+- SteeringAction: Proceed/Guide/Interrupt decisions
+
+Usage:
+    handler = LLMSteeringHandler(system_prompt="...")
+    agent = Agent(tools=[...], hooks=[handler])
+"""
+
+# Core primitives
+# Context providers
+from .context_providers.ledger_provider import (
+    LedgerAfterToolCall,
+    LedgerBeforeToolCall,
+    LedgerProvider,
+)
+from .core.action import Guide, Interrupt, Proceed, SteeringAction
+from .core.context import SteeringContextCallback, SteeringContextProvider
+from .core.handler import SteeringHandler
+
+# Handler implementations
+from .handlers.llm import LLMPromptMapper, LLMSteeringHandler
+
+__all__ = [
+    "SteeringAction",
+    "Proceed",
+    "Guide",
+    "Interrupt",
+    "SteeringHandler",
+    "SteeringContextCallback",
+    "SteeringContextProvider",
+    "LedgerBeforeToolCall",
+    "LedgerAfterToolCall",
+    "LedgerProvider",
+    "LLMSteeringHandler",
+    "LLMPromptMapper",
+]

src/strands/experimental/steering/context_providers/__init__.py

@@ -0,0 +1,13 @@
+"""Context providers for steering evaluation."""
+
+from .ledger_provider import (
+    LedgerAfterToolCall,
+    LedgerBeforeToolCall,
+    LedgerProvider,
+)
+
+__all__ = [
+    "LedgerAfterToolCall",
+    "LedgerBeforeToolCall",
+    "LedgerProvider",
+]

src/strands/experimental/steering/context_providers/ledger_provider.py

@@ -0,0 +1,85 @@
+"""Ledger context provider for comprehensive agent activity tracking.
+
+Tracks complete agent activity ledger including tool calls, conversation history,
+and timing information. This comprehensive audit trail enables steering handlers
+to make informed guidance decisions based on agent behavior patterns and history.
+
+Data captured:
+
+    - Tool call history with inputs, outputs, timing, success/failure
+    - Conversation messages and agent responses
+    - Session metadata and timing information
+    - Error patterns and recovery attempts
+
+Usage:
+    Use as context provider functions or mix into steering handlers.
+"""
+
+import logging
+from datetime import datetime
+from typing import Any
+
+from ....hooks.events import AfterToolCallEvent, BeforeToolCallEvent
+from ..core.context import SteeringContext, SteeringContextCallback, SteeringContextProvider
+
+logger = logging.getLogger(__name__)
+
+
+class LedgerBeforeToolCall(SteeringContextCallback[BeforeToolCallEvent]):
+    """Context provider for ledger tracking before tool calls."""
+
+    def __init__(self) -> None:
+        """Initialize the ledger provider."""
+        self.session_start = datetime.now().isoformat()
+
+    def __call__(self, event: BeforeToolCallEvent, steering_context: SteeringContext, **kwargs: Any) -> None:
+        """Update ledger before tool call."""
+        ledger = steering_context.data.get("ledger") or {}
+
+        if not ledger:
+            ledger = {
+                "session_start": self.session_start,
+                "tool_calls": [],
+                "conversation_history": [],
+                "session_metadata": {},
+            }
+
+        tool_call_entry = {
+            "timestamp": datetime.now().isoformat(),
+            "tool_name": event.tool_use.get("name"),
+            "tool_args": event.tool_use.get("arguments", {}),
+            "status": "pending",
+        }
+        ledger["tool_calls"].append(tool_call_entry)
+        steering_context.data.set("ledger", ledger)
+
+
+class LedgerAfterToolCall(SteeringContextCallback[AfterToolCallEvent]):
+    """Context provider for ledger tracking after tool calls."""
+
+    def __call__(self, event: AfterToolCallEvent, steering_context: SteeringContext, **kwargs: Any) -> None:
+        """Update ledger after tool call."""
+        ledger = steering_context.data.get("ledger") or {}
+
+        if ledger.get("tool_calls"):
+            last_call = ledger["tool_calls"][-1]
+            last_call.update(
+                {
+                    "completion_timestamp": datetime.now().isoformat(),
+                    "status": event.result["status"],
+                    "result": event.result["content"],
+                    "error": str(event.exception) if event.exception else None,
+                }
+            )
+            steering_context.data.set("ledger", ledger)
+
+
+class LedgerProvider(SteeringContextProvider):
+    """Combined ledger context provider for both before and after tool calls."""
+
+    def context_providers(self, **kwargs: Any) -> list[SteeringContextCallback]:
+        """Return ledger context providers with shared state."""
+        return [
+            LedgerBeforeToolCall(),
+            LedgerAfterToolCall(),
+        ]

src/strands/experimental/steering/core/__init__.py

@@ -0,0 +1,6 @@
+"""Core steering system interfaces and base classes."""
+
+from .action import Guide, Interrupt, Proceed, SteeringAction
+from .handler import SteeringHandler
+
+__all__ = ["SteeringAction", "Proceed", "Guide", "Interrupt", "SteeringHandler"]

src/strands/experimental/steering/core/action.py

@@ -0,0 +1,65 @@
+"""SteeringAction types for steering evaluation results.
+
+Defines structured outcomes from steering handlers that determine how tool calls
+should be handled. SteeringActions enable modular prompting by providing just-in-time
+feedback rather than front-loading all instructions in monolithic prompts.
+
+Flow:
+    SteeringHandler.steer() → SteeringAction → BeforeToolCallEvent handling
+                    ↓                     ↓                      ↓
+              Evaluate context      Action type           Tool execution modified
+
+SteeringAction types:
+    Proceed: Tool executes immediately (no intervention needed)
+    Guide: Tool cancelled, agent receives contextual feedback to explore alternatives
+    Interrupt: Tool execution paused for human input via interrupt system
+
+Extensibility:
+    New action types can be added to the union. Always handle the default
+    case in pattern matching to maintain backward compatibility.
+"""
+
+from typing import Annotated, Literal
+
+from pydantic import BaseModel, Field
+
+
+class Proceed(BaseModel):
+    """Allow tool to execute immediately without intervention.
+
+    The tool call proceeds as planned. The reason provides context
+    for logging and debugging purposes.
+    """
+
+    type: Literal["proceed"] = "proceed"
+    reason: str
+
+
+class Guide(BaseModel):
+    """Cancel tool and provide contextual feedback for agent to explore alternatives.
+
+    The tool call is cancelled and the agent receives the reason as contextual
+    feedback to help them consider alternative approaches while maintaining
+    adaptive reasoning capabilities.
+    """
+
+    type: Literal["guide"] = "guide"
+    reason: str
+
+
+class Interrupt(BaseModel):
+    """Pause tool execution for human input via interrupt system.
+
+    The tool call is paused and human input is requested through Strands'
+    interrupt system. The human can approve or deny the operation, and their
+    decision determines whether the tool executes or is cancelled.
+    """
+
+    type: Literal["interrupt"] = "interrupt"
+    reason: str
+
+
+# SteeringAction union - extensible for future action types
+# IMPORTANT: Always handle the default case when pattern matching
+# to maintain backward compatibility as new action types are added
+SteeringAction = Annotated[Proceed | Guide | Interrupt, Field(discriminator="type")]