[단순 LLM 호출]
질문 → LLM → 답변
(1회성, 기억 없음, 도구 없음)
[챗봇]
대화 → LLM → 대화
(맥락 유지, 기억 있음, 도구 없음)
[AI 에이전트]
목표 → [계획 → 도구 사용 → 결과 검토 → 재계획] → 목표 달성
(자율적, 기억 있음, 도구 있음, 루프 가능)

┌─────────────────────────────────────────────────┐
│                  AI 에이전트                        │
│                                                   │
│  ┌──────────┐  ┌──────────┐  ┌────────────────┐  │
│  │  인식층   │  │  추론 엔진 │  │    메모리 시스템  │  │
│  │(Perception)│→│(Reasoning)│←→│   (Memory)     │  │
│  └──────────┘  └─────┬────┘  └────────────────┘  │
│                      │                            │
│               ┌──────▼──────┐                     │
│               │  도구 실행층  │                     │
│               │(Tool Layer) │                     │
│               └─────────────┘                     │
└─────────────────────────────────────────────────┘

# 인식층 예시: 다양한 입력 처리
def perceive(input_data: dict) -> str:
    """
    입력 유형에 따라 적절하게 처리해서
    에이전트가 이해할 수 있는 텍스트로 변환
    """
    if input_data["type"] == "file":
        return parse_file(input_data["path"])
    elif input_data["type"] == "api_response":
        return json.dumps(input_data["data"], ensure_ascii=False)
    elif input_data["type"] == "text":
        return input_data["content"]
    else:
        return str(input_data)

# 추론 엔진: LLM이 다음 행동을 결정
def reason(context: str, available_tools: list) -> dict:
    """
    현재 상황을 분석하고 다음 행동을 결정
    반환: {"action": "tool_name", "input": "...", "reasoning": "..."}
    """
    prompt = f"""
현재 상황: {context}
사용 가능한 도구: {[t['name'] for t in available_tools]}
다음 중 하나를 결정하세요:
1. 특정 도구를 사용 → {{"action": "use_tool", "tool": "도구명", "input": "입력값"}}
2. 최종 답변 생성 → {{"action": "final_answer", "content": "답변 내용"}}
JSON으로만 응답하세요.
"""
    response = llm.invoke(prompt)
    return json.loads(response.content)

# 메모리 시스템: 3가지 유형
class AgentMemory:
    def __init__(self):
        # 단기: 현재 세션의 대화
        self.short_term = []
        # 장기: 세션 간 유지되는 사실
        self.long_term = {}
        # 에피소딕: 과거 작업 결과
        self.episodic = []
    def add_short_term(self, role: str, content: str):
        self.short_term.append({"role": role, "content": content})
        # 너무 길어지면 요약 압축
        if len(self.short_term) > 20:
            self._compress()
    def save_long_term(self, key: str, value: str):
        self.long_term[key] = {
            "value": value,
            "updated_at": datetime.now().isoformat()
        }
    def get_context(self) -> str:
        """추론 엔진에 주입할 컨텍스트 생성"""
        ctx = ""
        if self.long_term:
            ctx += f"[저장된 정보]\n{json.dumps(self.long_term, ensure_ascii=False)}\n\n"
        ctx += f"[현재 대화]\n"
        for msg in self.short_term[-10:]:  # 최근 10개만
            ctx += f"{msg['role']}: {msg['content']}\n"
        return ctx

# 도구 등록 패턴
from langchain.tools import tool
@tool
def search_web(query: str) -> str:
    """웹에서 최신 정보를 검색합니다."""
    # 실제 구현: Tavily, SerpAPI 등 연동
    return f"검색 결과: {query}에 대한 최신 정보..."
@tool
def execute_python(code: str) -> str:
    """Python 코드를 안전하게 실행하고 결과를 반환합니다."""
    import subprocess
    result = subprocess.run(
        ["python", "-c", code],
        capture_output=True, text=True, timeout=10
    )
    return result.stdout or result.stderr
@tool
def query_database(sql: str) -> str:
    """데이터베이스에서 정보를 조회합니다."""
    # 실제 구현: DB 연결 및 쿼리
    return f"쿼리 결과: {sql}"

[ReAct 루프]
목표 입력
    ↓
Thought: "이 문제를 해결하려면 먼저 X를 알아야 한다"
    ↓
Action: search_web("X에 대한 정보")
    ↓
Observation: "검색 결과: ..."
    ↓
Thought: "이제 Y를 계산해야 한다"
    ↓
Action: execute_python("Y 계산 코드")
    ↓
Observation: "결과: 42"
    ↓
Thought: "충분한 정보를 얻었다. 답변할 수 있다."
    ↓
Final Answer: "..."

[Plan-and-Execute 흐름]
목표: "경쟁사 3곳의 가격 분석 리포트 작성"
    ↓
[계획 단계]
Plan:
  Step 1: A사 웹사이트에서 가격 데이터 수집
  Step 2: B사 API에서 가격 데이터 수집
  Step 3: C사 스크래핑
  Step 4: 3개 데이터 비교 분석
  Step 5: 리포트 생성
    ↓
