syedkhizarrayaz/BM-AI-Analysis-And-Alert-Prioritization-Agent

1# imports2from sqlalchemy import create_engine3import pandas as pd4from openai import OpenAI5from llmeventhandler import EventHandler6from fastapi import APIRouter, HTTPException, Request7from config import OPENAI_API_KEY, DB_CONNECTION_STR, DATA_TABLE, PREDICTION_FLAG, ANALYSIS_TABLE, ASSISTANT_ID, PII_COLUMNS8from pydantic import BaseModel, ValidationError, Field9from typing import Optional, List, Dict, Any, Union10from pydantic import BaseModel, Field as PydanticField11import logging12from localLLM import LocalLLM13from fastapi.responses import StreamingResponse14import json15import asyncio16from functools import lru_cache17import threading18from hybrid_ultra_fast_aml import get_hybrid_system19from hybrid_template_llm_aml import get_hybrid_system as get_template_llm_system20 21 22def mask_pii_value(value: str) -> str:23    """24    Mask a PII value by keeping first 2 and last 2 characters, replacing middle with ***25    Handles all string lengths appropriately.26    27    Args:28        value: The value to mask29        30    Returns:31        Masked value with format: first 2 chars + *** + last 2 chars32        For very short values, appropriate masking is applied33    """34    if not value or not isinstance(value, str):35        return value36    37    value = str(value).strip()38    length = len(value)39    40    # Handle edge cases for very short values41    if length == 0:42        return value43    elif length == 1:44        return "*"45    elif length == 2:46        return "**"47    elif length == 3:48        return value[0] + "**"49    elif length == 4:50        return value[0] + "**" + value[-1]51    elif length == 5:52        return value[0] + "***" + value[-1]53    54    # For values with length >= 6, keep first 2 and last 2, mask the middle55    return value[:2] + "***" + value[-2:]56 57 58def mask_pii_data(data: Union[Dict[str, Any], List[Dict[str, Any]]], pii_columns: List[str]) -> Union[Dict[str, Any], List[Dict[str, Any]]]:59    """60    Mask PII data in a dictionary or list of dictionaries based on provided column names61    62    Args:63        data: Dictionary or list of dictionaries containing the data64        pii_columns: List of column names to mask65        66    Returns:67        Data with masked PII values68    """69    if not pii_columns:70        return data71    72    # Normalize column names (strip whitespace)73    pii_columns = [col.strip() for col in pii_columns if col.strip()]74    75    if not pii_columns:76        return data77    78    def mask_dict(d: Dict[str, Any]) -> Dict[str, Any]:79        """Mask PII in a single dictionary"""80        masked = d.copy()81        82        # Mask direct fields83        for col in pii_columns:84            if col in masked and masked[col] is not None:85                masked[col] = mask_pii_value(str(masked[col]))86        87        # Handle FilteredTransactions (JSON string)88        if 'FilteredTransactions' in masked and masked['FilteredTransactions']:89            try:90                transactions = json.loads(masked['FilteredTransactions']) if isinstance(masked['FilteredTransactions'], str) else masked['FilteredTransactions']91                if isinstance(transactions, list):92                    masked_transactions = []93                    for trans in transactions:94                        masked_trans = trans.copy() if isinstance(trans, dict) else trans95                        if isinstance(trans, dict):96                            for col in pii_columns:97                                if col in masked_trans and masked_trans[col] is not None:98                                    masked_trans[col] = mask_pii_value(str(masked_trans[col]))99                        masked_transactions.append(masked_trans)100                    masked['FilteredTransactions'] = json.dumps(masked_transactions) if isinstance(masked['FilteredTransactions'], str) else masked_transactions101            except (json.JSONDecodeError, TypeError):102                # If parsing fails, leave as is103                pass104        105        # Handle PreviousAlerts (list of dictionaries)106        if 'PreviousAlerts' in masked and isinstance(masked['PreviousAlerts'], list):107            masked_alerts = []108            for alert in masked['PreviousAlerts']:109                if isinstance(alert, dict):110                    masked_alert = alert.copy()111                    for col in pii_columns:112                        if col in masked_alert and masked_alert[col] is not None:113                            masked_alert[col] = mask_pii_value(str(masked_alert[col]))114                    masked_alerts.append(masked_alert)115                else:116                    masked_alerts.append(alert)117            masked['PreviousAlerts'] = masked_alerts118        119        # Handle Counterparties (list of dictionaries)120        if 'Counterparties' in masked and isinstance(masked['Counterparties'], list):121            masked_counterparties = []122            for cp in masked['Counterparties']:123                if isinstance(cp, dict):124                    masked_cp = cp.copy()125                    for col in pii_columns:126                        if col in masked_cp and masked_cp[col] is not None:127                            masked_cp[col] = mask_pii_value(str(masked_cp[col]))128                    masked_counterparties.append(masked_cp)129                else:130                    masked_counterparties.append(cp)131            masked['Counterparties'] = masked_counterparties132        133        # Handle BranchQueries (dictionary)134        if 'BranchQueries' in masked and isinstance(masked['BranchQueries'], dict):135            masked_queries = masked['BranchQueries'].copy()136            # Mask PII in Requested and Response text fields137            for key in ['Requested', 'Response']:138                if key in masked_queries and masked_queries[key]:139                    # Simple text masking - could be enhanced to detect PII patterns140                    text = str(masked_queries[key])141                    # For now, we'll mask known PII patterns in text142                    # This is a simple approach - could be enhanced with regex143                    for col in pii_columns:144                        # Try to find and mask the column value in text145                        # This is a basic implementation146                        pass147            masked['BranchQueries'] = masked_queries148        149        return masked150    151    # Handle list of dictionaries152    if isinstance(data, list):153        return [mask_dict(item) if isinstance(item, dict) else item for item in data]154    155    # Handle single dictionary156    if isinstance(data, dict):157        return mask_dict(data)158    159    return data160 161 162# analysis alert request model - all fields optional to handle systems with missing data163class AnalysisAlertRequest(BaseModel):164    AlertID: Optional[int] = Field(None, description="Unique alert identifier", example=1)165    FocusColumnValue: Optional[str] = Field(None, description="Customer/entity identifier", example="100001")166    KYCMonthlyIncome: Optional[str] = Field(None, description="Monthly income range", example="85,000 PKR")167    KYCNoOfCredits: Optional[str] = Field(None, description="Expected number of credits", example="3-5")168    KYCNoOfDebits: Optional[str] = Field(None, description="Expected number of debits", example="8-12")169    KYCRiskCategoryValue: Optional[str] = Field(None, description="KYC risk category", example="Low")170    KYCValueOfCredits: Optional[str] = Field(None, description="Expected value of credits", example="150,000 - 200,000 PKR")171    KYCValueOfDebits: Optional[str] = Field(None, description="Expected value of debits", example="80,000 - 120,000 PKR")172    OccupationValue: Optional[str] = Field(None, description="Customer occupation", example="Private Employee")173    FilteredTransactions: Optional[str] = Field(None, description="JSON array string with transaction details", example='[{"CUSTOMERID":"100001","TRANSACTIONAMOUNT":150000.0,"CURRENCY":"PKR"}]')174    ScenarioName: Optional[str] = Field(None, description="Alert scenario name", example="Unusually large installment")175    STRCount: Optional[int] = Field(0, description="Count of previous STRs", example=0)176    STRScenarioHistory: Optional[str] = Field("", description="History of STR scenarios", example="")177    # Additional optional fields from the JSON payload178    CustomerName: Optional[str] = Field(None, description="Customer full name", example="Muhammad Bilal Sheikh")179    CUSTOMERID: Optional[str] = Field(None, description="Customer ID", example="100001")180    BranchID: Optional[str] = Field(None, description="Branch identifier", example="KHI-DHA")181    Country: Optional[str] = Field(None, description="Customer country", example="Pakistan")182    CustomerType: Optional[str] = Field(None, description="Customer type", example="Retail")183    CustomerStatus: Optional[str] = Field(None, description="Customer status", example="Retail Customer")184    CreatedDate: Optional[str] = Field(None, description="Account creation date", example="2020-01-15")185    RelationshipStartDate: Optional[str] = Field(None, description="Relationship start date", example="2020-01-15")186    RiskScore: Optional[str] = Field(None, description="Risk score", example="4.2")187    PreviousAlerts: Optional[List[Dict[str, Any]]] = Field([], description="Array of previous alert objects", example=[])188    Counterparties: Optional[List[Dict[str, Any]]] = Field([], description="Array of counterparty objects", example=[])189    BranchQueries: Optional[Dict[str, Any]] = Field(None, description="Branch query request/response", example={"Requested": "Verification required...", "Response": "Customer stated..."})190    191    class Config:192        json_schema_extra = {193            "example": {194                "AlertID": 1,195                "FilteredTransactions": '[{"CUSTOMERID":"100001","IDENTITYNUMBERS":"42101-1234567-1","LOANID":"LN2024001","ACCOUNTID":"AC100001","CREATEDDATE":"2025-06-01 10:15:00","TRANSACTIONAMOUNT":150000.0,"CURRENCY":"PKR","INSTALLMENTNUMBER":1,"EXCESSAMOUNT":0.0}]',196                "FocusColumnValue": "100001",197                "KYCMonthlyIncome": "85,000 PKR",198                "KYCNoOfCredits": "3-5",199                "KYCNoOfDebits": "8-12",200                "KYCRiskCategoryValue": "Low",201                "KYCValueOfCredits": "150,000 - 200,000 PKR",202                "KYCValueOfDebits": "80,000 - 120,000 PKR",203                "OccupationValue": "Private Employee",204                "STRCount": 0,205                "STRScenarioHistory": "",206                "ScenarioName": "Unusually large installment",207                "CustomerName": "Muhammad Bilal Sheikh",208                "CUSTOMERID": "100001",209                "BranchID": "KHI-DHA",210                "Country": "Pakistan",211                "CustomerType": "Retail",212                "CustomerStatus": "Retail Customer",213                "CreatedDate": "2020-01-15",214                "RelationshipStartDate": "2020-01-15",215                "RiskScore": "4.2",216                "PreviousAlerts": [],217                "Counterparties": [],218                "BranchQueries": {219                    "Requested": "Verification required for loan installments totaling 530,000 PKR within one month, significantly exceeding declared monthly income of 85,000 PKR. Please confirm source of funds and provide documentation for additional income sources.",220                    "Response": "Customer stated installments are from remittances received from brother working in UAE, family savings from wedding expenses, and advance salary from employer for Eid holidays. Customer provided remittance receipts and employer letter."221                }222            }223        }224 225# Response models226class AnalysisItem(BaseModel):227    AlertID: int = Field(..., description="Alert identifier", example=2)228    FocusColumnValue: str = Field(..., description="Customer/entity identifier", example="100002")229    analysis: str = Field(..., description="Generated AML analysis report", example="AML Investigation Report...")230    response_time_ms: float = Field(..., description="Response time in milliseconds", example=1234.56)231    method: str = Field(..., description="Analysis method used", example="hybrid_template_only")232    model: Optional[str] = Field(None, description="LLM model used", example="granite3.1-moe:3b")233    thinking: Optional[str] = Field(None, description="Thinking/reasoning from audit mode (when audit=true)")234 235class AnalysisResponse(BaseModel):236    status: int = Field(..., description="HTTP status code", example=200)237    message: str = Field(..., description="Response message", example="Success")238    data: List[AnalysisItem] = Field(..., description="List of analysis results")239    240    class Config:241        json_schema_extra = {242            "example": {243                "status": 200,244                "message": "Success",245                "data": [246                    {247                        "AlertID": 2,248                        "FocusColumnValue": "100002",249                        "analysis": "AML Investigation Report - TMS Case: Structuring / Smurfing activity\n\nCustomer Name: Ayesha Malik\n...",250                        "response_time_ms": 1234.56,251                        "method": "hybrid_template_only",252                        "model": "granite3.1-moe:3b"253                    }254                ]255            }256        }257 258# api route instance259router = APIRouter()260 261# **Setup logging** (add this section to configure the logging)262logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[263    logging.FileHandler("analysisapi.log"),264    logging.StreamHandler()265])266 267logger = logging.getLogger(__name__)268 269# Global LLM instance for performance optimization270_global_llm = None271_llm_lock = threading.Lock()272 273def get_global_llm():274    """Get or create global LLM instance with thread safety"""275    global _global_llm276    if _global_llm is None:277        with _llm_lock:278            if _global_llm is None:279                _global_llm = LocalLLM()280    return _global_llm281 282# Template caching for faster response generation283@lru_cache(maxsize=100)284def get_analysis_template(scenario_name: Optional[str], kyc_info: str, transaction_data: Optional[str], 285                          str_count: Optional[int], str_history: Optional[str]) -> str:286    """Cached template generation for analysis prompts"""287    # Handle None or empty values288    scenario_name = scenario_name or "Unknown Scenario"289    transaction_data = transaction_data or "No transaction data available"290    str_count = str_count if str_count is not None else 0291    str_history = str_history or "No previous STR scenarios"292    kyc_info = kyc_info or "KYC profile not available"293    294    return (295        f"You are an AML analyst. An alert has been raised regarding potential suspicious activity involving the scenario: {scenario_name}.\n\n"296        "Conclusion Up Front\n"297        "- Provide a clear conclusion on whether the case can be closed or warrants escalation/STR.\n"298        "- Include a brief rationale highlighting red flags or justifications for closure.\n\n"299 300        "Initial Risk\n"301        f"- Alerted Scenario / Rule ID: {scenario_name}\n\n"302 303        "Reviewed Period\n"304        "- Define the transaction review period based on the data provided.\n\n"305 306        "Transaction Review\n"307        f"- Transaction Data:\n{transaction_data}\n"308        "- Summarize the number of transactions, total values in USD, dates, involved parties, locations, and any remarks indicating suspicious behavior.\n\n"309 310        "Research on Focus\n"311        f"- KYC Profile:\n{kyc_info}\n"312        "- Include Name, CIF, ID, DOB/Age, Country, Employment, Job Title, Monthly Income, Risk Level, Relationship Start Date, and any name screening results.\n\n"313 314        "Research on Parties / Counterparties Involved\n"315        "- Assess the profile of counterparties involved in the transactions.\n"316        "- Include any name screening or risk database results.\n\n"317 318        "History Related Cases\n"319        f"- Alerts raised Previously: {str_count}\n"320        f"- STR raised Scenarios: {str_history}\n"321        "- Summarize only if there are relevant concerns or patterns found.\n\n"322 323        "RFI / Enquiry Branch Observation\n"324        "- Summarize any branch or RM input, if available.\n\n"325 326        "Risk Search\n"327        "- Detail outcomes of internal or external screening and risk searches.\n"328        "- Mention any matches with internal watchlists, fraud portals, or regulatory flags.\n\n"329 330        "At the end no need for suggestion or conclusion, just return headings I have mentioned above. All the headings and their content must be present in the response.\n"331    )332 333# analysis class334class AnalysisProcessor:335    # constructor336    def __init__(self, db_connection_str, client, assistant_id):337        self.db_connection_str = db_connection_str338        # Only create database engine if connection string is provided and not empty339        if db_connection_str:340            try:341                self.engine = create_engine(self.db_connection_str)342                logger.info("Database engine initialized successfully")343            except Exception as e:344                logger.warning(f"Database connection not available: {e}. Running without database.")345                self.engine = None346        else:347            self.engine = None348            logger.info("Database connection string not provided. Running without database.")349        350        # Create a new thread for the conversation only if client is provided351        self.client = client352        if client is not None:353            try:354                self.thread = client.beta.threads.create()355            except Exception as e:356                logger.warning(f"Failed to create OpenAI thread: {e}. Running without OpenAI thread.")357                self.thread = None358        else:359            self.thread = None360        361        self.assistant_id = assistant_id or ""362        logger.info(f"Initialized AnalysisProcessor (assistant_id: {self.assistant_id}, db_available: {self.engine is not None}, client_available: {self.client is not None})")363 364 365    # query database for retrival of data366    def query_data(self, table_name, prediction_value):367        """368        Queries the database for records with Prediction = 1 and returns it as a DataFrame.369        """370        try:371            # Skip database operations if engine is not available372            if self.engine is None:373                logger.info(f"Database not configured. Skipping query from table {table_name}")374                return pd.DataFrame()  # Return empty DataFrame375            376            logger.info(f"Querying data from table: {table_name} with Prediction = {prediction_value}")377 378            query = f"SELECT * FROM {table_name} WHERE Prediction = {prediction_value}"  # Adjust table name as needed379            df = pd.read_sql(query, self.engine)380            logger.info(f"Retrieved {len(df)} rows from the database")381            return df382        except Exception as e:383            logger.error(f"Error querying data: {e}")384            # Return empty DataFrame instead of raising exception385            logger.warning(f"Returning empty DataFrame due to database error")386            return pd.DataFrame()387    388    # return kyc data in string389    def return_kyc_profile(self, customer_data):390        try:391            logger.debug("Generating KYC profile string")392 393            # Safely get values with defaults for missing fields394            kyc_monthly = customer_data.get('KYCMonthlyIncome', 'Not available')395            kyc_numcredits = customer_data.get('KYCNoOfCredits', 'Not available')396            kyc_numdebits = customer_data.get('KYCNoOfDebits', 'Not available')397            kyc_riskcategory = customer_data.get('KYCRiskCategoryValue', 'Not available')398            kyc_valuecredits = customer_data.get('KYCValueOfCredits', 'Not available')399            kyc_valuedebits = customer_data.get('KYCValueOfDebits', 'Not available')400            kyc_occupation = customer_data.get('OccupationValue', 'Not available')401            402            # Get additional customer info if available403            customer_name = customer_data.get('CustomerName', customer_data.get('CUSTOMERID', 'Not available'))404            customer_id = customer_data.get('CUSTOMERID', 'Not available')405            country = customer_data.get('Country', 'Not available')406            relationship_start = customer_data.get('RelationshipStartDate', 'Not available')407            risk_score = customer_data.get('RiskScore', 'Not available')408 409            return f'''410        Customer Name: - {customer_name}411        Customer ID: - {customer_id}412        Country: - {country}413        Relationship Start Date: - {relationship_start}414        Monthly Income Range: - {kyc_monthly}415        Occupation:- {kyc_occupation}416        Risk Category: - {kyc_riskcategory}417        Risk Score: - {risk_score}418        Expected Number of Credits: - {kyc_numcredits}419        Expected Total Value of Credits: -  {kyc_valuecredits}420        Expected Number of Debits: - {kyc_numdebits}421        Expected Total Value of Debits: - {kyc_valuedebits}422        '''423        except Exception as e:424                logger.error(f"Error generating KYC profile: {e}")425                # Return a minimal profile if there's an error426                return "KYC Profile: Data not available"427    428    # Send a comprehensive message to the assistant429    def send_message_to_assistant(self, transaction_data, scenario, prev_alerts, kyc_profile, str_scenarios_history):430        # Construct the full query for the assistant431        try:432            logger.info(f"Sending message to assistant for transaction: {transaction_data}")433 434            # Handle None or empty values435            scenario = scenario or "Unknown Scenario"436            transaction_data = transaction_data or "No transaction data available"437            prev_alerts = prev_alerts if prev_alerts is not None else 0438            str_scenarios_history = str_scenarios_history or "No previous STR scenarios"439            kyc_profile = kyc_profile or "KYC profile not available"440 441            full_query = (442                f"You are an AML analyst. An alert has been raised regarding potential suspicious activity involving the scenario: {scenario}.\n\n"443                "Conclusion Up Front\n"444                "- Provide a clear conclusion on whether the case can be closed or warrants escalation/STR.\n"445                "- Include a brief rationale highlighting red flags or justifications for closure.\n\n"446 447                "Initial Risk\n"448                f"- Alerted Scenario / Rule ID: {scenario}\n\n"449 450                "Reviewed Period\n"451                "- Define the transaction review period based on the data provided.\n\n"452 453                "Transaction Review\n"454                f"- Transaction Data:\n{transaction_data}\n"455                "- Summarize the number of transactions, total values in USD, dates, involved parties, locations, and any remarks indicating suspicious behavior.\n\n"456 457                "Research on Focus\n"458                f"- KYC Profile:\n{kyc_profile}\n"459                "- Include Name, CIF, ID, DOB/Age, Country, Employment, Job Title, Monthly Income, Risk Level, Relationship Start Date, and any name screening results.\n\n"460 461                "Research on Parties / Counterparties Involved\n"462                "- Assess the profile of counterparties involved in the transactions.\n"463                "- Include any name screening or risk database results.\n\n"464 465                "History Related Cases\n"466                f"- Alerts raised Previously: {prev_alerts}\n"467                f"- STR raised Scenarios: {str_scenarios_history}\n"468                "- Summarize only if there are relevant concerns or patterns found.\n\n"469 470                "RFI / Enquiry Branch Observation\n"471                "- Summarize any branch or RM input, if available.\n\n"472 473                "Risk Search\n"474                "- Detail outcomes of internal or external screening and risk searches.\n"475                "- Mention any matches with internal watchlists, fraud portals, or regulatory flags.\n\n"476 477                "At the end no need for suggestion or conclusion, just return headings I have mentioned above. All the headings and their content must be present in the response.\n"478 479            )480 481            # Send the message482            message = self.client.beta.threads.messages.create(483                thread_id=self.thread.id,484                role="user",485                content=full_query486            )487            logger.info("Message sent to assistant successfully")488 489            # Initialize a variable to store the final JSON response490            response_data = {}491            # Instantiate the event handler492            event_handler = EventHandler()493            # Stream the response494            with self.client.beta.threads.runs.stream(495                thread_id=self.thread.id,496                assistant_id=self.assistant_id,  # Use your existing assistant ID497                instructions="You are an AML Analyst. Be assertive in explanation. Don't use 'could', 'can'. Suggest actions instead. Your input prompt will be transaction details in XML. Analyze the details and explain the transaction in context of the document and categorize it in Scenario strictly within document mentioned scenarios only, but don't mention it is coming from the document explicitly in response. Just mention which Scenario the transaction lies in and its explanation with quoted transaction in the same paragraph where explanation is.",498                event_handler=event_handler,499            ) as stream:500                stream.until_done()  # Complete streaming501 502            response_data = event_handler.get_final_response()503            logger.info(f"Response received from assistant: {response_data}")504 505            return response_data506 507        except Exception as e:508            logger.error(f"Error sending message to assistant: {e}")509            raise510 511    def send_message_direct(self, transaction_data, scenario, prev_alerts, kyc_profile, str_scenarios_history):512        """513        Send transaction details directly to OpenAI's latest model without threads or agents.514        """515        try:516            logger.info(f"Sending direct request to OpenAI for transaction: {transaction_data}")517 518            # Handle None or empty values519            scenario = scenario or "Unknown Scenario"520            transaction_data = transaction_data or "No transaction data available"521            prev_alerts = prev_alerts if prev_alerts is not None else 0522            str_scenarios_history = str_scenarios_history or "No previous STR scenarios"523            kyc_profile = kyc_profile or "KYC profile not available"524 525            # Construct the query (same format you had for the threaded version)526            full_query = (527                f"You are an AML analyst. An alert has been raised regarding potential suspicious activity involving the scenario: {scenario}.\n\n"528                "Conclusion Up Front\n"529                "- Provide a clear conclusion on whether the case can be closed or warrants escalation/STR.\n"530                "- Include a brief rationale highlighting red flags or justifications for closure.\n\n"531 532                "Initial Risk\n"533                f"- Alerted Scenario / Rule ID: {scenario}\n\n"534 535                "Reviewed Period\n"536                "- Define the transaction review period based on the data provided.\n\n"537 538                "Transaction Review\n"539                f"- Transaction Data:\n{transaction_data}\n"540                "- Summarize the number of transactions, total values in USD, dates, involved parties, locations, and any remarks indicating suspicious behavior.\n\n"541 542                "Research on Focus\n"543                f"- KYC Profile:\n{kyc_profile}\n"544                "- Include Name, CIF, ID, DOB/Age, Country, Employment, Job Title, Monthly Income, Risk Level, Relationship Start Date, and any name screening results.\n\n"545 546                "Research on Parties / Counterparties Involved\n"547                "- Assess the profile of counterparties involved in the transactions.\n"548                "- Include any name screening or risk database results.\n\n"549 550                "History Related Cases\n"551                f"- Alerts raised Previously: {prev_alerts}\n"552                f"- STR raised Scenarios: {str_scenarios_history}\n"553                "- Summarize only if there are relevant concerns or patterns found.\n\n"554 555                "RFI / Enquiry Branch Observation\n"556                "- Summarize any branch or RM input, if available.\n\n"557 558                "Risk Search\n"559                "- Detail outcomes of internal or external screening and risk searches.\n"560                "- Mention any matches with internal watchlists, fraud portals, or regulatory flags.\n\n"561 562                "At the end no need for suggestion or conclusion, just return headings I have mentioned above. All the headings and their content must be present in the response.\n"563            )564 565            # Direct API call566            response = self.client.chat.completions.create(567                model="gpt-5-mini",  # or "gpt-4.1-mini" / "gpt-4.1-turbo" depending on your need568                messages=[569                    {"role": "system", "content": "You are an AML Analyst. Be assertive in explanation. Don't use 'could', 'can'. Suggest actions instead."},570                    {"role": "user", "content": full_query}571                ],572                temperature=0,   # deterministic output573            )574 575            # Extract the text output576            answer = response.choices[0].message.content577            logger.info(f"Direct response received: {answer}")578 579            return answer580 581        except Exception as e:582            logger.error(f"Error in send_message_direct: {e}")583            raise584    585    # convert text analysis to html586    def convert_to_html(self, response):587        # Placeholder conversion function (e.g., using an HTML template library)588        # Convert response content to HTML as needed589        return f"<html><body>{response}</body></html>"590    591    # store analysis in the database592    def store_analysis_data(self, df, responses, table_name):593        # Skip database operations if engine is not available594        if self.engine is None:595            logger.info(f"Database not configured. Skipping storage to table {table_name}")596            return True  # Return True to indicate "success" (no error, just skipped)597        598        # Prepare data for storage599        data_to_store = []600        for i, response in enumerate(responses):601            # Convert response to HTML format602            html_analysis = self.convert_to_html(response)603 604            # Gather necessary fields from df and the HTML response - safely get values with defaults605            data_to_store.append({606                "AlertID": df.get('AlertID', pd.Series([0])).iloc[i] if 'AlertID' in df.columns else 0,607                "FocusColumnValue": df.get('FocusColumnValue', pd.Series([""])).iloc[i] if 'FocusColumnValue' in df.columns else "",608                "html_analysis": html_analysis609            })610 611        # Convert the data to a DataFrame to store in the database612        store_df = pd.DataFrame(data_to_store)613        try:614            self.insert_data(table_name, store_df)615            return True616        except Exception as e:617            logger.error(f"Error storing data: {e}")618            return False619 620    # insert the data into database621    def insert_data(self, table_name, df):622        try:623            # Skip database operations if engine is not available624            if self.engine is None:625                logger.info(f"Database not configured. Skipping insert into table {table_name}")626                return627            628            # Insert the DataFrame into the specified table in the database629            df.to_sql(table_name, con=self.engine, if_exists='append', index=False)630            logger.info(f"Data inserted into table {table_name}")631        except Exception as e:632            logger.error(f"Error inserting data into {table_name}: {e}")633            raise634 635 636    # run the analysis637    def get_analysis(self):638        try:639            logger.info("Running analysis")640 641            data_table_name = DATA_TABLE642            predicition_flag = PREDICTION_FLAG643            store_table_name = ANALYSIS_TABLE644            df = self.query_data(data_table_name, predicition_flag)645            responses = []646            for i in range(len(df)):647                kyc_info = self.return_kyc_profile(df.iloc[i])648 649                # Send the full query to the assistant - safely get values with defaults650                response = self.send_message_to_assistant(651                    df.get('FilteredTransactions', pd.Series([None])).iloc[i] if 'FilteredTransactions' in df.columns else None,652                    df.get('ScenarioName', pd.Series([None])).iloc[i] if 'ScenarioName' in df.columns else None,653                    df.get('STRCount', pd.Series([0])).iloc[i] if 'STRCount' in df.columns else 0,654                    kyc_info,655                    df.get('STRScenarioHistory', pd.Series([""])).iloc[i] if 'STRScenarioHistory' in df.columns else ""656                )657                responses.append(response)658            659            # Store the responses in the table660            success = self.store_analysis_data(df, responses, store_table_name)661            # Return a JSON response based on the success662            if success:663                logger.info("Analysis completed and stored successfully")664 665                return {"status": 200, "message": "Data stored successfully"}666            else:667                logger.error("Failed to store analysis data")668 669                return {"status": "error", "message": "Failed to store data"}670            671        except Exception as e:672            logger.error(f"Error running analysis: {e}")673            return {"status": "error", "message": str(e)}674 675    676    # return the analysis in json677    def get_analysis_json(self, json_input):678        try:679            logger.info("Processing JSON input for analysis")680 681            df = json_input682            responses = []683            for i in range(len(df)):684                kyc_info = self.return_kyc_profile(df.iloc[i])685 686                # Send the full query to the assistant - safely get values with defaults687                response = self.send_message_direct(688                    df.get('FilteredTransactions', pd.Series([None])).iloc[i] if 'FilteredTransactions' in df.columns else None,689                    df.get('ScenarioName', pd.Series([None])).iloc[i] if 'ScenarioName' in df.columns else None,690                    df.get('STRCount', pd.Series([0])).iloc[i] if 'STRCount' in df.columns else 0,691                    kyc_info,692                    df.get('STRScenarioHistory', pd.Series([""])).iloc[i] if 'STRScenarioHistory' in df.columns else ""693                )694                logger.info(f"Response from assistant: {response}")695 696 697                cleaned_response = response.replace("Text(annotations=[], value='**')", "") if response else "Analysis not available"698 699                responses.append({700                    "AlertID": int(df.get('AlertID', pd.Series([0])).iloc[i]) if 'AlertID' in df.columns else 0,701                    "FocusColumnValue": str(df.get('FocusColumnValue', pd.Series([""])).iloc[i]) if 'FocusColumnValue' in df.columns else "",702                    "analysis": cleaned_response703                })704            705            logger.info("Returning analysis in JSON format")706 707            # Return the responses as a list of JSON objects708            return {"status": 200, "message": "Success", "data": responses}709        except Exception as e:710            logger.error(f"Error processing JSON input for analysis: {e}")711 712            # Return an error JSON response if conversion fails713            return {"status": "error", "message": str(e)}714 715    # return the analysis in json using local LLM (non-streaming) - OPTIMIZED716    def get_analysis_json_local_llm(self, json_input):717        try:718            logger.info("Processing JSON input for analysis using local LLM (OPTIMIZED)")719 720            df = json_input721            responses = []722            723            # Use global LLM instance for better performance724            local_llm = get_global_llm()725            726            for i in range(len(df)):727                kyc_info = self.return_kyc_profile(df.iloc[i])728 729                # Use cached template for faster generation - safely get values with defaults730                full_query = get_analysis_template(731                    df.get('ScenarioName', pd.Series([None])).iloc[i] if 'ScenarioName' in df.columns else None,732                    kyc_info,733                    df.get('FilteredTransactions', pd.Series([None])).iloc[i] if 'FilteredTransactions' in df.columns else None,734                    df.get('STRCount', pd.Series([0])).iloc[i] if 'STRCount' in df.columns else 0,735                    df.get('STRScenarioHistory', pd.Series([""])).iloc[i] if 'STRScenarioHistory' in df.columns else ""736                )737 738                # Use local LLM to generate response (non-streaming) with optimized settings739                response = local_llm.generate_response(740                    full_query, 741                    stream=False,742                    temperature=0.1,  # Lower temperature for faster, more deterministic responses743                    top_p=0.9,744                    num_predict=150  # Limit response length for speed745                )746                logger.info(f"Response from local LLM: {response[:100] if response else 'No response'}...")  # Log only first 100 chars747 748                cleaned_response = response.replace("Text(annotations=[], value='**')", "") if response else "Analysis not available"749 750                responses.append({751                    "AlertID": int(df.get('AlertID', pd.Series([0])).iloc[i]) if 'AlertID' in df.columns else 0,752                    "FocusColumnValue": str(df.get('FocusColumnValue', pd.Series([""])).iloc[i]) if 'FocusColumnValue' in df.columns else "",753                    "analysis": cleaned_response754                })755            756            logger.info("Returning analysis in JSON format using local LLM (OPTIMIZED)")757 758            return {"status": 200, "message": "Success", "data": responses}759        except Exception as e:760            logger.error(f"Error processing JSON input for analysis using local LLM: {e}")761            return {"status": "error", "message": str(e)}762 763    # return the analysis in json using local LLM (streaming) - OPTIMIZED764    def get_analysis_json_local_llm_streaming(self, json_input):765        try:766            logger.info("Processing JSON input for analysis using local LLM (streaming OPTIMIZED)")767 768            df = json_input769            770            # Use global LLM instance for better performance771            local_llm = get_global_llm()772            773            for i in range(len(df)):774                kyc_info = self.return_kyc_profile(df.iloc[i])775 776                # Use cached template for faster generation - safely get values with defaults777                full_query = get_analysis_template(778                    df.get('ScenarioName', pd.Series([None])).iloc[i] if 'ScenarioName' in df.columns else None,779                    kyc_info,780                    df.get('FilteredTransactions', pd.Series([None])).iloc[i] if 'FilteredTransactions' in df.columns else None,781                    df.get('STRCount', pd.Series([0])).iloc[i] if 'STRCount' in df.columns else 0,782                    df.get('STRScenarioHistory', pd.Series([""])).iloc[i] if 'STRScenarioHistory' in df.columns else ""783                )784 785                # Use local LLM to generate streaming response with optimized settings786                response_stream = local_llm.generate_response(787                    full_query, 788                    stream=True,789                    temperature=0.1,  # Lower temperature for faster responses790                    top_p=0.9,791                    num_predict=150  # Limit response length for speed792                )793                794                # Yield the streaming response795                yield {796                    "AlertID": int(df.get('AlertID', pd.Series([0])).iloc[i]) if 'AlertID' in df.columns else 0,797                    "FocusColumnValue": str(df.get('FocusColumnValue', pd.Series([""])).iloc[i]) if 'FocusColumnValue' in df.columns else "",798                    "analysis": response_stream799                }800            801            logger.info("Completed streaming analysis using local LLM (OPTIMIZED)")802 803        except Exception as e:804            logger.error(f"Error processing JSON input for analysis using local LLM (streaming): {e}")805            yield {"status": "error", "message": str(e)}806 807    # return the analysis in json using hybrid ultra-fast system (MILLISECOND RESPONSES)808    def get_analysis_json_ultra_fast(self, json_input):809        try:810            logger.info("Processing JSON input for analysis using HYBRID ULTRA-FAST system")811 812            df = json_input813            responses = []814            815            # Use hybrid system for ultra-fast responses816            hybrid_system = get_hybrid_system()817            818            for i in range(len(df)):819                # Convert DataFrame row to dictionary820                alert_data = df.iloc[i].to_dict()821                822                # Generate ultra-fast analysis823                result = hybrid_system.analyze_ultra_fast(alert_data)824                825                responses.append({826                    "AlertID": result["AlertID"],827                    "FocusColumnValue": result["FocusColumnValue"],828                    "analysis": result["analysis"],829                    "response_time_ms": result["response_time_ms"],830                    "risk_level": result["risk_level"],831                    "method": result["method"]832                })833            834            logger.info(f"Returning ULTRA-FAST analysis in JSON format - Average response time: {sum(r['response_time_ms'] for r in responses)/len(responses):.2f}ms")835 836            return {"status": 200, "message": "Success", "data": responses}837        except Exception as e:838            logger.error(f"Error processing JSON input for ULTRA-FAST analysis: {e}")839            return {"status": "error", "message": str(e)}840 841    # return the analysis in json using hybrid ultra-fast system (streaming)842    def get_analysis_json_ultra_fast_streaming(self, json_input):843        try:844            logger.info("Processing JSON input for analysis using HYBRID ULTRA-FAST streaming")845 846            df = json_input847            848            # Use hybrid system for ultra-fast responses849            hybrid_system = get_hybrid_system()850            851            for i in range(len(df)):852                # Convert DataFrame row to dictionary853                alert_data = df.iloc[i].to_dict()854                855                # Generate ultra-fast analysis856                result = hybrid_system.analyze_ultra_fast(alert_data)857                858                # Yield the ultra-fast streaming response859                yield {860                    "AlertID": result["AlertID"],861                    "FocusColumnValue": result["FocusColumnValue"],862                    "analysis": result["analysis"],863                    "response_time_ms": result["response_time_ms"],864                    "risk_level": result["risk_level"],865                    "method": result["method"]866                }867            868            logger.info("Completed ULTRA-FAST streaming analysis")869 870        except Exception as e:871            logger.error(f"Error processing JSON input for ULTRA-FAST streaming analysis: {e}")872            yield {"status": "error", "message": str(e)}873 874    # return the analysis in json using hybrid template+LLM system (TEMPLATES FOR DATA + LLM FOR ANALYSIS)875    def get_analysis_json_hybrid_template_llm(self, json_input, use_cloud: bool = False, llm_on_server: bool = False, url: str = "", evaluation: bool = False, audit: bool = False):876        try:877            method_name = "HYBRID TEMPLATE+LLM (CLOUD)" if use_cloud else "HYBRID TEMPLATE+LLM"878            logger.info(f"Processing JSON input for analysis using {method_name} system")879 880            df = json_input881            responses = []882 883            # Use hybrid template+LLM system884            hybrid_system = get_template_llm_system()885 886            for i in range(len(df)):887                # Convert DataFrame row to dictionary888                alert_data = df.iloc[i].to_dict()889 890                # Generate hybrid analysis (templates for data + LLM for analysis)891                # Use cloud method if flag is set, otherwise use local method892                if use_cloud:893                    result = hybrid_system.analyze_hybrid_cloud(alert_data, evaluation=evaluation, audit=audit)894                else:895                    result = hybrid_system.analyze_hybrid(alert_data, llm_on_server=llm_on_server, url=url, evaluation=evaluation, audit=audit)896 897                item = {898                    "AlertID": result["AlertID"],899                    "FocusColumnValue": result["FocusColumnValue"],900                    "analysis": result["analysis"],901                    "response_time_ms": result["response_time_ms"],902                    "method": result["method"],903                    "model": result.get("model", "N/A")904                }905                if result.get("thinking") is not None:906                    item["thinking"] = result["thinking"]907                responses.append(item)908 909            logger.info(f"Returning {method_name} analysis in JSON format - Average response time: {sum(r['response_time_ms'] for r in responses)/len(responses):.2f}ms")910 911            return {"status": 200, "message": "Success", "data": responses}912        except Exception as e:913            logger.error(f"Error processing JSON input for {method_name} analysis: {e}")914            return {"status": "error", "message": str(e)}915 916    # return the analysis in json using hybrid template+LLM system (streaming)917    def get_analysis_json_hybrid_template_llm_streaming(self, json_input):918        try:919            logger.info("Processing JSON input for analysis using HYBRID TEMPLATE+LLM streaming")920 921            df = json_input922 923            # Use hybrid template+LLM system924            hybrid_system = get_template_llm_system()925 926            for i in range(len(df)):927                # Convert DataFrame row to dictionary928                alert_data = df.iloc[i].to_dict()929 930                # Generate hybrid analysis (templates for data + LLM for analysis)931                result = hybrid_system.analyze_hybrid(alert_data)932 933                # Yield the hybrid streaming response934                yield {935                    "AlertID": result["AlertID"],936                    "FocusColumnValue": result["FocusColumnValue"],937                    "analysis": result["analysis"],938                    "response_time_ms": result["response_time_ms"],939                    "method": result["method"],940                    "model": result.get("model", "N/A")941                }942 943            logger.info("Completed HYBRID TEMPLATE+LLM streaming analysis")944 945        except Exception as e:946            logger.error(f"Error processing JSON input for HYBRID TEMPLATE+LLM streaming analysis: {e}")947            yield {"status": "error", "message": str(e)}948 949# api route for generating analysis and returning json response950@router.post(951    "/api/ai-service/generateamlanalysisoai",952    response_model=AnalysisResponse,953    summary="Generate AML Analysis (OpenAI)",954    description="Generate AML analysis using OpenAI GPT models. Requires OPENAI_API_KEY and ASSISTANT_ID configuration."955)956async def generate_analysis(alert_data: Union[AnalysisAlertRequest, List[AnalysisAlertRequest]]):957    try:958        logger.info("API call to /generateamlanalysisoai started")959 960        # Handle both single object and array of objects961        if isinstance(alert_data, list):962            json_data = [item.model_dump(exclude_none=True) for item in alert_data]963            df = pd.DataFrame(json_data)964        else:965            json_data = alert_data.model_dump(exclude_none=True)966            df = pd.DataFrame([json_data])967        db_connection_str = DB_CONNECTION_STR968        api_key = OPENAI_API_KEY969        client = OpenAI(api_key=api_key)970        analysis = AnalysisProcessor(db_connection_str, client, ASSISTANT_ID)971        972        logger.info("API call to /generateamlanalysisoai completed successfully")973 974        return analysis.get_analysis_json(df)975    976    except ValidationError as e:977        # Handle Pydantic validation errors and return an appropriate response978        logger.error(f"Error in /generateamlanalysisoai validation error: {e.errors()}")979        raise HTTPException(status_code=422, detail=f"Validation error: {e.errors()}")980    except Exception as e:981        logger.error(f"Error in /generateamlanalysisoai: {e}")982        raise HTTPException(status_code=500, detail="An error occurred while generating AML analysis using oai")983 984 985# Extended model for hybrid analysis with Cloud flags986class AnalysisAlertRequestWithFlags(AnalysisAlertRequest):987    Cloud: Optional[bool] = Field(False, description="Use cloud LLM (OpenRouter) instead of local Ollama. Set to true to use OpenRouter API.", example=False)988    llm_on_server: Optional[bool] = Field(False, description="Use LLM on remote server. Set to true to use a remote Ollama server instead of local.", example=False)989    url: Optional[str] = Field("", description="URL of remote Ollama server. Required if llm_on_server is true. Example: http://remote-ollama:11434", example="http://remote-ollama:11434")990    anonymous: Optional[bool] = Field(False, description="Mask PII (Personally Identifiable Information) in the data. When true, PII columns specified in PII_COLUMNS environment variable will be masked.", example=False)991    evaluation: Optional[bool] = Field(False, description="When true, evaluate the generated analysis and fix mistakes using an evaluator agent (cloud or local/remote Ollama).", example=False)992    audit: Optional[bool] = Field(False, description="When true, use a thinking model for generation and include 'thinking' in the response (cloud: OPENROUTER_THINKING_MODEL, local: OLLAMA_THINKING_MODEL).", example=False)993    994    class Config:995        json_schema_extra = {996            "example": {997                "AlertID": 1,998                "FilteredTransactions": '[{"CUSTOMERID":"100001","IDENTITYNUMBERS":"42101-1234567-1","LOANID":"LN2024001","ACCOUNTID":"AC100001","CREATEDDATE":"2025-06-01 10:15:00","TRANSACTIONAMOUNT":150000.0,"CURRENCY":"PKR","INSTALLMENTNUMBER":1,"EXCESSAMOUNT":0.0}]',999                "FocusColumnValue": "100001",1000                "KYCMonthlyIncome": "85,000 PKR",1001                "KYCNoOfCredits": "3-5",1002                "KYCNoOfDebits": "8-12",1003                "KYCRiskCategoryValue": "Low",1004                "KYCValueOfCredits": "150,000 - 200,000 PKR",1005                "KYCValueOfDebits": "80,000 - 120,000 PKR",1006                "OccupationValue": "Private Employee",1007                "STRCount": 0,1008                "STRScenarioHistory": "",1009                "ScenarioName": "Unusually large installment",1010                "CustomerName": "Muhammad Bilal Sheikh",1011                "CUSTOMERID": "100001",1012                "BranchID": "KHI-DHA",1013                "Country": "Pakistan",1014                "CustomerType": "Retail",1015                "CustomerStatus": "Retail Customer",1016                "CreatedDate": "2020-01-15",1017                "RelationshipStartDate": "2020-01-15",1018                "RiskScore": "4.2",1019                "PreviousAlerts": [],1020                "Counterparties": [],1021                "BranchQueries": {1022                    "Requested": "Verification required for loan installments totaling 530,000 PKR within one month, significantly exceeding declared monthly income of 85,000 PKR. Please confirm source of funds and provide documentation for additional income sources.",1023                    "Response": "Customer stated installments are from remittances received from brother working in UAE, family savings from wedding expenses, and advance salary from employer for Eid holidays. Customer provided remittance receipts and employer letter."1024                },1025                "Cloud": False,1026                "llm_on_server": False,1027                "url": "",1028                "anonymous": False,1029                "evaluation": False,1030                "audit": False1031            }1032        }1033 1034# api route for generating analysis and returning json response using HYBRID TEMPLATE+LLM system (TEMPLATES FOR DATA + LLM FOR ANALYSIS)1035@router.post(1036    "/api/ai-service/generateamlanalysis",1037    response_model=AnalysisResponse,1038    summary="Generate AML Analysis (Hybrid Template+LLM)",1039    description="""1040    Generate AML analysis using Hybrid Template+LLM system.1041    1042    **LLM Options:**1043    - **Local LLM (default)**: Set `Cloud: false` and `llm_on_server: false` - Uses local Ollama1044    - **Cloud LLM**: Set `Cloud: true` - Uses OpenRouter API (requires OPENROUTER_API_KEY)1045    - **Remote LLM**: Set `llm_on_server: true` and provide `url` - Uses remote Ollama server1046    1047    **Note**: Only one LLM option should be enabled at a time. If multiple are set, priority is: Cloud > llm_on_server > Local1048    1049    **PII Masking:**1050    - Set `anonymous: true` to mask Personally Identifiable Information (PII) in the data1051    - PII values are masked by keeping first 2 and last 2 characters, replacing middle with `***`1052    - Example: `"Muhammad Bilal Sheikh"` → `"Mu***kh"`, `"100001"` → `"10***01"`1053    - Masking applies to direct fields and nested structures (FilteredTransactions, PreviousAlerts, Counterparties)1054    - PII columns are configurable via `PII_COLUMNS` environment variable (comma-separated list)1055    - Default PII columns: `CustomerName`, `CUSTOMERID`, `FocusColumnValue`, `IDENTITYNUMBERS`, `ACCOUNTID`, `LOANID`1056    - The masking function handles all string lengths appropriately1057    1058    **Evaluation (evaluation: true):**1059    - When `evaluation: true`, the generated report is evaluated by an evaluator agent (cloud or local/remote Ollama)1060    - The LLM then decides whether corrections are needed; if YES, a fix step is applied; if NO, the original report is returned1061    - Works with both Cloud and Local/Remote LLM1062    1063    **Audit (audit: true):**1064    - When `audit: true`, a thinking model is used for generation (cloud: OPENROUTER_THINKING_MODEL, default moonshotai/kimi-k2-thinking; local: OLLAMA_THINKING_MODEL, default deepseek-r1:8b)1065    - The response includes an optional `thinking` field with the model's reasoning/reasoning_content when available1066    """,1067    response_description="Analysis results with generated AML report; optional 'thinking' when audit=true"1068)1069async def generate_aml_analysis(alert_data: Union[AnalysisAlertRequestWithFlags, List[AnalysisAlertRequestWithFlags]]):1070    try:1071        logger.info("API call to /generateamlanalysis started")1072 1073        # Extract flags and handle both single object and array1074        use_cloud = False1075        llm_on_server = False1076        url = ""1077        anonymous = False1078        evaluation = False1079        audit = False1080        _exclude = {'Cloud', 'llm_on_server', 'url', 'anonymous', 'evaluation', 'audit'}1081        1082        if isinstance(alert_data, list):1083            if len(alert_data) > 0:1084                use_cloud = alert_data[0].Cloud if alert_data[0].Cloud is not None else False1085                llm_on_server = alert_data[0].llm_on_server if alert_data[0].llm_on_server is not None else False1086                url = alert_data[0].url if alert_data[0].url else ""1087                anonymous = alert_data[0].anonymous if alert_data[0].anonymous is not None else False1088                evaluation = alert_data[0].evaluation if getattr(alert_data[0], 'evaluation', None) is not None else False1089                audit = alert_data[0].audit if getattr(alert_data[0], 'audit', None) is not None else False1090            json_data = [item.model_dump(exclude_none=True, exclude=_exclude) for item in alert_data]1091            1092            # Apply PII masking if anonymous flag is true1093            if anonymous:1094                json_data = mask_pii_data(json_data, PII_COLUMNS)1095            1096            df = pd.DataFrame(json_data)1097        else:1098            use_cloud = alert_data.Cloud if alert_data.Cloud is not None else False1099            llm_on_server = alert_data.llm_on_server if alert_data.llm_on_server is not None else False1100            url = alert_data.url if alert_data.url else ""1101            anonymous = alert_data.anonymous if alert_data.anonymous is not None else False1102            evaluation = alert_data.evaluation if getattr(alert_data, 'evaluation', None) is not None else False1103            audit = alert_data.audit if getattr(alert_data, 'audit', None) is not None else False1104            json_data = alert_data.model_dump(exclude_none=True, exclude=_exclude)1105            1106            # Apply PII masking if anonymous flag is true1107            if anonymous:1108                json_data = mask_pii_data(json_data, PII_COLUMNS)1109            1110            df = pd.DataFrame([json_data])1111        # Use None for database connection since it's not needed for hybrid template LLM1112        db_connection_str = None1113        analysis = AnalysisProcessor(db_connection_str, None, None)1114        1115        logger.info(f"API call to /generateamlanalysis completed successfully (Cloud: {use_cloud}, llm_on_server: {llm_on_server}, url: {url}, anonymous: {anonymous}, evaluation: {evaluation}, audit: {audit})")1116 1117        return analysis.get_analysis_json_hybrid_template_llm(df, use_cloud=use_cloud, llm_on_server=llm_on_server, url=url, evaluation=evaluation, audit=audit)1118 1119    except ValidationError as e:1120        logger.error(f"Error in /generateamlanalysis validation error: {e.errors()}")1121        raise HTTPException(status_code=422, detail=f"Validation error: {e.errors()}")1122    except Exception as e:1123        logger.error(f"Error in /generateamlanalysis: {e}")1124        raise HTTPException(status_code=500, detail="An error occurred while generating analysis")1125 1126# api route for generating analysis and returning streaming json response using HYBRID TEMPLATE+LLM system1127@router.post("/api/ai-service/generateamlanalysisstreaming")1128async def generate_aml_analysis_streaming(alert_data: Union[AnalysisAlertRequest, List[AnalysisAlertRequest]]):1129    try:1130        logger.info("API call to /generateamlanalysisstreaming started")1131 1132        # Handle both single object and array of objects1133        if isinstance(alert_data, list):1134            json_data = [item.model_dump(exclude_none=True) for item in alert_data]1135            df = pd.DataFrame(json_data)1136        else:1137            json_data = alert_data.model_dump(exclude_none=True)1138            df = pd.DataFrame([json_data])1139        # Use None for database connection since it's not needed for hybrid template LLM1140        db_connection_str = None1141        analysis = AnalysisProcessor(db_connection_str, None, None)1142        1143        logger.info("API call to /generateamlanalysisstreaming completed successfully")1144 1145        # Create a generator function for hybrid template+LLM streaming response1146        async def generate_hybrid_template_llm_stream():1147            try:1148                for response_chunk in analysis.get_analysis_json_hybrid_template_llm_streaming(df):1149                    yield f"data: {json.dumps(response_chunk)}\n\n"1150            except Exception as e:1151                logger.error(f"Error in hybrid template+LLM streaming response: {e}")1152                yield f"data: {json.dumps({'status': 'error', 'message': str(e)})}\n\n"1153 1154        return StreamingResponse(1155            generate_hybrid_template_llm_stream(),1156            media_type="text/plain",1157            headers={"Cache-Control": "no-cache", "Connection": "keep-alive"}1158        )1159    1160    except ValidationError as e:1161        logger.error(f"Error in /generateamlanalysisstreaming validation error: {e.errors()}")1162        raise HTTPException(status_code=422, detail=f"Validation error: {e.errors()}")1163    except Exception as e:1164        logger.error(f"Error in /generateamlanalysisstreaming: {e}")1165        raise HTTPException(status_code=500, detail="An error occurred while generating analysis streaming")1166 1167