Building AI-Powered Decision Engines: Implementation Guide with Code Examples

from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
from pydantic import BaseModel, Field
from typing import List, Optional
import json

class PolicyRule(BaseModel):
    condition: str = Field(description="The condition that triggers this rule")
    action: str = Field(description="The action to take when condition is met")
    priority: int = Field(description="Rule priority (1 = highest)")
    exceptions: List[str] = Field(default=[], description="Exceptions to this rule")

class PolicyDocument(BaseModel):
    title: str
    version: str
    effective_date: str
    rules: List[PolicyRule]
    escalation_criteria: List[str]

class PolicyParser:
    def __init__(self, llm_model: str = "gpt-4"):
        self.llm = OpenAI(model_name=llm_model, temperature=0.1)
        self.extraction_prompt = PromptTemplate(
            input_variables=["policy_text"],
            template="""
            Analyse the following business policy document and extract structured rules.

            Policy Document:
            {policy_text}

            Extract the following information in JSON format:
            1. Individual rules with conditions and actions
            2. Rule priorities based on context
            3. Exceptions to each rule
            4. Escalation criteria

            Return only valid JSON matching this schema:
            {{
                "rules": [
                    {{
                        "condition": "string describing when rule applies",
                        "action": "string describing what action to take",
                        "priority": integer,
                        "exceptions": ["list of exceptions"]
                    }}
                ],
                "escalation_criteria": ["list of escalation conditions"]
            }}
            """
        )
        self.chain = LLMChain(llm=self.llm, prompt=self.extraction_prompt)

    def parse_policy(self, policy_text: str) -> PolicyDocument:
        """Parse natural language policy into structured rules."""
        try:
            result = self.chain.run(policy_text=policy_text)
            parsed_data = json.loads(result)

            rules = [PolicyRule(**rule) for rule in parsed_data.get("rules", [])]

            return PolicyDocument(
                title="Extracted Policy",
                version="1.0",
                effective_date="2024-01-01",
                rules=rules,
                escalation_criteria=parsed_data.get("escalation_criteria", [])
            )
        except Exception as e:
            raise PolicyParsingError(f"Failed to parse policy: {str(e)}")

class PolicyParsingError(Exception):
    pass

from dataclasses import dataclass
from typing import Dict, Any, Optional
import asyncio
from datetime import datetime

@dataclass
class DecisionContext:
    request_id: str
    requester_id: str
    amount: float
    currency: str
    department: str
    vendor_id: Optional[str]
    urgency_level: str
    business_justification: str
    historical_approvals: List[Dict[str, Any]]
    current_budget_status: Dict[str, float]
    vendor_status: Optional[str]
    request_timestamp: datetime

class ContextBuilder:
    def __init__(self, db_connection, external_apis):
        self.db = db_connection
        self.apis = external_apis

    async def build_context(self, request_data: Dict[str, Any]) -> DecisionContext:
        """Gather all relevant context for decision making."""

        # Parallel data gathering for performance
        context_tasks = [
            self._get_historical_approvals(request_data["requester_id"]),
            self._get_budget_status(request_data["department"]),
            self._get_vendor_status(request_data.get("vendor_id")),
            self._check_business_calendar(),
        ]

        historical_approvals, budget_status, vendor_status, calendar_info = \
            await asyncio.gather(*context_tasks)

        return DecisionContext(
            request_id=request_data["id"],
            requester_id=request_data["requester_id"],
            amount=request_data["amount"],
            currency=request_data.get("currency", "GBP"),
            department=request_data["department"],
            vendor_id=request_data.get("vendor_id"),
            urgency_level=request_data.get("urgency", "normal"),
            business_justification=request_data["justification"],
            historical_approvals=historical_approvals,
            current_budget_status=budget_status,
            vendor_status=vendor_status,
            request_timestamp=datetime.now()
        )

    async def _get_historical_approvals(self, requester_id: str) -> List[Dict[str, Any]]:
        """Fetch recent approval history for pattern analysis."""
        query = """
        SELECT amount, vendor_id, approval_status, processing_time
        FROM approvals
        WHERE requester_id = %s
        AND created_at > NOW() - INTERVAL '90 days'
        ORDER BY created_at DESC LIMIT 10
        """
        return await self.db.fetch_all(query, [requester_id])

    async def _get_budget_status(self, department: str) -> Dict[str, float]:
        """Get current budget utilisation for department."""
        query = """
        SELECT budget_allocated, budget_spent, budget_committed
        FROM department_budgets
        WHERE department = %s AND fiscal_year = %s
        """
        result = await self.db.fetch_one(query, [department, datetime.now().year])
        return result or {"budget_allocated": 0, "budget_spent": 0, "budget_committed": 0}

    async def _get_vendor_status(self, vendor_id: Optional[str]) -> Optional[str]:
        """Check vendor approval status and risk rating."""
        if not vendor_id:
            return None

        query = """
        SELECT status, risk_rating, contract_expiry
        FROM vendors
        WHERE vendor_id = %s
        """
        result = await self.db.fetch_one(query, [vendor_id])
        return result["status"] if result else "unknown"

