gihyeon/factorio-ai-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: JSON 파싱 강화 + 3회 재시도 + 잘린 응답 복구

Browse Source 
- max_tokens 1500→2000 (잘림 방지)
- 중첩 괄호 카운팅 기반 JSON 추출
- _repair_truncated_json() 잘린 응답 복구
- 3회 실패 시 기본 탐색 행동 반환 (크래시 방지)
- 시스템 프롬프트 강화 (순수 JSON 강조)
This commit is contained in:
gihyeon
2026-03-25 15:11:00 +09:00
parent ea4be446bb
commit 6f220bc3f5
1 changed files with 119 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

143
ai_planner.py
View File

@@ -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] 생각 중...")
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. <think> 태그 제거
if "<think>" in text:
text = text.split("</think>")[-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()