[실행 단계: 각 스텝 순서대로 실행]
Execute Step 1 → Execute Step 2 → ... → Execute Step 5
    ↓
최종 리포트

[슈퍼바이저 패턴]
사용자 요청
    ↓
슈퍼바이저 에이전트
    ├─→ 리서치 에이전트 (정보 수집 전문)
    ├─→ 분석 에이전트   (데이터 처리 전문)
    ├─→ 작성 에이전트   (문서 생성 전문)
    └─→ 검증 에이전트   (품질 확인 전문)
    ↓
통합 결과물

# react_agent.py
# 의존성: pip install langchain langchain-anthropic python-dotenv
import os
import json
from dotenv import load_dotenv
from langchain_anthropic import ChatAnthropic
from langchain.tools import tool
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
load_dotenv()
# ── 도구 정의 ──────────────────────────────────────────
@tool
def calculator(expression: str) -> str:
    """수학 계산을 수행합니다. 예: '2 + 2 * 10'"""
    try:
        result = eval(expression, {"__builtins__": {}})
        return f"계산 결과: {result}"
    except Exception as e:
        return f"계산 오류: {str(e)}"
@tool
def get_current_time() -> str:
    """현재 시간을 반환합니다."""
    from datetime import datetime
    return f"현재 시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
@tool
def search_knowledge(query: str) -> str:
    """
    내부 지식베이스에서 정보를 검색합니다.
    실제 환경에서는 RAG 파이프라인이나 검색 API와 연결하세요.
    """
    knowledge_base = {
        "python": "Python은 2026년 TIOBE 1위, 점유율 26.14%의 언어입니다.",
        "langgraph": "LangGraph는 126,000+ GitHub 스타를 보유한 에이전트 프레임워크입니다.",
        "ai agent": "AI 에이전트는 기업의 51%가 프로덕션에서 운영 중인 기술입니다.",
    }
    for key, value in knowledge_base.items():
        if key.lower() in query.lower():
            return value
    return f"'{query}'에 대한 정보를 찾을 수 없습니다."
