Create agent.py

agent.py

@@ -0,0 +1,189 @@
+import numpy as np
+import json
+from datetime import datetime
+import random
+
+class ProcurementRLAgent:
+    def __init__(self, tools):
+        self.tools = {tool.__class__.__name__: tool for tool in tools}
+        self.q_table = self._load_or_initialize_q_table()
+        self.learning_rate = 0.1
+        self.discount_factor = 0.95
+        self.epsilon = 0.1  # exploration rate
+        
+    def _load_or_initialize_q_table(self):
+        """Initialize Q-table with realistic values based on historical data"""
+        try:
+            with open("demo_space/historical_procurement_data.json", "r") as f:
+                historical_data = json.load(f)
+            
+            # Initialize Q-table based on historical patterns
+            # States: (urgency, contract_status, external_disruption, inventory_sufficient)
+            # Actions: (auto_execute, human_review, transfer_inventory, extend_contract)
+            
+            q_table = {}
+            for urgency in ["Low", "Medium", "High"]:
+                for contract in ["Valid", "Expired"]:
+                    for disruption in [True, False]:
+                        for inventory in [True, False]:
+                            state = (urgency, contract, disruption, inventory)
+                            # Initialize with reasonable values based on business logic
+                            if contract == "Expired":
+                                q_table[state] = {"auto_execute": 0.2, "human_review": 0.8}
+                            elif disruption:
+                                q_table[state] = {"auto_execute": 0.3, "human_review": 0.7}
+                            elif urgency == "High" and not inventory:
+                                q_table[state] = {"auto_execute": 0.6, "human_review": 0.4}
+                            else:
+                                q_table[state] = {"auto_execute": 0.8, "human_review": 0.2}
+            
+            return q_table
+        except FileNotFoundError:
+            return self._default_q_table()
+    
+    def _default_q_table(self):
+        """Default Q-table initialization"""
+        return {
+            ("Low", "Valid", False, True): {"auto_execute": 0.9, "human_review": 0.1},
+            ("Medium", "Valid", False, True): {"auto_execute": 0.8, "human_review": 0.2},
+            ("High", "Valid", False, True): {"auto_execute": 0.7, "human_review": 0.3},
+            ("High", "Expired", True, False): {"auto_execute": 0.1, "human_review": 0.9},
+        }
+    
+    def _get_state(self, pr_data):
+        """Convert PR data to state representation"""
+        inventory_sufficient = pr_data["Current_Inventory"] >= pr_data["Quantity"]
+        return (
+            pr_data["Urgency"],
+            pr_data["Contract_Status"],
+            pr_data["External_Disruption"],
+            inventory_sufficient
+        )
+    
+    def _select_action(self, state):
+        """Select action using epsilon-greedy policy"""
+        if state in self.q_table:
+            if random.random() < self.epsilon:
+                # Exploration: random action
+                return random.choice(["auto_execute", "human_review"])
+            else:
+                # Exploitation: best action
+                return max(self.q_table[state].items(), key=lambda x: x[1])[0]
+        else:
+            # Default action for unknown states
+            return "human_review"
+    
+    def decide(self, pr_data):
+        """Main decision function"""
+        timestamp = datetime.now().strftime("%H:%M:%S")
+        logs = []
+        
+        # Get current state
+        state = self._get_state(pr_data)
+        logs.append({
+            "timestamp": timestamp,
+            "tool": "StateAnalyzer",
+            "result": f"State: {state}"
+        })
+        
+        # Execute tools to gather information
+        tool_results = {}
+        
+        # Check inventory
+        inventory_result = self.tools["InventoryTool"].run(pr_data)
+        tool_results["inventory"] = inventory_result
+        logs.append({
+            "timestamp": timestamp,
+            "tool": "InventoryTool",
+            "result": inventory_result["message"]
+        })
+        
+        # Check external risks
+        risk_result = self.tools["ExternalRiskTool"].run(pr_data)
+        tool_results["risk"] = risk_result
+        logs.append({
+            "timestamp": timestamp,
+            "tool": "ExternalRiskTool",
+            "result": risk_result["message"]
+        })
+        
+        # Check contract status
+        contract_result = self.tools["ContractTool"].run(pr_data)
+        tool_results["contract"] = contract_result
+        logs.append({
+            "timestamp": timestamp,
+            "tool": "ContractTool",
+            "result": contract_result["message"]
+        })
+        
+        # Select action using RL policy
+        selected_action = self._select_action(state)
+        
+        # Execute decision logic
+        if selected_action == "auto_execute" and self._can_auto_execute(tool_results):
+            # Execute PO automatically
+            po_result = self.tools["POExecutionTool"].run(pr_data)
+            logs.append({
+                "timestamp": timestamp,
+                "tool": "POExecutionTool",
+                "result": po_result["message"]
+            })
+            
+            confidence = self.q_table.get(state, {}).get("auto_execute", 0.5)
+            
+            return {
+                "action_type": "AUTO_EXECUTE",
+                "action": "Purchase Order created and sent to supplier",
+                "reason": "All conditions met for automatic execution",
+                "confidence": confidence,
+                "logs": logs
+            }
+        else:
+            # Require human review
+            human_result = self.tools["HumanNotificationTool"].run(pr_data)
+            logs.append({
+                "timestamp": timestamp,
+                "tool": "HumanNotificationTool",
+                "result": human_result["message"]
+            })
+            
+            reason = self._get_human_review_reason(tool_results)
+            
+            return {
+                "action_type": "HUMAN_REVIEW",
+                "action": "Escalated to procurement specialist",
+                "reason": reason,
+                "recommendation": self._get_recommendation(tool_results),
+                "logs": logs
+            }
+    
+    def _can_auto_execute(self, tool_results):
+        """Determine if automatic execution is possible"""
+        return (
+            not tool_results["contract"]["contract_expired"] and
+            not tool_results["risk"]["risk"] and
+            tool_results["inventory"]["inventory_available"]
+        )
+    
+    def _get_human_review_reason(self, tool_results):
+        """Get reason for human review requirement"""
+        reasons = []
+        if tool_results["contract"]["contract_expired"]:
+            reasons.append("Contract expired - requires renewal")
+        if tool_results["risk"]["risk"]:
+            reasons.append("External supply chain disruption detected")
+        if not tool_results["inventory"]["inventory_available"]:
+            reasons.append("Insufficient inventory - supplier sourcing required")
+        
+        return "; ".join(reasons) if reasons else "Complex procurement scenario requires expert review"
+    
+    def _get_recommendation(self, tool_results):
+        """Get recommendation for human reviewer"""
+        if tool_results["contract"]["contract_expired"]:
+            return "Extend contract with current supplier or initiate competitive bidding"
+        elif tool_results["risk"]["risk"]:
+            return "Evaluate alternative suppliers to mitigate supply chain risk"
+        elif not tool_results["inventory"]["inventory_available"]:
+            return "Check inventory at other locations or place emergency order"
+        else:
+            return "Review procurement parameters and approve if acceptable"