File size: 6,874 Bytes
571ff03
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
from __future__ import annotations

import os
import base64
import mimetypes
import argparse
from pathlib import Path
from typing import List, Dict, Any, Optional
from dotenv import load_dotenv
from huggingface_hub import InferenceClient

# Load environment variables (e.g., HF_TOKEN) from a .env file
load_dotenv()

# This is the system prompt you provided
AGENT_SYSTEM_PROMPT = """You are a general AI assistant. I will ask you a question. Report your thoughts, and finish your answer with the following template: FINAL ANSWER: [YOUR FINAL ANSWER]. YOUR FINAL ANSWER should be a number OR as few words as possible OR a comma separated list of numbers and/or strings. If you are asked for a number, don't use comma to write your number neither use units such as $ or percent sign unless specified otherwise. If you are asked for a string, don't use articles, neither abbreviations (e.g. for cities), and write the digits in plain text unless specified otherwise. If you are asked for a comma separated list, apply the above rules depending of whether the element to be put in the list is a number or a string."""


class ImageAgent:
    """

    Encapsulated agent for answering questions about an image.



    Usage:

        # Assumes HF_TOKEN is in your environment

        agent = ImageAgent()

        

        # Ask a question with an image

        question = "What is in this image?"

        image_path = "/path/to/your/image.png"

        answer = agent.answer(question, image_path)

        

        # Or use the callable shortcut

        answer = agent(question, image_path)

    """

    def __init__(

        self,

        model_id: Optional[str] = None,

        provider: Optional[str] = "novita",

        temperature: float = 0.2,

        api_key: Optional[str] = None,

    ):
        """

        Initializes the ImageAgent.



        Args:

            model_id: The repository ID of the model to use.

            provider: The provider for the InferenceClient.

            temperature: The generation temperature.

            api_key: The API key. Defaults to os.environ["HF_TOKEN"].

        """
        self.model_id = model_id or "meta-llama/Llama-4-Scout-17B-16E-Instruct"
        self.temperature = temperature
        
        # Get API key from arg or environment, raising an error if missing
        self.api_key = api_key or os.environ.get("HF_TOKEN")
        if not self.api_key:
            raise ValueError(
                "API key not found. Please set the HF_TOKEN environment variable "
                "or pass it as 'api_key' to the ImageAgent."
            )

        # Initialize the client
        self.client = InferenceClient(
            provider=provider,
            api_key=self.api_key,
        )

    # ---- Public API ----

    def answer(self, question: str, image_path: str) -> str:
        """

        Run the agent to answer a question about an image.



        Args:

            question: The text prompt to ask the model.

            image_path: The local file path to the image.



        Returns:

            The parsed "Final Answer" string from the model.

        """
        # 1. Convert image to data URL
        try:
            image_data_url = self._to_data_url(image_path)
        except FileNotFoundError:
            return f"Error: Image file not found at {image_path}"
        except Exception as e:
            return f"Error processing image: {e}"

        # 2. Build the messages payload
        messages = self._build_messages(question, image_data_url)

        # 3. Call the API
        try:
            completion = self.client.chat.completions.create(
                model=self.model_id,
                messages=messages,
                temperature=self.temperature,
            )
            response_text = completion.choices[0].message.content
        except Exception as e:
            return f"Error calling API: {e}"

        # 4. Parse and return the final answer
        return self._parse_answer(response_text)

    def __call__(self, question: str, image_path: str) -> str:
        """Callable shortcut for self.answer()."""
        return self.answer(question, image_path)

    # ---- Internals ----

    def _to_data_url(self, path: str) -> str:
        """Converts a local file path to a base64 data URL."""
        p = Path(path)
        if not p.exists():
            raise FileNotFoundError(f"No file found at {path}")
            
        mime = mimetypes.guess_type(str(p))[0] or "image/png"
        b64 = base64.b64encode(p.read_bytes()).decode("utf-8")
        return f"data:{mime};base64,{b64}"

    def _build_messages(self, question: str, image_data_url: str) -> List[Dict[str, Any]]:
        """Constructs the message payload for the API."""
        return [
            {
                "role": "system",
                "content": [
                    {"type": "text", "text": AGENT_SYSTEM_PROMPT},
                ],
            },
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": question},
                    {
                        "type": "image_url",
                        "image_url": {"url": image_data_url},
                    },
                ],
            },
        ]

    def _parse_answer(self, response: str) -> str:
        """Extracts the text after 'FINAL ANSWER: '."""
        try:
            # Use rpartition for robustness, splits on the *last* occurrence
            _before, _marker, final_answer = response.rpartition("FINAL ANSWER: ")
            
            if _marker:
                return final_answer.strip()
            else:
                # If marker isn't found, return the whole response with a warning
                print("Warning: Could not find 'FINAL ANSWER:' marker. Returning full response.")
                return response.strip()

        except Exception as e:
            print(f"Error parsing response: {e}")
            return response.strip()


# ---- CLI Entrypoint ----
def main():
    """

    Allows running the agent from the command line.

    

    Example:

       python image_agent.py "What move is this?" "path/to/my_image.png"

    """
    parser = argparse.ArgumentParser(description="Class-based agent for image questions.")
    parser.add_argument("question", type=str, help="Your question (quoted).")
    parser.add_argument("image_path", type=str, help="Path to the image file.")
    args = parser.parse_args()

    try:
        image_agent = ImageAgent()
        answer = image_agent.answer(args.question, args.image_path)
        print(answer)
    except Exception as e:
        print(f"An error occurred: {e}")

if __name__ == "__main__":
    main()