57 lines
2.1 KiB
Python
57 lines
2.1 KiB
Python
"""Bound model-facing observations while keeping full, task-scoped evidence."""
|
||
|
||
from pathlib import Path
|
||
import re
|
||
from typing import Any
|
||
|
||
from artemis.config.constants import PLUGIN_PROMPT_TEXT_CHARS
|
||
|
||
|
||
def result_path(engine: Any, call_id: str) -> Path:
|
||
if not re.fullmatch(r"[a-f0-9]{32}", call_id):
|
||
raise ValueError("插件结果编号无效")
|
||
root = Path(engine.base_dir).resolve()
|
||
path = root / "evidence" / "plugin-results" / f"{call_id}.txt"
|
||
if not path.resolve().is_relative_to(root):
|
||
raise ValueError("插件结果路径超出任务目录")
|
||
return path
|
||
|
||
|
||
def save_result_text(engine: Any, call_id: str, text: str) -> None:
|
||
path = result_path(engine, call_id)
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(text, encoding="utf-8")
|
||
|
||
|
||
def bounded_result_text(
|
||
text: str, *, call_id: str = "", limit: int = PLUGIN_PROMPT_TEXT_CHARS
|
||
) -> str:
|
||
if len(text) <= limit:
|
||
return text
|
||
reference = (
|
||
f'完整结果可用 plugin_read_result(call_id="{call_id}", offset=0) 分页读取。'
|
||
if call_id
|
||
else "完整结果保存在本步骤的 trace；需要目标控件时可缩小观察范围重新读取。"
|
||
)
|
||
notice = f"\n[观察文本已节选，原文 {len(text)} 字符。{reference}]\n"
|
||
available = max(0, limit - len(notice))
|
||
if not available:
|
||
return notice[:limit]
|
||
head = available * 3 // 4
|
||
return text[:head] + notice + text[-(available - head) :]
|
||
|
||
|
||
def read_result_text(engine: Any, call_id: str, offset: int = 0) -> str:
|
||
if isinstance(offset, bool) or not isinstance(offset, int) or offset < 0:
|
||
raise ValueError("offset 必须为非负整数")
|
||
text = result_path(engine, call_id).read_text(encoding="utf-8")
|
||
if offset > len(text):
|
||
raise ValueError(f"offset 超出结果长度 {len(text)}")
|
||
end = min(len(text), offset + PLUGIN_PROMPT_TEXT_CHARS)
|
||
return (
|
||
f"插件结果 {call_id}，字符 {offset}:{end}/{len(text)}；"
|
||
+ (f"下一页 offset={end}" if end < len(text) else "已到结尾")
|
||
+ "\n"
|
||
+ text[offset:end]
|
||
)
|