npc_social_network/models/llm_helper.py

# npc_social_network/models/llm_helper.py

from npc_social_network.npc.emotion_config import EMOTION_LIST
from npc_social_network.npc.npc_memory import Memory
from .. import simulation_core
from typing import List
import json
import re
import time
from google.api_core import exceptions as google_exceptions
from typing import TYPE_CHECKING, List

if TYPE_CHECKING:
    from ..npc.npc_base import NPC
    from ..manager.conversation_manager import Conversation

def query_llm_for_emotion(user_input):
    """
    LLM을 통해 플레이어 입력에서 감정 추출
    """
    prompt = f"""
    플레이어 입력: "{user_input}"

    아래 감정 리스트 중 가장 잘 해당되는 감정을 한 개만 골라 반환하세요.
    감정 리스트: {', '.join(EMOTION_LIST)}

    반드시 감정 이름을 한 개만 출력하세요. (예: joy)
    """
    try:
        response = simulation_core.active_llm_model.generate_content(prompt)
        return response.text.strip()
    except Exception as e:
        return f"[LLM Error] {str(e)}"

def query_llm_with_prompt(prompt: str) -> str:
    """
    prompt 문자열을 받아 LLM 호출
     - api 사용량 오류를 감지하고 자동 대처
    """
    if not simulation_core.active_llm_model:
        return "[LLM Error] API 모델이 정상적으로 설정되지 않았습니다."
    
    try:
        # 1. 첫 번째 API 호출 시도
        response = simulation_core.active_llm_model.generate_content(prompt)
        return response.text.strip()
    except google_exceptions.ResourceExhausted as e:
        error_message = str(e)

        # 2. 오류 메시지를 분석하여 원인 파악
        if "PerMinute" in error_message:
            # -- 분당 사용량 초과 --
            simulation_core.add_log("[API 경고] 분당 사용량을 초과했습니다. 60초 후 자동으로 재시도 합니다.")
            time.sleep(60)

            try:
                # 2-1. 두 번째 API 호출 시도
                simulation_core.add_log("[API 정보] API 호출을 재시도합니다.")
                response = simulation_core.active_llm_model.generate_content(prompt)
                return response.text.strip()
            except Exception as retry_e:
                # 재시도마저 실패할 경우, 최종 에러 반환
                simulation_core.add_log("========= [API Error] =========")

                # 시물레이션을 안전하게 '일시정지' 상태로 변경
                with simulation_core.simulation_lock:
                    simulation_core.simulation_paused = True

                return f"[API 에러] 재시도에 실패했습니다: {retry_e}"
            
        elif "PerDay" in error_message:
            # -- 일일 사용량 초과 시 --
            simulation_core.add_log("[API Error] 일일 사용량을 모두 소진했습니다.")
            simulation_core.add_log("다른 API 키를 사용하거나, 내일 다시 시도해주세요.")

            # 시뮬레이션을 안전하게 '일시정지' 상태로 변경
            with simulation_core.simulation_lock:
                simulation_core.simulation_paused = True
            return "[LLM Error] Daily quota exceeded"
        
        else:
            # 그 외 다른 ResourceExhausted 오류
            with simulation_core.simulation_lock:
                simulation_core.simulation_paused = True
            return f"[LLM Error] {error_message}"
        
    except Exception as e:
        # 그 외 모든 종류의 오류
        return f"[LLM Error] {str(e)}"
    
def summarize_text(text: str) -> str:
    """주어진 긴 텍스트(대화 내용 등)를 한 문장으로 요약"""
    if not text.strip():
        return "아무 내용 없는 대화였음"
    
    prompt = f"""
# Instruction
다음 텍스트를 가장 핵심적인 내용을 담아 한국어 한 문장으로 요약해주세요.

# Text
{text}

# Summary (a single sentence):
"""
    summary = query_llm_with_prompt(prompt)
    return summary if summary and "[LLM Error]" not in summary else "대화 요약에 실패함"

