chenmingxuan
e1ea099592
Install and update artemis-use with task diagnostics and plugin guidance. Preserve cross-device file receipts, review canonical reports, guard stage transitions and reduce small history summaries. Improve exploration lifecycle, request recovery and browser live connections.
60 lines
2.4 KiB
Python
60 lines
2.4 KiB
Python
"""Deliver persisted plugin results without substituting the Android observation."""
|
|
|
|
import base64
|
|
import io
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from langchain_core.messages import BaseMessage
|
|
from PIL import Image
|
|
|
|
from artemis.tools.tool_wrapper import tool_result_messages
|
|
from artemis.plugins.delivery import result_blocks
|
|
|
|
|
|
def plugin_result_messages(report: Any, engine: Any) -> dict[str, list[BaseMessage]]:
|
|
if not isinstance(report, dict):
|
|
return {}
|
|
results = {}
|
|
for action in report.get("execution", []):
|
|
if not isinstance(action, dict) or not str(action.get("action", "")).startswith("plug_"):
|
|
continue
|
|
call_id = action.get("tool_call_id")
|
|
if not call_id:
|
|
continue
|
|
status = action.get("plugin_status")
|
|
text = action.get("result") or "此插件操作未完成或未执行；请重新观察后再决定下一步。"
|
|
blocks = result_blocks(
|
|
str(text),
|
|
call_id=str(action.get("text_ref") or ""),
|
|
attachments=action.get("attachments"),
|
|
)
|
|
names = action.get("image_names") or [action.get("post_image_name")]
|
|
for name in dict.fromkeys(names):
|
|
if not name or engine is None:
|
|
continue
|
|
try:
|
|
image = Path(engine.get_image_path(name)).read_bytes()
|
|
# DataEngine keeps the original bytes under a .jpg filename;
|
|
# computer plugins may actually return PNG or WebP.
|
|
with Image.open(io.BytesIO(image)) as captured:
|
|
mime = Image.MIME.get(captured.format)
|
|
if mime not in {"image/jpeg", "image/png", "image/webp"}:
|
|
raise ValueError("Unsupported plugin image format")
|
|
except (OSError, ValueError):
|
|
blocks.append({"type": "text", "text": "本次插件截图无法读取，请重新观察。"})
|
|
continue
|
|
blocks.append(
|
|
{
|
|
"type": "image_url",
|
|
"image_url": {"url": f"data:{mime};base64," + base64.b64encode(image).decode()},
|
|
}
|
|
)
|
|
results[str(call_id)] = tool_result_messages(
|
|
str(call_id),
|
|
blocks,
|
|
name=action["action"],
|
|
status="success" if status == "success" else "error",
|
|
)
|
|
return results
|