LRS-Agents: Resilient AI Agents via Active Inference

from lrs import create_lrs_agent
from langchain_anthropic import ChatAnthropic

# Create agent with automatic adaptation
llm = ChatAnthropic(model="claude-sonnet-4-20250514")
agent = create_lrs_agent(llm, tools=[api_tool, cache_tool])

# Agent automatically adapts when tools fail
result = agent.invoke({"messages": [{"role": "user", "content": "Fetch data"}]})
# API fails → Precision drops → Agent explores cache → Task completed! ✨

from lrs import create_lrs_agent
from lrs.multi_agent import MultiAgentCoordinator, SocialPrecisionTracker, CommunicationLens
from lrs.core.lens import ToolLens, ExecutionResult
from langchain_anthropic import ChatAnthropic
import random

# Define tools as ToolLens objects
class APITool(ToolLens):
    """Unreliable API that fails 30% of the time."""
    
    def __init__(self):
        super().__init__(
            name="api_call",
            input_schema={"type": "object", "properties": {"query": {"type": "string"}}},
            output_schema={"type": "object", "properties": {"data": {"type": "string"}}}
        )
    
    def get(self, state):
        if random.random() < 0.3:
            return ExecutionResult(
                success=False, 
                value=None, 
                error="API timeout", 
                prediction_error=0.9
            )
        return ExecutionResult(
            success=True, 
            value={"data": "API result"}, 
            error=None, 
            prediction_error=0.1
        )

class CacheTool(ToolLens):
    """Reliable cache that always works."""
    
    def __init__(self):
        super().__init__(
            name="cache_lookup",
            input_schema={"type": "object", "properties": {"key": {"type": "string"}}},
            output_schema={"type": "object", "properties": {"data": {"type": "string"}}}
        )
    
    def get(self, state):
        return ExecutionResult(
            success=True, 
            value={"data": "Cached result"}, 
            error=None, 
            prediction_error=0.05
        )

# Create LRS agent
llm = ChatAnthropic(model="claude-sonnet-4-20250514")
agent = create_lrs_agent(llm, tools=[APITool(), CacheTool()])

# Run task - agent automatically adapts when API fails
result = agent.invoke({
    "messages": [
        {"role": "user", "content": "Fetch data for query 'test'"}
    ]
})

print(result['messages'][-1]['content'])
# Output: Successfully retrieved data via cache (after API failures)

# Multi-agent coordination example
from lrs.multi_agent import MultiAgentCoordinator

coordinator = MultiAgentCoordinator()
coordinator.register_agent("agent_a", agent)
coordinator.register_agent("agent_b", agent)

# Run coordination with social intelligence
results = coordinator.run(
    task="Coordinate warehouse operations",
    max_rounds=10
)

print(f"Rounds: {results['total_rounds']}")
print(f"Messages: {results['total_messages']}")
print(f"Social Precisions: {results['social_precisions']}")

from lrs.multi_agent import SocialPrecisionTracker

tracker = SocialPrecisionTracker(agent_id="agent_a")
tracker.register_agent("agent_b")

# Agent A observes Agent B's behavior
tracker.update_social_precision(
    other_agent_id="agent_b",
    predicted_action="fetch_data", 
    observed_action="use_cache"  # Surprise!
)

print(tracker.get_social_precision("agent_b"))  # Decreased

from lrs.multi_agent import MultiAgentCoordinator

coordinator = MultiAgentCoordinator()
coordinator.register_agent("planner", planner_agent)
coordinator.register_agent("executor", executor_agent) 
coordinator.register_agent("monitor", monitor_agent)

# Agents coordinate with turn-taking and messaging
results = coordinator.run(
    task="Process customer order end-to-end",
    max_rounds=20
)

# Get coordination insights
print(f"Social trust levels: {results['social_precisions']}")
print(f"Communication efficiency: {results['total_messages']}")

# Low social precision + high environmental precision = communicate
should_message = tracker.should_communicate(
    other_agent_id="agent_b",
    env_precision=0.8  # Environment is reliable
)

if should_message:
    comm_tool = CommunicationLens("agent_a", shared_state)
    result = comm_tool.get({
        "to_agent": "agent_b",
        "message_type": "query",
        "content": "What's your current status?"
    })

