deerflow-code/offline-backend-20260512/backend/packages/harness/deerflow/models/vision.py

"""Image recognition (识图) via a configurable vision model.

When ``config.yaml`` defines ``vision.model_name``, the lead agent's
``view_image`` tool delegates image understanding to that dedicated model and
returns a text description — so a text-only chat model can still "see" uploaded
images. When no dedicated model is configured, vision falls back to the main
chat model, which only works when that model has ``supports_vision: true``.
"""

import logging

from deerflow.config import get_app_config
from deerflow.config.app_config import AppConfig
from deerflow.config.vision_config import DEFAULT_VISION_PROMPT

logger = logging.getLogger(__name__)


def _configured_vision_model(config: AppConfig) -> str | None:
    """Return a valid dedicated vision model name from config, or None."""
    vision_cfg = getattr(config, "vision", None)
    name = getattr(vision_cfg, "model_name", None) if vision_cfg else None
    if not name:
        return None
    if config.get_model_config(name) is None:
        logger.warning("vision.model_name %r is not defined in models[]; ignoring", name)
        return None
    return name


def resolve_vision_model_name(model_name: str | None, *, app_config: AppConfig | None = None) -> str | None:
    """Resolve which model should handle image recognition for this run.

    Priority:
    1. A dedicated ``vision.model_name`` configured in ``config.yaml``.
    2. The main chat model, only if it declares ``supports_vision: true``.
    3. ``None`` — vision is unavailable.
    """
    config = app_config or get_app_config()
    configured = _configured_vision_model(config)
    if configured:
        return configured
    if model_name:
        mc = config.get_model_config(model_name)
        if mc is not None and mc.supports_vision:
            return model_name
    return None


def vision_enabled(model_name: str | None, *, app_config: AppConfig | None = None) -> bool:
    """Whether the ``view_image`` tool/middleware should be available this run."""
    return resolve_vision_model_name(model_name, app_config=app_config) is not None


def is_delegated_vision(*, app_config: AppConfig | None = None) -> bool:
    """Whether image recognition is delegated to a dedicated vision model.

    True when ``vision.model_name`` is configured and valid — in that case
    ``view_image`` calls the dedicated model and returns a text description
    instead of injecting raw image data into the main chat model.
    """
    config = app_config or get_app_config()
    return _configured_vision_model(config) is not None


def _extract_text(message) -> str:
    """Flatten an LLM response message into plain text."""
    content = getattr(message, "content", message)
    if isinstance(content, str):
        return content.strip()
    if isinstance(content, list):
        parts: list[str] = []
        for block in content:
            if isinstance(block, str):
                parts.append(block)
            elif isinstance(block, dict) and block.get("type") == "text":
                parts.append(str(block.get("text", "")))
        return "\n".join(p for p in parts if p).strip()
    return str(content).strip()


def recognize_image(
    image_base64: str,
    mime_type: str,
    *,
    question: str | None = None,
    app_config: AppConfig | None = None,
) -> str:
    """Describe/analyze an image using the configured dedicated vision model.

    Returns the model's text answer. Raises ``ValueError`` if no dedicated
    vision model is configured, and propagates model/network errors to the
    caller for handling.
    """
    from langchain_core.messages import HumanMessage

    from deerflow.models.factory import create_chat_model

    config = app_config or get_app_config()
    model_name = _configured_vision_model(config)
    if model_name is None:
        raise ValueError("No dedicated vision model is configured (vision.model_name).")

    vision_cfg = getattr(config, "vision", None)
    default_prompt = getattr(vision_cfg, "prompt", None) or DEFAULT_VISION_PROMPT
    prompt = (question or "").strip() or default_prompt

    extra: dict = {}
    max_tokens = getattr(vision_cfg, "max_tokens", None) if vision_cfg else None
    if max_tokens:
        extra["max_tokens"] = max_tokens

    model = create_chat_model(name=model_name, thinking_enabled=False, app_config=config, **extra)
    message = HumanMessage(
        content=[
            {"type": "text", "text": prompt},
            {"type": "image_url", "image_url": {"url": f"data:{mime_type};base64,{image_base64}"}},
        ]
    )
    result = model.invoke([message])
    return _extract_text(result)