def summarize_memories(memories: List[Memory]) -> str:
    """
    Memory 객체 리스트를 받아 LLM을 통해 1~2문장으로 요약
    """
    if not memories:
        return ""
    
    # Memory 객체에서 .content를 추출하여 사용
    memory_contents = [mem.content for mem in memories]
    joined = "\n".join([f"- {content}" for content in memory_contents])

    prompt = f"""다음은 한 인물의 과거 기억들입니다:

    {joined}

    → 이 사람은 어떤 경험을 했는지 1~2문장으로 요약해 주세요."""

    # LLM 호출 함수를 사용하여 요약 요청
    summary = query_llm_with_prompt(prompt)
    return summary

def analyze_gossip(gossip_content: str) -> dict:
    """LLM을 이용해 소문을 분석하고, 관련 인물과 긍/부정을 JSON으로 반환합니다."""
    prompt="""
# 지시사항
다음 소문 내용에서 핵심 인물 2명과, 그들 사이의 상호작용이 긍정적인지, 부정적인지, 중립적인지 분석해줘.
결과를 반드시 아래 JSON 형식으로만 응답해줘, 다른 설명은 붙이지 마.

# 소문 내용
"{gossip_content}"

# 분석 결과 (JSON 형식)
{{
    "person1": "첫 번째 인물 이름",
    "person2": "두 번째 인물 이름",
    "sentiment": "positive or negative or neutral",
    "reason": "한 문장으로 요약한 이유"
}}
"""
    try:
        response_text = query_llm_with_prompt(prompt)
        # LLM 응답에서 JSON 부분만 추출
        json_match = re.search(r'\{.*\}', response_text, re.DOTALL)
        if json_match:
            return json.loads(json_match.group())
        return {}
    except (json.JSONDecodeError, Exception) as e:
        print(f"소문 분석 중 오류 발생: {e}")
        return {}

def classify_and_extract_plan_details(plan_step: str) -> dict:
    """
    계획 단계를 분석하여 행동 타입, 대상, 주제를 추출.
    """
    prompt = f"""
# Instruction
Analyze the following action plan. Classify its type and extract the target NPC and topic if applicable.
Your response MUST be in JSON format.
Action Types can be "TALK" (for conversing with others), "HELP" (for helping someone), or "SOLO_ACTION" (for acting alone).

# Action Plan
"{plan_step}"

# Analysis (JSON format only)
{{
    "action_type": "TALK or SOLO_ACTION",
    "target": "NPC's name or null",
    "topic": "topic of conversation or null"
}}
"""
    
    try:
        response_text = query_llm_with_prompt(prompt)
        json_match = re.search(r'\{.*\}', response_text, re.DOTALL)

        if json_match:
            return json.loads(json_match.group())
        
        # 분석 실패 시 혼자 하는 행동으로 간주
        return {"action_type": "SOLO_ACTION", "target": None, "topic": plan_step}
    
    except Exception:
        return {"action_type": "SOLO_ACTION", "target": None, "topic": plan_step}
    
