File size: 3,592 Bytes
64e5ee2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | import os
import requests
from dotenv import load_dotenv
import logging
load_dotenv()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class LLMExplainer:
def __init__(self):
self.api_key = os.environ.get("MISTRAL_API_KEY")
self.api_url = "https://api.mistral.ai/v1/chat/completions"
self.explanation_cache = {}
def explain_discrepancy(self, row, match_status, books_vendor, gst_vendor, books_amount, gst_amount):
"""
Generates a human-readable explanation for a discrepancy using Mistral AI.
"""
if not self.api_key:
return "API Key not configured. Discrepancy: " + match_status
is_anomaly = row.get('IsAnomaly', False)
if match_status == "Exact Match" and not is_anomaly:
return "Records match perfectly."
cache_key = (match_status, books_vendor, gst_vendor, books_amount, gst_amount, is_anomaly)
if cache_key in self.explanation_cache:
logger.info("Returning cached LLM explanation.")
return self.explanation_cache[cache_key]
prompt = f"""
You are an expert AI financial auditor. Explain the following transaction in 1-2 concise, professional sentences. Provide a recommended action.
Reconciliation Status: {match_status}
Is Flagged as Anomaly: {is_anomaly}
Anomaly Score: {row.get('AnomalyScore', 0.0)}
Invoice ID: {row.get('InvoiceID', 'Unknown')}
Books Vendor: {books_vendor}
GST Vendor: {gst_vendor}
Books Amount: {books_amount}
GST Amount: {gst_amount}
Explanation and Recommendation:
"""
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
"Authorization": f"Bearer {self.api_key}"
}
data = {
"model": "mistral-tiny", # Using tiny for faster responses, can upgrade to small/medium
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.3
}
try:
response = requests.post(self.api_url, headers=headers, json=data, timeout=10)
response.raise_for_status()
result = response.json()
explanation = result['choices'][0]['message']['content'].strip()
self.explanation_cache[cache_key] = explanation
return explanation
except requests.exceptions.RequestException as e:
logger.error(f"Error calling Mistral API: {e}")
return f"API Error ({match_status}). Please check amounts and vendor names manually."
except (KeyError, IndexError) as e:
logger.error(f"Unexpected response format from Mistral API: {e}")
return f"Error parsing AI response ({match_status})."
def generate_explanations_batch(self, discrepancies_df):
"""
Generates explanations for a dataframe of discrepancies.
"""
explanations = []
for _, row in discrepancies_df.iterrows():
status = row.get('MatchStatus', 'Unknown')
b_vendor = row.get('VendorName_books', 'N/A')
g_vendor = row.get('VendorName_gst', 'N/A')
b_amount = row.get('Amount_books', 0)
g_amount = row.get('Amount_gst', 0)
explanation = self.explain_discrepancy(row, status, b_vendor, g_vendor, b_amount, g_amount)
explanations.append(explanation)
return explanations
|