from langchain.agents import Tool, AgentExecutor, create_react_agent
from langchain.prompts import PromptTemplate
from langchain.schema import BaseOutputParser
import re

class DecisionOutputParser(BaseOutputParser):
    """Parse AI agent decision output into structured format."""

    def parse(self, text: str) -> Dict[str, Any]:
        decision_pattern = r"DECISION:\s*(APPROVED|REJECTED|ESCALATE)"
        reasoning_pattern = r"REASONING:\s*(.*?)(?=\n[A-Z]+:|$)"
        next_steps_pattern = r"NEXT_STEPS:\s*(.*?)(?=\n[A-Z]+:|$)"

        decision_match = re.search(decision_pattern, text, re.DOTALL)
        reasoning_match = re.search(reasoning_pattern, text, re.DOTALL)
        next_steps_match = re.search(next_steps_pattern, text, re.DOTALL)

        return {
            "decision": decision_match.group(1) if decision_match else "ESCALATE",
            "reasoning": reasoning_match.group(1).strip() if reasoning_match else "",
            "next_steps": next_steps_match.group(1).strip() if next_steps_match else "",
            "confidence": self._extract_confidence(text)
        }

    def _extract_confidence(self, text: str) -> float:
        confidence_pattern = r"CONFIDENCE:\s*(\d+(?:\.\d+)?)"
        match = re.search(confidence_pattern, text)
        return float(match.group(1)) / 100 if match else 0.5

class AIDecisionEngine:
    def __init__(self, policy_parser: PolicyParser):
        self.policy_parser = policy_parser
        self.llm = OpenAI(model_name="gpt-4", temperature=0.1)
        self.output_parser = DecisionOutputParser()

        # Define tools for the agent
        self.tools = [
            Tool(
                name="check_budget_limit",
                description="Check if request amount exceeds department budget limits",
                func=self._check_budget_limits
            ),
            Tool(
                name="verify_vendor_status",
                description="Verify vendor approval status and compliance",
                func=self._verify_vendor_status
            ),
            Tool(
                name="check_approval_history",
                description="Check requester's historical approval patterns",
                func=self._check_approval_history
            )
        ]

        self.agent_prompt = PromptTemplate(
            input_variables=["policies", "context", "tools", "tool_names", "agent_scratchpad"],
            template="""
            You are an AI agent responsible for making purchase approval decisions based on company policies.

            COMPANY POLICIES:
            {policies}

            REQUEST CONTEXT:
            {context}

            AVAILABLE TOOLS:
            {tools}

            Tool Names: {tool_names}

            DECISION FRAMEWORK:
            1. Analyse the request against all applicable policies
            2. Use available tools to gather additional information if needed
            3. Consider exceptions and edge cases mentioned in policies
            4. Make a decision: APPROVED, REJECTED, or ESCALATE
            5. Provide clear reasoning for your decision
            6. Suggest next steps

            Your response must include:
            DECISION: [APPROVED/REJECTED/ESCALATE]
            REASONING: [Detailed explanation of decision logic]
            NEXT_STEPS: [What should happen next]
            CONFIDENCE: [0-100 representing confidence in decision]

            {agent_scratchpad}
            """
        )

        self.agent = create_react_agent(
            llm=self.llm,
            tools=self.tools,
            prompt=self.agent_prompt
        )

        self.agent_executor = AgentExecutor(
            agent=self.agent,
            tools=self.tools,
            max_iterations=5,
            verbose=True
        )

    async def make_decision(self, context: DecisionContext,
                          policies: PolicyDocument) -> Dict[str, Any]:
        """Make an approval decision based on context and policies."""

        policies_text = self._format_policies(policies)
        context_text = self._format_context(context)

        try:
            result = await self.agent_executor.arun(
                policies=policies_text,
                context=context_text
            )

            parsed_decision = self.output_parser.parse(result)

            # Add metadata
            parsed_decision.update({
                "request_id": context.request_id,
                "timestamp": datetime.now().isoformat(),
                "policy_version": policies.version,
                "processing_time_ms": 0  # Would be calculated in production
            })

            return parsed_decision

        except Exception as e:
            # Fallback to escalation on errors
            return {
                "decision": "ESCALATE",
                "reasoning": f"Error during processing: {str(e)}",
                "next_steps": "Manual review required",
                "confidence": 0.0,
                "request_id": context.request_id,
                "timestamp": datetime.now().isoformat(),
                "error": True
            }

    def _format_policies(self, policies: PolicyDocument) -> str:
        """Format policy rules for LLM consumption."""
        formatted_rules = []
        for rule in policies.rules:
            rule_text = f"RULE (Priority {rule.priority}): {rule.condition} → {rule.action}"
            if rule.exceptions:
                rule_text += f"\nExceptions: {', '.join(rule.exceptions)}"
            formatted_rules.append(rule_text)

        return "\n\n".join(formatted_rules)

    def _format_context(self, context: DecisionContext) -> str:
        """Format request context for LLM consumption."""
        return f"""
        Request ID: {context.request_id}
        Requester: {context.requester_id}
        Amount: {context.currency} {context.amount:,.2f}
        Department: {context.department}
        Vendor: {context.vendor_id or 'New/Unknown'}
        Urgency: {context.urgency_level}
        Justification: {context.business_justification}
        Budget Status: Allocated: {context.current_budget_status.get('budget_allocated', 0):,.2f},
                     Spent: {context.current_budget_status.get('budget_spent', 0):,.2f}
        Vendor Status: {context.vendor_status or 'Unknown'}
        Historical Approvals: {len(context.historical_approvals)} recent requests
        """

    def _check_budget_limits(self, amount_and_department: str) -> str:
        """Tool function to check budget limits."""
        # In production, this would query actual budget data
        return "Budget check: Within limits"

    def _verify_vendor_status(self, vendor_id: str) -> str:
        """Tool function to verify vendor status."""
        # In production, this would query vendor database
        return "Vendor status: Approved"

    def _check_approval_history(self, requester_id: str) -> str:
        """Tool function to check approval history."""
        # In production, this would analyse historical patterns
        return "Approval history: Good standing"

