File size: 3,886 Bytes
0ff3cf2 70eea9d 0ff3cf2 ee0c758 | 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 | import base64
from crewai import Agent, Task, Crew, Process
from crewai.tools import tool
from langchain_openai import ChatOpenAI
from openai import OpenAI
## client for vision
vision_client = OpenAI()
@tool("Invoice Image Reader")
def read_invoice_image(image_path:str)->str:
"""
Read the image and extract the raw text from it
"""
with open(image_path,'rb') as f:
image_base64 = base64.b64encode(f.read()).decode('utf-8')
response = vision_client.responses.create(
model="gpt-4.1-mini",
input=[{
"role": "user",
"content": [
{"type": "input_text", "text": ("Extract the vendor name, tax id , invoice number"
"invoice date, items table (descriptiopn, quantity, net price ),"
"and total gross from this invoice"
),
},
{
"type": "input_image",
"image_url": f"data:image/jpeg;base64,{image_base64}",
},
],
}],
)
return response.output_text
def extract_invoice(image_path:str)->str:
"""
Extract the information in JSON structure
"""
llm = ChatOpenAI(model = "gpt-4.1-mini", temperature=0)
### Agent 1 OCR specialist
visual_reader = Agent(
role="OCR Specialist",
goal= "Extract the invoice data fram images",
backstory= ("you cann't see the images directly."
"you must always use the Invoice Image Reader tool"),
llm = llm,
tools = [read_invoice_image],
verbose=True
)
#### Agent 2 JSON
json_architect = Agent(
role="Data Engineer",
goal= "Convert extracted invoice text to structured JSON",
backstory= "You normalize numbers and dates and output strict to JSON",
llm = llm,
verbose=True
)
## Task 1 extraction task
extraction_task = Task (
description = (
f"Use the Invoice Reader tool to read the invoice image "
f"at the path '{image_path}'. extract the vendor name , tax id, invoice number, "
f"date, item rows, total gross"
),
expected_output = "structured invoice text",
agent = visual_reader
)
### Task 2 JSON
formatting_task = Task (
description = (
"Convert the extracted invoice text into JSON. \n\n"
"{\n"
"'invoice_no' : str, \n"
"'date': 'YYYY-MM-DD',\n"
" 'vendor':{'name':str,'tax_id':str},\n "
" 'items' :[{'description':str,'quantity':float,'net_price':float}] ,\n"
"}\n\n"
"Rules:\n"
"- if missing value use null \n"
" - Output only JSON"
),
expected_output = "Valid JSON Only",
agent = json_architect,
context = [extraction_task]
)
crew = Crew(
agents = [visual_reader,json_architect],
tasks = [extraction_task,formatting_task],
process = Process.sequential,
verbose =True)
return crew.kickoff()
|