def _query_llm_for_json_robustly(prompt: str) -> dict | list:
    """LLM에 JSON을 요청하고, 실패 시 자가 교정을 시도하는 '가디언' 함수"""

    # 1단계: LLM에 JSON 형식 요청
    response_text = query_llm_with_prompt(prompt)

    # Markdown 코드 블록(```json ... ```) 안의 내용부터 추출 시도
    markdown_match = re.search(r'```(?:json)?\s*(\{.*\}|\[.*\])\s*```', response_text, re.DOTALL)
    if markdown_match:
        text_to_parse = markdown_match.group(1) # 코드 블록 안의 내용만 사용
    else:
        # Markdown이 없으면, 전체 텍스트에서 JSON을 직접 찾음
        # 응답이 Markdown 코드 블록(```json...```)으로 감싸여 있을 경우를 대비
        json_match = re.search(r'\{.*\}|\[.*\]', response_text, re.DOTALL)
        if json_match:
            text_to_parse = json_match.group(0)
        else:
            text_to_parse = response_text   # 최후의 경우 원본 텍스트 사용

    # 불리언 값의 대소문자를 표준에 맞게 수정 (True -> true)
    text_to_parse = text_to_parse.replace(": True", ": true").replace(": False", ": false")
    # 딕셔너리나 리스트의 마지막 요소 뒤에 붙은 꼬리표 쉼표 제거
    text_to_parse = re.sub(r',\s*(\}|\])', r'\1', text_to_parse)
    
    # 추출된 텍스트로 파싱 및 자가 교정 시도
    try:
        return json.loads(text_to_parse)    # 파싱 성공 시 즉시 반환
    except json.JSONDecodeError:
        # 3단계: 파싱 실패 시, LLM에게 '자가 교정' 요청
        correction_prompt = f"""
# Instruction
The following text is not a valid JSON.
Correct the syntax errors and return ONLY the valid JSON object or array.
Do not include any other text, explanations, or markdown.

# Invalid Text
"{response_text}"

# Corrected JSON:
"""
        corrected_response = query_llm_with_prompt(correction_prompt)
        # 자가 교정된 텍스트도 한번 더 정리
        corrected_response = corrected_response.replace(": True", ": true").replace(": False", ": false")
        corrected_response = re.sub(r',\s*(\}|\])', r'\1', corrected_response)
        try:
            return json.loads(corrected_response)   # 교정된 응답 파싱 시도
        except json.JSONDecodeError:
            # 4단계: 최종 실패 시, 안전한 기본값 반환
            print(f"LLM 자가 교정 실패. 원본 응답: {response_text}")
            return {}
    
def generate_plan_from_goal(npc_name: str, goal: str) -> List[str]:
    """목표가 주어지면, 가디언 함수를 이용해 안정적으로 계획을 생성"""
    prompt = f"""
# Instruction
- 당신은 '{npc_name}'이라는 NPC를 위한 AI 조수입니다.
- 이 NPC의 목표는 "{goal}" 입니다.
- 이 목표를 달성하기 위한 3~5단계의 구체적인 행동 계획을 만들어주세요.
- 당신의 응답은 반드시 각 단계가 문자열로 이루어진 유효한 JSON 배열이어야 합니다.
- 행동 계획에 대한 내용만 작성해주세요. ('1단계'와 같은 불필요한 단어는 필요없습니다.)

# 계획 (문자열로 구성된 JSON 배열 형식):
"""
    result = _query_llm_for_json_robustly(prompt)
    if isinstance(result, list) and all(isinstance(s, str) for s in result):
        return result
    return ["[계획 생성에 실패했습니다]"] # 최종 실패 시 안전한 값 반환