from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from typing import Optional
import asyncio
import logging

app = FastAPI(title="AI Approval Controller", version="1.0.0")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Configure appropriately for production
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

security = HTTPBearer()

class ApprovalRequest(BaseModel):
    amount: float
    currency: str = "GBP"
    department: str
    vendor_id: Optional[str] = None
    urgency_level: str = "normal"
    business_justification: str
    requester_id: str

class ApprovalResponse(BaseModel):
    request_id: str
    decision: str
    reasoning: str
    next_steps: str
    confidence: float
    processing_time_ms: int

# Global instances (would be properly configured in production)
policy_parser = PolicyParser()
context_builder = ContextBuilder(db_connection=None, external_apis=None)
decision_engine = AIDecisionEngine(policy_parser)

@app.post("/v1/approvals", response_model=ApprovalResponse)
async def process_approval_request(
    request: ApprovalRequest,
    background_tasks: BackgroundTasks,
    credentials: HTTPAuthorizationCredentials = Depends(security)
):
    """Process a purchase approval request through AI decision engine."""

    start_time = time.time()
    request_id = str(uuid.uuid4())

    try:
        # Authenticate request (simplified for example)
        if not await authenticate_token(credentials.credentials):
            raise HTTPException(status_code=401, detail="Invalid authentication")

        # Build decision context
        request_data = request.dict()
        request_data["id"] = request_id
        context = await context_builder.build_context(request_data)

        # Load applicable policies
        policies = await load_policies_for_department(request.department)

        # Make decision
        decision_result = await decision_engine.make_decision(context, policies)

        # Log decision for audit
        background_tasks.add_task(log_decision, request_id, decision_result, context)

        # Send notifications
        background_tasks.add_task(send_notifications, decision_result, request)

        processing_time_ms = int((time.time() - start_time) * 1000)

        return ApprovalResponse(
            request_id=request_id,
            decision=decision_result["decision"],
            reasoning=decision_result["reasoning"],
            next_steps=decision_result["next_steps"],
            confidence=decision_result["confidence"],
            processing_time_ms=processing_time_ms
        )

    except Exception as e:
        logging.error(f"Error processing approval request {request_id}: {str(e)}")
        raise HTTPException(status_code=500, detail="Internal processing error")

@app.get("/v1/policies/{department}")
async def get_department_policies(department: str):
    """Retrieve current policies for a department."""
    policies = await load_policies_for_department(department)
    return {"policies": policies.dict()}

