Skip to content

Decision Engines

Decision engines determine whether your AI agent should respond based on various factors like context, timing, and engagement.

DecisionEngine (Base Class)

proactiveagent.decision_engines.base.DecisionEngine

Bases: ABC

Abstract base class for response decision strategies

should_respond abstractmethod async

Python
should_respond(messages: List[Dict[str, str]], last_user_message_time: float, context: Dict[str, Any], config: Dict[str, Any], triggered_by_user_message: bool = False) -> tuple[bool, str]

Determine if AI should respond and provide reasoning

Parameters:

Name Type Description Default
messages List[Dict[str, str]]

Conversation history

required
last_user_message_time float

Timestamp of last user message

required
context Dict[str, Any]

Current context information

required
config Dict[str, Any]

Full configuration dictionary

required
triggered_by_user_message bool

True if this evaluation was triggered by a user message

False

Returns:

Type Description
tuple[bool, str]

Tuple of (should_respond: bool, reason: str)

AIBasedDecisionEngine

Uses AI to make intelligent decisions about when to respond.

proactiveagent.decision_engines.ai_based.AIBasedDecisionEngine

Python
AIBasedDecisionEngine(provider: BaseProvider)

Bases: DecisionEngine

AI provider-based decision engine (default behavior)

Initialize with an AI provider

Parameters:

Name Type Description Default
provider BaseProvider

AI provider instance

required

should_respond async

Python
should_respond(messages: List[Dict[str, str]], last_user_message_time: float, context: Dict[str, Any], config: Dict[str, Any], triggered_by_user_message: bool = False) -> tuple[bool, str]

Determine if AI should respond and provide reasoning

Parameters:

Name Type Description Default
messages List[Dict[str, str]]

Conversation history

required
last_user_message_time float

Timestamp of last user message

required
context Dict[str, Any]

Current context information

required
config Dict[str, Any]

Full configuration dictionary

required
triggered_by_user_message bool

True if this evaluation was triggered by a user message

False

Returns:

Type Description
tuple[bool, str]

Tuple of (should_respond: bool, reason: str)

SimpleDecisionEngine

Simple time-based decision logic.

proactiveagent.decision_engines.simple.SimpleDecisionEngine

Python
SimpleDecisionEngine()

Bases: DecisionEngine

Simple time-based decision engine without AI

Initialize simple decision engine

should_respond async

Python
should_respond(messages: List[Dict[str, str]], last_user_message_time: float, context: Dict[str, Any], config: Dict[str, Any], triggered_by_user_message: bool = False) -> tuple[bool, str]

Make simple time-based decision

ThresholdDecisionEngine

Priority-based timing with different response times for different message types.

proactiveagent.decision_engines.threshold_based.ThresholdDecisionEngine

Python
ThresholdDecisionEngine(response_thresholds: Dict[str, int] = None)

Bases: DecisionEngine

Threshold-based decision engine with configurable parameters

Initialize with response thresholds

Parameters:

Name Type Description Default
response_thresholds Dict[str, int]

Dictionary mapping context types to response intervals in seconds

None

should_respond async

Python
should_respond(messages: List[Dict[str, str]], last_user_message_time: float, context: Dict[str, Any], config: Dict[str, Any], triggered_by_user_message: bool = False) -> tuple[bool, str]

Make threshold-based decision

FunctionBasedDecisionEngine

Custom decision logic using Python functions.

proactiveagent.decision_engines.function_based.FunctionBasedDecisionEngine

Python
FunctionBasedDecisionEngine(decision_function: Callable)

Bases: DecisionEngine

Adapter for function-based decision engines

Initialize with a decision function

Parameters:

Name Type Description Default
decision_function Callable

Function that takes (messages, last_user_message_time, context, config, triggered_by_user_message) and returns (should_respond: bool, reason: str)

required

should_respond async

Python
should_respond(messages: List[Dict[str, str]], last_user_message_time: float, context: Dict[str, Any], config: Dict[str, Any], triggered_by_user_message: bool = False) -> tuple[bool, str]

Make decision using the provided function