def generate_dialogue_action(speaker: "NPC", target: "NPC", conversation: "Conversation",
                             memories: List["Memory"], is_final_turn: bool = False) -> dict:
    """NPC의 모든 내면 정보를 바탕으로 다음 대사와 행동을 생성."""
    history_str = "\n".join(conversation.conversation_history)
    memory_str = "\n".join([f"- {m.content}" for m in memories]) if memories else "관련된 특별한 기억 없음"

    # 평판 정보 생성
    reputation_info = ""
    reputation_score = speaker.reputation.get(target.name, 0)
    if reputation_score > 5:
        reputation_info = f"- '{target.korean_name}'에 대한 나의 평판: 신뢰할 만한 사람 같다. (점수: {reputation_score})"
    elif reputation_score < -5:
        reputation_info = f"- '{target.korean_name}'에 대한 나의 평판: 좋지 않은 소문을 들었다. (점수: {reputation_score})"

    # 상징 기억 조회
    symbolic_memories = [mem for mem in speaker.memory_store.get_all_memories() if mem.memory_type == "Symbolic"]
    symbolic_section = ""
    if symbolic_memories:
        symbolic_section_title = "# Core Symbolic Values (Symbolic Memory)"
        symbolic_contents = "\n".join([f"- {mem.content}" for mem in symbolic_memories])
        # 프롬프트에 예쁘게 들어갈 수 있도록 앞뒤로 줄바꿈 추가
        symbolic_section = f"\n{symbolic_section_title}\n{symbolic_contents}"

    # 상대방에 대해 알고 있는 정보를 프롬프트에 추가
    known_info_summary = "아직 아는 정보가 없음"
    if target.name in speaker.knowledge:
        known_facts = []
        if "job" in speaker.knowledge[target.name]:
            known_facts.append(f"직업은 {speaker.knowledge[target.name]['job']}이다.")
        if "age" in speaker.knowledge[target.name]:
            known_facts.append(f"나이는 {speaker.knowledge[target.name]['age']}이다.")
        if known_facts:
            known_info_summary = ", ".join(known_facts)

    if is_final_turn:
        task_instruction = """
- 상대방이 대화를 마무리하려고 합니다. 상대방의 마지막 말에 어울리는 자연스러운 작별 인사를 생성하세요. (예: "네, 조심히 가세요.", "다음에 또 봬요.", "알겠습니다. 좋은 하루 보내세요.")
- 이 대사를 끝으로 대화를 마무리("END")하세요.
"""
    else:
        task_instruction = """
- 위 대화의 흐름과 주제에 맞춰, 당신이 이번 턴에 할 자연스러운 대사를 생성하세요.
- 대화를 계속 이어갈지(`action: "CONTINUE"`), 아니면 이 대사를 끝으로 대화를 마무리할지(`action: "END"`) 결정하세요.
- 만약 상대방의 말에 예시와 같이 작별 인사 또는 대화 종료를 의미한다면, 반드시 당신도 그에 맞는 작별 인사를 하고 "action"을 "END"로 설정합니다. (예시: "그럼 이만.", "다음에 또 봬요.", "알겠습니다. 조심히가세요.", "네 감사합니다.", "안녕히 가세요.", "안녕히 계세요.")
- **대화를 끝내기로 결정했다면 (`action: "END"`), 당신의 대사는 반드시 "이제 가봐야겠어요", "다음에 또 이야기 나눠요" 와 같이 대화의 마무리를 암시하는 내용이어야 합니다.**
- 단순히 상대방의 말에 동의만 하지 말고, 당신의 기억이나 성격을 바탕으로 새로운 생각이나 질문, 화제를 꺼내어 대화를 풍부하게 만드세요.
"""

    # 대화 전략(Conversation Strategies) 프롬프트 도입
    prompt = f"""
# Persona
당신은 다음 프로필을 가진 "{speaker.korean_name}"이라는 인물입니다.
- 직업: {speaker.job}
- 성격: {speaker.personality.get_narrative_summary()}
당신은 지금 "{target.korean_name}"와 대화하고 있습니다.
당신의 성격과 상대방과의 관계, 관련된 과거 기억, 그리고 지금까지의 대화 흐름을 모두 고려하여 가장 현실적이고 깊이 있는 다음 대사를 만드세요.
아래 설정에 완벽하게 몰입하여 대답하세요.

# Inner State
- "{target.korean_name}"와의 관계: {speaker.relationships.get_relationship_summary(target.name)}
- {reputation_info}
- "{target.korean_name}"에 대해 내가 아는 정보: {known_info_summary}

{symbolic_section}

# Relevant Memories
{memory_str}

# Conversation Context
- 현재 대화 주제: "{conversation.topic}"
- 지금까지의 대화 내용:
{history_str}

# Instruction
{task_instruction}
- 당신의 응답은 반드시 아래 JSON 형식이어야 합니다.
- 대사에는 이름, 행동 묘사, 따옴표를 절대 포함하지 마세요.
- 'Relevant Memories'은 당신의 생각과 감정의 배경이 되는 정보입니다. 기억의 내용 자체를 그대로 말하는 것이 아니라, 그 기억을 통해 형성된 당신의 생각이나 느낌을 대화에 자연스럽게 녹여내세요.
- 만약 'Relevant Memories'이 현재 대화 상대('{target.korean_name}')와 직접적인 관련이 없는 다른 사람에 대한 것이라면, 절대로 현재 대화 상대에게 일어난 일인 것처럼 착각해서 말하지 마세요.** (예: 밥에 대한 기억을 플레이어에게 말할 때는 "사실 얼마 전에 밥 씨 때문에..." 와 같이 명확하게 주체를 밝히세요.)
- 아래의 'Conversation Strategies'과 같이 현재 대화의 맥락과 당신의 성격에 가장 적절한 행동 하나를 선택하세요.

# Conversation Strategies
1. **공감 및 위로 (Empathy & Comfort):** 상대방의 감정에 공감하거나 따뜻한 말로 위로합니다. (예: "정말 힘드셨겠어요. 저라도 그랬을 거예요.")
2. **경험 공유 (Share Experience):** 자신의 비슷한 경험을 이야기하며 유대감을 형성합니다. (예: "저도 예전에 비슷한 일로 고생한 적이 있어서 그 마음 잘 알아요.")
3. **해결책 제안 (Suggest Solution):** 문제 상황에 대해 구체적인 조언이나 해결책을 제안합니다. (예: "혹시 이렇게 해보는 건 어떨까요? 제가 도와드릴 수도 있고요.")
4. **심화 질문 (Deeper Question):** 상대방의 말에 대해 더 깊이 이해하기 위한 구체적인 질문을 합니다. (예: "그렇게 생각하게 된 특별한 계기가 있으셨나요?")
5. **화제 전환 (Change Topic):** 대화가 충분히 이루어졌다고 생각되면 자연스럽게 다른 주제로 넘어갑니다. (예: "그건 그렇고, 혹시 다른 재미있는 소식은 없나요?")
6. **유머 사용 (Use of Humor):** 긴장된 분위기를 완화하거나 분위기를 부드럽게 바꾸기 위해 웃음을 유도합니다. (예: "이쯤 되면 우리 둘 다 자격증 있어야 하는 거 아닌가요? 고민 전문가!)
7. **회피 및 무시 (Avoidance/Deflection):**  불편하거나 복잡한 주제를 일부러 피하거나 무시합니다. (예: "그 얘긴 다음에 하자. 지금 그 얘기를 할 기분 아니야.")
8. **감정 무시 (Invalidation):** 상대의 감정을 사소하게 여기거나 인정하지 않습니다. (예: "그 정도 가지고 뭘 그래.")
9. **비난/책임 전가 (Blame/Accusation):** 문제의 원인을 상대에게 돌리며 공격적으로 반응합니다. (예: "그건 너 잘못이잖아. 왜 나한테 그래?")
10. **비꼬기 및 조롱 (Sarcasm/Mockery):** 직접적으로 공격하지 않지만, 비꼬는 말투로 상대를 깎아내립니다. (예: "오, 이제야 그걸 깨달았다고? 역시 빠르시네.")
11. **화제 회피를 위한 과도한 일반화 (Over-Generalization):** 구체적인 문제에서 벗어나기 위해 '원래 다 그래' 식으로 일반화 합니다. (예: "사람들이 다 그렇지 뭐. 기대하지 마.")

# Response (JSON format only)
{{
    "dialogue": "Conversation Strategies 전략에 따라 생성된 실제 말할 대사 내용",
    "action": "CONTINUE or END"
}}
"""
    result = _query_llm_for_json_robustly(prompt)   # 3중 방어 시스템 재활용

    # 결과 포맷 검증 및 기본값 처리
    if isinstance(result, dict) and "dialogue" in result and "action" in result:
        # 후처리 로직 (이름 중복이나 따옴표 제거)
        dialogue = result.get("dialogue", "...").strip().strip('"')
        name_prefix = f"{speaker.name}:"
        if dialogue.startswith(name_prefix):
            dialogue = dialogue[len(name_prefix):].strip().strip('"')
        result["dialogue"] = dialogue
        return result
    
    return {"dialogue": "...", "action": "END"} # 최종 실패 시 안전하게 대화 종료

