aishastr.live

Agentic Architecture Pattern: Model Based Agent

Written by

admin

in


    """
Pattern 46: Model-Based Agent Pattern

Agent maintains internal world model to predict outcomes and plan actions.
Features state transition modeling, prediction, and model updating.
"""

from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class StateType(Enum):
    """Types of state variables."""
    NUMERIC = "numeric"
    BOOLEAN = "boolean"
    CATEGORICAL = "categorical"


@dataclass
class WorldState:
    """Current state of the world."""
    variables: Dict[str, Any]
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class Action:
    """Action that can be performed."""
    name: str
    parameters: Dict[str, Any] = field(default_factory=dict)


@dataclass
class Transition:
    """State transition record."""
    from_state: WorldState
    action: Action
    to_state: WorldState
    reward: float = 0.0


class WorldModel:
    """Internal model of environment dynamics."""
    
    def __init__(self, llm: ChatOpenAI):
        self.llm = llm
        self.transitions: List[Transition] = []
        self.state_variables: Dict[str, StateType] = {}
    
    def define_variable(self, name: str, var_type: StateType):
        """Define a state variable."""
        self.state_variables[name] = var_type
    
    def predict_next_state(self, current_state: WorldState, action: Action) -> WorldState:
        """Predict next state after action using LLM."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a world model predicting state transitions. "
                      "Predict the next state after an action is performed."),
            ("human", """Current State: {current_state}
Action: {action_name} with parameters {action_params}

