Core Components

Note

This section documents the core Active Inference components of LRS-Agents.

Precision Tracking

Precision tracking for Active Inference agents.

class lrs.core.precision.PrecisionParameters(alpha: float = 1.0, beta: float = 1.0, gain_learning_rate: float = 0.1, loss_learning_rate: float = 0.2, adaptation_threshold: float = 0.4)

Bases: object

Precision parameters using Beta distribution.

Precision γ = α/(α+β) represents confidence in predictions.

Parameters:

• alpha – Success parameter (default: 1.0)

• beta – Failure parameter (default: 1.0)

• gain_learning_rate – Learning rate for successes (default: 0.1)

• loss_learning_rate – Learning rate for failures (default: 0.2)

• adaptation_threshold – Threshold below which adaptation triggers (default: 0.4)

Example

>>> precision = PrecisionParameters()
>>> precision.value  # 0.5 (maximum uncertainty)
>>> precision.update(prediction_error=0.1)  # Success
>>> precision.value  # ~0.52 (slight increase)

alpha: float = 1.0

beta: float = 1.0

gain_learning_rate: float = 0.1

loss_learning_rate: float = 0.2

adaptation_threshold: float = 0.4

property value: float

Get current precision value γ = α/(α+β).

Returns:

Precision in [0,1]

property variance: float

Get variance of Beta distribution.

Returns:

Variance of precision estimate

update(prediction_error: float) → None

Update precision based on prediction error.

Uses asymmetric learning rates: - Low error (success) → small increase in α - High error (failure) → larger increase in β

Parameters:

prediction_error – Prediction error δ ∈ [0,1]

Example

>>> precision = PrecisionParameters()
>>> precision.update(0.1)  # Success
>>> precision.update(0.9)  # Failure

should_adapt() → bool

Check if precision is below adaptation threshold.

Returns:

True if adaptation should be triggered

reset() → None

Reset to initial uniform prior.

get_all() → Dict[str, float]

Get all precision statistics.

Returns:

Dictionary with value, alpha, beta, variance

property learning_rate_gain: float

Alias for gain_learning_rate.

property learning_rate_loss: float

Alias for loss_learning_rate.

property threshold: float

Alias for adaptation_threshold.

get_all_values() → Dict[str, float]

Alias for get_all().

__init__(alpha: float = 1.0, beta: float = 1.0, gain_learning_rate: float = 0.1, loss_learning_rate: float = 0.2, adaptation_threshold: float = 0.4) → None

class lrs.core.precision.HierarchicalPrecision(_abstract: ~lrs.core.precision.PrecisionParameters = <factory>, _planning: ~lrs.core.precision.PrecisionParameters = <factory>, _execution: ~lrs.core.precision.PrecisionParameters = <factory>, propagation_threshold: float = 0.7, attenuation_factor: float = 0.5)

Bases: object

Hierarchical precision tracking across abstraction levels.

Precision is maintained at three levels: - Abstract: Long-term goals and strategies - Planning: Policy sequences - Execution: Individual tool calls

Errors propagate upward with attenuation.

Example

>>> hp = HierarchicalPrecision()
>>> hp.update('execution', 0.9)  # High error
>>> hp.execution  # Decreased
>>> hp.planning   # Also decreased (propagation)

propagation_threshold: float = 0.7

attenuation_factor: float = 0.5

property abstract: float

Get abstract level precision value.

property planning: float

Get planning level precision value.

property execution: float

Get execution level precision value.

get_level(level: str) → PrecisionParameters

Get PrecisionParameters object for a specific level.

Parameters:

level – One of ‘abstract’, ‘planning’, or ‘execution’

Returns:

PrecisionParameters object for that level

Example

>>> hp = HierarchicalPrecision()
>>> exec_params = hp.get_level('execution')
>>> exec_params.value  # 0.5

update(level: str, prediction_error: float) → None

Update precision at a specific level with upward propagation.

High prediction errors (>0.7) propagate upward with attenuation.

Parameters:

• level – Level to update (‘abstract’, ‘planning’, ‘execution’)

• prediction_error – Prediction error δ ∈ [0,1]