@app.put("/v1/policies/{department}")
async def update_department_policies(
    department: str,
    policy_text: str,
    credentials: HTTPAuthorizationCredentials = Depends(security)
):
    """Update department policies (admin only)."""
    if not await authenticate_admin_token(credentials.credentials):
        raise HTTPException(status_code=403, detail="Admin access required")

    # Parse and validate new policy
    try:
        parsed_policy = policy_parser.parse_policy(policy_text)
        await save_policy(department, parsed_policy)
        return {"status": "success", "message": "Policy updated successfully"}
    except PolicyParsingError as e:
        raise HTTPException(status_code=400, detail=f"Policy parsing error: {str(e)}")

async def authenticate_token(token: str) -> bool:
    """Authenticate API token (simplified for example)."""
    return True  # Implement proper authentication

async def authenticate_admin_token(token: str) -> bool:
    """Authenticate admin API token."""
    return True  # Implement proper admin authentication

async def load_policies_for_department(department: str) -> PolicyDocument:
    """Load current policies for department."""
    # Implementation would load from database
    return PolicyDocument(
        title=f"{department} Policies",
        version="1.0",
        effective_date="2024-01-01",
        rules=[],
        escalation_criteria=[]
    )

async def save_policy(department: str, policy: PolicyDocument):
    """Save policy to database with versioning."""
    pass  # Implementation would save to database

async def log_decision(request_id: str, decision: Dict[str, Any], context: DecisionContext):
    """Log decision for audit trail."""
    pass  # Implementation would log to audit system

async def send_notifications(decision: Dict[str, Any], request: ApprovalRequest):
    """Send appropriate notifications based on decision."""
    pass  # Implementation would send emails/Slack notifications

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

import pytest
from unittest.mock import AsyncMock, MagicMock
from typing import List, Dict, Any

class PolicyTestCase(BaseModel):
    name: str
    policy_text: str
    test_scenarios: List[Dict[str, Any]]
    expected_outcomes: List[str]

class AIDecisionEngineTest:
    @pytest.fixture
    def decision_engine(self):
        mock_parser = MagicMock(spec=PolicyParser)
        return AIDecisionEngine(mock_parser)

    @pytest.fixture
    def sample_context(self):
        return DecisionContext(
            request_id="test-123",
            requester_id="user-456",
            amount=1500.0,
            currency="GBP",
            department="Engineering",
            vendor_id="vendor-789",
            urgency_level="normal",
            business_justification="Software licence renewal",
            historical_approvals=[],
            current_budget_status={"budget_allocated": 10000, "budget_spent": 5000},
            vendor_status="approved",
            request_timestamp=datetime.now()
        )

    @pytest.mark.asyncio
    async def test_standard_approval_flow(self, decision_engine, sample_context):
        """Test standard approval flow with valid request."""

        mock_policies = PolicyDocument(
            title="Engineering Policies",
            version="1.0",
            effective_date="2024-01-01",
            rules=[
                PolicyRule(
                    condition="Amount under £2000 and approved vendor",
                    action="Auto-approve",
                    priority=1,
                    exceptions=[]
                )
            ],
            escalation_criteria=[]
        )

        decision = await decision_engine.make_decision(sample_context, mock_policies)

        assert decision["decision"] == "APPROVED"
        assert decision["confidence"] > 0.7
        assert "test-123" in decision["request_id"]

    @pytest.mark.asyncio
    async def test_policy_exception_handling(self, decision_engine, sample_context):
        """Test handling of policy exceptions."""

        # Modify context to trigger exception
        sample_context.vendor_status = "pending_review"

        mock_policies = PolicyDocument(
            title="Engineering Policies",
            version="1.0",
            effective_date="2024-01-01",
            rules=[
                PolicyRule(
                    condition="Amount under £2000",
                    action="Auto-approve",
                    priority=1,
                    exceptions=["Pending vendor review requires manual approval"]
                )
            ],
            escalation_criteria=[]
        )

        decision = await decision_engine.make_decision(sample_context, mock_policies)

        assert decision["decision"] in ["ESCALATE", "REJECTED"]
        assert "vendor review" in decision["reasoning"].lower()

# Integration test for full API flow
@pytest.mark.asyncio
async def test_full_approval_api_flow():
    """Test complete approval flow through API."""
    from httpx import AsyncClient

    async with AsyncClient(app=app, base_url="http://test") as ac:
        approval_request = {
            "amount": 1500.0,
            "currency": "GBP",
            "department": "Engineering",
            "vendor_id": "test-vendor",
            "urgency_level": "normal",
            "business_justification": "Software licence renewal",
            "requester_id": "test-user"
        }

        response = await ac.post(
            "/v1/approvals",
            json=approval_request,
            headers={"Authorization": "Bearer test-token"}
        )

        assert response.status_code == 200
        result = response.json()
        assert "decision" in result
        assert "reasoning" in result
        assert "request_id" in result