feat: enhance AIPlanner with compact system prompt and improved JSON parsing for input-only mode responses
This commit is contained in:
@@ -116,6 +116,13 @@ state_reader가 상태 요약에 포함하는 `마지막 행동(기억)` 섹션
|
||||
반드시 아래 JSON 형식으로만 응답하세요. 마크다운이나 설명 텍스트 없이 순수 JSON만 출력하세요.
|
||||
{"thinking": "상황 분석 내용", "current_goal": "현재 목표", "actions": [{"action": "액션명", "params": {...}, "reason": "이유"}], "after_this": "다음 계획"}"""
|
||||
|
||||
SYSTEM_PROMPT_COMPACT = """당신은 팩토리오 순수 플레이 AI입니다.
|
||||
치트 없이 실제 이동/채굴/제작/건설 제약을 지키세요.
|
||||
출력은 반드시 JSON 객체 1개만:
|
||||
{"thinking":"...","current_goal":"...","actions":[{"action":"...","params":{},"reason":"..."}],"after_this":"..."}
|
||||
절대 마크다운/설명문/사고과정 텍스트를 출력하지 마세요.
|
||||
actions는 1~4개, 가능한 짧고 실행 가능한 계획만 반환하세요."""
|
||||
|
||||
|
||||
def _debug_enabled() -> bool:
|
||||
v = os.environ.get("AI_DEBUG", "").strip().lower()
|
||||
@@ -123,10 +130,13 @@ def _debug_enabled() -> bool:
|
||||
|
||||
|
||||
class AIPlanner:
|
||||
_last_payload_mode = "auto"
|
||||
|
||||
def __init__(self):
|
||||
self.step = 0
|
||||
self.feedback_log: list[dict] = []
|
||||
self._fallback_explore_turn = 0
|
||||
self._payload_mode_hint = "auto"
|
||||
self.long_term_goal = (
|
||||
"완전 자동화 달성: "
|
||||
"석탄 채굴 → 철 채굴+제련 자동화 → 구리 채굴+제련 → "
|
||||
@@ -207,6 +217,10 @@ class AIPlanner:
|
||||
)
|
||||
if resp.status_code < 400:
|
||||
data = resp.json()
|
||||
if "messages" in payload:
|
||||
AIPlanner._last_payload_mode = "legacy_chat"
|
||||
else:
|
||||
AIPlanner._last_payload_mode = "input_only"
|
||||
break
|
||||
if resp.status_code == 400 and i < len(candidates) - 1:
|
||||
err_msg = self._extract_http_error_message(resp)
|
||||
@@ -230,7 +244,13 @@ class AIPlanner:
|
||||
print(f"[AI] 응답 수신 ({dt:.2f}s, {len(content)}자)")
|
||||
if _debug_enabled():
|
||||
print(f"[AI][디버그] raw={content[:300]}")
|
||||
return json.loads(content)
|
||||
try:
|
||||
return json.loads(content)
|
||||
except json.JSONDecodeError:
|
||||
repaired = self._extract_json_object_text(content)
|
||||
if repaired is None:
|
||||
raise
|
||||
return json.loads(repaired)
|
||||
|
||||
@staticmethod
|
||||
def _build_chat_payload(user_message: str) -> dict:
|
||||
@@ -251,7 +271,7 @@ class AIPlanner:
|
||||
|
||||
@staticmethod
|
||||
def _build_input_only_payload(user_message: str) -> dict:
|
||||
merged = f"{SYSTEM_PROMPT}\n\n{user_message}"
|
||||
merged = f"{SYSTEM_PROMPT_COMPACT}\n\n{user_message}"
|
||||
return {
|
||||
"model": OLLAMA_MODEL,
|
||||
"input": merged,
|
||||
@@ -261,10 +281,16 @@ class AIPlanner:
|
||||
@staticmethod
|
||||
def _build_payload_candidates(user_message: str) -> list[dict]:
|
||||
# 서버 구현체마다 스키마가 달라 자동 호환을 위해 후보를 순차 시도한다.
|
||||
return [
|
||||
default = [
|
||||
AIPlanner._build_chat_payload(user_message),
|
||||
AIPlanner._build_input_only_payload(user_message),
|
||||
]
|
||||
# 이전 호출에서 성공한 모드를 다음 턴에 우선 사용해 불필요한 400 재시도를 줄인다.
|
||||
if getattr(AIPlanner, "_last_payload_mode", "auto") == "input_only":
|
||||
return [default[1], default[0]]
|
||||
if getattr(AIPlanner, "_last_payload_mode", "auto") == "legacy_chat":
|
||||
return default
|
||||
return default
|
||||
|
||||
@staticmethod
|
||||
def _extract_http_error_message(resp: httpx.Response) -> str:
|
||||
@@ -310,6 +336,10 @@ class AIPlanner:
|
||||
for item in output:
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
# LM Studio Responses 스타일: {"type":"reasoning","content":"..."}
|
||||
content_str = item.get("content")
|
||||
if isinstance(content_str, str) and content_str.strip():
|
||||
return content_str
|
||||
content_items = item.get("content")
|
||||
if not isinstance(content_items, list):
|
||||
continue
|
||||
@@ -322,6 +352,34 @@ class AIPlanner:
|
||||
|
||||
raise ValueError(f"응답에서 텍스트 콘텐츠를 찾지 못했습니다. keys={list(data.keys())}")
|
||||
|
||||
@staticmethod
|
||||
def _extract_json_object_text(text: str) -> str | None:
|
||||
start = text.find("{")
|
||||
if start < 0:
|
||||
return None
|
||||
depth = 0
|
||||
in_str = False
|
||||
esc = False
|
||||
for i in range(start, len(text)):
|
||||
ch = text[i]
|
||||
if in_str:
|
||||
if esc:
|
||||
esc = False
|
||||
elif ch == "\\":
|
||||
esc = True
|
||||
elif ch == '"':
|
||||
in_str = False
|
||||
continue
|
||||
if ch == '"':
|
||||
in_str = True
|
||||
elif ch == "{":
|
||||
depth += 1
|
||||
elif ch == "}":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return text[start:i + 1]
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _ensure_move_before_build_actions(actions: list[dict]) -> list[dict]:
|
||||
"""
|
||||
|
||||
Reference in New Issue
Block a user