How do I optimize a multimodal model?

[phukon]

Pardon me for I'm a noobie in the AI space. So I made a custom model for processing multimodal queries. However, I now want to optimize the program that uses the custom model such that for a given image, I can make it output a specific set of answers.

If I use an optimizer like MIPROv2 using a dataset that includes the base64 image string, the few shot examples (the compiled progam) will also have the base64 image string in the prompt, and this is useless. How should I go about associating an image with each example, for the LLM to have context for what input images were the set of example outputs generated?

Custom DSPy multimodal model for processing multimodal tasks

    import requests
    import re
    from dsp import LM
    from utils.dspy.tracer import CustomLangfuseTracker
    from portkey_ai import PORTKEY_GATEWAY_URL, createHeaders
    from langsmith import traceable
    from functools import wraps
    
    
    class GPTVision(LM):
        def __init__(self, model, api_key, portkey_api_key):
            super().__init__("gpt-4o")
            self.model = model
            self.api_key = api_key
            self.portkey_api_key = portkey_api_key
            self.provider = "openai"
            self.history = []
            self.request_price = 0
            self.base_url = PORTKEY_GATEWAY_URL + "/chat/completions"
            self.tracker = CustomLangfuseTracker()
            self.session_id = None
            self.function_type = None
            self.update_portkey_headers()
    
        def update_metadata(self, session_id=None, function_type=None):
            if session_id:
                self.session_id = session_id
            if function_type:
                self.function_type = function_type
            self.update_portkey_headers()
    
        def update_portkey_headers(self):
            metadata = {}
            if self.session_id:
                metadata["session_id"] = self.session_id
            self.portkey_headers = createHeaders(api_key=self.portkey_api_key, metadata=metadata, provider="openai")
    
        def filter_prompt(self, prompt):
            pattern = r"^Base 64 Image: .*"
            matches = re.findall(pattern, prompt, re.MULTILINE)
            base64_image = None
    
            if matches:
                base64_image = matches[1].replace("Base 64 Image: ", "")
                for match in matches:
                    prompt = prompt.replace(f"\n{match}\n", "")
    
            return prompt, base64_image
    
        @property
        def traceable_basic_request(self):
            @traceable(
                run_type="llm",
                metadata={"ls_provider": "openai", "ls_model_name": "gpt-4o(vision)", "session_id": self.session_id, "function_type": self.function_type},
            )
            @wraps(self.basic_request)
            def wrapped(*args, **kwargs):
                return self.basic_request(*args, **kwargs)
    
            return wrapped
    
        def basic_request(self, prompt, base64_image=None, **kwargs):
            headers = {"Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}", **self.portkey_headers}
    
            if base64_image:
                data = {
                    **kwargs,
                    "model": "gpt-4o",
                    "messages": [
                        {
                            "role": "user",
                            "content": [
                                {"type": "text", "text": prompt},
                                {"type": "image_url", "image_url": {"url": base64_image}},
                            ],
                        }
                    ],
                    "max_tokens": 4096,
                }
            else:
                data = {**kwargs, "model": "gpt-4o-mini", "messages": [{"role": "user", "content": prompt}], "max_tokens": 8000}
           
            response = requests.post(self.base_url, headers=headers, json=data)
            response = response.json()
            self.history.append({"prompt": prompt, "response": response, "kwargs": kwargs})
    
            self.request_price = self.calculate_cost(response)
            self.tracker.call(i=prompt, o=response, name="custom-vision-model")
            return response
    
        def calculate_cost(self, response):
            input_token_cost_per_million = 5.00
            output_token_cost_per_million = 15.00
    
            prompt_tokens = response["usage"]["prompt_tokens"]
            completion_tokens = response["usage"]["completion_tokens"]
    
            input_cost = (prompt_tokens / 1_000_000) * input_token_cost_per_million
            output_cost = (completion_tokens / 1_000_000) * output_token_cost_per_million
    
            total_cost = input_cost + output_cost
            return total_cost
    
        def __call__(self, prompt, only_completed=True, return_sorted=False, **kwargs):
            filtered_prompt, base64_image = self.filter_prompt(prompt)
            responses = self.traceable_basic_request(filtered_prompt, base64_image, **kwargs)
            completions = [choice["message"]["content"] for choice in responses["choices"]]
            return completions

This is the signature that I use with the program:

class PredictPropertySignature(dspy.Signature):
    event_name = dspy.InputField(desc="Name of the event to be used in a tracking plan, 1-3 word separated by underscores")
    outer_html = dspy.InputField(desc="the outer html of the target where the click event was registered")
    referer_url = dspy.InputField(desc="the url where the previous event occured in the flow")
    current_page_url = dspy.InputField(desc="the url where this event occured")
    base64_image = dspy.InputField()
    answer = dspy.OutputField(
        desc="""[
    {
      "property_name": "some_property_name",
      "data_type": "some_data_type",
      "sample_value": "some_sample_value",
      "description": "some_description"
    }
    ]"""
    )

