ImageToolkit

@smolagents/examples/open_deep_research/scripts/visual_qa.py @camel/camel/toolkits/image_analysis_toolkit.py https://platform.openai.com/docs/guides/images-vision?api-mode=chat

ImageToolkit

Bases: AsyncBaseToolkit

源代码位于： utu/tools/image_toolkit.py

class ImageToolkit(AsyncBaseToolkit):
    def __init__(self, config: ToolkitConfig = None) -> None:
        super().__init__(config)
        image_llm_config = {
            "type": EnvUtils.get_env("UTU_IMAGE_LLM_TYPE", "chat.completions"),
            "model": EnvUtils.get_env("UTU_IMAGE_LLM_MODEL"),  # NOTE: you should set these envs in .env
            "api_key": EnvUtils.get_env("UTU_IMAGE_LLM_API_KEY"),
            "base_url": EnvUtils.get_env("UTU_IMAGE_LLM_BASE_URL"),
        }
        self.llm = SimplifiedAsyncOpenAI(**image_llm_config)

    def _load_image(self, image_path: str) -> str:
        parsed = urlparse(image_path)
        image: Image.Image = None

        if parsed.scheme in ("http", "https"):
            logger.debug(f"Fetching image from URL: {image_path}")
            try:
                response = requests.get(image_path, timeout=15)
                response.raise_for_status()
                image = Image.open(BytesIO(response.content)).convert("RGB")
            except requests.exceptions.RequestException as e:
                logger.error(f"URL fetch failed: {e}")
                raise
        else:
            logger.debug(f"Loading local image: {image_path}")
            try:
                image = Image.open(image_path).convert("RGB")
            except Exception as e:  # pylint: disable=broad-except
                logger.error(f"Image loading failed: {e}")
                raise ValueError(f"Invalid image file: {image_path}") from e
        # Convert the image to a base64 string
        buffer = BytesIO()
        image.save(buffer, format="JPEG")  # Use the appropriate format (e.g., JPEG, PNG)
        base64_image = base64.b64encode(buffer.getvalue()).decode("utf-8")

        # add string formatting required by the endpoint
        image_string = f"data:image/jpeg;base64,{base64_image}"
        return image_string

    @register_tool
    async def image_qa(self, image_path: str, question: str | None = None) -> str:
        """Generate textual description or answer questions about attached image.

        Args:
            image_path (str): Local path or URL to an image.
            question (str, optional): The question to answer. If not provided, return a description of the image.
        """
        image_str = self._load_image(image_path)
        if not question:
            messages = [
                {"role": "system", "content": TOOL_PROMPTS["image_summary"]},
                {"role": "user", "content": [{"type": "image_url", "image_url": {"url": image_str}}]},
            ]
            output = await self.llm.query_one(messages=messages, **self.config.config_llm.model_params.model_dump())
            output = f"You did not provide a particular question, so here is a detailed caption for the image: {output}"
        else:
            messages = [
                {"role": "system", "content": TOOL_PROMPTS["image_qa"]},
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": question},
                        {"type": "image_url", "image_url": {"url": image_str}},
                    ],
                },
            ]
            output = await self.llm.query_one(messages=messages, **self.config.config_llm.model_params.model_dump())
        return output

tools_map property

tools_map: dict[str, Callable]

Lazy loading of tools map. - collect tools registered by @register_tool

image_qa async

image_qa(image_path: str, question: str | None = None) -> str

Generate textual description or answer questions about attached image.

参数：

名称	类型	描述	默认
image_path	str	Local path or URL to an image.	必需
question	str	The question to answer. If not provided, return a description of the image.	None

源代码位于： utu/tools/image_toolkit.py

@register_tool
async def image_qa(self, image_path: str, question: str | None = None) -> str:
    """Generate textual description or answer questions about attached image.

    Args:
        image_path (str): Local path or URL to an image.
        question (str, optional): The question to answer. If not provided, return a description of the image.
    """
    image_str = self._load_image(image_path)
    if not question:
        messages = [
            {"role": "system", "content": TOOL_PROMPTS["image_summary"]},
            {"role": "user", "content": [{"type": "image_url", "image_url": {"url": image_str}}]},
        ]
        output = await self.llm.query_one(messages=messages, **self.config.config_llm.model_params.model_dump())
        output = f"You did not provide a particular question, so here is a detailed caption for the image: {output}"
    else:
        messages = [
            {"role": "system", "content": TOOL_PROMPTS["image_qa"]},
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": question},
                    {"type": "image_url", "image_url": {"url": image_str}},
                ],
            },
        ]
        output = await self.llm.query_one(messages=messages, **self.config.config_llm.model_params.model_dump())
    return output

get_tools_map_func

get_tools_map_func() -> dict[str, Callable]

Get tools map. It will filter tools by config.activated_tools if it is not None.

源代码位于： utu/tools/base.py

def get_tools_map_func(self) -> dict[str, Callable]:
    """Get tools map. It will filter tools by config.activated_tools if it is not None."""
    if self.config.activated_tools:
        assert all(tool_name in self.tools_map for tool_name in self.config.activated_tools), (
            f"Error config activated tools: {self.config.activated_tools}! available tools: {self.tools_map.keys()}"
        )
        tools_map = {tool_name: self.tools_map[tool_name] for tool_name in self.config.activated_tools}
    else:
        tools_map = self.tools_map
    return tools_map

get_tools_in_agents

get_tools_in_agents() -> list[FunctionTool]

Get tools in openai-agents format.

源代码位于： utu/tools/base.py

def get_tools_in_agents(self) -> list[FunctionTool]:
    """Get tools in openai-agents format."""
    tools_map = self.get_tools_map_func()
    tools = []
    for _, tool in tools_map.items():
        tools.append(
            function_tool(
                tool,
                strict_mode=False,  # turn off strict mode
            )
        )
    return tools

get_tools_in_openai

get_tools_in_openai() -> list[dict]

Get tools in OpenAI format.

源代码位于： utu/tools/base.py

def get_tools_in_openai(self) -> list[dict]:
    """Get tools in OpenAI format."""
    tools = self.get_tools_in_agents()
    return [ChatCompletionConverter.tool_to_openai(tool) for tool in tools]

get_tools_in_mcp

get_tools_in_mcp() -> list[Tool]

Get tools in MCP format.

源代码位于： utu/tools/base.py

def get_tools_in_mcp(self) -> list[types.Tool]:
    """Get tools in MCP format."""
    tools = self.get_tools_in_agents()
    return [MCPConverter.function_tool_to_mcp(tool) for tool in tools]

call_tool async

call_tool(name: str, arguments: dict) -> str

Call a tool by its name.

源代码位于： utu/tools/base.py

async def call_tool(self, name: str, arguments: dict) -> str:
    """Call a tool by its name."""
    tools_map = self.get_tools_map_func()
    if name not in tools_map:
        raise ValueError(f"Tool {name} not found")
    tool = tools_map[name]
    return await tool(**arguments)