Skip to content
Documentation

StateManager Class

The StateManager class manages conversation state and policy violation history for the guardrails system. It provides methods for storing and retrieving state information.

Class Definition

class StateManager:
    """Manages conversation state and policy violation history."""

Instance Methods

get_state

def get_state(
    conversation_id: str,
    context: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]

Get the state for a conversation.

Parameters

ParameterTypeDefaultDescription
conversation_idstr-Unique identifier for the conversation
contextOptional[Dict[str, Any]]NoneAdditional context

Returns

  • Dict[str, Any]: Conversation state

Example

state = state_manager.get_state(
    conversation_id="conv123",
    context={"user_id": "user123"}
)


if "policy_violations" in state:
    print(f"Violations: {state['policy_violations']}")

update_state

def update_state(
    conversation_id: str,
    updates: Dict[str, Any],
    context: Optional[Dict[str, Any]] = None
) -> None

Update the state for a conversation.

Parameters

ParameterTypeDefaultDescription
conversation_idstr-Unique identifier for the conversation
updatesDict[str, Any]-State updates to apply
contextOptional[Dict[str, Any]]NoneAdditional context

Example

state_manager.update_state(
    conversation_id="conv123",
    updates={
        "policy_violations": [
            {
                "policy_id": "content_moderation",
                "timestamp": "2024-03-22T12:00:00Z",
                "details": "Inappropriate content detected"
            }
        ]
    }
)

add_violation

def add_violation(
    conversation_id: str,
    policy_id: str,
    details: Dict[str, Any],
    context: Optional[Dict[str, Any]] = None
) -> None

Add a policy violation to the conversation state.

Parameters

ParameterTypeDefaultDescription
conversation_idstr-Unique identifier for the conversation
policy_idstr-ID of the violated policy
detailsDict[str, Any]-Violation details
contextOptional[Dict[str, Any]]NoneAdditional context

Example

state_manager.add_violation(
    conversation_id="conv123",
    policy_id="content_moderation",
    details={
        "message": "Inappropriate content",
        "severity": "high",
        "timestamp": "2024-03-22T12:00:00Z"
    }
)

get_violations

def get_violations(
    conversation_id: str,
    context: Optional[Dict[str, Any]] = None
) -> List[Dict[str, Any]]

Get all policy violations for a conversation.

Parameters

ParameterTypeDefaultDescription
conversation_idstr-Unique identifier for the conversation
contextOptional[Dict[str, Any]]NoneAdditional context

Returns

  • List[Dict[str, Any]]: List of policy violations

Example

violations = state_manager.get_violations("conv123")
for violation in violations:
    print(f"Policy {violation['policy_id']} violated at {violation['timestamp']}")

clear_state

def clear_state(
    conversation_id: str,
    context: Optional[Dict[str, Any]] = None
) -> None

Clear the state for a conversation.

Parameters

ParameterTypeDefaultDescription
conversation_idstr-Unique identifier for the conversation
contextOptional[Dict[str, Any]]NoneAdditional context

Example

state_manager.clear_state("conv123")

Properties

PropertyTypeDescription
statesDict[str, Dict[str, Any]]Dictionary of conversation states
tracerTracerThe OpenTelemetry tracer

State Structure

The conversation state is a dictionary with the following structure:

{
    "conversation_id": str,           # Unique conversation identifier
    "created_at": str,               # ISO timestamp of creation
    "updated_at": str,               # ISO timestamp of last update
    "policy_violations": [           # List of policy violations
        {
            "policy_id": str,        # ID of violated policy
            "timestamp": str,        # ISO timestamp of violation
            "details": dict,         # Violation details
            "severity": str          # Violation severity
        }
    ],
    "metadata": dict                 # Additional state metadata
}