And this is how I use the signature with the DSPy program meant to process multimodal queries:

class PredictPropertyModule(dspy.Module):
    def __init__(self):
        context = fetch_context("properties")
        # 👇🏽 the prompt that instructs the model to use the provided image to perform the task
        PredictPropertySignature.__doc__ = predict_property_signature_prompt.format(additional_context=json.dumps(context, indent=2))
        self.predictor = dspy.Predict(PredictPropertySignature)

    def forward(self, event_name, base64_image, outer_html, referer_url, current_page_url):
        result = self.predictor(
            event_name=event_name, base64_image=base64_image, outer_html=outer_html, referer_url=referer_url, current_page_url=current_page_url
        )
        enforced_properties = property_list_format_enforcer(result.answer)
        return dspy.Prediction(answer=enforced_properties)

[tejasunku]

I'm going to work on something similar myself. I don't have any code, but from my research, it seems like you also need to create a custom optimizer as well. You shouldn't need to do anything fancy; rather, you just need to use an existing optimizer for the prompt. ChatGPT gave me this, though I'm not sure it will work:

Got it! Here’s the complete code, starting from the GPTVision subclass, including the PredictPropertySignature, PredictPropertyModule, and the custom Optimizer, all in one place.

1. GPTVision Subclass

import requests
import re
from dsp import LM
from utils.dspy.tracer import CustomLangfuseTracker
from portkey_ai import PORTKEY_GATEWAY_URL, createHeaders
from langsmith import traceable
from functools import wraps

class GPTVision(LM):
    def __init__(self, model, api_key, portkey_api_key):
        super().__init__("gpt-4o")
        self.model = model
        self.api_key = api_key
        self.portkey_api_key = portkey_api_key
        self.provider = "openai"
        self.history = []
        self.request_price = 0
        self.base_url = PORTKEY_GATEWAY_URL + "/chat/completions"
        self.tracker = CustomLangfuseTracker()
        self.session_id = None
        self.function_type = None
        self.update_portkey_headers()

    def update_metadata(self, session_id=None, function_type=None):
        if session_id:
            self.session_id = session_id
        if function_type:
            self.function_type = function_type
        self.update_portkey_headers()

    def update_portkey_headers(self):
        metadata = {}
        if self.session_id:
            metadata["session_id"] = self.session_id
        self.portkey_headers = createHeaders(api_key=self.portkey_api_key, metadata=metadata, provider="openai")

    def filter_prompt(self, prompt):
        pattern = r"^Base 64 Image: .*"
        matches = re.findall(pattern, prompt, re.MULTILINE)
        base64_image = None

        if matches:
            base64_image = matches[1].replace("Base 64 Image: ", "")
            for match in matches:
                prompt = prompt.replace(f"\n{match}\n", "")

        return prompt, base64_image

    @property
    def traceable_basic_request(self):
        @traceable(
            run_type="llm",
            metadata={"ls_provider": "openai", "ls_model_name": "gpt-4o(vision)", "session_id": self.session_id, "function_type": self.function_type},
        )
        @wraps(self.basic_request)
        def wrapped(*args, **kwargs):
            return self.basic_request(*args, **kwargs)

        return wrapped

    def basic_request(self, prompt, base64_image=None, **kwargs):
        headers = {"Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}", **self.portkey_headers}

        if base64_image:
            data = {
                **kwargs,
                "model": "gpt-4o",
                "messages": [
                    {
                        "role": "user",
                        "content": [
                            {"type": "text", "text": prompt},
                            {"type": "image_url", "image_url": {"url": base64_image}},
                        ],
                    }
                ],
                "max_tokens": 4096,
            }
        else:
            data = {**kwargs, "model": "gpt-4o-mini", "messages": [{"role": "user", "content": prompt}], "max_tokens": 8000}
        
        response = requests.post(self.base_url, headers=headers, json=data)
        response = response.json()
        self.history.append({"prompt": prompt, "response": response, "kwargs": kwargs})

        self.request_price = self.calculate_cost(response)
        self.tracker.call(i=prompt, o=response, name="custom-vision-model")
        return response

    def calculate_cost(self, response):
        input_token_cost_per_million = 5.00
        output_token_cost_per_million = 15.00

        prompt_tokens = response["usage"]["prompt_tokens"]
        completion_tokens = response["usage"]["completion_tokens"]

        input_cost = (prompt_tokens / 1_000_000) * input_token_cost_per_million
        output_cost = (completion_tokens / 1_000_000) * output_token_cost_per_million

        total_cost = input_cost + output_cost
        return total_cost

    def __call__(self, prompt, only_completed=True, return_sorted=False, **kwargs):
        filtered_prompt, base64_image = self.filter_prompt(prompt)
        responses = self.traceable_basic_request(filtered_prompt, base64_image, **kwargs)
        completions = [choice["message"]["content"] for choice in responses["choices"]]
        return completions