Example

>>> hp = HierarchicalPrecision()
>>> hp.update('execution', 0.95)  # High error
>>> # Execution precision drops AND planning is affected

get_all_values() → Dict[str, float]

Get all precision values as a dictionary.

Returns:

Dictionary with abstract, planning, execution values

get_all() → Dict[str, float]

Alias for get_all_values().

reset() → None

Reset all levels to initial values.

should_adapt(level: str = 'execution') → bool

Check if adaptation is needed at specified level.

Parameters:

level – Level to check (default: ‘execution’)

Returns:

True if adaptation should be triggered

__init__(_abstract: ~lrs.core.precision.PrecisionParameters = <factory>, _planning: ~lrs.core.precision.PrecisionParameters = <factory>, _execution: ~lrs.core.precision.PrecisionParameters = <factory>, propagation_threshold: float = 0.7, attenuation_factor: float = 0.5) → None

lrs.core.precision.beta_mean(alpha: float, beta: float) → float

Calculate mean of Beta distribution.

Parameters:

• alpha – Alpha parameter

• beta – Beta parameter

Returns:

Mean = α/(α+β)

lrs.core.precision.beta_variance(alpha: float, beta: float) → float

Calculate variance of Beta distribution.

Parameters:

• alpha – Alpha parameter

• beta – Beta parameter

Returns:

Variance

Tool Lenses

ToolLens: Categorical abstraction for tools.

A lens is a bidirectional morphism: - get: Execute the tool (forward) - set: Update belief state (backward)

Lenses compose via the >> operator, creating pipelines with automatic error propagation.

class lrs.core.lens.ExecutionResult(success: bool, value: Any | None, error: str | None, prediction_error: float)

Bases: object

Result of executing a tool.

success

Whether execution succeeded

Type:

bool

value

Return value (None if failed)

Type:

Any | None

error

Error message (None if succeeded)

Type:

str | None

prediction_error

How surprising this outcome was [0, 1]

Type:

float

Examples

>>> # Successful execution
>>> result = ExecutionResult(
...     success=True,
...     value="Data fetched",
...     error=None,
...     prediction_error=0.1  # Expected success
... )
>>>
>>> # Failed execution
>>> result = ExecutionResult(
...     success=False,
...     value=None,
...     error="API timeout",
...     prediction_error=0.9  # Unexpected failure
... )

success: bool

value: Any | None

error: str | None

prediction_error: float

__init__(success: bool, value: Any | None, error: str | None, prediction_error: float) → None

class lrs.core.lens.ToolLens(name: str, input_schema: Dict[str, Any], output_schema: Dict[str, Any])

Bases: ABC

Abstract base class for tools as lenses.

A lens has two operations: 1. get(state) → ExecutionResult: Execute the tool 2. set(state, observation) → state: Update belief state

Lenses compose via >> operator:

lens_a >> lens_b >> lens_c

This creates a pipeline where: - Data flows forward through get operations - Belief updates flow backward through set operations - Errors propagate automatically

name

Tool identifier

input_schema

JSON schema for inputs

output_schema

JSON schema for outputs

call_count

Number of times get() has been called

failure_count

Number of times get() has failed

Examples

>>> class FetchTool(ToolLens):
...     def get(self, state):
...         data = fetch(state['url'])
...         return ExecutionResult(True, data, None, 0.1)
...
...     def set(self, state, observation):
...         return {**state, 'data': observation}
>>>
>>> class ParseTool(ToolLens):
...     def get(self, state):
...         parsed = json.loads(state['data'])
...         return ExecutionResult(True, parsed, None, 0.05)
...
...     def set(self, state, observation):
...         return {**state, 'parsed': observation}
>>>
>>> # Compose
>>> pipeline = FetchTool() >> ParseTool()
>>> result = pipeline.get({'url': 'api.com/data'})

__init__(name: str, input_schema: Dict[str, Any], output_schema: Dict[str, Any])

Initialize tool lens.

Parameters:

• name – Unique tool identifier

• input_schema – JSON schema for expected inputs

• output_schema – JSON schema for expected outputs

abstractmethod get(state: Dict[str, Any]) → ExecutionResult

