Integrate DSPy into the form generation

.gitignore

@@ -10,9 +10,6 @@ samples/*
 # Logs
 .logs
 
-# Logs
-.logs
-
 # VS Code
 .vscode
 

app.py

@@ -1,10 +1,14 @@
 import os
-from http  import HTTPStatus
+import json
+from http import HTTPStatus
 from dotenv import load_dotenv
 from auth import Token
+from backend.form import FertiliserForm
+from azure.core.exceptions import HttpResponseError
 from werkzeug.utils import secure_filename
-from backend import OCR, GPT, LabelStorage
-from flask import Flask, request, render_template
+from backend import OCR, GPT, LabelStorage, save_text_to_file
+from datetime import datetime
+from flask import Flask, request, render_template, jsonify
 from flask_cors import CORS
 
 # Load environment variables
@@ -19,8 +23,9 @@
 app = Flask(__name__)
 app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
 # CORS configuration limited to the frontend URL
-cors = CORS(app,resources={"*",FRONTEND_URL}) 
+cors = CORS(app, resources={"*", FRONTEND_URL})
 app.config['CORS_HEADERS'] = 'Content-Type'
+
 # Configuration for Azure Form Recognizer
 API_ENDPOINT = os.getenv('AZURE_API_ENDPOINT')
 API_KEY = os.getenv('AZURE_API_KEY')
@@ -35,54 +40,67 @@
 def main_page():
     return render_template('index.html')
 
-# Example request
-# curl -X POST http://localhost:5000/analyze \
-#     -H "Authorization: Basic <your_encoded_credentials>" \
-#     -F "images=@/path/to/image1.jpg" \
-#     -F "images=@/path/to/image2.jpg"
 @app.route('/analyze', methods=['POST'])
 def analyze_document():
-    files = request.files.getlist('images')
-
-    # The authorization scheme is still unsure.
-    #
-    # Current format: user_id:session_id
-    # Initialize a token instance from the request authorization header
-    auth_header = request.headers.get("Authorization")
-    # Currently we are not using the token. It might change in the future.
-    Token(auth_header) if request.authorization else Token()
-
-    # Initialize the storage for the user
-    label_storage = LabelStorage()
-
-    for file in files:
-        if file:
-            filename = secure_filename(file.filename)
-            file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
-            file.save(file_path)
-
-            # Add image to label storage
-            label_storage.add_image(file_path)
-
-    document = label_storage.get_document()
-    if not document:
-        return "No documents to analyze", HTTPStatus.BAD_REQUEST
-
-    result = ocr.extract_text(document=document)
-
-    # Generate form from extracted text
-    # Send the JSON if we have more token.
-    # form = language_model.generate_form(result_json)
-    form = language_model.generate_form(result.content)
-
-    # Clear the label cache
-    label_storage.clear()
-
-    return app.response_class(
-        response=form,
-        status=HTTPStatus.OK,
-        mimetype="application/json"
-    )
+    try:
+        files = request.files.getlist('images')
+
+        if not files:
+            raise ValueError("No files provided for analysis")
+
+        # The authorization scheme is still unsure.
+        auth_header = request.headers.get("Authorization")
+        Token(auth_header) if request.authorization else Token()
+
+        # Initialize the storage for the user
+        label_storage = LabelStorage()
+
+        for file in files:
+            if file:
+                filename = secure_filename(file.filename)
+                file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
+                file.save(file_path)
+                label_storage.add_image(file_path)
+
+        document = label_storage.get_document()
+        result = ocr.extract_text(document=document)
+
+        # Logs the results from document intelligence
+        now = datetime.now()
+        if not os.path.exists('./.logs'):
+            os.mkdir('./.logs')
+        save_text_to_file(result.content, f"./.logs/{now}.md")
+
+        # Generate form from extracted text
+        raw_form = language_model.generate_form(result.content)
+
+        # Logs the results from GPT
+        save_text_to_file(raw_form, f"./.logs/{now}.json")
+
+        # Clear the label cache
+        label_storage.clear()
+
+        # Check the conformity of the JSON
+        form = FertiliserForm(**json.loads(raw_form))
+        return app.response_class(
+            response=form.model_dump_json(indent=2),
+            status=HTTPStatus.OK,
+            mimetype="application/json"
+        )
+    except ValueError as err:
+        return jsonify(error=str(err)), HTTPStatus.BAD_REQUEST
+    except HttpResponseError as err:
+        return jsonify(error=err.message), err.status_code
+    except Exception as err:
+        return jsonify(error=str(err)), HTTPStatus.INTERNAL_SERVER_ERROR
+
+@app.errorhandler(404)
+def not_found(error):
+    return jsonify(error="Not Found"), HTTPStatus.NOT_FOUND
+
+@app.errorhandler(500)
+def internal_error(error):
+    return jsonify(error=str(error)), HTTPStatus.INTERNAL_SERVER_ERROR
 
 if __name__ == "__main__":
-    app.run(host="0.0.0.0", debug=True)
+    app.run(host='0.0.0.0', debug=True)

backend/__init__.py

@@ -1,6 +1,7 @@
 from .label import LabelStorage  # noqa: F401
 from .ocr import OCR  # noqa: F401
 from .gpt import GPT  # noqa: F401
+from .form import FertiliserForm  # noqa: F401
 import requests
 
 def curl_file(url:str, path: str):

backend/form.py

@@ -0,0 +1,58 @@
+from typing import List, Optional
+from pydantic import BaseModel, Field
+
+class NutrientAnalysis(BaseModel):
+    nutrient: str
+    percentage: str
+
+class Micronutrient(BaseModel):
+    name: str
+    percentage: str
+
+class OrganicIngredient(BaseModel):
+    name: str
+    percentage: str
+
+class Specification(BaseModel):
+    humidity: Optional[str] = Field(..., alias='humidity')
+    ph: Optional[str] = Field(..., alias='ph')
+    solubility: str
+
+class FertiliserForm(BaseModel):
+    company_name: Optional[str] = ""
+    company_address: Optional[str] = ""
+    company_website: Optional[str] = ""
+    company_phone_number: Optional[str] = ""
+    manufacturer_name: Optional[str] = ""
+    manufacturer_address: Optional[str] = ""
+    manufacturer_website: Optional[str] = ""
+    manufacturer_phone_number: Optional[str] = ""
+    fertiliser_name: Optional[str] = ""
+    fertiliser_registration_number: Optional[str] = ""
+    fertiliser_lot_number: Optional[str] = ""
+    fertiliser_weight_kg: Optional[str] = None
+    fertiliser_weight_lb: Optional[str] = None
+    fertiliser_density: Optional[str] = None
+    fertiliser_volume: Optional[str] = None
+    warranty: Optional[str] = ""
+    fertiliser_npk: str = Field(..., pattern=r'^(\d+-\d+-\d+)?$')
+    precautions_en: Optional[str] = None
+    instructions_en: List[str] = []
+    micronutrients_en: List[Micronutrient] = []
+    organic_ingredients_en: List[OrganicIngredient] = []
+    inert_ingredients_en: List[str] = []
+    specifications_en: List[Specification] = []
+    cautions_en: Optional[str] = None
+    first_aid_en: Optional[str] = None
+    precautions_fr: Optional[str] = None
+    instructions_fr: List[str] = []
+    micronutrients_fr: List[Micronutrient] = []
+    organic_ingredients_fr: List[OrganicIngredient] = []
+    inert_ingredients_fr: List[str] = []
+    specifications_fr: List[Specification] = []
+    cautions_fr: Optional[str] = None
+    first_aid_fr: Optional[str] = None
+    fertiliser_guaranteed_analysis: List[NutrientAnalysis] = []
+
+    class Config:
+        populate_by_name = True

backend/gpt.py

@@ -1,24 +1,46 @@
 import os
-from openai import AzureOpenAI
+import dspy
 from openai.types.chat.completion_create_params import ResponseFormat
 
 # Constants
 MODELS_WITH_RESPONSE_FORMAT = [
     "ailab-llm"
 ]  # List of models that support the response_format option
 
+class ProduceLabelForm(dspy.Signature):
+    """
+    You are a fertilizer label inspector working for the Canadian Food Inspection Agency. 
+    Your task is to classify all information present in the provided text using the specified keys.
+    Your response should be accurate, formatted in JSON, and contain all the text from the provided text without modifications.
+    """
+
+    text = dspy.InputField(desc="The text of the fertilizer label extracted using OCR.")
+    specification = dspy.InputField(desc="The specification containing the fields to highlight and their requirements.")
+    form = dspy.OutputField(desc="A complete JSON with all fields occupied. Do not return any note or additional text that isn't in the JSON.")
+
 class GPT:
     def __init__(self, api_endpoint, api_key, deployment="ailab-gpt-35-turbo-16k"):
         if not api_endpoint or not api_key:
-            raise ValueError(
-                "API endpoint and key are required to instantiate the GPT class."
-            )
+            raise ValueError("API endpoint and key are required to instantiate the GPT class.")
+
+        # self.model = deployment
+
+        response_format = None
+        if deployment in MODELS_WITH_RESPONSE_FORMAT:
+            response_format = ResponseFormat(type='json_object')
+
+        max_token = 12000
+        if deployment == 'ailab-llm':
+            max_token = 3500
 
-        self.model = deployment
-        self.client = AzureOpenAI(
+
+        self.dspy_client = dspy.AzureOpenAI(
+            api_base=api_endpoint,
             api_key=api_key,
-            azure_endpoint=api_endpoint,  # Your Azure OpenAI resource's endpoint value.
+            deployment_id=deployment,
             api_version="2024-02-01",
+            max_tokens=max_token,
+            response_format=response_format,
         )
 
     def generate_form(self, prompt):
@@ -27,17 +49,9 @@ def generate_form(self, prompt):
         system_prompt = prompt_file.read()
         prompt_file.close()
 
-        response_format = None
-        if self.model in MODELS_WITH_RESPONSE_FORMAT:
-            response_format = ResponseFormat(type='json_object')
+        dspy.configure(lm=self.dspy_client)
+        signature = dspy.ChainOfThought(ProduceLabelForm)
+        prediction = signature(specification=system_prompt, text=prompt)
 
-        response = self.client.chat.completions.create(
-            model=self.model, # model = "deployment_name".
-            messages=[
-                {"role": "system", "content": system_prompt},
-                {"role": "user", "content": prompt}
-            ],
-            response_format=response_format,
-            temperature=0,
-        )
-        return response.choices[0].message.content
+        # print(prediction)
+        return prediction.form