File size: 3,301 Bytes
6f90f5c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 | import importlib.util
import json
import math
import os
import sysconfig
import time
from pathlib import Path
from typing import Any, Dict, Optional
# ---------------------------------------------------------------------------
# Stdlib logging compatibility
# ---------------------------------------------------------------------------
# This file is intentionally named logging.py.
# When scripts under eval_agent/ are executed directly, `import logging` may
# resolve to this file instead of Python stdlib logging.
# To avoid breaking stdlib users, load stdlib logging and re-export its API.
_stdlib_logging_path = os.path.join(sysconfig.get_path("stdlib"), "logging", "__init__.py")
_stdlib_spec = importlib.util.spec_from_file_location("_stdlib_logging", _stdlib_logging_path)
_stdlib_logging = importlib.util.module_from_spec(_stdlib_spec)
assert _stdlib_spec is not None and _stdlib_spec.loader is not None
_stdlib_spec.loader.exec_module(_stdlib_logging)
for _name, _value in vars(_stdlib_logging).items():
if _name not in globals():
globals()[_name] = _value
def _sanitize_for_json(value: Any) -> Any:
if isinstance(value, float):
return value if math.isfinite(value) else None
if isinstance(value, dict):
return {k: _sanitize_for_json(v) for k, v in value.items()}
if isinstance(value, list):
return [_sanitize_for_json(v) for v in value]
if isinstance(value, tuple):
return [_sanitize_for_json(v) for v in value]
if hasattr(value, "item") and callable(getattr(value, "item")):
try:
return _sanitize_for_json(value.item())
except Exception:
return str(value)
return value
def _normalize_experiment_root(results_dir: str) -> Path:
root = Path(results_dir).resolve()
if root.name == "results":
return root.parent.parent
if root.name.startswith("gen_"):
return root.parent
return root
def _ensure_memory_dir(results_dir: str) -> Path:
memory_dir = _normalize_experiment_root(results_dir) / "eval_agent_memory"
memory_dir.mkdir(parents=True, exist_ok=True)
return memory_dir
def log_event(results_dir: str, event_type: str, payload: Optional[Dict[str, Any]] = None) -> None:
"""Append one structured event to eval_agent_memory/agent_behavior.jsonl."""
if not results_dir:
return
try:
memory_dir = _ensure_memory_dir(results_dir)
event = {
"timestamp": time.time(),
"event_type": event_type,
**(payload or {}),
}
event = _sanitize_for_json(event)
with open(memory_dir / "agent_behavior.jsonl", "a", encoding="utf-8") as f:
f.write(json.dumps(event, ensure_ascii=False) + "\n")
except Exception:
# Never fail caller because of logging.
pass
def save_text_artifact(results_dir: str, relative_path: str, content: str) -> None:
"""Save text artifact under eval_agent_memory/<relative_path>."""
if not results_dir:
return
try:
memory_dir = _ensure_memory_dir(results_dir)
artifact_path = memory_dir / relative_path
artifact_path.parent.mkdir(parents=True, exist_ok=True)
artifact_path.write_text(content, encoding="utf-8")
except Exception:
pass
|