feat: 추가된 메모리 기능으로 광맥 및 마지막 행동 저장
- 에이전트가 발견한 광맥 좌표를 `ore_patch_memory.json`에 저장하여 재시작 시 활용 가능 - 마지막 실행한 행동과 결과를 `agent_last_action_memory.json`에 저장하여 다음 상태 요약에서 참고 가능 - `state_reader.py`에서 메모리 로드 및 상태 요약에 포함 - `ai_planner.py`에서 시스템 프롬프트에 기억된 광맥 및 마지막 행동 관련 가이드 추가 - `README.md`에 새로운 메모리 기능 설명 추가
This commit is contained in:
141
ore_patch_memory.py
Normal file
141
ore_patch_memory.py
Normal file
@@ -0,0 +1,141 @@
|
||||
"""
|
||||
ore_patch_memory.py
|
||||
|
||||
에이전트가 "발견"한 광맥(자원 엔티티) 좌표를 파일로 기억하기 위한 유틸.
|
||||
|
||||
메모리 포맷:
|
||||
- key: ore name (예: "iron-ore", "stone", "coal")
|
||||
- value: patch 좌표들의 리스트
|
||||
- { tile_x, tile_y, last_seen, count? }
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
|
||||
ORE_PATCH_CACHE_FILE = "ore_patch_memory.json"
|
||||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
|
||||
|
||||
def _cache_path(cache_dir: str | None = None) -> str:
|
||||
base = cache_dir or BASE_DIR
|
||||
return os.path.join(base, ORE_PATCH_CACHE_FILE)
|
||||
|
||||
|
||||
def load_ore_patch_memory(cache_dir: str | None = None) -> dict[str, list[dict[str, Any]]]:
|
||||
path = _cache_path(cache_dir)
|
||||
try:
|
||||
if not os.path.exists(path):
|
||||
return {}
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
raw = f.read().strip()
|
||||
if not raw:
|
||||
return {}
|
||||
data = json.loads(raw)
|
||||
if not isinstance(data, dict):
|
||||
return {}
|
||||
|
||||
# 과거(단일 좌표) 포맷 호환: ore -> dict
|
||||
converted: dict[str, list[dict[str, Any]]] = {}
|
||||
for ore, v in data.items():
|
||||
if isinstance(v, list):
|
||||
converted[ore] = [x for x in v if isinstance(x, dict)]
|
||||
elif isinstance(v, dict):
|
||||
converted[ore] = [v]
|
||||
return converted
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
|
||||
def save_ore_patch_memory(
|
||||
memory: dict[str, list[dict[str, Any]]],
|
||||
cache_dir: str | None = None,
|
||||
) -> None:
|
||||
path = _cache_path(cache_dir)
|
||||
try:
|
||||
if not isinstance(memory, dict):
|
||||
return
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
json.dump(memory, f, ensure_ascii=False)
|
||||
except Exception:
|
||||
# 메모리는 부가 기능이므로 실패해도 전체 동작이 깨지면 안 됨
|
||||
pass
|
||||
|
||||
|
||||
def update_ore_patch_memory(
|
||||
ore_patch_memory: dict[str, list[dict[str, Any]]],
|
||||
ore: str,
|
||||
tile_x: int,
|
||||
tile_y: int,
|
||||
*,
|
||||
last_seen: float | None = None,
|
||||
count: int | None = None,
|
||||
max_patches_per_ore: int = 8,
|
||||
dedupe_if_same_tile: bool = True,
|
||||
) -> dict[str, list[dict[str, Any]]]:
|
||||
"""
|
||||
memory[ore]의 patch 리스트에 (tile_x, tile_y)를 추가/갱신한다.
|
||||
|
||||
- 같은 타일이면 last_seen 갱신(+ count가 있으면 count도 더 큰 값을 우선)
|
||||
- 다른 타일이면 리스트에 추가하고 최근(last_seen) 기준으로 max_patches_per_ore까지 유지
|
||||
"""
|
||||
if not isinstance(ore_patch_memory, dict):
|
||||
ore_patch_memory = {}
|
||||
if not ore:
|
||||
return ore_patch_memory
|
||||
|
||||
ore = str(ore)
|
||||
tile_x = int(tile_x)
|
||||
tile_y = int(tile_y)
|
||||
now = time.time() if last_seen is None else float(last_seen)
|
||||
|
||||
patches = ore_patch_memory.get(ore)
|
||||
if not isinstance(patches, list):
|
||||
patches = []
|
||||
else:
|
||||
patches = [p for p in patches if isinstance(p, dict)]
|
||||
|
||||
# 같은 타일이면 해당 patch를 업데이트한다.
|
||||
updated = False
|
||||
for p in patches:
|
||||
px = p.get("tile_x")
|
||||
py = p.get("tile_y")
|
||||
if not dedupe_if_same_tile:
|
||||
continue
|
||||
if isinstance(px, int) and isinstance(py, int) and px == tile_x and py == tile_y:
|
||||
p["last_seen"] = now
|
||||
if count is not None:
|
||||
prev_count = p.get("count")
|
||||
if prev_count is None or not isinstance(prev_count, int):
|
||||
p["count"] = int(count)
|
||||
else:
|
||||
p["count"] = int(count) if int(count) >= int(prev_count) else prev_count
|
||||
updated = True
|
||||
break
|
||||
|
||||
if not updated:
|
||||
patch: dict[str, Any] = {
|
||||
"tile_x": tile_x,
|
||||
"tile_y": tile_y,
|
||||
"last_seen": now,
|
||||
}
|
||||
if count is not None:
|
||||
patch["count"] = int(count)
|
||||
patches.append(patch)
|
||||
|
||||
# 최신 순으로 자르고 다시 저장
|
||||
patches.sort(key=lambda x: float(x.get("last_seen", 0.0)), reverse=True)
|
||||
ore_patch_memory[ore] = patches[: int(max_patches_per_ore)]
|
||||
|
||||
return ore_patch_memory
|
||||
|
||||
|
||||
def compute_distance_sq(px: float, py: float, tile_x: int, tile_y: int) -> float:
|
||||
dx = float(tile_x) - float(px)
|
||||
dy = float(tile_y) - float(py)
|
||||
return dx * dx + dy * dy
|
||||
|
||||
Reference in New Issue
Block a user