153f02f5e9
- `_append_memory_sections` 함수를 추가하여 압축 모드에서 기억된 광맥과 마지막 행동 정보를 상태 요약에 포함하도록 개선 - `main.py`에서 상태 요약 생성 시 메모리 섹션을 자동으로 추가하여 AI의 의사결정에 필요한 정보를 제공 - `README.md`에 새로운 메모리 기능 및 사용 방법에 대한 설명 추가 - `state_reader.py`에서 인벤토리 판독 로직을 개선하여 캐시 사용 조건을 명확히 하여 안정성 향상
247 lines
9.1 KiB
Python
247 lines
9.1 KiB
Python
"""
|
|
main.py — 순수 AI 플레이 버전
|
|
|
|
치트 없이 실제 게임 메커니즘으로 플레이.
|
|
걷기/채굴/제작에 실제 시간이 걸리므로 ACTION_DELAY를 설정.
|
|
|
|
핵심 변경:
|
|
- 시작 시 플레이어 접속 여부 확인
|
|
- game.player → game.players[1] (RCON 호환)
|
|
- JSON 파싱 실패 시 재시도 (ai_planner 내부)
|
|
"""
|
|
import time
|
|
import os
|
|
import json
|
|
from factorio_rcon import FactorioRCON
|
|
from state_reader import StateReader
|
|
from context_compressor import ContextCompressor
|
|
from ai_planner import AIPlanner
|
|
from action_executor import ActionExecutor
|
|
from agent_last_action_memory import save_last_action_memory, load_last_action_memory
|
|
from ore_patch_memory import load_ore_patch_memory, compute_distance_sq
|
|
|
|
RCON_HOST = os.getenv("FACTORIO_HOST", "127.0.0.1")
|
|
RCON_PORT = int(os.getenv("FACTORIO_PORT", "25575"))
|
|
RCON_PASSWORD = os.getenv("FACTORIO_PASSWORD", "factorio_ai")
|
|
ACTION_DELAY = 0.2
|
|
LOG_FILE = "agent_log.jsonl"
|
|
|
|
# 건물 N개 이상이면 압축 모드로 전환
|
|
COMPRESS_THRESHOLD = 50
|
|
|
|
|
|
def check_player(rcon: FactorioRCON) -> bool:
|
|
"""서버에 접속한 플레이어가 있는지 확인하고 안내"""
|
|
print("\n[초기화] 플레이어 접속 확인 중...")
|
|
|
|
for attempt in range(30): # 최대 30초 대기
|
|
result = rcon.check_player()
|
|
if result:
|
|
info = rcon.lua("""
|
|
local p = game.players[1]
|
|
if p then
|
|
rcon.print(p.name .. " @ " .. string.format("%.0f, %.0f", p.position.x, p.position.y))
|
|
end
|
|
""")
|
|
print(f"[초기화] OK 플레이어 발견: {info}")
|
|
return True
|
|
|
|
if attempt == 0:
|
|
print("[초기화] WARN 접속한 플레이어가 없습니다!")
|
|
print("[초기화] 팩토리오 클라이언트로 이 서버에 접속하세요.")
|
|
print(f"[초기화] 서버 주소: {RCON_HOST}")
|
|
print(f"[초기화] 대기 중... (최대 30초)")
|
|
|
|
time.sleep(1)
|
|
|
|
print("[오류] 30초 내에 플레이어가 접속하지 않았습니다.")
|
|
print("[오류] 팩토리오 클라이언트로 서버에 접속한 후 다시 실행하세요.")
|
|
return False
|
|
|
|
|
|
def _append_memory_sections(summary: str) -> str:
|
|
"""
|
|
압축 모드(중반/후반)에서는 context_compressor 출력만 들어가서
|
|
state_reader의 "기억된 광맥/마지막 행동" 섹션이 사라질 수 있다.
|
|
이 정보를 프롬프트에 다시 덧붙여 폴백/복구를 가능하게 한다.
|
|
"""
|
|
ore_mem = load_ore_patch_memory()
|
|
last_action = load_last_action_memory()
|
|
|
|
# compressed summary에도 "위치: (x, y)"가 포함되므로, 여기서 플레이어 좌표만 뽑는다.
|
|
m = None
|
|
try:
|
|
import re
|
|
|
|
m = re.search(r"위치:\s*\(\s*(-?\d+)\s*,\s*(-?\d+)\s*\)", summary)
|
|
except Exception:
|
|
m = None
|
|
|
|
px = py = None
|
|
if m:
|
|
try:
|
|
px = int(m.group(1))
|
|
py = int(m.group(2))
|
|
except Exception:
|
|
px = py = None
|
|
|
|
mem_lines: list[str] = []
|
|
|
|
mem_lines.append("### 기억된 광맥 (좌표)")
|
|
if ore_mem:
|
|
known: list[tuple[int, str, int, int, object]] = []
|
|
for ore, patches in ore_mem.items():
|
|
if not isinstance(patches, list):
|
|
continue
|
|
for patch in patches:
|
|
if not isinstance(patch, dict):
|
|
continue
|
|
tx = patch.get("tile_x")
|
|
ty = patch.get("tile_y")
|
|
if isinstance(tx, int) and isinstance(ty, int):
|
|
if px is not None and py is not None:
|
|
dist = int(compute_distance_sq(px, py, tx, ty) ** 0.5)
|
|
else:
|
|
dist = 10**18
|
|
known.append((dist, ore, tx, ty, patch.get("count", "?")))
|
|
|
|
known.sort(key=lambda x: x[0])
|
|
for dist, ore, tx, ty, cnt in known[:10]:
|
|
if dist >= 10**18:
|
|
mem_lines.append(f"- {ore}: ({tx},{ty}) [거리: ~?타일] count={cnt}")
|
|
else:
|
|
mem_lines.append(f"- {ore}: ({tx},{ty}) [거리: ~{dist}타일] count={cnt}")
|
|
else:
|
|
mem_lines.append("- 없음")
|
|
|
|
mem_lines.append("")
|
|
mem_lines.append("### 마지막 행동(기억)")
|
|
if last_action and isinstance(last_action, dict) and last_action.get("action"):
|
|
act = last_action.get("action", "")
|
|
success = last_action.get("success", None)
|
|
msg = last_action.get("message", "")
|
|
mem_lines.append(f"- action={act} success={success} message={msg}")
|
|
params = last_action.get("params", {})
|
|
if params:
|
|
try:
|
|
mem_lines.append(f"- params={json.dumps(params, ensure_ascii=False)}")
|
|
except Exception:
|
|
pass
|
|
else:
|
|
mem_lines.append("- 없음")
|
|
|
|
return summary + "\n\n" + "\n".join(mem_lines)
|
|
|
|
|
|
def run():
|
|
print("=" * 60)
|
|
print(" 팩토리오 순수 AI 에이전트 (치트 없음)")
|
|
print(" - 실제 걷기 / 실제 채굴 / 실제 제작 / 건설 거리 제한")
|
|
print(" - RCON 호환: game.players[1] 사용")
|
|
print("=" * 60)
|
|
|
|
with FactorioRCON(RCON_HOST, RCON_PORT, RCON_PASSWORD) as rcon:
|
|
# ── 플레이어 접속 확인 ──
|
|
if not check_player(rcon):
|
|
return
|
|
|
|
reader = StateReader(rcon)
|
|
compressor = ContextCompressor(rcon)
|
|
planner = AIPlanner()
|
|
executor = ActionExecutor(rcon)
|
|
|
|
total_actions = 0
|
|
queue: list[dict] = []
|
|
entity_count = 0
|
|
|
|
while True:
|
|
try:
|
|
if not queue:
|
|
print(f"\n{'='*50}")
|
|
print(f"[재계획] 총 {total_actions}개 실행 완료")
|
|
|
|
# ── 압축 레벨 자동 선택 ──
|
|
if entity_count < COMPRESS_THRESHOLD:
|
|
print("[상태] 초반 모드 - 상세 상태 수집")
|
|
state = reader.get_full_state()
|
|
summary = reader.summarize_for_ai(state)
|
|
entity_count = sum(
|
|
v for v in state.get("buildings", {}).values()
|
|
if isinstance(v, int)
|
|
)
|
|
elif entity_count < 200:
|
|
print(f"[상태] 중반 모드 - 구역 압축 (건물 {entity_count}개)")
|
|
summary = compressor.get_compressed_state(detail_level=1)
|
|
summary = _append_memory_sections(summary)
|
|
else:
|
|
print(f"[상태] 후반 모드 - 글로벌 압축 (건물 {entity_count}개)")
|
|
summary = compressor.get_compressed_state(detail_level=0)
|
|
summary = _append_memory_sections(summary)
|
|
|
|
global_info = compressor._get_global_summary()
|
|
entity_count = global_info.get("total_entities", entity_count)
|
|
|
|
# ── AI 계획 ──
|
|
queue = planner.decide(summary)
|
|
if not queue:
|
|
print("[경고] AI가 행동 반환 안 함. 10초 후 재시도")
|
|
time.sleep(10)
|
|
continue
|
|
|
|
# ── 행동 실행 ──
|
|
action = queue.pop(0)
|
|
total_actions += 1
|
|
act = action.get("action", "")
|
|
reason = action.get("reason", "")
|
|
|
|
print(f"\n[{total_actions}] {act}")
|
|
print(f" 이유: {reason}")
|
|
print(f" params: {json.dumps(action.get('params',{}), ensure_ascii=False)}")
|
|
|
|
success, message = executor.execute(action)
|
|
status = "OK" if success else "FAIL"
|
|
print(f" 결과: {status} {message}")
|
|
|
|
# 재시작 시 직전 행동을 참고할 수 있도록 메모리에 저장
|
|
try:
|
|
save_last_action_memory({
|
|
"action": act,
|
|
"params": action.get("params", {}),
|
|
"success": success,
|
|
"message": message,
|
|
"timestamp": time.time(),
|
|
})
|
|
except Exception:
|
|
pass
|
|
|
|
planner.record_feedback(action, success, message)
|
|
|
|
if not success:
|
|
print(" -> 실패. 재계획 요청")
|
|
queue.clear()
|
|
|
|
_log(total_actions, action, success, message)
|
|
time.sleep(ACTION_DELAY)
|
|
|
|
except KeyboardInterrupt:
|
|
print(f"\n종료 (총 {total_actions}개 실행)")
|
|
break
|
|
except Exception as e:
|
|
print(f"[오류] {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
time.sleep(5)
|
|
|
|
|
|
def _log(step, action, success, message):
|
|
with open(LOG_FILE, "a", encoding="utf-8") as f:
|
|
f.write(json.dumps({
|
|
"step": step, "action": action,
|
|
"success": success, "message": message,
|
|
"timestamp": time.time()
|
|
}, ensure_ascii=False) + "\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run()
|