def evaluate_social_action_outcome(initiator: "NPC", target: "NPC", action_type: str) -> dict:
    """두 NPC의 성격과 관계를 바탕으로 사회적 행동의 결과를 추론합니다."""
    prompt = f"""
# Context
- 행동자: "{initiator.korean_name}" (성격: {initiator.personality.get_narrative_summary()})
- 대상자: "{target.korean_name}" (성격: {target.personality.get_narrative_summary()})
- 둘의 관계: {initiator.relationships.get_relationship_summary(target.name)}
- 행동: 행동자('{initiator.korean_name}')는 대상자('{target.korean_name}')를 도와주려고 합니다. (Action: {action_type})

# Your Task
위 상황을 바탕으로, 이 행동의 결과를 현실적으로 추론해주세요.
결과는 반드시 아래 JSON 형식이어야 합니다.

# Outcome (JSON format only)
{{
    "success": "True or False"
    "outcome_summary": "행동의 결과를 한 문장으로 요약 (예: '밥은 찰리의 서투른 도움에 고마워하면서도 조금 답답해했다.')",
    "initiator_emotion": "행동을 한 후 행동자가 느낄 감정 (아래 'Emotion_List에서 찾아서 한 단어로 표현)",
    "target_emotion": "행동을 받은 후 대상자가 느낄 감정 (아래 'Emotion_List에서 찾아서 한 단어로 표현)"
}}

# Emotion_List
{EMOTION_LIST}
"""
    result = _query_llm_for_json_robustly(prompt)   # 3중 방어 시스템 재활용
    
    if isinstance(result, dict) and "success" in result:
        return result
    return {# 최종 실패 시 안전한 기본값 반환
            "success": False,
            "outcome_summary": "행동의 결과를 평가하는 데 실패했습니다.",
            "initiator_emotion": "confusion",
            "target_emotion": "neutral"
        }