Execute the tool (forward operation).

Parameters:

state – Current agent state

Returns:

ExecutionResult with value and prediction error

Note

Implementations should update call_count and failure_count

abstractmethod set(state: Dict[str, Any], observation: Any) → Dict[str, Any]

Update belief state with observation (backward operation).

Parameters:

• state – Current state

• observation – Tool output

Returns:

Updated state

property success_rate: float

Calculate success rate from history

class lrs.core.lens.ComposedLens(left: ToolLens, right: ToolLens)

Bases: ToolLens

Composition of two lenses.

Created via >> operator. Handles: - Forward data flow (left.get then right.get) - Backward belief update (right.set then left.set) - Error short-circuiting (stop on first failure)

left

First lens in composition

right

Second lens in composition

__init__(left: ToolLens, right: ToolLens)

Create composed lens.

Parameters:

• left – First lens

• right – Second lens

get(state: Dict[str, Any]) → ExecutionResult

Execute composed lens (left then right).

If left fails, short-circuit and return left’s error. Otherwise, execute right with left’s output.

Parameters:

state – Input state

Returns:

ExecutionResult from final lens (or first failure)

set(state: Dict[str, Any], observation: Any) → Dict[str, Any]

Update state (right then left, backward flow).

Parameters:

• state – Current state

• observation – Final observation

Returns:

Fully updated state

Tool Registry

Tool registry with natural transformation discovery.

Manages tools and their fallback chains. Automatically discovers alternative tools based on schema compatibility.

class lrs.core.registry.ToolRegistry

Bases: object

Registry for managing tools and their alternatives.

Features: - Register tools with explicit fallback chains - Discover compatible alternatives via schema matching - Track tool statistics for Free Energy calculation

tools

Dict mapping tool names to ToolLens objects

alternatives

Dict mapping tool names to lists of alternative names

statistics

Dict tracking execution history per tool

Examples

>>> registry = ToolRegistry()
>>>
>>> # Register primary tool with alternatives
>>> registry.register(
...     api_tool,
...     alternatives=["cache_tool", "fallback_tool"]
... )
>>>
>>> # Register alternatives
>>> registry.register(cache_tool)
>>> registry.register(fallback_tool)
>>>
>>> # Find alternatives when primary fails
>>> alts = registry.find_alternatives("api_tool")
>>> print(alts)
['cache_tool', 'fallback_tool']

__init__()

Initialize empty registry

register(tool: ToolLens, alternatives: List[str] | None = None)

Register a tool with optional alternatives.

Parameters:

• tool – ToolLens to register

• alternatives – List of alternative tool names (fallback chain)

Examples

>>> registry.register(
...     APITool(),
...     alternatives=["CacheTool", "LocalTool"]
... )

get_tool(name: str) → ToolLens | None

Retrieve tool by name.

Parameters:

name – Tool name

Returns:

ToolLens or None if not found

find_alternatives(tool_name: str) → List[str]

Find registered alternatives for a tool.

Parameters:

tool_name – Name of primary tool

Returns:

List of alternative tool names (may be empty)

Examples

>>> alts = registry.find_alternatives("api_tool")
>>> for alt_name in alts:
...     alt_tool = registry.get_tool(alt_name)
...     result = alt_tool.get(state)

discover_compatible_tools(input_schema: Dict[str, Any], output_schema: Dict[str, Any]) → List[str]

Discover tools compatible with given schemas.

Uses structural matching to find tools that could serve as natural transformations (alternatives).

Parameters:

• input_schema – Required input schema

• output_schema – Required output schema

Returns:

List of compatible tool names

Examples

>>> compatible = registry.discover_compatible_tools(
...     input_schema={'type': 'object', 'required': ['url']},
...     output_schema={'type': 'string'}
... )

update_statistics(tool_name: str, success: bool, prediction_error: float)

Update execution statistics for a tool.

Used by Free Energy calculation to estimate success probabilities and epistemic values.

Parameters:

• tool_name – Name of executed tool

• success – Whether execution succeeded

• prediction_error – Observed prediction error

Examples

