Delete agent.py

agent.py

@@ -1,189 +0,0 @@
-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"