def evaluate_goal_achievement(initiator_name: str, goal: str, conversation_transcript: str) -> dict:
    """대화의 목표와 전체 내용을 바탕으로, 목표 달성 여부를 평가"""
    prompt = f"""
# Context
- 대화 시작자: "{initiator_name}"
- 대화 시작자의 목표 (대화 주제): "{goal}"
- 실제 대화 내용:
{conversation_transcript}

# Your Task
위 정보를 바탕으로, 대화 시작자("{initiator_name}")가 자신의 원래 목표를 성공적으로 달성했는지 평가해주세요.
결과는 반드시 아래 JSON 형식이어야 합니다.

# Evaluation (JSON format only)
{{
    "goal_achieved": True,
    "reason": "목표 달성 또는 실패에 대한 간결한 이유 (예: '상대방을 위로하고 지지하며 긍정적인 반응을 이끌어냈다.)",
    "initiator_emotion": "이 결과를 얻은 후 대화 시작자가 느낄 감정 (아래 'Emotion_List에서 찾아서 한 단어로 표현)
}}

# Emotion_List
{EMOTION_LIST}
"""
    result = _query_llm_for_json_robustly(prompt)

    if isinstance(result, dict) and "goal_achieved" in result:
        return result
    return {
        # 최종 실패 시 안전한 기본값 반환
        "goal_achieved": False,
        "reason": "목표 달성 여부 평가에 실패했습니다.",
        "initiator_emotion": "confusion"
    }

def generate_reflection_topic(memories: List["Memory"], personality_summary: str) -> str:
    """최근 기억과 성격을 바탕으로, 깊이 있는 성찰 주제를 동적으로 생성"""
    if not memories:
        return "나의 인생에서 가장 중요했던 순간은 언제였을까?"
    
    # 기억 목록을 프롬프트에 넣기 좋은 형태로 변환
    memory_details = "\n".join([f"- {mem.content} (감정: {mem.emotion})" for mem in memories])

    prompt = f"""
# Instructions
- 아래 'Personality'와 'Recent Memory List'는 각각 어떤 사람의 '성격'과 '최근 기억 목록'입니다.
- 이 기억들을 종합적으로 분석하여, 이 사람이 지금 자신에 대해 깊이 성찰해볼 만한 가장 의미 있는 '질문'을 한국어로 한 문장 만들어주세요.

# Personality
{personality_summary}

# Recent Memory List
{memory_details}

# Self-Reflective Question (in Korean, as a single sentence):
"""
    topic = query_llm_with_prompt(prompt).strip()
    return topic if topic and "[LLM Error]" not in topic else "나의 인생에서 가장 중요한 가치는 무엇일까?"