Artemis-Eapil/artemis/agents/image_processor/image_processor.py

# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Universal Multi-Model Image Processor for Artemis."""

import base64
import json
from pathlib import Path
import shutil
import time
import uuid

from langchain_core.messages import (
    BaseMessage,
    HumanMessage,
    SystemMessage,
    ToolMessage,
)

from artemis.config import settings
from artemis.context import ArtemisContext
from artemis.core.tool_declaration import ToolDeclaration
from artemis.data_engine.trace import TraceSpan, trace
from artemis.services.llm import get_llm
from artemis.utils.logger import get_logger
from artemis.utils.python_executor import PythonExecutor

logger = get_logger(__name__)

EXECUTE_PYTHON_TOOL = ToolDeclaration(
    name="execute_python",
    description=(
        "Executes the provided Python code within an isolated"
        " Jupyter environment and returns the resulting stdout,"
        " stderr, and traceback outputs."
    ),
    parameters={
        "type": "object",
        "properties": {
            "code": {
                "type": "string",
                "description": "The Python code to execute",
            }
        },
        "required": ["code"],
    },
)

SUBMIT_RESULT_TOOL = ToolDeclaration(
    name="submit_result",
    description=("Submits a summary of the modifications to successfully complete the task."),
    parameters={
        "type": "object",
        "properties": {
            "summary": {
                "type": "string",
                "description": "Summary of what you found or changed.",
            }
        },
        "required": ["summary"],
    },
)


