修复多轮/引用识图串图:历史不再持久化原图,失败图片 URL 丢弃,请求图本地优先#9308
Conversation
- 历史落库与加载时剥离 image_url,替换为 [Image] 占位 - OpenAI source 图片解析失败时丢弃原 URL,改为 [Image unavailable] - 请求图片本地优先、去重并限制数量,降低远程死链干扰 - 增加 tests/test_image_history_strip.py
There was a problem hiding this comment.
Hey - I've found 3 issues, and left some high level feedback:
- The image-stripping logic in
dump_messages_with_checkpointslargely duplicatesstrip_images_from_content_parts; consider refactoring to reuse the shared helper so behavior stays aligned and easier to maintain. _finalize_request_image_urlsis called withnormalize_and_dedupe_strings(req.image_urls)but also callsnormalize_and_dedupe_stringsinternally, causing redundant work; you can remove one of these calls to simplify and avoid double processing.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The image-stripping logic in `dump_messages_with_checkpoints` largely duplicates `strip_images_from_content_parts`; consider refactoring to reuse the shared helper so behavior stays aligned and easier to maintain.
- `_finalize_request_image_urls` is called with `normalize_and_dedupe_strings(req.image_urls)` but also calls `normalize_and_dedupe_strings` internally, causing redundant work; you can remove one of these calls to simplify and avoid double processing.
## Individual Comments
### Comment 1
<location path="astrbot/core/astr_main_agent.py" line_range="1558" />
<code_context>
)
)
- req.image_urls = normalize_and_dedupe_strings(req.image_urls)
+ req.image_urls = _finalize_request_image_urls(
+ normalize_and_dedupe_strings(req.image_urls),
+ max_total=max(int(getattr(config, "max_quoted_fallback_images", 4) or 4), 4),
+ )
req.audio_urls = normalize_and_dedupe_strings(req.audio_urls)
</code_context>
<issue_to_address>
**suggestion:** `req.image_urls` is being normalized twice and the `max_total` expression could be simplified.
`req.image_urls` is normalized before being passed to `_finalize_request_image_urls`, which normalizes again, so the outer `normalize_and_dedupe_strings` call can be removed. Also, consider simplifying the `max_total` computation, e.g.:
```python
raw = getattr(config, "max_quoted_fallback_images", 4)
max_total = max(int(raw or 4), 4)
```
and then reuse `max_total` in the call to `_finalize_request_image_urls` to avoid duplicated logic.
```suggestion
if isinstance(req.contexts, str):
)
)
)
raw_max_quoted_fallback_images = getattr(config, "max_quoted_fallback_images", 4)
max_quoted_fallback_images = max(int(raw_max_quoted_fallback_images or 4), 4)
req.image_urls = _finalize_request_image_urls(
req.image_urls,
max_total=max_quoted_fallback_images,
)
req.audio_urls = normalize_and_dedupe_strings(req.audio_urls)
```
</issue_to_address>
### Comment 2
<location path="astrbot/core/provider/sources/openai_source.py" line_range="276-278" />
<code_context>
url, image_detail = self._extract_image_part_info(part)
if not url:
- return part
+ return {
+ "type": "text",
+ "text": "[Image unavailable]",
+ }
</code_context>
<issue_to_address>
**suggestion:** Replacing failed image parts with a hardcoded text placeholder may drop useful metadata and is not easily configurable.
On missing URL, this now replaces the whole `image_url` part with `{type: 'text', text: '[Image unavailable]'}`, which drops any existing metadata (alt text, descriptions, etc.) and hardcodes the string. Please centralize the placeholder in a shared constant (ideally near the history placeholder) and consider including minimal context from the original part (e.g., an alt/description field) so downstream consumers still know what was intended.
Suggested implementation:
```python
if part.get("type") == "image_url":
url, image_detail = self._extract_image_part_info(part)
if not url:
# Build a text placeholder while preserving minimal context from the original part.
# Prefer explicit alt/description fields if available.
alt_text = (
part.get("alt")
or part.get("description")
or (part.get("metadata") or {}).get("alt")
or (part.get("metadata") or {}).get("description")
)
if alt_text:
placeholder_text = f"{IMAGE_UNAVAILABLE_PLACEHOLDER} (original description: {alt_text})"
else:
placeholder_text = IMAGE_UNAVAILABLE_PLACEHOLDER
return {
"type": "text",
"text": placeholder_text,
}
try:
resolved_part = await self._resolve_image_part(
)
except Exception as exc:
logger.warning(
"Image preprocess failed; dropping image content. url=%s err=%s",
url,
exc,
)
```
To fully implement the centralization of the placeholder text:
1. Define a module-level constant `IMAGE_UNAVAILABLE_PLACEHOLDER` near the existing history placeholder constant in `astrbot/core/provider/sources/openai_source.py`, for example:
```python
IMAGE_UNAVAILABLE_PLACEHOLDER = "[Image unavailable]"
```
2. Ensure that this constant is imported/used only within this module (no circular imports), and update any other hardcoded `"[Image unavailable]"` occurrences, if present, to use `IMAGE_UNAVAILABLE_PLACEHOLDER` instead.
</issue_to_address>
### Comment 3
<location path="astrbot/core/agent/message.py" line_range="321" />
<code_context>
+ return {"type": "text", "text": _IMAGE_HISTORY_PLACEHOLDER}
+
+
+def strip_images_from_content_parts(content: Any) -> Any:
+ """Replace image parts with a short text placeholder.
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider reusing the shared image-stripping helper in dump_messages_with_checkpoints so the image placeholder logic is implemented only once.
You can remove the duplicated image-stripping logic in `dump_messages_with_checkpoints` by reusing `strip_images_from_content_parts` and keeping `_no_save` handling separate.
Right now you have two copies of the “collapse consecutive images to a single `[Image]` placeholder” loop:
- `strip_images_from_content_parts`
- the inline `persisted_parts` loop inside `dump_messages_with_checkpoints`
You can keep behavior identical and reduce complexity by:
1. Filtering out `_no_save` parts in `dump_messages_with_checkpoints`
2. Delegating all image / placeholder logic to `strip_images_from_content_parts`
Example change:
```python
def dump_messages_with_checkpoints(messages: list[Message]) -> list[dict]:
"""Dump runtime messages and reinsert bound checkpoint segments."""
dumped: list[dict] = []
for message in messages:
message_data = message.model_dump()
if isinstance(message.content, list):
# 1) Remove _no_save parts here
persisted_parts = [
part for part in message.content
if not getattr(part, "_no_save", False)
]
# 2) Reuse shared image-stripping / placeholder logic
stripped_parts = strip_images_from_content_parts(persisted_parts)
# 3) Normalized model_dump / dict handling in one place
message_data["content"] = [
part.model_dump() if isinstance(part, ContentPart) else part
for part in stripped_parts
]
dumped.append(message_data)
if message._checkpoint_after is not None:
dumped.append(
CheckpointMessageSegment(content=message._checkpoint_after).model_dump()
)
return dumped
```
This:
- Keeps `_no_save` semantics intact
- Ensures the “image → collapsed placeholder” rules live in a single function
- Eliminates the second `previous_was_image_placeholder` state machine
If you later need a pure-dict variant, consider a tiny wrapper instead of reimplementing the loop:
```python
def strip_images_from_content_parts_to_dict(parts: list[Any]) -> list[dict]:
stripped = strip_images_from_content_parts(parts)
return [
p if isinstance(p, dict) else p.model_dump()
for p in stripped
]
```
Then callers needing dicts can just call this helper and not worry about how placeholders are created.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| return { | ||
| "type": "text", | ||
| "text": "[Image unavailable]", |
There was a problem hiding this comment.
suggestion: Replacing failed image parts with a hardcoded text placeholder may drop useful metadata and is not easily configurable.
On missing URL, this now replaces the whole image_url part with {type: 'text', text: '[Image unavailable]'}, which drops any existing metadata (alt text, descriptions, etc.) and hardcodes the string. Please centralize the placeholder in a shared constant (ideally near the history placeholder) and consider including minimal context from the original part (e.g., an alt/description field) so downstream consumers still know what was intended.
Suggested implementation:
if part.get("type") == "image_url":
url, image_detail = self._extract_image_part_info(part)
if not url:
# Build a text placeholder while preserving minimal context from the original part.
# Prefer explicit alt/description fields if available.
alt_text = (
part.get("alt")
or part.get("description")
or (part.get("metadata") or {}).get("alt")
or (part.get("metadata") or {}).get("description")
)
if alt_text:
placeholder_text = f"{IMAGE_UNAVAILABLE_PLACEHOLDER} (original description: {alt_text})"
else:
placeholder_text = IMAGE_UNAVAILABLE_PLACEHOLDER
return {
"type": "text",
"text": placeholder_text,
}
try:
resolved_part = await self._resolve_image_part(
)
except Exception as exc:
logger.warning(
"Image preprocess failed; dropping image content. url=%s err=%s",
url,
exc,
)To fully implement the centralization of the placeholder text:
- Define a module-level constant
IMAGE_UNAVAILABLE_PLACEHOLDERnear the existing history placeholder constant inastrbot/core/provider/sources/openai_source.py, for example:IMAGE_UNAVAILABLE_PLACEHOLDER = "[Image unavailable]"
- Ensure that this constant is imported/used only within this module (no circular imports), and update any other hardcoded
"[Image unavailable]"occurrences, if present, to useIMAGE_UNAVAILABLE_PLACEHOLDERinstead.
There was a problem hiding this comment.
Code Review
This pull request introduces functionality to strip image payloads from conversation history messages, replacing them with placeholders to prevent image data leakage in subsequent turns. It also prioritizes local image references over remote URLs and handles image preprocessing failures gracefully. The review feedback suggests refactoring duplicated image-stripping logic to reuse helper functions, improving URL scheme parsing to be case-insensitive, and correcting double-escaped backslashes in raw strings within the tests.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| persisted_parts: list[Any] = [] | ||
| previous_was_image_placeholder = False | ||
| for part in message.content: | ||
| if getattr(part, "_no_save", False): | ||
| continue | ||
| if _is_image_content_part(part): | ||
| if previous_was_image_placeholder: | ||
| continue | ||
| persisted_parts.append(TextPart(text=_IMAGE_HISTORY_PLACEHOLDER)) | ||
| previous_was_image_placeholder = True | ||
| continue | ||
| previous_was_image_placeholder = False | ||
| persisted_parts.append(part) | ||
| message_data["content"] = [ | ||
| part.model_dump() | ||
| for part in message.content | ||
| if not getattr(part, "_no_save", False) | ||
| part.model_dump() if isinstance(part, ContentPart) else part | ||
| for part in persisted_parts | ||
| ] |
There was a problem hiding this comment.
在 dump_messages_with_checkpoints 中,剥离图片并合并连续占位符的逻辑与 strip_images_from_content_parts 函数高度重复。可以通过先过滤掉 _no_save 的部分,然后直接复用 strip_images_from_content_parts 来简化代码,避免重复实现。
filtered_parts = [part for part in message.content if not getattr(part, "_no_save", False)]
persisted_parts = strip_images_from_content_parts(filtered_parts)
message_data["content"] = [
part.model_dump() if isinstance(part, ContentPart) else part
for part in persisted_parts
]References
- When implementing similar functionality for different cases (e.g., direct vs. quoted attachments), refactor the logic into a shared helper function to avoid code duplication.
| def _is_remote_image_ref(image_ref: str) -> bool: | ||
| return image_ref.startswith("http://") or image_ref.startswith("https://") |
There was a problem hiding this comment.
为了更健壮地处理可能包含大写字母的 URL 协议头(例如 HTTP:// 或 HTTPS://),建议在判断时先转换为小写,并使用 startswith 接收元组参数来简化代码。
| def _is_remote_image_ref(image_ref: str) -> bool: | |
| return image_ref.startswith("http://") or image_ref.startswith("https://") | |
| def _is_remote_image_ref(image_ref: str) -> bool: | |
| return image_ref.lower().startswith(("http://", "https://")) |
| r"C:\\tmp\\a.jpg", | ||
| "https://example.com/a", | ||
| r"C:\\tmp\\b.jpg", |
There was a problem hiding this comment.
- dump_messages 复用 strip_images_from_content_parts,去掉重复逻辑 - finalize 不再双重 normalize,remote/data scheme 大小写不敏感 - Image unavailable 抽成常量;同步更新 openai_source 既有测试预期 - ruff format / import 整理
问题
在多轮对话和引用消息识图场景中,模型容易“看错图”或“看到不存在的图”,常见表现包括:
image_url(含data:image...base64)被反复回放到后续请求,旧图与当前图混在一起。multimedia.nt.qq.com.cn等易过期 CDN),模型继续尝试“看”死链。改动
1. 历史落库 / 加载时剥离图片载荷
astrbot/core/agent/message.pydump_messages_with_checkpoints:持久化时将image_url替换为文本占位[Image]bind_checkpoint_messages/strip_images_from_history_messages:加载历史时同样剥离,兼容旧库中已污染的历史2. OpenAI source:解析失败不再回退保留原 URL
astrbot/core/provider/sources/openai_source.py_transform_content_part在 URL 为空、预处理异常、或 resolve 结果为空时,统一改为文本[Image unavailable],避免死链继续进入模型请求3. 请求图片本地优先 + 去重限量
astrbot/core/astr_main_agent.py_finalize_request_image_urls:本地路径 / embedded base64 优先于远程 URL,并限制总数build_main_agent加载conversation.history时走剥离逻辑4. 测试
tests/test_image_history_strip.pytests/test_openai_source.py中“解析失败仍保留原图”的旧预期行为变化说明(有意为之)
[Image]文本占位。image_urls/ 当前 content)。未改动默认
max_quoted_fallback_images配置值,避免影响既有部署策略;本 PR 只做防串图的核心路径修复。根据审查意见的后续更新(
b2407e0d)已根据 Sourcery / Gemini Code Assist 的 review,以及 CI 失败项做了第二轮收敛:
dump_messages_with_checkpoints改为先过滤_no_save,再复用strip_images_from_content_parts,避免两套“连续 image 折叠为[Image]”状态机分叉。_finalize_request_image_urls内部已normalize_and_dedupe_strings,调用侧不再包一层。http(s)/data:/base64://判断统一.lower()。IMAGE_UNAVAILABLE_PLACEHOLDER,避免魔法字符串散落。test_image_history_strip.py路径写法,并覆盖本地优先断言test_prepare_chat_payload_keeps_original_context_image_when_materialization_fails更新为 drop 行为预期ruff format/ import 整理,修复format-check与Run pytest suite关于“失败图片是否保留 alt/description 元数据”的建议:当前 AstrBot 图 part 主路径基本不携带稳定 alt 字段,为控制范围未额外拼接描述文本;若后续 content part 协议补齐 metadata,可再扩展占位文案。
测试计划
tests/test_image_history_strip.py相关用例通过format-check通过Run pytest suite通过相关背景
桌面安装版实测:历史库清理 + 上述代码修复后,引用图可正确识别;仅清库不修代码,后续仍可能因落库/死链回退再次复发。