from lrs.core.precision import PrecisionParameters

precision = PrecisionParameters()
print(precision.value)  # 0.5 (initial)

# Update after success
precision.update(prediction_error=0.1)
print(precision.value)  # 0.518 (slight increase)

# Update after failure
precision.update(prediction_error=0.9)
print(precision.value)  # 0.424 (larger decrease)

# Check if adaptation needed
if precision.should_adapt():
    print("Time to explore alternatives!")

from lrs.core.free_energy import calculate_expected_free_energy

# Reliable tool (high success rate, low info gain)
G_exploit = calculate_expected_free_energy(
    policy=[reliable_tool],
    registry=registry,
    preferences={"success": 5.0, "error": -3.0}
)
# G ≈ 0.2 - 4.5 = -4.3 (very negative = good)

# Novel tool (low success rate, high info gain)
G_explore = calculate_expected_free_energy(
    policy=[novel_tool],
    registry=registry,
    preferences={"success": 5.0, "error": -3.0}
)
# G ≈ 2.0 - 1.0 = 1.0 (positive = uncertain)

# At low precision, exploration becomes more valuable!

from lrs.core.lens import ToolLens, ExecutionResult

class SearchTool(ToolLens):
    def get(self, state):
        """Forward: Execute search."""
        try:
            results = search_api(state['query'])
            return ExecutionResult(
                success=True,
                value=results,
                error=None,
                prediction_error=0.1  # Low error = high confidence
            )
        except Exception as e:
            return ExecutionResult(
                success=False,
                value=None,
                error=str(e),
                prediction_error=0.9  # High error = low confidence
            )
    
    def set(self, state, obs):
        """Backward: Update state with results."""
        return {**state, 'search_results': obs}

# Compose tools with >> operator
pipeline = search_tool >> filter_tool >> format_tool
result = pipeline.get(state)

from lrs.core.precision import HierarchicalPrecision

hp = HierarchicalPrecision()

# High execution error propagates upward
hp.update('execution', prediction_error=0.95)

print(hp.execution)  # 0.42 (dropped significantly)
print(hp.planning)   # 0.46 (attenuated propagation)
print(hp.abstract)   # 0.49 (minimal impact)

from langchain_community.tools import DuckDuckGoSearchRun
from lrs.integration.langchain_adapter import wrap_langchain_tool

# Wrap LangChain tool
search = wrap_langchain_tool(
    DuckDuckGoSearchRun(),
    timeout=10.0,
    error_fn=lambda result, schema: 0.1 if result else 0.8
)

# Use in LRS agent
agent = create_lrs_agent(llm, tools=[search, other_tools])

from lrs.integration.openai_assistants import create_openai_lrs_agent
from openai import OpenAI

client = OpenAI(api_key="...")
agent = create_openai_lrs_agent(
    client=client,
    assistant_id="asst_...",
    tools=[file_tool, search_tool]
)

# Temperature automatically adapts based on precision
# High precision → Low temperature (exploit)
# Low precision → High temperature (explore)

from lrs.integration.autogpt_adapter import LRSAutoGPTAgent

agent = LRSAutoGPTAgent(
    llm=llm,
    commands=[read_file, write_file, web_search, execute_code],
    goals=["Research topic X", "Write summary", "Save to file"]
)

# AutoGPT now automatically adapts when commands fail
agent.run()

from lrs.monitoring.dashboard import run_dashboard
from lrs.monitoring.tracker import LRSStateTracker

tracker = LRSStateTracker()

# Track agent execution
tracker.track_state({
    'precision': precision.get_all_values(),
    'prediction_errors': [0.1, 0.3, 0.8],
    'tool_history': ['api_call', 'api_call', 'cache_lookup']
})

# Launch dashboard
run_dashboard(tracker, port=8501)
# Open http://localhost:8501

from lrs.monitoring.structured_logging import create_logger_for_agent

logger = create_logger_for_agent('my_agent')

# All events logged as structured JSON
logger.log_tool_execution(
    tool_name='api_call',
    success=False,
    prediction_error=0.9,
    execution_time=1.2
)

logger.log_adaptation_event(
    level='execution',
    old_precision=0.45,
    new_precision=0.38,
    trigger='precision_threshold'
)