| 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", |
| "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 |
|
|
