Building AI Data Agents

# Walmart Supply Chain: ReAct Agent Simulation
class ReActAgent:
    def __init__(self, tools):
        self.tools = tools
        self.thought_log = []

    def think(self, thought):
        self.thought_log.append(("THINK", thought))
        print("THOUGHT: {0}".format(thought))

    def act(self, tool_name, params):
        self.thought_log.append(("ACT", tool_name))
        if tool_name in self.tools:
            result = self.tools[tool_name](params)
            print("ACTION : {0}({1}) -> {2}".format(tool_name, params, result))
            return result
        print("ACTION : {0} -> Tool not found!".format(tool_name))

    def run(self, goal):
        print("AGENT GOAL: '{0}'".format(goal))
        print("=" * 50)
        
        self.think("I need to find which routes had delays > 24 hours")
        delays = self.act("query_db", "SELECT route, hours FROM delays WHERE hours > 24")
        
        self.think("Found {0} delayed routes. Now I need to calculate the average".format(len(delays)))
        avg = self.act("calculator", sum(delays) / len(delays))
        
        self.think("Average delay is {0} hours. I should generate a report.".format(avg))
        self.act("generate_report", "Q4 Supply Chain Delay Report")
        
        print("\nAGENT COMPLETE: {0} steps executed.".format(len(self.thought_log)))

# Enterprise tools
tools = {
    "query_db": lambda q: [28, 36, 42],
    "calculator": lambda x: round(x, 1),
    "generate_report": lambda title: "Report '{0}' generated".format(title)
}
agent = ReActAgent(tools)
agent.run("Analyze Q4 supply chain delays and generate a report")

# Visa Fraud Investigation: Agent with Tools
class ToolRegistry:
    def __init__(self):
        self.tools = {}

    def register(self, name, description, func):
        self.tools[name] = {"desc": description, "func": func}
        print("REGISTERED TOOL: {0} -> {1}".format(name, description))

    def execute(self, name, params):
        if name in self.tools:
            return self.tools[name]["func"](params)
        return "Tool not found"

class InvestigationAgent:
    def __init__(self, registry):
        self.registry = registry

    def investigate(self, case_id):
        print("\nINVESTIGATION AGENT: Case {0}".format(case_id))
        
        # Step 1: Fetch transaction details
        tx = self.registry.execute("fetch_transaction", case_id)
        print(" Step 1: Transaction -> {0}".format(tx))
        
        # Step 2: Check merchant reputation
        rep = self.registry.execute("check_merchant", tx["merchant"])
        print(" Step 2: Merchant Rep -> {0}".format(rep))
        
        # Step 3: Calculate risk score
        risk = self.registry.execute("calc_risk", {"amount": tx["amount"], "rep": rep})
        print(" Step 3: Risk Score -> {0}".format(risk))
        
        verdict = "BLOCK" if risk > 0.7 else "ALLOW"
        print("\nVERDICT: {0} (Risk: {1})".format(verdict, risk))

# Setup tools
registry = ToolRegistry()
registry.register("fetch_transaction", "Get transaction by ID", lambda id: {"amount": 2500, "merchant": "UnknownShop", "country": "NG"})
registry.register("check_merchant", "Check merchant reputation", lambda m: 0.2 if m == "Amazon" else 0.9)
registry.register("calc_risk", "Calculate fraud risk", lambda d: round(d["amount"] / 3000 * d["rep"], 2))

agent = InvestigationAgent(registry)
agent.investigate(case_id="CASE-7742")

# Enterprise Agent: Safety Guardrails
class SafeAgent:
    def __init__(self, budget_limit):
        self.budget = budget_limit
        self.spent = 0
        self.audit_log = []

    def request_action(self, action, risk_level):
        self.audit_log.append(action)
        
        # Guardrail 1: Budget check
        cost = 1
        if self.spent + cost > self.budget:
            print("BLOCKED: Budget limit reached ({0}/{1})".format(self.spent, self.budget))
            return False
        
        # Guardrail 2: Approval gate for high-risk
        if risk_level == "HIGH":
            print("APPROVAL REQUIRED: '{0}' is HIGH risk.".format(action))
            print(" -> Waiting for human approval... [SIMULATED: APPROVED]")
        
        self.spent += cost
        print("EXECUTED: '{0}' (Risk: {1}, Budget: {2}/{3})".format(action, risk_level, self.spent, self.budget))
        return True

    def show_audit(self):
        print("\nAUDIT LOG ({0} actions):".format(len(self.audit_log)))
        for i, action in enumerate(self.audit_log):
            print(" [{0}] {1}".format(i + 1, action))

# Scenario: Visa fraud investigation agent with safety
agent = SafeAgent(budget_limit=3)
agent.request_action("Query transaction database", "LOW")
agent.request_action("Flag account for review", "MEDIUM")
agent.request_action("Reverse transaction $2,500", "HIGH")
agent.request_action("Send email notification", "LOW")
agent.show_audit()