2. PredictPropertySignature with Editable Flags

class PredictPropertySignature(dspy.Signature):
    event_name = dspy.InputField(desc="Name of the event to be used in a tracking plan, 1-3 words separated by underscores", editable=False)
    outer_html = dspy.InputField(desc="The outer HTML of the target where the click event was registered", editable=False)
    referer_url = dspy.InputField(desc="The URL where the previous event occurred in the flow", editable=False)
    current_page_url = dspy.InputField(desc="The URL where this event occurred", editable=False)
    base64_image = dspy.InputField(editable=False)  # Ensure the image isn't optimized
    prompt = dspy.InputField(desc="The prompt to be optimized", editable=True)  # Add the prompt here for optimization
    answer = dspy.OutputField(
        desc="""[
    {
      "property_name": "some_property_name",
      "data_type": "some_data_type",
      "sample_value": "some_sample_value",
      "description": "some_description"
    }
    ]""", editable=True  # Only optimize this field if needed
    )

3. PredictPropertyModule (Unchanged)

class PredictPropertyModule(dspy.Module):
    def __init__(self):
        context = fetch_context("properties")
        # 👇🏽 the prompt that instructs the model to use the provided image to perform the task
        PredictPropertySignature.__doc__ = predict_property_signature_prompt.format(additional_context=json.dumps(context, indent=2))
        self.predictor = dspy.Predict(PredictPropertySignature)

    def forward(self, event_name, base64_image, outer_html, referer_url, current_page_url, prompt):
        result = self.predictor(
            event_name=event_name,
            base64_image=base64_image,
            outer_html=outer_html,
            referer_url=referer_url,
            current_page_url=current_page_url,
            prompt=prompt  # Include the prompt in the forward method
        )
        enforced_properties = property_list_format_enforcer(result.answer)
        return dspy.Prediction(answer=enforced_properties)

4. CustomOptimizer Class

from dsp import Optimizer
from functools import wraps

class CustomOptimizer(Optimizer):
    def __init__(self, vision_model: GPTVision, signature: PredictPropertySignature, module: PredictPropertyModule, base_optimizer):
        self.vision_model = vision_model
        self.signature = signature
        self.module = module
        self.base_optimizer = base_optimizer

    def optimize(self, input_data, num_iterations=100):
        # Ensure all fields are passed through, but only optimize the editable ones
        fields = {key: value for key, value in input_data.items()}

        # Step 2: Pass all fields (including the prompt) through the PredictPropertyModule
        initial_prediction = self.module.forward(
            event_name=fields.get("event_name"),
            base64_image=fields.get("base64_image"),
            outer_html=fields.get("outer_html"),
            referer_url=fields.get("referer_url"),
            current_page_url=fields.get("current_page_url"),
            prompt=fields.get("prompt")  # Optimize the prompt
        )
        
        # Step 3: Use the base optimizer to optimize the prompt and answer
        optimized_prediction = self.base_optimizer.optimize(
            initial_prediction.answer, num_iterations=num_iterations
        )
        
        # Step 4: Return the optimized prediction, keeping non-editable fields intact
        return optimized_prediction
    
    # Optionally wrap the method for traceability
    @wraps(optimize)
    def traceable_optimize(self, *args, **kwargs):
        return self.optimize(*args, **kwargs)

5. Example Usage

# Assuming MIPROv2 or any other optimizer is already available
mipro_optimizer = MIPROv2()

# Initialize the vision model, signature, and module
vision_model = GPTVision(model="gpt-4o", api_key="your_openai_api_key", portkey_api_key="your_portkey_api_key")
signature = PredictPropertySignature()
module = PredictPropertyModule()

# Create the custom optimizer using the base optimizer (MIPROv2 in this case)
custom_optimizer = CustomOptimizer(vision_model, signature, module, base_optimizer=mipro_optimizer)

# Sample input data
input_data = {
    'event_name': 'click_event',
    'base64_image': 'base64_encoded_string',
    'outer_html': '<div>...</div>',
    'referer_url': 'https://previous.page',
    'current_page_url': 'https://current.page',
    'prompt': 'Please extract properties from this image.'  # The prompt to be optimized
}

# Optimize
optimized_output = custom_optimizer.optimize(input_data)

Summary:

• The GPTVision model processes both text and images, using the filter_prompt method to separate image and text content.
• The PredictPropertySignature includes

both the prompt (marked as editable) and other necessary fields (marked as non-editable).

• The PredictPropertyModule passes all inputs to the predictor, with special attention to the prompt.
• The CustomOptimizer ensures that all fields are passed but only optimizes the fields marked as editable (like the prompt).

Let me know if this works for your setup!