>>> registry.update_statistics("api_tool", success=True, prediction_error=0.1)

get_statistics(tool_name: str) → Dict[str, Any] | None

Retrieve statistics for a tool.

Parameters:

tool_name – Tool name

Returns:

Statistics dict or None

list_tools() → List[str]

List all registered tool names

Free Energy

Free energy calculations for Active Inference.

class lrs.core.free_energy.PolicyEvaluation(epistemic_value: float, pragmatic_value: float, total_G: float, expected_success_prob: float, components: Dict[str, Any])

Bases: object

Results of policy evaluation.

epistemic_value: float

pragmatic_value: float

total_G: float

expected_success_prob: float

components: Dict[str, Any]

__init__(epistemic_value: float, pragmatic_value: float, total_G: float, expected_success_prob: float, components: Dict[str, Any]) → None

lrs.core.free_energy.calculate_epistemic_value(policy: List[ToolLens], registry: ToolRegistry | None = None, historical_stats: Dict[str, Dict[str, float]] | None = None) → float

Calculate epistemic value (information gain) of a policy.

Higher values indicate more information gain from exploration.

Parameters:

• policy – Sequence of tools to execute

• registry – Tool registry with statistics

Returns:

Epistemic value (information gain)

Example

>>> epistemic = calculate_epistemic_value([novel_tool])
>>> # High value for unexplored tools

lrs.core.free_energy.calculate_pragmatic_value(policy: List[ToolLens], preferences: Dict[str, float], registry: ToolRegistry | None = None, historical_stats: Dict[str, Dict[str, float]] | None = None, discount_factor: float = 0.95) → float

Calculate pragmatic value (expected reward) of a policy.

Higher values indicate higher expected utility.

Parameters:

• policy – Sequence of tools to execute

• preferences – Reward/cost for outcomes (success, error, step_cost)

• registry – Tool registry with statistics

• discount – Temporal discount factor (default: 0.95)

Returns:

Pragmatic value (expected reward)

Example

>>> pragmatic = calculate_pragmatic_value(
...     [reliable_tool],
...     preferences={'success': 5.0, 'error': -3.0}
... )
>>> # High value for reliable tools

lrs.core.free_energy.calculate_expected_free_energy(policy: List[ToolLens], registry: ToolRegistry | None = None, preferences: Dict[str, float] | None = None, precision: float | None = 0.5, historical_stats: Dict[str, Dict[str, float]] | None = None, epistemic_weight: float | None = None) → float

Calculate Expected Free Energy G(π) for a policy.

G(π) = Epistemic Value - Pragmatic Value

Lower G is better (minimization objective).

Parameters:

• policy – Sequence of tools to execute

• registry – Tool registry with statistics

• preferences – Reward structure

• precision – Current precision γ ∈ [0,1]

• epistemic_weight – Override for epistemic term weight

Returns:

Expected Free Energy G

Example

>>> G = calculate_expected_free_energy(
...     policy=[search_tool, filter_tool],
...     preferences={'success': 5.0, 'error': -3.0},
...     precision=0.7
... )
>>> # Low G indicates good policy

lrs.core.free_energy.evaluate_policy(policy: List[ToolLens], registry: ToolRegistry | None = None, preferences: Dict[str, float] | None = None, historical_stats: Dict[str, Dict[str, float]] | None = None, precision: float | None = 0.5) → PolicyEvaluation

Evaluate a single policy comprehensively.

Returns:

PolicyEvaluation with all metrics

lrs.core.free_energy.precision_weighted_selection(policy_evaluations: List[PolicyEvaluation], precision: float = 0.5, temperature: float = 1.0, evaluations: List[PolicyEvaluation] | None = None) → int

Select policy using precision-weighted softmax.

P(π) ∝ exp(-β * G(π))

where β = precision (inverse temperature).

Parameters:

• policy_evaluations – Evaluated policies

• precision – Current precision γ ∈ [0,1]

• temperature – Additional temperature parameter

Returns:

Index of selected policy

Example

>>> selected_idx = precision_weighted_selection(
...     evaluations,
...     precision=0.7
... )
>>> best_policy = policies[selected_idx]