File size: 7,614 Bytes
7850a69 |
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 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 |
import anthropic
import os
from remittance_pdf_processing_types import InvoiceNumbers, PaymentAmount
from remittance_pdf_processing_utils import remittance_logger, remove_duplicate_lists
from anthropic.types import ContentBlock, ImageBlockParam
def extract_invoice_numbers_with_anthropic_ai(base64_images: list[str], multi_hop: bool = False) -> list[InvoiceNumbers]:
"""
Extracts invoice numbers from one or more base64-encoded images using Anthropic's Claude 3.5 Sonnet model.
Args:
base64_images (list[str]): A list of base64-encoded image strings.
multi_hop (bool): Whether to use multi-hop processing.
Returns:
list[InvoiceNumbers]: A list containing lists of extracted invoice numbers.
"""
if multi_hop:
return extract_invoice_numbers_with_anthropic_ai_multi_hop(base64_images)
else:
return extract_invoice_numbers_with_anthropic_ai_single_hop(base64_images)
def extract_invoice_numbers_with_anthropic_ai_single_hop(base64_images: list[str]) -> list[InvoiceNumbers]:
client = anthropic.Anthropic(
api_key=os.environ.get("ANTHROPIC_API_KEY"),
)
content: list[ContentBlock] = [
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": image
}
} for image in base64_images
]
message = client.messages.create(
model="claude-3-5-sonnet-20240620",
max_tokens=1024,
temperature=0,
system="Given the remittance letter images, extract all invoice numbers. Respond with a comma-separated list only.",
messages=[
{
"role": "user",
"content": content
}
]
)
remittance_logger.debug(f'Anthropic (raw) response: {message.content}')
invoice_numbers = parse_anthropic_response(message.content[0].text)
return [invoice_numbers]
def extract_invoice_numbers_with_anthropic_ai_multi_hop(base64_images: list[str]) -> list[InvoiceNumbers]:
# First hop: Extract column headers
column_headers = extract_column_headers_from_images(base64_images)
remittance_logger.debug(f"Extracted column headers: {column_headers}")
# Second hop: Extract invoice numbers for each column (up to 3 columns)
all_invoice_numbers = []
for column_name in column_headers[:3]:
invoice_numbers = extract_invoice_numbers_for_column_from_images(base64_images, column_name)
remittance_logger.debug(f"Extracted invoice numbers for column '{column_name}': {invoice_numbers}")
if invoice_numbers: # Only add non-empty lists
all_invoice_numbers.append(invoice_numbers)
# Remove duplicate lists using the utility function
unique_invoice_numbers = remove_duplicate_lists(all_invoice_numbers)
return unique_invoice_numbers
def extract_column_headers_from_images(base64_images: list[str]) -> list[str]:
client = anthropic.Anthropic(
api_key=os.environ.get("ANTHROPIC_API_KEY"),
)
content: list[ContentBlock] = [
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": image
}
} for image in base64_images
]
message = client.messages.create(
model="claude-3-5-sonnet-20240620",
max_tokens=1024,
temperature=0,
system="Given the remittance letter images, extract all column header names that could contain invoice numbers. Respond with a comma-separated list only.",
messages=[
{
"role": "user",
"content": content
}
]
)
remittance_logger.debug(f'Anthropic (raw) response for column headers: {message.content}')
return parse_anthropic_response(message.content[0].text)
def extract_invoice_numbers_for_column_from_images(base64_images: list[str], column_name: str) -> InvoiceNumbers:
client = anthropic.Anthropic(
api_key=os.environ.get("ANTHROPIC_API_KEY"),
)
content: list[ContentBlock] = [
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": image
}
} for image in base64_images
]
message = client.messages.create(
model="claude-3-5-sonnet-20240620",
max_tokens=1024,
temperature=0,
system=f"Given the remittance letter images, extract all invoice numbers from the column '{column_name}'. Respond with a comma-separated list only.",
messages=[
{
"role": "user",
"content": content
}
]
)
remittance_logger.debug(f'Anthropic (raw) response for invoice numbers in column {column_name}: {message.content}')
return parse_anthropic_response(message.content[0].text)
def parse_anthropic_response(response: str) -> list[str]:
"""
Parses the response from Claude 3.5 Sonnet model and extracts a list of items.
Args:
response (str): The response string from Claude 3.5 Sonnet model.
Returns:
list[str]: A list of extracted items (invoice numbers or column headers).
"""
return [item.strip() for item in response.split(',') if item.strip()]
def extract_invoice_numbers_from_single_base64_image(base64_image: str, multi_hop: bool = False) -> list[InvoiceNumbers]:
"""
Extracts invoice numbers from a single base64-encoded image using Anthropic's Claude 3.5 Sonnet model.
Args:
base64_image (str): The base64-encoded image string.
multi_hop (bool): Whether to use multi-hop processing.
Returns:
list[InvoiceNumbers]: A list containing lists of extracted invoice numbers.
"""
return extract_invoice_numbers_with_anthropic_ai([base64_image], multi_hop)
def extract_invoice_numbers_from_multi_page_images(base64_images: list[str], multi_hop: bool = False) -> list[InvoiceNumbers]:
"""
Extracts invoice numbers from multiple base64-encoded images using Anthropic's Claude 3.5 Sonnet model.
Args:
base64_images (list[str]): A list of base64-encoded image strings.
multi_hop (bool): Whether to use multi-hop processing.
Returns:
list[InvoiceNumbers]: A list containing lists of extracted invoice numbers.
"""
return extract_invoice_numbers_with_anthropic_ai(base64_images, multi_hop)
def extract_payment_amounts_with_anthropic_ai(base64_images: list[str]) -> list[PaymentAmount]:
"""
Extracts payment amounts from one or more base64-encoded images using Anthropic's Claude 3.5 Sonnet model.
Args:
base64_images (list[str]): A list of base64-encoded image strings.
Returns:
list[PaymentAmount]: A list containing extracted payment amounts.
"""
client = anthropic.Anthropic(
api_key=os.environ.get("ANTHROPIC_API_KEY"),
)
# Prepare the message content
content = []
for image in base64_images:
content.append({
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": image
}
})
# Call the Anthropic API
message = client.messages.create(
model="claude-3-5-sonnet-20240620",
max_tokens=1024,
temperature=0,
system="You are a precise payment amount extractor. Given remittance letter images, extract only the total payment amount. Respond with the numerical amount only, including any decimal places and currency symbols if present. Do not include any additional text or explanations.",
messages=[
{
"role": "user",
"content": content
}
]
)
remittance_logger.debug(f'Anthropic (raw) response for payment amount: {message.content}')
# assert(isinstance(message.content, anthropic.TextBlock))
# Parse the response
payment_amount = parse_anthropic_payment_response(message.content[0].text)
return payment_amount
def parse_anthropic_payment_response(response: str) -> list[PaymentAmount]:
"""
Parses the response from Claude 3.5 Sonnet model and extracts the payment amount.
Args:
response (str): The response string from Claude 3.5 Sonnet model.
Returns:
list[PaymentAmount]: A list containing the extracted payment amount.
"""
# Strip whitespace and return as a single-item list
return [response.strip()] |