Predict the next state. Return as JSON-like dict with same variable names.
Only respond with the predicted state dict.""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "current_state": str(current_state.variables),
            "action_name": action.name,
            "action_params": str(action.parameters)
        })
        
        try:
            # Parse LLM response
            import ast
            predicted_vars = ast.literal_eval(response.content.strip())
            return WorldState(variables=predicted_vars)
        except:
            # Fallback: return current state
            return WorldState(variables=current_state.variables.copy())
    
    def update_model(self, transition: Transition):
        """Update model with observed transition."""
        self.transitions.append(transition)
    
    def evaluate_action(self, current_state: WorldState, action: Action) -> float:
        """Evaluate expected reward of action."""
        predicted_state = self.predict_next_state(current_state, action)
        
        # Use LLM to estimate reward
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Evaluate how good a state is. Return a number from -10 to 10."),
            ("human", "State: {state}\nRate this state's desirability:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"state": str(predicted_state.variables)})
        
        try:
            reward = float(response.content.strip())
            return max(-10, min(10, reward))
        except:
            return 0.0


class ModelBasedAgent:
    """Agent that uses world model for decision-making."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.world_model = WorldModel(self.llm)
        self.current_state: Optional[WorldState] = None
        self.action_history: List[Tuple[Action, WorldState]] = []
    
    def perceive(self, state: WorldState):
        """Update agent's perception of world state."""
        if self.current_state and self.action_history:
            # Learn from transition
            last_action, _ = self.action_history[-1]
            transition = Transition(
                from_state=self.current_state,
                action=last_action,
                to_state=state
            )
            self.world_model.update_model(transition)
        
        self.current_state = state
        print(f"   {self.name} perceived state: {state.variables}")
    
    def plan_action(self, available_actions: List[Action]) -> Action:
        """Select best action based on model predictions."""
        if not self.current_state:
            return available_actions[0]
        
        print(f"   {self.name} evaluating {len(available_actions)} possible actions...")
        
        # Evaluate each action
        action_scores = []
        for action in available_actions:
            score = self.world_model.evaluate_action(self.current_state, action)
            action_scores.append((action, score))
            print(f"     - {action.name}: score {score:.2f}")
        
        # Select best action
        action_scores.sort(key=lambda x: x[1], reverse=True)
        best_action = action_scores[0][0]
        
        print(f"   {self.name} selected: {best_action.name}")
        return best_action
    
    def execute(self, action: Action):
        """Execute action and record."""
        self.action_history.append((action, self.current_state))
        print(f"   {self.name} executing: {action.name}")


class ModelBasedAgentPattern:
    """
    Model-Based Agent Pattern - World model for predictions.
    
    Agent maintains internal model of environment dynamics.
    Uses model to predict outcomes and select optimal actions.
    """
    
    def __init__(self):
        self.agents: Dict[str, ModelBasedAgent] = {}
    
    def register_agent(self, agent: ModelBasedAgent):
        """Register model-based agent."""
        self.agents[agent.name] = agent
    
    def run(self, simulation_steps: List[Dict[str, Any]]) -> Dict[str, Any]:
        """Run simulation with model-based agents."""
        print("=== Model-Based Agent Pattern Demo ===\n")
        
        for step_num, step in enumerate(simulation_steps, 1):
            print(f"\nStep {step_num}:")
            
            # Update state
            state = WorldState(variables=step["state"])
            available_actions = [
                Action(name=a["name"], parameters=a.get("params", {}))
                for a in step["actions"]
            ]
            
            for agent in self.agents.values():
                agent.perceive(state)
                action = agent.plan_action(available_actions)
                agent.execute(action)
        
        # Statistics
        total_transitions = sum(
            len(agent.world_model.transitions) 
            for agent in self.agents.values()
        )
        
        return {
            "pattern": "ModelBasedAgent",
            "total_agents": len(self.agents),
            "simulation_steps": len(simulation_steps),
            "learned_transitions": total_transitions
        }


# Example usage
if __name__ == "__main__":
    print("=== Model-Based Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = ModelBasedAgentPattern()
    
    # Create agent
    print("1. Creating model-based agent...")
    agent = ModelBasedAgent("RobotAgent")
    
    # Define state variables
    agent.world_model.define_variable("battery", StateType.NUMERIC)
    agent.world_model.define_variable("location", StateType.CATEGORICAL)
    agent.world_model.define_variable("task_complete", StateType.BOOLEAN)
    
    pattern.register_agent(agent)
    print(f"   Agent created with world model")
    
    # Simulation
    print("\n2. Running simulation...")
    
    simulation = [
        {
            "state": {"battery": 100, "location": "start", "task_complete": False},
            "actions": [
                {"name": "move_to_target"},
                {"name": "charge_battery"},
                {"name": "wait"}
            ]
        },
        {
            "state": {"battery": 80, "location": "target", "task_complete": False},
            "actions": [
                {"name": "perform_task"},
                {"name": "return_to_base"},
                {"name": "charge_battery"}
            ]
        },
        {
            "state": {"battery": 60, "location": "target", "task_complete": True},
            "actions": [
                {"name": "return_to_base"},
                {"name": "charge_battery"},
                {"name": "wait"}
            ]
        },
    ]
    
    summary = pattern.run(simulation)
    
    print(f"\n3. Summary:")
    print(f"   Simulation Steps: {summary['simulation_steps']}")
    print(f"   Learned Transitions: {summary['learned_transitions']}")
    print(f"   Actions Taken: {len(agent.action_history)}")
    
    print("\nāœ“ Agent used world model to predict and plan!")

Agentic Architecture Pattern: Model Based Agent

Pattern 46: 46 Model Based Agent – Line-by-Line Explanation

Overview

Pattern 46: Model-Based Agent Pattern

Agent maintains internal world model to predict outcomes and plan actions. Features state transition modeling, prediction, and model updating.

Code Structure Analysis

File Information

  • Pattern Number: 46
  • Pattern Name: 46 Model Based Agent
  • File Path: patterns\decision_making\46_model_based_agent.py
  • Total Lines: 267

Detailed Code Explanation

Module Docstring

"""
Pattern 46: Model-Based Agent Pattern

Agent maintains internal world model to predict outcomes and plan actions.
Features state transition modeling, prediction, and model updating.
"""

Explanation: This module docstring describes the purpose and key features of the 46 Model Based Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
            import ast  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Tuple: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

StateType (Line 16)

class StateType(Enum):
    NUMERIC
    BOOLEAN
    CATEGORICAL

Purpose: Types of state variables.

Values: – NUMERIC: Represents numeric – BOOLEAN: Represents boolean – CATEGORICAL: Represents categorical

Classes

StateType (Line 16)

Description: Types of state variables.

WorldState (Line 24)

Description: Current state of the world.

Action (Line 31)

Description: Action that can be performed.

Transition (Line 38)

Description: State transition record.

WorldModel (Line 46)

Description: Internal model of environment dynamics.

Methods:

__init__ (Line 49)
def __init__(self, llm):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – llm: Parameter for llm

define_variable (Line 54)
def define_variable(self, name, var_type):
    """Define a state variable."""

Purpose: Define a state variable.

Parameters: – self: Instance reference – name: Identifier or label – var_type: Parameter for var_type

predict_next_state (Line 58)
def predict_next_state(self, current_state, action):
    """Predict next state after action using LLM."""

Purpose: Predict next state after action using LLM.

Parameters: – self: Instance reference – current_state: Parameter for current_state – action: Parameter for action

update_model (Line 86)
def update_model(self, transition):
    """Update model with observed transition."""

Purpose: Update model with observed transition.

Parameters: – self: Instance reference – transition: Parameter for transition

evaluate_action (Line 90)
def evaluate_action(self, current_state, action):
    """Evaluate expected reward of action."""

Purpose: Evaluate expected reward of action.

Parameters: – self: Instance reference – current_state: Parameter for current_state – action: Parameter for action

ModelBasedAgent (Line 110)

Description: Agent that uses world model for decision-making.

Methods:

__init__ (Line 113)
def __init__(self, name, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – model_name: LLM model to use (e.g., “gpt-4”)

perceive (Line 120)
def perceive(self, state):
    """Update agent's perception of world state."""

Purpose: Update agent’s perception of world state.

Parameters: – self: Instance reference – state: Current workflow state

plan_action (Line 135)
def plan_action(self, available_actions):
    """Select best action based on model predictions."""

Purpose: Select best action based on model predictions.

Parameters: – self: Instance reference – available_actions: Parameter for available_actions

execute (Line 156)
def execute(self, action):
    """Execute action and record."""

Purpose: Execute action and record.

Parameters: – self: Instance reference – action: Parameter for action

ModelBasedAgentPattern (Line 162)

Description: Model-Based Agent Pattern – World model for predictions.

Agent maintains internal model of environment dynamics. Uses model to predict outcomes and select optimal actions.

Methods:

__init__ (Line 170)
def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 173)
def register_agent(self, agent):
    """Register model-based agent."""

Purpose: Register model-based agent.

Parameters: – self: Instance reference – agent: Parameter for agent

run (Line 177)
def run(self, simulation_steps):
    """Run simulation with model-based agents."""

Purpose: Run simulation with model-based agents.

Parameters: – self: Instance reference – simulation_steps: Parameter for simulation_steps

Design Patterns Used

  • Strategy Pattern: Different execution strategies for various scenarios
  • State Pattern: Managing different states in the workflow
  • Chain of Responsibility: Sequential processing through workflow nodes
  • Observer Pattern: Tracking and monitoring pattern execution
  • Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = StateType(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

  • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
  • Workflow Orchestration: LangGraph manages execution flow through nodes
  • State Management: TypedDict defines workflow state structure
  • Error Handling: Comprehensive error handling and fallback mechanisms
  • Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts