Criando um agente para ler um recibo com LangChain

Criando um agente para ler um recibo com LangChain

Nos últimos tempos tenho sido bastante atingido por notícias sobre LLM-agents e como eles serão responsáveis pelo próximo grande passo das IAs. Especialmente no que se refere ao LangGraph para criação de multi-agent workflows. Desenvolvi algumas coisas e estou bastante animado pra continuar explorando.

LangGraph é uma biblioteca para construir aplicações multi-agents com LLMs, permitindo criar agentes e fluxos de trabalho entre eles. Cada nó desse grafo é um agent capaz de realizar alguma tarefa específica ou múltiplas tarefas, its up to you. Isso significa que, antes de começar com LangGraph, é necessário aprender a criar um agent. É aí que a biblioteca LangChain ajuda bastante. Vou mostrar como criar um agent capaz de transformar a imagem de um recibo de compras em um objeto JSON estruturado.

Transformações com LangChain

No contexto do LangChain, uma Chain é uma estrutura que permite encadear várias transformações ou operações de forma sequencial. Cada etapa da cadeia recebe a saída da etapa anterior como entrada, permitindo assim a composição de transformações complexas a partir de operações mais simples. Isso é particularmente útil em fluxos de dados onde múltiplas etapas de processamento são necessárias. No nosso exemplo, vou criar uma Chain bastante simples responsável por fazer o encode da imagem de entrada.

from langchain.chains import TransformChain
import base64


def load_image(inputs: dict) -> dict:
    image_path = inputs["image_path"]

    def encode_image(image_path):
        with open(image_path, "rb") as image_file:
            return base64.b64encode(image_file.read()).decode("utf-8")

    image_base64 = encode_image(image_path)
    return {"image": image_base64}


load_image_chain = TransformChain(
    input_variables=["image_path"],
    output_variables=["image"],
    transform=load_image
)

Em seguida, vou criar classes que irão representar o output esperado para o recibo. Nesse caso, eu quero algumas informações como:

  1. Descrição do recibo
  2. Data do recibo
  3. Total do recibo
  4. Lista de itens do recibo, em que cada contém nome, valor original, quantidade, desconto, valor final e soma do valor final

Definindo o Schema de Saída

Esse é apenas um exemplo de estruturação da saída, mas você pode estruturar de qualquer outra forma.

from typing import List
from langchain_core.pydantic_v1 import BaseModel, Field
import datetime

class Items(BaseModel):
    name: str = Field(
        description="The name of the item"
    )
    original_value: float = Field(
        description="The original value of the item unit"
    )
    quantity: int = Field(
        description="The quantity of the item"
    )
    discount: float = Field(
        description="The discount on the item"
    )
    final_value: float = Field(
        description="The final value of the item unit"
    )
    final_total_value: float = Field(
        description="The final total value of the item"
    )

class ImageInformation(BaseModel):
    """Information about an image."""
    date: datetime.datetime = Field(
        description="The date of the receipt"
    )
    description: str = Field(
        description="A short description of the receipt"
    )
    items: List[Items] = Field(
        description="The list of items from the receipt"
    )
    total_amount: float = Field(
        description="The total amount of the receipt"
    )

Em posse das classes, eu criei um parser que me permite especificar um esquema JSON arbitrário de forma que o output das LLMs estejam em conformidade com esse esquema. Nesse caso, meu parser é instanciado com a classe recém criada ImageInformation.

from langchain_core.output_parsers import JsonOutputParser

parser = JsonOutputParser(pydantic_object=ImageInformation)

Por enquanto sou capaz de converter a imagem em base64 e tenho um parser que garante meu schema de saída. Entretanto, ainda preciso conectar essas ferramentas.

Para isso, defino uma função que irá receber os parâmetros da imagem e irá invocar uma LLM com esses dados. No meu caso, estou usando o GPT-4o da OpenAI, mas você pode optar por alguma outra LLM. Essa função recebe um dicionário com 2 parâmetros importantes: um prompt para a LLM e uma image já em base64.

from langchain_core.messages import HumanMessage

def image_model(inputs: dict):
    """Invoke model with image and prompt."""
    model = ChatOpenAI(
        temperature=0, model="gpt-4o", max_tokens=1024
    )
    msg = model.invoke(
            [
                HumanMessage(
                    content=[
                        {"type": "text", "text": inputs["prompt"]},
                        {
                            "type": "text",
                            "text": parser.get_format_instructions(),
                        },
                        {
                            "type": "image_url",
                            "image_url": {
                                "url": f"data:image/jpeg;base64,{inputs['image']}"
                            },
                        },
                    ]
                )
            ]
    )
    return msg.content

Conectando via Chain

Essa função irá enviar a imagem para a LLM e retornar sua resposta. Mas ainda falta criar o prompt e definir a chain. A função get_image_informations irá receber o path da imagem e invocar a chain completa para realizar a transformação. Nossa chain será uma sequência com essas etapas:

  1. load_image_chain: recebe um path de imagem e converte para a base64.
  2. image_model: recebe uma imagem em base64 e um prompt para enviar a LLM.
  3. parser: garante que a saída respeita o schema definido.
def get_image_informations(image_path: str) -> dict:
    vision_prompt = """
    Given the image, provide the following information:
        date: the date of the receipt
        description: a short description of the receipt
        itens: the list of itens from the receipt
        total_amount the total amount of the receipt
    """
    vision_chain = load_image_chain | image_model | parser
    print(vision_chain)
    return vision_chain.invoke(
        {"image_path": f"{image_path}", "prompt": vision_prompt}
    )

Testando o Agent

Vamos testar. Baixei uma imagem de recibo que encontrei online e salvei na mesma pasta do projeto.

result = get_image_informations("recibo.png")
print(result)

O resultado, como esperado, é um objeto JSON no schema que definimos.

{
   "date":"2018-03-28T16:31:00",
   "description":"Receipt from Empresa Teste",
   "itens":[
      {
         "item_name":"CERVEJA LONGNECK UN",
         "item_original_value":6.0,
         "item_amount":1,
         "item_discount":0.0,
         "item_final_value":6.0,
         "item_final_total_value":6.0
      },
      {
         "item_name":"GATORADE UN",
         "item_original_value":4.5,
         "item_amount":1,
         "item_discount":0.0,
         "item_final_value":4.5,
         "item_final_total_value":4.5
      },
      {
         "item_name":"MONSTER UN",
         "item_original_value":8.0,
         "item_amount":1,
         "item_discount":0.0,
         "item_final_value":8.0,
         "item_final_total_value":8.0
      },
      {
         "item_name":"REDBULL UN",
         "item_original_value":10.0,
         "item_amount":1,
         "item_discount":0.0,
         "item_final_value":10.0,
         "item_final_total_value":10.0
      },
      {
         "item_name":"SUNDAE UN",
         "item_original_value":6.5,
         "item_amount":1,
         "item_discount":0.0,
         "item_final_value":6.5,
         "item_final_total_value":6.5
      },
      {
         "item_name":"CACHORRO 1 SALSICHA UN",
         "item_original_value":8.0,
         "item_amount":1,
         "item_discount":0.0,
         "item_final_value":8.0,
         "item_final_total_value":8.0
      }
   ],
   "total_amount":43.0
}

Embora bastante simples, esse exemplo mostra o potencial de um agent como ferramenta para compor uma aplicação maior. Esse agent poderia ser um nó em um LangGraph que executa uma ação mais complexa, como enviar essas informações para algum sistema ou mesmo gerar algum insight a partir dela.