122 lines
4.5 KiB
Python
122 lines
4.5 KiB
Python
"""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)
|