diff --git a/ai_planner.py b/ai_planner.py index 0309d9c..3474c64 100644 --- a/ai_planner.py +++ b/ai_planner.py @@ -7,31 +7,36 @@ ai_planner.py — 순수 AI 플레이 버전 - 제작은 재료가 인벤토리에 있어야 함 → 재료 확보 순서 중요 - 건설은 건설 거리 내에서만 가능 → 배치 전 move 필수 - AI가 이 제약을 이해하고 행동 순서를 계획해야 함 + +JSON 파싱 강화: + - GLM 응답이 잘리거나 마크다운으로 감싸져도 복구 + - 최대 2회 재시도 """ import json import os +import re import urllib.request import urllib.error GLM_API_URL = "https://api.z.ai/api/coding/paas/v4/chat/completions" -GLM_MODEL = "GLM-4.7" +GLM_MODEL = "GLM-4-7" SYSTEM_PROMPT = """당신은 팩토리오 게임을 순수하게 플레이하는 AI 에이전트입니다. -치트나 텔레포트 없이, 실제 게임 메커닉만 사용합니다. +치트나 텔레포트 없이, 실제 게임 메커니즘만 사용합니다. 게임 상태와 이전 행동 결과를 분석해서 스스로 판단하고 계획을 세웁니다. ## 핵심 제약 사항 (반드시 준수!) 1. **이동은 실제 걷기** — 먼 거리는 시간이 오래 걸림. 불필요한 왕복 최소화 2. **채굴은 자원 패치 근처에서만 가능** — 반드시 자원 위치로 move한 후 mine_resource -3. **제작은 재료가 있어야 함** — iron-plate 없이 iron-gear-wheel 못 만듦. +3. **제작은 재료가 있어야 함** — iron-plate 없이 iron-gear-wheel 못 만듬. 재료 확인 후 craft_item. 재료 부족하면 먼저 채굴→제련 4. **건설은 건설 거리 내에서만 가능** — 배치할 좌표 근처로 move한 후 place_entity 5. **자원은 유한** — 직접 채굴해야 하고, 제련소에 넣어야 plate가 됨 ## 효율적인 행동 패턴 -- 같은 구역 작업을 몰아서 (이동 최소화) +- 같은 구역 작업을 묶어서 (이동 최소화) - move → mine/place/insert 순서로 항상 위치 먼저 확보 - 채굴 → 제련 → 제작 → 건설 흐름 유지 - 한 번에 넉넉히 채굴 (왕복 줄이기) @@ -45,7 +50,7 @@ SYSTEM_PROMPT = """당신은 팩토리오 게임을 순수하게 플레이하는 - 자동화 연구팩: iron-gear-wheel + iron-plate → assembling-machine - 건물 배치 전 반드시: 1) 인벤토리에 아이템 있는지 2) 가까이 있는지 확인 -## 응답 형식 — JSON만 반환, 다른 텍스트 절대 금지 +## 응답 형식 — 반드시 순수 JSON만 반환, 다른 텍스트 절대 금지 { "thinking": "현재 상태 분석. 인벤토리/위치/자원 확인 후 판단 (자유롭게 서술)", "current_goal": "지금 달성하려는 목표", @@ -64,37 +69,28 @@ SYSTEM_PROMPT = """당신은 팩토리오 게임을 순수하게 플레이하는 ### 채굴 (자원 패치 근처에서만 작동) - "mine_resource" → {"ore": "iron-ore", "count": int} - 주의: 자원 패치 근처에 있어야 함. 없으면 move 먼저 채굴 가능: 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 - inserter: iron-gear-wheel 1 + iron-plate 1 + electronic-circuit 1 - small-electric-pole: copper-cable 2 + wood 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칸 내) - 배치 가능: burner-mining-drill, electric-mining-drill, stone-furnace, - burner-inserter, inserter, transport-belt, small-electric-pole, - pipe, offshore-pump, boiler, steam-engine, assembling-machine-1, lab -### 벨트 라인 (걸어가며 한 칸씩 배치 — 시간 많이 걸림) +### 벨트 라인 (걸어다니면서 하나씩 배치 — 시간 많이 걸림) - "place_belt_line" → {"from_x": int, "from_y": int, "to_x": int, "to_y": int} - 주의: 벨트 수량 충분히 craft_item 먼저 ### 연료/아이템 삽입 (건설 거리 내에서) - "insert_to_entity" → {"x": int, "y": int, "item": "coal", "count": int} - 주의: 가까이 있어야 하고, 플레이어 인벤토리에서 차감됨 ### 조립기 레시피 설정 - "set_recipe" → {"x": int, "y": int, "recipe": str} @@ -104,7 +100,8 @@ SYSTEM_PROMPT = """당신은 팩토리오 게임을 순수하게 플레이하는 ### 대기 - "wait" → {"seconds": int} -""" + +## 절대 중요: 순수 JSON만 출력하세요. ```json 같은 마크다운 블록, 설명 텍스트, 주석 없이 오직 { } 만.""" class AIPlanner: @@ -122,10 +119,6 @@ class AIPlanner: ) def decide(self, state_summary: str) -> list[dict]: - """ - 게임 상태를 받아 GLM이 스스로 생각하고 - 실행할 행동 시퀀스(여러 개)를 반환 - """ self.step += 1 feedback_text = self._format_feedback() @@ -137,12 +130,29 @@ class AIPlanner: "현재 상태를 분석하고, 장기 목표를 향해 지금 해야 할 행동 시퀀스를 계획하세요.\n" "⚠️ 순수 플레이입니다. 건설/채굴/삽입 전에 반드시 move로 가까이 이동하세요.\n" "⚠️ 제작은 재료가 있어야 합니다. 인벤토리를 확인하세요.\n" - "반드시 JSON만 반환하세요." + "반드시 JSON만 반환하세요. 마크다운 블록(```)이나 설명 텍스트 없이 순수 JSON만." ) print(f"\n[GLM] 생각 중...") - raw = self._call_glm(user_message) - plan = self._parse_json(raw) + + for attempt in range(3): + try: + raw = self._call_glm(user_message) + plan = self._parse_json(raw) + break + except (ValueError, json.JSONDecodeError) as e: + if attempt < 2: + print(f"[경고] JSON 파싱 실패 (시도 {attempt+1}/3), 재시도...") + continue + print(f"[오류] JSON 파싱 3회 실패. 기본 탐색 행동 사용.") + plan = { + "thinking": "API 응답 파싱 실패로 기본 탐색 수행", + "current_goal": "주변 탐색", + "actions": [ + {"action": "move", "params": {"x": 20, "y": 0}, "reason": "탐색을 위해 이동"}, + ], + "after_this": "자원 발견 후 채굴 시작" + } thinking = plan.get("thinking", "") if thinking: @@ -156,7 +166,6 @@ class AIPlanner: return actions def record_feedback(self, action: dict, success: bool, message: str = ""): - """행동 결과를 기록 (다음 판단에 활용)""" self.feedback_log.append({ "action": action.get("action", ""), "params": action.get("params", {}), @@ -187,7 +196,7 @@ class AIPlanner: {"role": "user", "content": user_message}, ], "temperature": 0.3, - "max_tokens": 1500, + "max_tokens": 2000, # 1500 → 2000으로 증가 (잘림 방지) }).encode("utf-8") req = urllib.request.Request( @@ -207,22 +216,104 @@ class AIPlanner: raise ConnectionError(f"GLM API 오류 {e.code}: {e.read().decode()}") def _parse_json(self, raw: str) -> dict: + """GLM 응답에서 JSON을 안전하게 추출. 여러 형태 대응.""" text = raw.strip() + + # 1. 태그 제거 if "" in text: text = text.split("")[-1].strip() + + # 2. 마크다운 코드블록 제거 if text.startswith("```"): + # ```json ... ``` 패턴 text = "\n".join( l for l in text.splitlines() if not l.strip().startswith("```") ).strip() + + # 3. 순수 JSON 시도 try: return json.loads(text) except json.JSONDecodeError: - s, e = text.find("{"), text.rfind("}") + 1 - if s != -1 and e > s: - return json.loads(text[s:e]) + pass + + # 4. { } 범위 추출 (중첩 괄호 고려) + start = text.find("{") + if start == -1: + raise ValueError(f"JSON 파싱 실패 ('{' 없음):\n{raw[:300]}") + + # 중첩 괄호 카운팅으로 정확한 끝 위치 찾기 + depth = 0 + in_string = False + escape = False + end = start + + for i in range(start, len(text)): + c = text[i] + if escape: + escape = False + continue + if c == '\\' and in_string: + escape = True + continue + if c == '"' and not escape: + in_string = not in_string + continue + if in_string: + continue + if c == '{': + depth += 1 + elif c == '}': + depth -= 1 + if depth == 0: + end = i + 1 + break + + if depth != 0: + # 괄호가 안 닫힘 — 잘린 응답. 복구 시도 + partial = text[start:] + # 잘린 actions 배열 복구 + partial = self._repair_truncated_json(partial) + try: + return json.loads(partial) + except json.JSONDecodeError: + raise ValueError(f"JSON 파싱 실패 (잘린 응답 복구 불가):\n{raw[:400]}") + + try: + return json.loads(text[start:end]) + except json.JSONDecodeError: raise ValueError(f"JSON 파싱 실패:\n{raw[:400]}") + def _repair_truncated_json(self, text: str) -> str: + """잘린 JSON 응답 복구 시도""" + # 마지막 완전한 action 항목까지만 유지 + # actions 배열에서 마지막 완전한 } 를 찾아서 배열 닫기 + + # "actions" 키가 있는지 확인 + if '"actions"' not in text: + # actions도 없으면 기본 구조로 대체 + return '{"thinking":"응답 잘림","current_goal":"탐색","actions":[],"after_this":"재시도"}' + + # 마지막으로 완전한 action 객체의 } 위치 찾기 + last_complete = -1 + # "reason": "..." } 패턴의 마지막 위치 + for m in re.finditer(r'"reason"\s*:\s*"[^"]*"\s*\}', text): + last_complete = m.end() + + if last_complete > 0: + # 그 지점까지 자르고 배열과 객체 닫기 + result = text[:last_complete] + # 열린 괄호 세기 + open_brackets = result.count('[') - result.count(']') + open_braces = result.count('{') - result.count('}') + result += ']' * open_brackets + # after_this 추가 + result += ',"after_this":"계속 진행"' + result += '}' * open_braces + return result + + return '{"thinking":"응답 잘림","current_goal":"탐색","actions":[],"after_this":"재시도"}' + def set_goal(self, goal: str): self.long_term_goal = goal self.feedback_log.clear()