class ImageProcessor:
    def __init__(self, ctx: ArtemisContext):
        self.ctx = ctx

    @trace(type="agent", name="image_processor")
    async def run(self, instruction: str, target_image_path: str) -> dict:
        try:
            llm = get_llm(self.ctx, name="image_processor")
        except Exception:
            llm = get_llm(self.ctx, name="operator")

        base_dir = settings.TRACES_PATH
        image_processor_dir = base_dir / "images" / "image_processor"
        image_processor_dir.mkdir(parents=True, exist_ok=True)
        intermediate_artifact_save_path = image_processor_dir / f"intermediates_{int(time.time())}"
        intermediate_artifact_save_path.mkdir(parents=True, exist_ok=True)
        intermediate_artifact_save_path_str = str(intermediate_artifact_save_path.resolve())

        max_iterations = 5

        prompt_path = Path(__file__).parent / "image_processor.md"
        prompt_template = prompt_path.read_text(encoding="utf-8")
        prompt_template = (
            prompt_template.replace("{target_image_path}", target_image_path)
            .replace("{instruction}", instruction)
            .replace(
                "{intermediate_artifact_save_path}",
                intermediate_artifact_save_path_str,
            )
            .replace("{max_iterations}", str(max_iterations))
        )

        tools_declaration = [EXECUTE_PYTHON_TOOL, SUBMIT_RESULT_TOOL]

        with open(target_image_path, "rb") as f:
            img_bytes = f.read()
        img_b64 = base64.b64encode(img_bytes).decode("utf-8")

        messages: list[BaseMessage] = [
            SystemMessage(content=prompt_template),
            HumanMessage(
                content=[
                    {"type": "text", "text": "Here is the target image. Begin writing your code."},
                    {
                        "type": "image_url",
                        "image_url": {"url": f"data:image/jpeg;base64,{img_b64}"},
                    },
                ]
            ),
        ]

        img_0_path = intermediate_artifact_save_path / "img_0.jpg"
        try:
            shutil.copy2(target_image_path, img_0_path)
        except Exception as e:
            logger.error(f"Failed to copy original screenshot to intermediate pool: {e}")
            img_0_path = Path(target_image_path)

        self.intermediate_image_pool = {
            "img_0": {
                "image_id": "img_0",
                "path": str(img_0_path.resolve()),
                "turn_id": 0,
                "transform": {
                    "offset_x": 0.0,
                    "offset_y": 0.0,
                    "scale_x": 1.0,
                    "scale_y": 1.0,
                },
                "annotations": {},
            }
        }

        pool_json_path = intermediate_artifact_save_path / "intermediate_transforms.json"
        sync_pool = {
            "img_0": {
                "image_id": "img_0",
                "path": self.intermediate_image_pool["img_0"]["path"],
                "transform": self.intermediate_image_pool["img_0"]["transform"],
                "annotations": {},
            }
        }
        with open(pool_json_path, "w") as f:
            json.dump(sync_pool, f)

        executor = PythonExecutor(session_dir=image_processor_dir)
        iterations = 0
        final_result = None

        try:
            while iterations < max_iterations:
                iterations += 1
                logger.info(f"ImageProcessor Iteration {iterations}")

                # Inject newly created intermediate images from last turn
                if self.intermediate_image_pool:
                    image_blocks = []
                    last_turn_id = iterations - 1
                    for img_id, entry in self.intermediate_image_pool.items():
                        if entry["turn_id"] == last_turn_id:
                            try:
                                path = Path(entry["path"])
                                if path.exists():
                                    b64_data = base64.b64encode(path.read_bytes()).decode("utf-8")
                                    is_output = entry.get("is_output", False)
                                    label = (
                                        f"img_id: {img_id} (Output)"
                                        if is_output
                                        else f"img_id: {img_id}"
                                    )
                                    image_blocks.append(
                                        {
                                            "type": "image_url",
                                            "image_url": {
                                                "url": f"data:image/jpeg;base64,{b64_data}"
                                            },
                                        }
                                    )
                                    image_blocks.append({"type": "text", "text": f"\n{label}\n"})
                            except Exception as e:
                                logger.error(
                                    f"Failed to read intermediate image {entry['path']}: {e}"
                                )

                    if image_blocks:
                        if (
                            messages
                            and isinstance(messages[-1], HumanMessage)
                            and isinstance(messages[-1].content, list)
                        ):
                            messages[-1].content.extend(image_blocks)
                        else:
                            messages.append(HumanMessage(content=image_blocks))

                if iterations == max_iterations - 1:
                    warning_msg = (
                        "\n[WARNING] This is your next-to-last iteration. You"
                        " must complete your task and call `submit_result` now"
                        " or in the next turn, as no further code execution"
                        " will be allowed after that."
                    )
                    messages.append(HumanMessage(content=warning_msg))
                elif iterations == max_iterations:
                    warning_msg = (
                        "\n[WARNING] This is your final iteration. You MUST"
                        " call `submit_result` now to complete the task. No"
                        " other tools are available."
                    )
                    messages.append(HumanMessage(content=warning_msg))

                current_tools = tools_declaration
                if iterations == max_iterations:
                    current_tools = [t for t in tools_declaration if t.name == "submit_result"]

                bound_llm = llm.bind_tools(current_tools)

                with TraceSpan(name="image_processor_call", ctx=self.ctx):
                    response = await bound_llm.ainvoke(messages)

                tool_calls = response.tool_calls or []
                if not tool_calls:
                    break

                messages.append(response)

                submit_called = False
                for tc in tool_calls:
                    name = tc["name"].split(":")[-1] if ":" in tc["name"] else tc["name"]
                    args = tc.get("args") or {}
                    tc_id = tc.get("id") or str(uuid.uuid4())

                    if name == "submit_result":
                        final_result = args
                        submit_called = True
                        messages.append(
                            ToolMessage(
                                tool_call_id=tc_id,
                                name=name,
                                content=json.dumps({"status": "submitted"}),
                                status="success",
                            )
                        )
                        break

                    elif name == "execute_python":
                        code = args.get("code", "")
                        logger.info("Executing Python code via PythonExecutor")
                        exec_output = executor.execute(code)

                        messages.append(
                            ToolMessage(
                                tool_call_id=tc_id,
                                name=name,
                                content=exec_output,
                                status="success",
                            )
                        )

                        if pool_json_path.exists():
                            try:
                                with open(pool_json_path) as f:
                                    loaded_pool = json.load(f)
                                for img_id, entry in loaded_pool.items():
                                    if img_id not in self.intermediate_image_pool:
                                        self.intermediate_image_pool[img_id] = {
                                            "image_id": img_id,
                                            "path": entry["path"],
                                            "turn_id": iterations,
                                            "transform": entry["transform"],
                                            "is_output": entry.get("is_output", False),
                                            "annotations": entry.get("annotations", {}),
                                        }
                            except Exception as e:
                                logger.error(f"Failed to read pool_json_path: {e}")

                if submit_called:
                    break

        finally:
            executor.close()

        outputs = [
            entry
            for entry in self.intermediate_image_pool.values()
            if entry.get("is_output") is True
        ]

        if not outputs:
            sorted_keys = sorted(
                self.intermediate_image_pool.keys(),
                key=lambda k: (
                    int(k.split("_")[1])
                    if k.startswith("img_") and k.split("_")[1].isdigit()
                    else -1
                ),
            )
            if sorted_keys:
                last_key = sorted_keys[-1]
                outputs = [self.intermediate_image_pool[last_key]]

        summary = (
            final_result.get("summary", "")
            if final_result
            else "Failed to submit result within iterations"
        )

        return {"outputs": outputs, "summary": summary}