fix(agents): bound viewed image context per model request (#6828)

* fix(agents): bound viewed image context per model request

Keep recent image payloads within a count and byte budget, and move re-viewed paths to the newest position. Add focused regressions for budget enforcement and state ordering.

* test(agents): verify cumulative image budgets and re-view recovery

* fix(agents): clarify image omission and read-attempt limits

* docs(middleware): shorten image context guidance for CI budget

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
stanGSA 2026-10-14 10:10:25 +08:00 committed by GitHub
parent 4e2ba71a6c
commit 4fc0d8fb31
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 218 additions and 34 deletions

View File

@ -1735,6 +1735,7 @@ Advanced deployments can enable pluggable authorization with `authorization.enab
Follow-up suggestions also check `model:use` before calling the selected model, including the default model when no name is supplied. A denied model returns HTTP 403 without an LLM call; authorization-provider failures follow the configured `fail_closed` policy.
For vision-capable agents, `view_image` and the subsequent model-context image read also require `sandbox:execute`. Allowing the tool name alone does not grant image-file access; a role denied sandbox execution cannot reread previously recorded image metadata after its permissions change. External run input and thread-state updates cannot set `viewed_images` or `thread_data`; when a saved image is read from a host copy, its path must resolve to the current user's thread and the recorded virtual image path.
Each model request includes at most four recently viewed images and 28 MiB of base64 image data. Images excluded by these limits are reported as omitted; the agent can call `view_image` again to bring one back into context.
Advanced deployments can also extend the agent runtime itself by declaring `AgentMiddleware` classes under `extensions.middlewares` in `config.yaml` or `extensions_config.json`. Each entry is a `module.path:ClassName` string (zero-argument constructor) or an object `{class, kwargs}` whose `kwargs` are passed to the constructor. `kwargs` values must be JSON types (object, array, string, number, boolean, or null); YAML dates and timestamps are coerced to ISO strings so they match JSON. DeerFlow loads the same configured list into the lead-agent and subagent pipelines after their built-in runtime middlewares and loop/token guards, but before the terminal-response/safety/clarification tail, so enterprise forks can add domain guardrails, tool-call governance, or observability hooks without patching the built-in middleware builders. Missing packages, invalid classes, broken modules, and constructor errors fail loudly at agent creation. Treat `config.yaml` and `extensions_config.json` as trusted operator-controlled files: middleware paths are code execution, just like custom tool, model, sandbox, guardrail, MCP server, and MCP interceptor declarations. Gateway skill/MCP toggle endpoints preserve this field but do not expose an API write path for `extensions.middlewares`. Separate lead-only/subagent-only middleware lists are not supported yet.

View File

@ -119,7 +119,7 @@ Before changing a later authorization phase, read the [authorization RFC](../../
TODO attribution skips non-string statuses before membership; preserve args/TODO state for normal tool validation.
24. **TitleMiddleware** - Auto-generates the thread title after the first complete exchange and normalizes structured message content before prompting the title model. If a first-turn run is interrupted before this middleware can write a title, `runtime/runs/worker.py` keeps the run in a finalizing state, persists a local fallback title from the latest checkpoint or original run input, and then syncs it to `threads_meta.display_name`. Replacement runs admitted by `multitask_strategy="interrupt"` / `"rollback"` wait for older same-thread finalization before entering the graph; the interrupted run only skips the fallback title write once a later run has started and may have advanced the checkpoint.
25. **MemoryMiddleware** - Queues conversations for async memory update (filters to user + final AI responses); captures the runtime-resolved user so standalone LangGraph Server reads and writes stay in the same bucket
26. **ViewImageMiddleware** - *(vision models only)* Enforces the shared vision cap on remote rereads (see [download contract](../../sandbox/AGENTS.md)). Rebuilds hidden base64 context in `ModelRequest.messages`, keeping lightweight `viewed_images` in checkpoints. Validated blob refs precede compatibility paths. `_image_injection_plan` removes stale server-marked context, including old `before_model`/`after_model` payloads, then shares eligibility for sync/async injection. Removal requires both the reserved ID prefix and marker, preserving user and unmarked legacy messages. Both paths recheck `sandbox:execute` before I/O. Host copies require `actual_path` to match the current user/thread's canonical virtual path, even for old checkpoints or replacement sandboxes; absent/mismatched identity skips reads. Only the async wrapper passes `authorization_checked=True` after its gate to avoid duplicate checks in its worker.
26. **ViewImageMiddleware** - *(vision models only)* Enforces the shared vision cap on remote rereads (see [download contract](../../sandbox/AGENTS.md)). Rebuilds hidden base64 context in `ModelRequest.messages`, keeping lightweight `viewed_images` in checkpoints. Caps each request at four recent images and 28 MiB base64; constructor limits may tighten caps. Skips over-budget reads. Re-views refresh reducer recency. Validated blob refs precede compatibility paths. `_image_injection_plan` removes stale server-marked context, including old `before_model`/`after_model` payloads, then shares eligibility for sync/async injection. Removal requires both the reserved ID prefix and marker, preserving user and unmarked legacy messages. Both paths recheck `sandbox:execute` before I/O. Host copies require `actual_path` to match the current user/thread's canonical virtual path, even for old checkpoints or replacement sandboxes; absent/mismatched identity skips reads. Only the async wrapper passes `authorization_checked=True` after its gate to avoid duplicate checks in its worker.
27. **McpRoutingMiddleware** - Auto-promotes deferred schemas matching the latest real user message before `DeferredToolFilterMiddleware`, without executing tools. New names emit `middleware:tool_promotion` with `source=routing_hint`; repeat passes emit nothing
28. **DeferredToolPromotionAuditMiddleware** - Observes final `tool_search` `Command`s; keep it outer of `SkillToolPolicyMiddleware` so denied names are excluded. It atomically claims new names per lead run or subagent execution to dedupe parallel searches, derives subagent attribution from the server-installed recorder, returns the original `Command`, and omits private payloads
29. **DeferredToolFilterMiddleware** - *(optional, if `tool_search.enabled`)* Hides deferred (MCP) tool schemas from the bound model until `tool_search` or `McpRoutingMiddleware` promotes them (reads per-thread promotions from `ThreadState.promoted`, hash-scoped)

View File

@ -19,6 +19,8 @@ from deerflow.sandbox.lease import run_sync_lifecycle_operation
logger = logging.getLogger(__name__)
_MAX_CONTEXT_IMAGES = 4
_MAX_CONTEXT_BASE64_BYTES = 28 * 1024 * 1024
_IMAGE_CONTEXT_MESSAGE_ID_PREFIX = "view-image-context:"
_IMAGE_CONTEXT_MESSAGE_MARKER_KEY = "deerflow_view_image_context"
@ -34,7 +36,7 @@ class ViewImageMiddleware(AgentMiddleware[ViewImageMiddlewareState]):
1. Wraps each LLM call
2. Checks if the last assistant message contains view_image tool calls
3. Verifies all tool calls in that message have been completed (have corresponding ToolMessages)
4. If conditions are met, appends a human message with all viewed image details (including base64 data)
4. If conditions are met, appends a human message with recent viewed image details (including base64 data)
5. Hands the augmented request to the model so it can see and analyze the images
This enables the LLM to automatically receive and analyze images that were loaded via view_image tool,
@ -53,7 +55,7 @@ class ViewImageMiddleware(AgentMiddleware[ViewImageMiddlewareState]):
def __init__(
self,
*,
max_injected_images: int = 10,
max_injected_images: int = 4,
max_injected_total_chars: int = 32_000_000,
) -> None:
"""Bound the per-request image context injected after view_image rounds.
@ -62,7 +64,8 @@ class ViewImageMiddleware(AgentMiddleware[ViewImageMiddlewareState]):
``view_image`` round re-encodes and re-sends every entry ever viewed,
so a long PDF-inspection thread pushes the request body past provider
limits (e.g. Anthropic's 32 MB) even though each single image is
capped. These knobs cap how many images and how much base64 payload
capped. A request admits at most four images and 28 MiB of base64.
These knobs may further cap how many images and how much payload
(in data-url characters) may be injected per model call; admission
runs newest-first so the most recently viewed images survive, and
over-budget entries keep their text bullet with a placeholder, so
@ -335,7 +338,7 @@ class ViewImageMiddleware(AgentMiddleware[ViewImageMiddlewareState]):
*,
host_path_allowed: Callable[[str, str], bool] | None = None,
) -> list[str | dict]:
"""Create a formatted message with the viewed image details that fit the injection budget.
"""Create bounded image context, prioritizing the most recently viewed paths.
Reads image files on-demand from the active sandbox when available and
encodes them as base64 for the model. The base64 data is NOT persisted in
@ -359,29 +362,16 @@ class ViewImageMiddleware(AgentMiddleware[ViewImageMiddlewareState]):
# Return a properly formatted text block, not a plain string array
return [{"type": "text", "text": "No images have been viewed."}]
# Budget admission runs newest-first so the most recently viewed
# images survive the caps, and the oldest entries over budget keep
# their text bullet with a placeholder in place (#6825). Admission is
# decided from the recorded byte size before any file is read: base64
# expands 4/3 and the data-url prefix adds ~100 characters, so the
# estimate tracks the encoded payload closely. Omitted entries are
# never read from the sandbox, which also bounds the work this
# middleware performs per model call.
admitted: set[str] = set()
admitted_chars = 0
admitted_images = 0
for image_path, image_data in reversed(viewed_images.items()):
estimated_chars = int(image_data.get("size", 0) or 0) * 4 // 3 + 128
if admitted_images >= self._max_injected_images or admitted_chars + estimated_chars > self._max_injected_total_chars:
continue
admitted_images += 1
admitted_chars += estimated_chars
admitted.add(image_path)
# Build the message with image information
# Select newest first so older entries cannot consume the budget before
# the image that triggered this model call. Preserve chronological
# output and the upstream per-image omission placeholders.
content_blocks: list[str | dict] = [{"type": "text", "text": "Here are the images you've viewed:"}]
for image_path, image_data in viewed_images.items():
selected: list[list[dict]] = []
encoded_bytes = 0
admitted_chars = 0
attempted_images = 0
omitted = 0
for image_path, image_data in reversed(list(viewed_images.items())):
mime_type = image_data.get("mime_type", "unknown")
actual_path = image_data.get("actual_path", "")
expected_size = image_data.get("size", 0)
@ -389,12 +379,17 @@ class ViewImageMiddleware(AgentMiddleware[ViewImageMiddlewareState]):
source_sandbox_id = image_data.get("source_sandbox_id")
blob_ref = image_data.get("blob_ref")
# Add text description
content_blocks.append({"type": "text", "text": f"\n- **{image_path}** ({mime_type})"})
if image_path not in admitted:
content_blocks.append({"type": "text", "text": " (omitted: image injection budget reached; call view_image again to re-view this image)"})
estimated_bytes = 4 * ((expected_size + 2) // 3) if isinstance(expected_size, int) and expected_size >= 0 else _MAX_CONTEXT_BASE64_BYTES + 1
estimated_chars = estimated_bytes + len(f"data:{mime_type};base64,")
blocks = [{"type": "text", "text": f"\n- **{image_path}** ({mime_type})"}]
# Failed reads also consume slots, bounding potentially remote read attempts.
if attempted_images >= min(_MAX_CONTEXT_IMAGES, self._max_injected_images) or encoded_bytes + estimated_bytes > _MAX_CONTEXT_BASE64_BYTES or admitted_chars + estimated_chars > self._max_injected_total_chars:
omitted += 1
blocks.append({"type": "text", "text": " (omitted: image injection budget reached; call view_image again to re-view this image)"})
selected.append(blocks)
continue
attempted_images += 1
admitted_chars += estimated_chars
# Read the image file on-demand and encode as base64 for the model
data_url = self._read_image_as_data_url(
@ -409,14 +404,21 @@ class ViewImageMiddleware(AgentMiddleware[ViewImageMiddlewareState]):
allow_host_copy=host_path_allowed(image_path, actual_path) if host_path_allowed is not None else False,
)
if data_url:
content_blocks.append(
encoded_bytes += estimated_bytes
blocks.append(
{
"type": "image_url",
"image_url": {"url": data_url},
}
)
else:
content_blocks.append({"type": "text", "text": f" (file unavailable or changed: {image_path})"})
blocks.append({"type": "text", "text": f" (file unavailable or changed: {image_path})"})
selected.append(blocks)
if omitted:
content_blocks.append({"type": "text", "text": f"Omitted viewed images: {omitted}. Call view_image again to inspect a specific image."})
for blocks in reversed(selected):
content_blocks.extend(blocks)
return content_blocks

View File

@ -10,6 +10,11 @@ ownership and capture cleanup. Retain the real XLSX conversion coverage.
Backend tests must preserve the runtime invariants they exercise without changing production execution topology.
Image-context budget regressions use real PNG files through `view_image`, the
state reducer, and both model-call wrappers. Check decoded payloads, cumulative
base64 size and skipped file reads; keep count and byte limits independent.
Re-view coverage must first observe omission, then restoration in model input.
Local nginx path fixtures convert MSYS output through the selected Git Bash's
PATH and decode both renderer and converter output as UTF-8. POSIX Git-layout
stubs exercise Windows-style locale decoding and missing/failed/empty converters;

View File

@ -150,6 +150,13 @@ class TestMergeViewedImages:
existing = {"k1": {"mime_type": "image/png", "size": 1, "actual_path": "/a"}}
assert merge_viewed_images(existing, {}) == {}
def test_revisited_path_becomes_most_recent(self):
existing = {"old": {"size": 1}, "recent": {"size": 2}}
merged = merge_viewed_images(existing, {"old": {"size": 3}})
assert list(merged) == ["recent", "old"]
assert merged["old"]["size"] == 3
assert list(existing) == ["old", "recent"]
class TestMergeDelegations:
"""Reducer for completed subagent/task delegation records."""

View File

@ -23,6 +23,7 @@ Covered behavior:
import base64
import hashlib
from io import BytesIO
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock
@ -33,6 +34,7 @@ from langchain.agents.middleware.types import ModelRequest
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, SystemMessage, ToolMessage
from PIL import Image
from support.symlinks import symlink_or_skip
from deerflow.agents.middlewares.view_image_middleware import (
@ -42,6 +44,108 @@ from deerflow.agents.middlewares.view_image_middleware import (
from deerflow.agents.thread_state import merge_viewed_images
from deerflow.config.paths import Paths
from deerflow.storage import BlobRef
from deerflow.tools.builtins.view_image_tool import view_image_tool
@pytest.fixture
def budget_image_runtime(tmp_path, monkeypatch):
paths = Paths(tmp_path)
monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: paths)
monkeypatch.setattr("deerflow.storage.get_blob_store_if_enabled", lambda: None)
outputs = paths.sandbox_outputs_dir("thread-test", user_id="user-test")
outputs.mkdir(parents=True)
payloads = {}
for index in range(5):
buffer = BytesIO()
Image.new("RGB", (2, 2), (index * 40, 80, 160)).save(buffer, format="PNG")
payloads[f"/mnt/user-data/outputs/{index}.png"] = buffer.getvalue()
(outputs / f"{index}.png").write_bytes(buffer.getvalue())
runtime = SimpleNamespace(
state={"thread_data": {"outputs_path": str(outputs)}},
context={"user_id": "user-test", "thread_id": "thread-test"},
config={},
)
return runtime, payloads
def _view_budget_image(runtime, image_path, call_id):
result = view_image_tool.func(runtime=runtime, image_path=image_path, tool_call_id=call_id)
assert result.update["messages"][0].content == "Successfully read image"
runtime.state["viewed_images"] = merge_viewed_images(runtime.state.get("viewed_images"), result.update["viewed_images"])
request = _model_request([AIMessage(content="", tool_calls=[_view_image_call(call_id, image_path)]), *result.update["messages"]])
return request.override(state={**runtime.state, "messages": request.messages}, runtime=runtime)
async def _capture_budget_request(request, use_async):
seen = []
def handler(prepared):
seen.append(prepared)
return AIMessage(content="ok")
async def async_handler(prepared):
return handler(prepared)
middleware = ViewImageMiddleware()
if use_async:
await middleware.awrap_model_call(request, async_handler)
else:
middleware.wrap_model_call(request, handler)
assert len(seen) == 1
assert _image_context_messages(request.messages) == []
assert _image_context_messages(request.state["messages"]) == []
return _image_urls(seen[0])
@pytest.mark.anyio
@pytest.mark.parametrize("use_async", [False, True])
@pytest.mark.parametrize("budget_delta", [0, -1], ids=["exact-fit", "one-byte-short"])
async def test_cumulative_image_bytes_reach_model_within_budget(budget_image_runtime, monkeypatch, use_async, budget_delta):
from deerflow.agents.middlewares import view_image_middleware as module
runtime, payloads = budget_image_runtime
selected_paths = list(payloads)[:3]
for index, path in enumerate(selected_paths):
request = _view_budget_image(runtime, path, f"view-{index}")
encoded = [base64.b64encode(payloads[path]).decode("ascii") for path in selected_paths]
budget = len(encoded[1]) + len(encoded[2]) + budget_delta
assert all(len(data) <= budget for data in encoded)
# Add attributes on the unfixed baseline too: RED must be behavioral,
# not an AttributeError caused by a missing budget constant.
monkeypatch.setattr(module, "_MAX_CONTEXT_IMAGES", 4, raising=False)
monkeypatch.setattr(module, "_MAX_CONTEXT_BASE64_BYTES", budget, raising=False)
reads = []
original_open = open
def record_open(file, *args, **kwargs):
if isinstance(file, (str, Path)) and Path(file).suffix == ".png":
reads.append(Path(file).name)
return original_open(file, *args, **kwargs)
monkeypatch.setattr("builtins.open", record_open)
urls = await _capture_budget_request(request, use_async)
expected = selected_paths[1:] if budget_delta == 0 else selected_paths[2:]
actual = [url.split(",", 1)[1] for url in urls]
assert sum(len(data) for data in actual) <= budget
assert [base64.b64decode(data) for data in actual] == [payloads[path] for path in expected]
assert reads == [Path(path).name for path in reversed(expected)]
@pytest.mark.anyio
@pytest.mark.parametrize("use_async", [False, True])
async def test_revisited_image_returns_to_model_context(budget_image_runtime, use_async):
runtime, payloads = budget_image_runtime
paths = list(payloads)
for index, path in enumerate(paths):
request = _view_budget_image(runtime, path, f"initial-{index}")
before = await _capture_budget_request(request, use_async)
assert [base64.b64decode(url.split(",", 1)[1]) for url in before] == [payloads[path] for path in paths[1:]]
saved = runtime.state["viewed_images"]
request = _view_budget_image(runtime, paths[0], "revisit-oldest")
after = await _capture_budget_request(request, use_async)
assert [base64.b64decode(url.split(",", 1)[1]) for url in after] == [payloads[path] for path in [*paths[2:], paths[0]]]
assert runtime.state["viewed_images"] == saved
assert list(saved) == paths
def _view_image_call(call_id: str = "call_1", path: str = "/mnt/user-data/uploads/img.png") -> dict:
@ -246,6 +350,71 @@ class TestAllToolsCompleted:
class TestCreateImageDetailsMessage:
def test_keeps_recent_images_within_count_and_byte_budget(self, tmp_path, monkeypatch):
from deerflow.agents.middlewares import view_image_middleware as module
monkeypatch.setattr(module, "_MAX_CONTEXT_IMAGES", 2)
monkeypatch.setattr(module, "_MAX_CONTEXT_BASE64_BYTES", 16)
viewed = {f"/{name}.png": _make_viewed_image(tmp_path, f"{name}.png", data=b"123456") for name in ("old", "middle", "new")}
reads = []
def read(_cls, _state, image_path, *_args, **_kwargs):
reads.append(image_path)
return "data:image/png;base64,MTIzNDU2"
monkeypatch.setattr(ViewImageMiddleware, "_read_image_as_data_url", classmethod(read))
blocks = ViewImageMiddleware()._create_image_details_message({"viewed_images": viewed})
assert reads == ["/new.png", "/middle.png"]
assert [block["type"] for block in blocks].count("image_url") == 2
assert "Omitted viewed images: 1." in str(blocks)
@pytest.mark.parametrize("omitted_count", [1, 2])
def test_skips_oversized_image_without_reading_it(self, tmp_path, monkeypatch, omitted_count):
from deerflow.agents.middlewares import view_image_middleware as module
monkeypatch.setattr(module, "_MAX_CONTEXT_BASE64_BYTES", 8)
viewed = {f"/large-{i}.png": _make_viewed_image(tmp_path, f"large-{i}.png", data=b"1234567") for i in range(omitted_count)}
monkeypatch.setattr(
ViewImageMiddleware,
"_read_image_as_data_url",
lambda *_args, **_kwargs: pytest.fail("oversized image was read"),
)
blocks = ViewImageMiddleware()._create_image_details_message({"viewed_images": viewed})
assert all(block["type"] != "image_url" for block in blocks)
assert f"Omitted viewed images: {omitted_count}." in str(blocks)
def test_omission_notice_does_not_assume_omitted_image_is_older(self, tmp_path, monkeypatch):
from deerflow.agents.middlewares import view_image_middleware as module
monkeypatch.setattr(module, "_MAX_CONTEXT_BASE64_BYTES", 8)
viewed = {
"/old.png": _make_viewed_image(tmp_path, "old.png", data=b"123456"),
"/new.png": _make_viewed_image(tmp_path, "new.png", data=b"1234567"),
}
blocks = ViewImageMiddleware()._create_image_details_message({"viewed_images": viewed}, host_path_allowed=lambda _, actual: Path(actual).is_relative_to(tmp_path))
urls = [block["image_url"]["url"] for block in blocks if block["type"] == "image_url"]
assert [base64.b64decode(url.split(",", 1)[1]) for url in urls] == [b"123456"]
assert blocks[1] == {"type": "text", "text": "Omitted viewed images: 1. Call view_image again to inspect a specific image."}
def test_failed_reads_consume_slots_before_readable_older_image(self, tmp_path, monkeypatch):
viewed = {"/old.png": _make_viewed_image(tmp_path, "old.png")}
viewed.update({f"/stale-{i}.png": {"mime_type": "image/png", "size": 1} for i in range(4)})
middleware = ViewImageMiddleware()
original_read = middleware._read_image_as_data_url
reads = []
def record_read(state, image_path, *args, **kwargs):
reads.append(image_path)
return original_read(state, image_path, *args, **kwargs)
monkeypatch.setattr(middleware, "_read_image_as_data_url", record_read)
blocks = middleware._create_image_details_message({"viewed_images": viewed}, host_path_allowed=lambda _, actual: Path(actual).is_relative_to(tmp_path))
assert reads == [f"/stale-{i}.png" for i in reversed(range(4))]
assert sum("file unavailable or changed" in block.get("text", "") for block in blocks) == 4
assert all(block["type"] != "image_url" for block in blocks)
assert "Omitted viewed images: 1." in str(blocks)
def test_returns_placeholder_when_no_images(self):
mw = ViewImageMiddleware()
state = {"viewed_images": {}}