413 lines
17 KiB
Python
413 lines
17 KiB
Python
"""
|
|
ai_planner.py — 순수 AI 플레이 버전
|
|
|
|
핵심 변경사항 (치트 모드 대비):
|
|
- 이동에 실제 시간이 걸림 → 불필요한 장거리 이동 최소화
|
|
- 채굴은 자원 패치 근처에서만 가능 → 반드시 move 후 mine
|
|
- 제작은 재료가 인벤토리에 있어야 함 → 재료 확보 순서 중요
|
|
- 건설은 건설 거리 내에서만 가능 → 배치 전 move 필수
|
|
- AI가 이 제약을 이해하고 행동 순서를 계획해야 함
|
|
|
|
LLM 백엔드: 로컬 Ollama (structured output으로 JSON 스키마 강제)
|
|
- OLLAMA_HOST: Ollama 서버 주소 (기본값: http://192.168.50.67:11434)
|
|
- OLLAMA_MODEL: 사용할 모델 (기본값: qwen3.5:9b)
|
|
"""
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
import threading
|
|
import traceback
|
|
import ollama
|
|
|
|
|
|
OLLAMA_MODEL = os.environ.get("OLLAMA_MODEL", "qwen3.5:9b-nothink")
|
|
OLLAMA_HOST = os.environ.get("OLLAMA_HOST", "http://192.168.50.67:11434")
|
|
|
|
|
|
SYSTEM_PROMPT = """당신은 팩토리오 게임을 순수하게 플레이하는 AI 에이전트입니다.
|
|
치트나 텔레포트 없이, 실제 게임 메커니즘만 사용합니다.
|
|
게임 상태와 이전 행동 결과를 분석해서 스스로 판단하고 계획을 세웁니다.
|
|
|
|
## 재시작/마지막 행동 메모리
|
|
state_reader가 상태 요약에 포함하는 `마지막 행동(기억)` 섹션을 확인하세요.
|
|
직전에 실패한 action을 그대로 반복하지 말고, 실패 메시지/상태 변화에 맞춰 원인을 먼저 해결한 뒤 다음 action을 선택하세요.
|
|
|
|
## 핵심 제약 사항 (반드시 준수!)
|
|
1. **이동은 실제 걷기** — 먼 거리는 시간이 오래 걸림. 불필요한 왕복 최소화
|
|
2. **채굴은 자원 패치 근처에서만 가능** — 반드시 자원 위치로 move한 후 mine_resource
|
|
3. **제작은 재료가 있어야 함** — iron-plate 없이 iron-gear-wheel 못 만듬.
|
|
재료 확인 후 craft_item. 재료 부족하면 먼저 채굴→제련
|
|
4. **건설은 건설 거리 내에서만 가능** — 배치할 좌표 근처로 move한 후 place_entity
|
|
5. **자원은 유한** — 직접 채굴해야 하고, 제련소에 넣어야 plate가 됨
|
|
|
|
## 효율적인 행동 패턴
|
|
- 같은 구역 작업을 묶어서 (이동 최소화)
|
|
- move → mine/place/insert 순서로 항상 위치 먼저 확보
|
|
- 채굴 → 제련 → 제작 → 건설 흐름 유지
|
|
- 한 번에 넉넉히 채굴 (왕복 줄이기)
|
|
|
|
## 팩토리오 자동화 핵심 지식
|
|
- 채굴기(burner-mining-drill)는 광맥 위에 배치해야 작동
|
|
- 제련소(stone-furnace)에 ore + 석탄 넣으면 plate 생산
|
|
- 채굴기 출력 → inserter → 벨트 → inserter → 제련소/조립기
|
|
- 제련소/보일러/채굴기는 석탄 연료 필요
|
|
- 전력: offshore-pump → pipe → boiler → steam-engine → small-electric-pole
|
|
- 자동화 연구팩: iron-gear-wheel + iron-plate → assembling-machine
|
|
- 건물 배치 전 반드시: 1) 인벤토리에 아이템 있는지 2) 가까이 있는지 확인
|
|
|
|
## 전체 action 목록
|
|
|
|
### 탐색 (★ 자원 없을 때 최우선! 걸으면서 자원 스캔)
|
|
- "explore" → {"direction": "east|west|north|south|...", "max_steps": 200, "wanted_ores": ["stone","coal", ...]} (선택)
|
|
★ 자원이 보이지 않을 때 반드시 explore 사용! move 대신!
|
|
★ `wanted_ores`가 있으면: 해당 자원이 발견될 때까지 계속 걷고, 다른 자원(예: iron-ore)만 계속 발견되더라도 즉시 멈추지 말 것
|
|
★ 상태 요약에 "기억된 광맥" 좌표가 있으면, 그 좌표로 먼저 이동(move)해 채굴(mine_resource)을 시도
|
|
★ 방향으로 걸으면서 반경 50타일 자원 스캔, 발견 즉시 멈춤
|
|
★ 장애물 자동 감지. 막히면 다른 방향 시도
|
|
★ 한 방향 실패 시 다음 방향 (east→north→south→west)
|
|
|
|
### 이동 (자원 좌표를 알 때만 사용)
|
|
- "move" → {"x": int, "y": int}
|
|
주의: 자원/건물의 정확한 좌표를 알 때만 사용. 탐색에는 explore!
|
|
|
|
### 채굴 (자원 패치 근처에서만 작동)
|
|
- "mine_resource" → {"ore": "iron-ore", "count": int}
|
|
채굴 가능: iron-ore, copper-ore, coal, stone
|
|
권장: count는 20~50 단위로 (작으면 비효율, 크면 오래 걸림)
|
|
|
|
### 제작 (인벤토리에 재료 필요!)
|
|
- "craft_item" → {"item": str, "count": int}
|
|
레시피 예시:
|
|
stone-furnace: stone 5개
|
|
burner-mining-drill: iron-gear-wheel 3 + iron-plate 3 + stone-furnace 1
|
|
transport-belt: iron-gear-wheel 1 + iron-plate 1
|
|
burner-inserter: iron-gear-wheel 1 + iron-plate 1
|
|
iron-gear-wheel: iron-plate 2
|
|
pipe: iron-plate 1
|
|
|
|
### 건물 배치 (건설 거리 내에서만!)
|
|
- "place_entity" → {"name": str, "x": int, "y": int, "direction": "north|south|east|west"}
|
|
주의: 1) 인벤토리에 아이템 필요 2) 가까이 있어야 함 (약 10칸 내)
|
|
|
|
### 건설 자동화 (Blueprint)
|
|
- "build_smelting_line" → {"ore": "iron-ore|copper-ore", "x": int, "y": int, "furnace_count": int}
|
|
★ 제련소 라인을 한 번에 배치. furnace_count개의 stone-furnace를 y축 방향으로 일렬 배치.
|
|
★ 각 furnace에 석탄 + 광석 자동 투입. 인벤토리에 stone-furnace와 연료가 있어야 함.
|
|
★ 예: {"action": "build_smelting_line", "params": {"ore": "iron-ore", "x": -90, "y": -70, "furnace_count": 4}}
|
|
|
|
### 벨트 라인 (걸어다니면서 하나씩 배치 — 시간 많이 걸림)
|
|
- "place_belt_line" → {"from_x": int, "from_y": int, "to_x": int, "to_y": int}
|
|
|
|
### 연료/아이템 삽입 (건설 거리 내에서)
|
|
- "insert_to_entity" → {"x": int, "y": int, "item": "coal", "count": int}
|
|
|
|
### 조립기 레시피 설정
|
|
- "set_recipe" → {"x": int, "y": int, "recipe": str}
|
|
|
|
### 연구
|
|
- "start_research" → {"tech": "automation"}
|
|
|
|
### 대기
|
|
- "wait" → {"seconds": int}"""
|
|
|
|
|
|
def _debug_enabled() -> bool:
|
|
v = os.environ.get("AI_DEBUG", "").strip().lower()
|
|
return v in ("1", "true", "yes", "on")
|
|
|
|
|
|
class AIPlanner:
|
|
def __init__(self):
|
|
self.step = 0
|
|
self.feedback_log: list[dict] = []
|
|
self._fallback_explore_turn = 0
|
|
self.long_term_goal = (
|
|
"완전 자동화 달성: "
|
|
"석탄 채굴 → 철 채굴+제련 자동화 → 구리 채굴+제련 → "
|
|
"전력 구축 → automation 연구 → 빨간 과학팩 자동 생산"
|
|
)
|
|
|
|
def decide(self, state_summary: str) -> list[dict]:
|
|
self.step += 1
|
|
feedback_text = self._format_feedback()
|
|
|
|
user_message = (
|
|
f"## 스텝 {self.step}\n\n"
|
|
f"### 현재 게임 상태\n{state_summary}\n\n"
|
|
f"{feedback_text}"
|
|
f"### 장기 목표\n{self.long_term_goal}\n\n"
|
|
"현재 상태를 분석하고, 장기 목표를 향해 지금 해야 할 행동 시퀀스를 계획하세요.\n"
|
|
"⚠️ 순수 플레이입니다. 건설/채굴/삽입 전에 반드시 move로 가까이 이동하세요.\n"
|
|
"⚠️ 제작은 재료가 있어야 합니다. 인벤토리를 확인하세요."
|
|
)
|
|
|
|
print(f"\n[AI] 요청 시작 (model={OLLAMA_MODEL})")
|
|
|
|
try:
|
|
plan = self._call_ollama(user_message)
|
|
except Exception as e:
|
|
print(f"[오류] Ollama 호출 실패: {e}")
|
|
if _debug_enabled():
|
|
traceback.print_exc()
|
|
plan = self._fallback_plan_from_summary(state_summary, last_error=str(e))
|
|
|
|
thinking = plan.get("thinking", "")
|
|
if thinking:
|
|
print(f"\n[AI] 판단:\n{thinking}\n")
|
|
|
|
actions = plan.get("actions", [])
|
|
if not isinstance(actions, list):
|
|
actions = []
|
|
if not actions:
|
|
plan = self._fallback_plan_from_summary(
|
|
state_summary,
|
|
last_error="LLM returned empty actions",
|
|
)
|
|
thinking = plan.get("thinking", "")
|
|
if thinking:
|
|
print(f"\n[AI] 판단:\n{thinking}\n")
|
|
actions = plan.get("actions", [])
|
|
|
|
print(f"[AI] 현재 목표: {plan.get('current_goal', '')}")
|
|
print(f"[AI] 완료 후: {plan.get('after_this', '')}")
|
|
|
|
actions = self._ensure_move_before_build_actions(actions)
|
|
print(f"[AI] {len(actions)}개 행동 계획됨")
|
|
return actions
|
|
|
|
def _call_ollama(self, user_message: str) -> dict:
|
|
t0 = time.perf_counter()
|
|
stop_event = threading.Event()
|
|
|
|
def _spinner():
|
|
while not stop_event.is_set():
|
|
elapsed = time.perf_counter() - t0
|
|
print(f"\r[AI] 생각 중... {elapsed:.0f}s", end="", flush=True)
|
|
stop_event.wait(1.0)
|
|
print("\r", end="")
|
|
|
|
spinner = threading.Thread(target=_spinner, daemon=True)
|
|
spinner.start()
|
|
|
|
try:
|
|
client = ollama.Client(host=OLLAMA_HOST)
|
|
response = client.chat(
|
|
model=OLLAMA_MODEL,
|
|
messages=[
|
|
{"role": "system", "content": SYSTEM_PROMPT},
|
|
{"role": "user", "content": user_message},
|
|
],
|
|
format={
|
|
"type": "object",
|
|
"properties": {
|
|
"thinking": {"type": "string"},
|
|
"current_goal": {"type": "string"},
|
|
"actions": {
|
|
"type": "array",
|
|
"items": {
|
|
"type": "object",
|
|
"properties": {
|
|
"action": {"type": "string"},
|
|
"params": {"type": "object"},
|
|
"reason": {"type": "string"},
|
|
},
|
|
"required": ["action", "params"],
|
|
},
|
|
},
|
|
"after_this": {"type": "string"},
|
|
},
|
|
"required": ["actions"],
|
|
},
|
|
options={"temperature": 0.3},
|
|
)
|
|
finally:
|
|
stop_event.set()
|
|
spinner.join()
|
|
|
|
dt = time.perf_counter() - t0
|
|
content = response.message.content
|
|
print(f"[AI] 응답 수신 ({dt:.2f}s, {len(content)}자)")
|
|
if _debug_enabled():
|
|
print(f"[AI][디버그] raw={content[:300]}")
|
|
return json.loads(content)
|
|
|
|
@staticmethod
|
|
def _ensure_move_before_build_actions(actions: list[dict]) -> list[dict]:
|
|
"""
|
|
LLM이 "move 후 배치" 순서를 놓치는 경우가 있어,
|
|
place_entity/insert_to_entity/set_recipe 앞에 최근 move가 같은 좌표가 아니면
|
|
안전장치로 move를 끼워 넣는다.
|
|
"""
|
|
out: list[dict] = []
|
|
last_move_xy: tuple[object, object] | None = None
|
|
|
|
for a in actions:
|
|
if not isinstance(a, dict):
|
|
continue
|
|
act = a.get("action")
|
|
params = a.get("params") if isinstance(a.get("params"), dict) else {}
|
|
|
|
if act == "move":
|
|
mx = params.get("x")
|
|
my = params.get("y")
|
|
last_move_xy = (mx, my)
|
|
out.append(a)
|
|
continue
|
|
|
|
if act in ("place_entity", "insert_to_entity", "set_recipe"):
|
|
tx = params.get("x")
|
|
ty = params.get("y")
|
|
if tx is not None and ty is not None:
|
|
if last_move_xy != (tx, ty):
|
|
out.append({
|
|
"action": "move",
|
|
"params": {"x": tx, "y": ty},
|
|
"reason": "안전장치: 건설/삽입 거리 제한 때문에 move 먼저",
|
|
})
|
|
last_move_xy = (tx, ty)
|
|
|
|
out.append(a)
|
|
|
|
return out
|
|
|
|
def record_feedback(self, action: dict, success: bool, message: str = ""):
|
|
self.feedback_log.append({
|
|
"action": action.get("action", ""),
|
|
"params": action.get("params", {}),
|
|
"success": success,
|
|
"message": message,
|
|
})
|
|
if len(self.feedback_log) > 15:
|
|
self.feedback_log.pop(0)
|
|
|
|
def _format_feedback(self) -> str:
|
|
if not self.feedback_log:
|
|
return ""
|
|
lines = ["### 이전 행동 결과 (성공/실패)\n"]
|
|
for fb in self.feedback_log[-8:]:
|
|
status = "OK" if fb["success"] else "FAIL"
|
|
msg = f" — {fb['message']}" if fb["message"] else ""
|
|
lines.append(
|
|
f" {status} {fb['action']} "
|
|
f"{json.dumps(fb['params'], ensure_ascii=False)}{msg}"
|
|
)
|
|
return "\n".join(lines) + "\n\n"
|
|
|
|
def set_goal(self, goal: str):
|
|
self.long_term_goal = goal
|
|
self.feedback_log.clear()
|
|
print(f"[AI] 새 목표: {goal}")
|
|
|
|
def _fallback_plan_from_summary(self, state_summary: str, last_error: str = "") -> dict:
|
|
"""
|
|
Ollama 실패 시에도 상태 요약(주변 패치·기억된 광맥 좌표)이 있으면
|
|
무한 explore 루프 대신 mine_resource / move 를 선택한다.
|
|
"""
|
|
pos = self._parse_player_position(state_summary)
|
|
if pos is None:
|
|
px = py = None
|
|
else:
|
|
px, py = pos
|
|
anchors = self._parse_ore_anchors(state_summary)
|
|
|
|
ore_priority = {"iron-ore": 0, "copper-ore": 1, "coal": 2, "stone": 3}
|
|
best: tuple[str, int, int, float] | None = None
|
|
best_key: tuple | None = None
|
|
|
|
for ore, ox, oy in anchors:
|
|
if px is not None and py is not None:
|
|
dist = ((ox - px) ** 2 + (oy - py) ** 2) ** 0.5
|
|
else:
|
|
dist = 0.0
|
|
key = (ore_priority.get(ore, 9), dist)
|
|
if best_key is None or key < best_key:
|
|
best_key = key
|
|
best = (ore, ox, oy, dist)
|
|
|
|
if best is not None:
|
|
ore, ox, oy, dist = best
|
|
move_threshold = float(os.environ.get("FALLBACK_MOVE_THRESHOLD", "200"))
|
|
actions: list[dict] = []
|
|
if px is not None and py is not None and dist > move_threshold:
|
|
actions.append({
|
|
"action": "move",
|
|
"params": {"x": ox, "y": oy},
|
|
"reason": f"폴백: 광맥까지 약 {dist:.0f}타일 — 먼저 이동",
|
|
})
|
|
actions.append({
|
|
"action": "mine_resource",
|
|
"params": {"ore": ore, "count": 35},
|
|
"reason": "폴백: 상태에 표시된 인근 광맥 채굴",
|
|
})
|
|
err_note = f" ({last_error})" if last_error else ""
|
|
return {
|
|
"thinking": (
|
|
f"Ollama를 사용할 수 없어 상태 요약의 광맥({ore}, 앵커 {ox},{oy})으로 "
|
|
f"채굴을 시도합니다.{err_note}"
|
|
),
|
|
"current_goal": f"{ore} 채굴 (휴리스틱)",
|
|
"actions": actions,
|
|
"after_this": "Ollama 복구 시 정상 계획으로 복귀",
|
|
}
|
|
|
|
dirs = [
|
|
"east", "north", "south", "west",
|
|
"northeast", "northwest", "southeast", "southwest",
|
|
]
|
|
self._fallback_explore_turn += 1
|
|
direction = dirs[(self._fallback_explore_turn - 1) % len(dirs)]
|
|
err_note = f" ({last_error})" if last_error else ""
|
|
return {
|
|
"thinking": (
|
|
f"Ollama 실패이며 상태 요약에서 광맥 좌표를 찾지 못해 "
|
|
f"{direction} 방향으로 탐색합니다.{err_note}"
|
|
),
|
|
"current_goal": "주변 탐색 (휴리스틱)",
|
|
"actions": [
|
|
{
|
|
"action": "explore",
|
|
"params": {"direction": direction, "max_steps": 200},
|
|
"reason": "폴백: 광맥 정보 없음 — 탐색",
|
|
},
|
|
],
|
|
"after_this": "자원 발견 후 채굴",
|
|
}
|
|
|
|
@staticmethod
|
|
def _parse_player_position(text: str) -> tuple[float, float] | None:
|
|
m = re.search(
|
|
r"-\s*위치:\s*\(\s*(-?\d+(?:\.\d+)?)\s*,\s*(-?\d+(?:\.\d+)?)\s*\)",
|
|
text,
|
|
)
|
|
if not m:
|
|
return None
|
|
return float(m.group(1)), float(m.group(2))
|
|
|
|
@staticmethod
|
|
def _parse_ore_anchors(text: str) -> list[tuple[str, int, int]]:
|
|
"""주변 스캔 '앵커' 좌표와 기억된 광맥 (tile_x, tile_y) 파싱."""
|
|
seen: set[tuple[str, int, int]] = set()
|
|
out: list[tuple[str, int, int]] = []
|
|
|
|
for m in re.finditer(
|
|
r"-\s*(iron-ore|copper-ore|coal|stone):\s*\d+타일\s*\(\s*앵커:\s*(-?\d+)\s*,\s*(-?\d+)\s*\)",
|
|
text,
|
|
):
|
|
t = (m.group(1), int(m.group(2)), int(m.group(3)))
|
|
if t not in seen:
|
|
seen.add(t)
|
|
out.append(t)
|
|
|
|
for m in re.finditer(
|
|
r"-\s*(iron-ore|copper-ore|coal|stone):\s*\(\s*(-?\d+)\s*,\s*(-?\d+)\s*\)",
|
|
text,
|
|
):
|
|
t = (m.group(1), int(m.group(2)), int(m.group(3)))
|
|
if t not in seen:
|
|
seen.add(t)
|
|
out.append(t)
|
|
|
|
return out
|