from time import sleep
from airflow.utils.decorators import apply_defaults
from operators.bedrock_automation_base_operator import BedrockAutomationBaseOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from config import Config
from hooks.grapi_hook import GrapiHook
import numpy as np
[docs]
class GrapiBedrockAutomationOperator(BedrockAutomationBaseOperator):
dest_db = None
MODEL_ACTIVITIES = {
"AiInvoiceProcessFinished": "AiInvoiceProcessFinished",
"AiInvoiceProcessStarted": "AiInvoiceProcessStarted",
"AiInvoiceProcessFailed": "AiInvoiceProcessFailed",
}
@apply_defaults
def __init__(
self,
input_pdf_key,
output_uri,
tenant_name,
env_name,
application_resource_id,
application_id,
**kwargs,
):
super().__init__(
input_pdf_key=input_pdf_key,
output_uri=output_uri,
blueprint="arn:aws:bedrock:us-east-1:463002504035:blueprint/82b8b0d229f1",
blueprint_version="2",
blueprint_stage="LIVE",
**kwargs,
)
self.dest_db = (
Config.BOSS_PLATFORM_GRAPI_STAGE_DB
if env_name == "stage"
else Config.BOSS_PLATFORM_GRAPI_DB
)
self.db_hook = PostgresHook(postgres_conn_id=self.dest_db)
self.tenant = tenant_name
self.env = env_name
self.application_resource_id = application_resource_id
self.application_id=application_id
self.graphi_hook = GrapiHook(tenant=tenant_name, stage=env_name)
[docs]
def is_there_any_running_processes(self):
sql=f"""
WITH application_activity AS (
SELECT
type, attributes
FROM {self.tenant}."ApplicationActivity"
WHERE "applicationId" = '{self.application_id}'
)
SELECT DISTINCT s.applicationResourceId
FROM (
SELECT attributes ->> 'applicationResourceId' AS applicationResourceId
FROM application_activity
WHERE type = 'AiInvoiceProcessStarted'
) s
LEFT JOIN (
SELECT attributes ->> 'applicationResourceId' AS applicationResourceId
FROM application_activity
WHERE type in ('AiInvoiceProcessFinished','AiInvoiceProcessFailed')
) f
ON s.applicationResourceId = f.applicationResourceId
WHERE f.applicationResourceId IS NULL;
"""
print(sql)
df= self.db_hook.get_pandas_df(sql)
nothing_is_runing=df.empty
return not nothing_is_runing
[docs]
def execute(self, context):
from time import sleep
current_iteration = 0
max_iteration = 10
while self.is_there_any_running_processes() and current_iteration < max_iteration:
print(f"Waiting for application to finish processing. Try {current_iteration}/{max_iteration}")
current_iteration += 1
sleep(10)
return super().execute(context)
[docs]
def generate_hash_id(self, invoice_data):
import hashlib
import json
# Extract key fields for hash
key_fields = {
"DATE": invoice_data["inference_result"]["DATE"],
"RECIPIENTADDRESS": invoice_data["inference_result"]["RECIPIENTADDRESS"],
"VENDORNAME": invoice_data["inference_result"]["VENDORNAME"],
"PO": invoice_data["inference_result"]["PO"],
"TOTAL": invoice_data["inference_result"]["TOTAL"],
}
# Convert to JSON string for consistency
key_string = json.dumps(key_fields, sort_keys=True)
# Generate SHA-256 hash
hash_object = hashlib.sha256(key_string.encode())
return hash_object.hexdigest()
[docs]
def identify_similar_invoices(self, invoice_hash):
db_hook = PostgresHook(postgres_conn_id=self.dest_db)
sql = f"""
-- select id,attributes->'textExtraction' document from {self.tenant}."ApplicationResource" where --attributes->'textExtraction'->>'similarity_hash'='{invoice_hash}'
WITH duplicate_invoices AS
(SELECT id,
replace(attributes->>'question', '-invoice', '') expense_line_item,
attributes->'textExtraction' document
FROM {self.tenant}."ApplicationResource"
WHERE attributes->'textExtraction'->>'similarity_hash'='{invoice_hash}'
AND id::TEXT <> '{self.application_resource_id}' ),
expense_lines AS
(SELECT ans.*,
ans.answer->>'cost' AS cost,
ans.answer->>'id' AS expense_id,
'1' AS GROUP
FROM {self.tenant}."ApplicationAnswer" ans
JOIN {self.tenant}."Application" app ON app.id = ans."applicationId"
WHERE ans.question LIKE 'expense_%'
AND ans.question != 'expensesList' -- Exclude expensesList
AND (ans.answer->>'status') IS DISTINCT
FROM 'deleted'
AND NOT EXISTS
(SELECT 1
FROM {self.tenant}."Application" child
WHERE child."parentApplicationId" = app.id )
AND ans.question IN
(SELECT expense_line_item
FROM duplicate_invoices)
ORDER BY ans."applicationId",
ans.question)
SELECT elis.*, di.document, di.id as application_resource_id
FROM expense_lines elis left join duplicate_invoices di on elis.question = di.expense_line_item
"""
print(sql)
df = db_hook.get_pandas_df(sql)
return df
[docs]
def get_langfuse_agent(self):
import boto3
import json
from langfuse.langchain import CallbackHandler
secret_name = f"{self.env}/langfuse/invoice_deduplication"
region_name = "us-east-1"
# Create a Secrets Manager client
session = boto3.session.Session()
client = session.client(service_name="secretsmanager", region_name=region_name)
try:
get_secret_value_response = client.get_secret_value(SecretId=secret_name)
except Exception as e:
# For a list of exceptions thrown, see
# https://docs.aws.amazon.com/secretsmanager/latest/apireference/API_GetSecretValue.html
raise e
secret = json.loads(get_secret_value_response["SecretString"])
from langfuse import Langfuse, get_client
import httpx
httpx_client = httpx.Client(verify=False)
Langfuse(
public_key=secret["public_key"],
secret_key=secret["secure_key"],
host=secret["host"], # Optional: defaults to https://cloud.langfuse.com
httpx_client=httpx_client,
environment=self.env,
)
langfuse = get_client()
langfuse_handler = CallbackHandler()
return langfuse_handler, langfuse
[docs]
def get_llm(self):
import boto3
from botocore.config import Config as BotocoreConfig
from langchain_aws import ChatBedrockConverse
from typing import Optional
from typing_extensions import Annotated, TypedDict
class InvoiceDeduplicationResult(TypedDict):
"""
Result of invoice deduplication analysis.
This class represents the outcome of comparing invoices to determine
if they are duplicates. It includes the reasoning behind the decision,
a confidence score, and the final determination.
"""
rational: Annotated[
str,
...,
"Detailed reasoning explaining why the invoices are or are not considered duplicates. 300 words maximum.",
]
confidence: Annotated[
float,
...,
"Confidence score between 0.0 and 1.0 indicating certainty in the duplicate determination",
]
is_duplicate: Annotated[
bool,
...,
"True if the invoices are determined to be duplicates, False otherwise",
]
config = BotocoreConfig(read_timeout=600)
llm = ChatBedrockConverse(
model=Config.CLAUDE_BEDROCK_SONNET,
# temperature=self.inference_config.get("temperature"),
# top_p=self.inference_config.get("topP"),
# max_tokens=self.inference_config.get("maxTokens"),
config=config,
)
structured_llm = llm.with_structured_output(InvoiceDeduplicationResult)
return structured_llm
[docs]
def generate_ai_message(self, similar_records, current_record):
pass
from langchain_core.messages import HumanMessage
llm = self.get_llm()
langfuse_handler, langfuse = self.get_langfuse_agent()
results = []
for _, row in similar_records.iterrows():
invoice_a = row["document"]
invoice_b = current_record
# Prepare the message content for the AI model
message_content = f"""
# Invoice Deduplication Analysis Prompt
You are an expert invoice analyst tasked with determining if two invoices are duplicates. You must analyze both invoices carefully and return a structured result.
## Task
Compare the two invoices provided and determine if they represent the same transaction (duplicates) or different transactions.
## Analysis Criteria
Consider these key factors when making your determination:
**Primary Identifiers (Highest Weight):**
- Invoice number/ID
- Vendor/supplier information
- Invoice amount (exact match or very close)
- Invoice date
- Purchase order number (if present)
**Secondary Identifiers (Medium Weight):**
- Line items and quantities
- Tax amounts
- Payment terms
- Billing address
- Description of goods/services
**Contextual Factors (Lower Weight):**
- Document formatting differences
- Minor clerical variations (typos, spacing)
- Different file formats or submission methods
- Time zone differences in timestamps
## Decision Logic
**Consider as DUPLICATES if:**
- Invoice numbers match exactly AND vendor matches
- Same vendor, amount, date, and similar line items (even with minor formatting differences)
- One invoice appears to be a corrected/updated version of another with same core transaction details
- Clear evidence of resubmission or duplicate processing
**Consider as SEPARATE invoices if:**
- Different invoice numbers from same vendor for different amounts/dates
- Same vendor but clearly different transactions (different PO numbers, dates, amounts)
- Partial payments or installments of a larger transaction
- Recurring charges (monthly subscriptions, etc.) with different periods
## Confidence Scoring Guidelines
- **0.9-1.0**: Virtually certain (exact matches on all primary identifiers)
- **0.7-0.89**: High confidence (strong evidence with minor discrepancies)
- **0.5-0.69**: Moderate confidence (mixed signals requiring judgment)
- **0.3-0.49**: Low confidence (limited matching criteria)
- **0.0-0.29**: Very uncertain (insufficient information or conflicting data)
## Input Format
You will receive two invoices in the following format:
```
INVOICE A:
[Invoice A content/details]
INVOICE B:
[Invoice B content/details]
```
## Required Output Format
Return your analysis as a JSON object matching this exact structure:
```json
{{
"rational": "Detailed explanation of your analysis, including which factors you considered, what matched or differed, and the reasoning behind your decision. Be specific about the evidence that led to your conclusion.",
"confidence": 0.85,
"is_duplicate": true
}}
```
## Important Notes
- Be thorough but concise in your reasoning
- Always explain your confidence level
- Consider both false positive and false negative risks
- When in doubt, lean slightly toward marking as NOT duplicates to avoid incorrectly rejecting legitimate invoices
- Focus on business logic - minor formatting differences don't make invoices different if the underlying transaction is the same
- keep your response in maximum 300 words maximum
Now analyze the following invoices:
---
**INVOICE A:**
{invoice_a}
**INVOICE B:**
{invoice_b}
"""
ai_response = llm.invoke(
[
HumanMessage(
content=[
{
"text": message_content,
},
],
)
],
config={
"callbacks": [langfuse_handler],
"metadata": {
"langfuse_user_id": self.tenant,
# "langfuse_session_id": "random-session",
# "langfuse_tags": ["random-tag-1", "random-tag-2"],
},
},
)
trace_id = langfuse_handler.last_trace_id
print(f"Trace ID: {trace_id}")
ai_response["trace_id"] = trace_id
ai_response["expense_id"] = row["expense_id"]
ai_response["application_id"] = row["applicationId"]
ai_response["application_resource_id"] = row["application_resource_id"]
ai_response["id"] = row["id"]
ai_response["question"] = row["question"]
ai_response["cost"] = row["cost"]
results.append(ai_response)
langfuse.flush()
return results
[docs]
def make_hashable(self, obj):
"""Convert lists to tuples to make them hashable"""
if isinstance(obj, dict):
return {k: self.make_hashable(v) for k, v in obj.items()}
elif isinstance(obj, list):
return tuple(self.make_hashable(x) for x in obj)
return self.convert_numpy_types(obj)
[docs]
def convert_numpy_types(self,data):
if isinstance(data, dict):
return {k: self.convert_numpy_types(v) for k, v in data.items()}
elif isinstance(data, list):
return [self.convert_numpy_types(item) for item in data]
elif isinstance(data, tuple):
return tuple(self.convert_numpy_types(item) for item in data)
elif isinstance(data, np.integer):
return int(data)
elif isinstance(data, np.floating):
return float(data)
return data
[docs]
def is_total_exceeds_elis(self, total_amount, invoice_hash):
import pandas as pd
db_hook = PostgresHook(postgres_conn_id=self.dest_db)
sql = f"""
with duplicate_invoices as (
select id,replace(attributes->>'question','-invoice','') expense_line_item,attributes->'textExtraction' document from {self.tenant}."ApplicationResource" where attributes->'textExtraction'->>'similarity_hash'='{invoice_hash}'
--and id::TEXT <> '{invoice_hash}'
),
expense_lines as (
SELECT
ans.*,
ans.answer->>'cost' as cost,
ans.answer->>'id' as expense_id
FROM {self.tenant}."ApplicationAnswer" ans
JOIN {self.tenant}."Application" app ON app.id = ans."applicationId"
WHERE ans.question LIKE 'expense_%'
AND ans.question != 'expensesList' -- Exclude expensesList
AND (ans.answer->>'status') IS DISTINCT FROM 'deleted'
AND NOT EXISTS (
SELECT 1 FROM {self.tenant}."Application" child
WHERE child."parentApplicationId" = app.id
)
AND ans.question in (select expense_line_item from duplicate_invoices)
ORDER BY ans."applicationId", ans.question
)
select * FROM expense_lines
"""
print(sql)
df = db_hook.get_pandas_df(sql)
print(df.head())
df["cost"] = pd.to_numeric(df["cost"], errors="coerce")
total_claimed = df["cost"].sum()
print(f"total_claimed:{total_claimed}, total_amount:{total_amount}")
is_exceed = self.safe_cast(total_claimed, float, float("inf")) > self.safe_cast(
total_amount, float, 0.0
)
return is_exceed, total_claimed, total_amount
[docs]
def safe_cast(self, val, to_type, default=None):
try:
return to_type(val)
except (ValueError, TypeError):
return default
[docs]
def get_current_flag_info(self):
sql = f"""
WITH duplicate_invoices AS
(SELECT id,
replace(attributes->>'question', '-invoice', '') expense_line_item,
attributes->'textExtraction' document
FROM {self.tenant}."ApplicationResource"
WHERE id::TEXT = '{self.application_resource_id}' ),
expense_lines AS
(SELECT ans.*,
ans.answer->>'cost' AS cost,
ans.answer->>'id' AS expense_id
FROM {self.tenant}."ApplicationAnswer" ans
JOIN {self.tenant}."Application" app ON app.id = ans."applicationId"
WHERE ans.question LIKE 'expense_%'
AND ans.question != 'expensesList' -- Exclude expensesList
AND NOT EXISTS
(SELECT 1
FROM {self.tenant}."Application" child
WHERE child."parentApplicationId" = app.id )
AND ans.question IN
(SELECT expense_line_item
FROM duplicate_invoices)
ORDER BY ans."applicationId",
ans.question)
SELECT elis.id, elis."applicationId" as application_id, elis.question, di.id as application_resource_id
FROM expense_lines elis left join duplicate_invoices di on elis.question = di.expense_line_item
"""
print("Getting current flag info with SQL:", sql)
db_hook = PostgresHook(postgres_conn_id=self.dest_db)
df = db_hook.get_pandas_df(sql)
return df.to_dict(orient="records")[0] if len(df) > 0 else {}
[docs]
def store_data(self, data):
if not data:
# something failed
raise Exception("No data returned from OCR processing")
self.graphi_hook.update_application_resource(
id=self.application_resource_id,
data={
"activity": self.MODEL_ACTIVITIES["AiInvoiceProcessFinished"],
"append": True,
"attributes": {"textExtraction": data},
"activityAttributes": {
"applicationResourceId": self.application_resource_id
},
},
)
[docs]
def post_process(self, context, process_model_output, response):
print(f"Post-processing the model output: {process_model_output}")
print(f"Response from the model: {response}")
total_value = process_model_output["inference_result"]["TOTAL"]
if not total_value or total_value == "":
raise Exception("Error in OCR extraction")
invoice_hash = self.generate_hash_id(process_model_output)
similar_invoices_df = self.identify_similar_invoices(invoice_hash)
print(f"Found {len(similar_invoices_df)} similar invoices")
current_flag_info = self.get_current_flag_info()
##TODO: CAll AI
ai_dedup = self.generate_ai_message(similar_invoices_df, process_model_output)
is_total_exceeds_elis, total_claimed, total_amount = self.is_total_exceeds_elis(
invoice_hash=invoice_hash, total_amount=total_value
)
process_model_output["similarity_hash"] = invoice_hash
process_model_output["similar_invoices"] = similar_invoices_df[
[
"question",
"applicationId",
"id",
"application_resource_id",
"expense_id",
"cost",
]
].to_dict(orient="records")
process_model_output["ai_deduplication_result"] = ai_dedup
process_model_output["is_total_exceeds_elis"] = is_total_exceeds_elis
process_model_output["total_claimed"] = total_claimed
process_model_output["total_amount"] = total_amount
process_model_output["current_flag_info"] = current_flag_info
print(process_model_output)
cleaned_data = self.make_hashable(process_model_output)
self.store_data(cleaned_data)
return cleaned_data