TOOLS = [calculator, get_current_time, search_knowledge]
TOOL_MAP = {t.name: t for t in TOOLS}
# ── ReAct 에이전트 ─────────────────────────────────────
class ReActAgent:
    def __init__(self):
        self.llm = ChatAnthropic(
            model="claude-sonnet-4-20250514",
            api_key=os.getenv("ANTHROPIC_API_KEY"),
            temperature=0,
            max_tokens=2048
        )
        self.max_iterations = 10  # 무한루프 방지
    def _build_system_prompt(self) -> str:
        tool_descriptions = "\n".join(
            f"- {t.name}: {t.description}" for t in TOOLS
        )
        return f"""당신은 도구를 사용해 문제를 해결하는 AI 에이전트입니다.
사용 가능한 도구:
{tool_descriptions}
반드시 아래 형식으로만 응답하세요:
도구를 사용할 때:
```json
{{"action": "tool_name", "input": "입력값", "thought": "이 도구를 선택한 이유"}}

{{"action": "final_answer", "content": "최종 답변 내용"}}

def run(self, goal: str) -> str:
    """목표를 받아 ReAct 루프를 실행합니다."""
    print(f"\n{'='*60}")
    print(f"🎯 목표: {goal}")
    print(f"{'='*60}\n")
    messages = [
        SystemMessage(content=self._build_system_prompt()),
        HumanMessage(content=f"목표: {goal}")
    ]
    for iteration in range(self.max_iterations):
        # LLM에게 다음 행동 결정
        response = self.llm.invoke(messages)
        raw = response.content.strip()
        # JSON 파싱
        try:
            # 마크다운 코드블록 제거
            if "```json" in raw:
                raw = raw.split("```json")[1].split("```")[0].strip()
            elif "```" in raw:
                raw = raw.split("```")[1].split("```")[0].strip()
            decision = json.loads(raw)
        except json.JSONDecodeError:
            print(f"⚠️ JSON 파싱 실패. 원본: {raw[:100]}")
            break
        action = decision.get("action")
        thought = decision.get("thought", "")
        # 최종 답변
        if action == "final_answer":
            final = decision.get("content", "")
            print(f"✅ 최종 답변: {final}")
            return final
        # 도구 실행
        tool_name = action
        tool_input = decision.get("input", "")
        if thought:
            print(f"💭 생각: {thought}")
        print(f"🔧 도구 사용: {tool_name}({tool_input!r})")
        if tool_name not in TOOL_MAP:
            observation = f"오류: '{tool_name}' 도구가 없습니다."
        else:
            try:
                observation = TOOL_MAP[tool_name].invoke(tool_input)
            except Exception as e:
                observation = f"도구 실행 오류: {str(e)}"
        print(f"👁️ 관찰: {observation}\n")
        # 대화에 결과 추가
        messages.append(AIMessage(content=response.content))
        messages.append(HumanMessage(
            content=f"도구 실행 결과: {observation}\n\n계속 진행하세요."
        ))
    return "최대 반복 횟수 초과. 작업을 완료하지 못했습니다."

# 테스트 1: 복합 계산
result = agent.run(
    "현재 시간을 확인하고, (2026 - 1991) × 3 을 계산한 다음 "
    "Python 언어에 대한 정보도 알려줘"
)
# 테스트 2: 지식 검색
result = agent.run("AI 에이전트의 현재 기업 도입 현황을 알려줘")

# ❌ 너무 많은 역할
@tool
def do_everything(task: str) -> str:
    """검색도 하고, 계산도 하고, 파일도 저장하는 함수"""
    ...
# ✅ 역할 분리
@tool
def search_web(query: str) -> str:
    """웹에서 정보를 검색합니다."""
    ...
@tool
def save_to_file(filename: str, content: str) -> str:
    """내용을 파일에 저장합니다."""
    ...

@tool
def call_external_api(endpoint: str) -> str:
    """외부 API를 호출합니다."""
    try:
        response = requests.get(endpoint, timeout=5)
        response.raise_for_status()
        return response.json()
    except requests.Timeout:
        return "오류: API 응답 시간 초과 (5초). 나중에 다시 시도하세요."
    except requests.HTTPError as e:
        return f"오류: HTTP {e.response.status_code} — {e.response.text[:200]}"
    except Exception as e:
        return f"오류: 예상치 못한 문제 발생 — {str(e)}"

# ❌ 문제
while True:
    action = agent.decide()
    result = execute(action)
# ✅ 해결: 반복 횟수 제한 + 진행 감지
MAX_ITER = 10
seen_actions = []
for i in range(MAX_ITER):
    action = agent.decide()
    # 같은 행동을 3번 이상 반복하면 중단
    if seen_actions.count(action) >= 3:
        return "루프 감지: 동일 행동 3회 반복으로 중단"
    seen_actions.append(action)
    result = execute(action)

# ✅ 해결: 슬라이딩 윈도우 + 요약 압축
def get_trimmed_context(messages: list, max_tokens: int = 4000) -> list:
    """최근 메시지만 유지, 초과 시 요약 압축"""
    if count_tokens(messages) <= max_tokens:
        return messages
    # 오래된 메시지 요약
    old_messages = messages[:-10]
    recent_messages = messages[-10:]
    summary = summarize(old_messages)
    return [SystemMessage(content=f"이전 대화 요약: {summary}")] + recent_messages

# ✅ 해결: 재시도 + 대안 경로
import time
def execute_with_retry(tool, input_data, max_retries=3):
    for attempt in range(max_retries):
        try:
            return tool.invoke(input_data)
        except Exception as e:
            if attempt < max_retries - 1:
                wait = 2 ** attempt  # 지수 백오프: 1초, 2초, 4초
                print(f"재시도 {attempt + 1}/{max_retries} — {wait}초 후")
                time.sleep(wait)
            else:
                return f"도구 실패 (3회 시도): {str(e)}. 대안을 찾겠습니다."

# ✅ 해결: 행동 전 검증 레이어
def validate_action(action: dict, available_tools: list) -> tuple[bool, str]:
    """에이전트의 행동을 실행 전에 검증"""
    tool_names = [t.name for t in available_tools]
    if action.get("action") not in tool_names + ["final_answer"]:
        return False, f"없는 도구 호출: '{action.get('action')}'"
    if action.get("action") == "final_answer" and not action.get("content"):
        return False, "최종 답변 내용이 비어있음"
    return True, "OK"

# 고객 서비스 에이전트 도구 예시
@tool
def check_order_status(order_id: str) -> str:
    """주문 ID로 배송 상태와 예상 도착일을 조회합니다."""
    ...
@tool
def process_refund(order_id: str, reason: str) -> str:
    """환불 요청을 처리합니다. 3~5 영업일 내 처리됩니다."""
    ...
@tool
def escalate_to_human(ticket_id: str, priority: str) -> str:
    """복잡한 문의를 인간 상담사에게 에스컬레이션합니다."""
    ...

@tool
def query_analytics(nl_question: str) -> str:
    """자연어 질문을 SQL로 변환해서 분석 DB를 조회합니다.
    예: '지난 달 가장 많이 팔린 제품 5개는?'"""
    sql = text_to_sql(nl_question)
    return execute_query(sql)
@tool
def generate_chart(data: str, chart_type: str) -> str:
    """데이터를 차트로 시각화하고 파일 경로를 반환합니다."""
    ...

@tool
def analyze_code_diff(diff: str) -> str:
    """코드 변경사항에서 버그, 보안 취약점, 성능 이슈를 분석합니다."""
    ...
@tool
def check_test_coverage(file_path: str) -> str:
    """파일의 테스트 커버리지를 확인하고 미커버 라인을 반환합니다."""
    ...