AI 논문 읽기: Agentic Reasoning 구현 가이드 2026

"""
ReAct 패턴의 핵심 루프 구현.

LLM이 자연어로 추론하고, 도구를 호출하고, 결과를 관찰하는
사이클을 max_steps까지 반복한다.
"""
from dataclasses import dataclass, field
from typing import Callable, Optional
from enum import Enum
import json
import re


class StepType(Enum):
    THOUGHT = "thought"
    ACTION = "action"
    OBSERVATION = "observation"
    FINAL_ANSWER = "final_answer"


@dataclass
class AgentStep:
    step_type: StepType
    content: str
    tool_name: Optional[str] = None
    tool_input: Optional[dict] = None
    token_count: int = 0


@dataclass
class AgentTrace:
    """에이전트 실행의 전체 기록."""
    question: str
    steps: list[AgentStep] = field(default_factory=list)
    final_answer: Optional[str] = None
    total_tokens: int = 0
    total_tool_calls: int = 0

    def add_step(self, step: AgentStep):
        self.steps.append(step)
        self.total_tokens += step.token_count
        if step.step_type == StepType.ACTION:
            self.total_tool_calls += 1


class ReActAgent:
    """ReAct 패턴 에이전트.

    LLM과 도구 집합을 받아, 질문에 대해 반복적으로
    추론-행동-관찰 루프를 수행한다.
    """

    SYSTEM_PROMPT = """You are a helpful assistant that solves problems step by step.
For each step, you MUST output exactly one of:
- Thought: <your reasoning about what to do next>
- Action: <tool_name>({"param": "value"})
- Final Answer: <your final response to the user>

Available tools:
{tool_descriptions}

Rules:
- Always think before acting.
- After observing a tool result, think about what it means before the next action.
- When you have enough information, provide Final Answer.
"""

    def __init__(
        self,
        llm: Callable,   # (messages: list[dict]) -> str
        tools: dict[str, Callable],
        tool_descriptions: dict[str, str],
        max_steps: int = 10,
        max_tokens_per_step: int = 1024,
    ):
        self.llm = llm
        self.tools = tools
        self.tool_descriptions = tool_descriptions
        self.max_steps = max_steps
        self.max_tokens_per_step = max_tokens_per_step

    def run(self, question: str) -> AgentTrace:
        trace = AgentTrace(question=question)

        # 시스템 프롬프트에 도구 설명 삽입
        tool_desc_text = "\n".join(
            f"- {name}: {desc}"
            for name, desc in self.tool_descriptions.items()
        )
        system_msg = self.SYSTEM_PROMPT.format(tool_descriptions=tool_desc_text)

        messages = [
            {"role": "system", "content": system_msg},
            {"role": "user", "content": question},
        ]

        for step_num in range(self.max_steps):
            # LLM에게 다음 단계를 생성하도록 요청
            response = self.llm(messages)
            parsed = self._parse_response(response)

            if parsed.step_type == StepType.FINAL_ANSWER:
                trace.final_answer = parsed.content
                trace.add_step(parsed)
                break

            trace.add_step(parsed)
            messages.append({"role": "assistant", "content": response})

            if parsed.step_type == StepType.ACTION and parsed.tool_name:
                # 도구 실행
                observation = self._execute_tool(
                    parsed.tool_name, parsed.tool_input or {}
                )
                obs_step = AgentStep(
                    step_type=StepType.OBSERVATION,
                    content=observation,
                )
                trace.add_step(obs_step)
                messages.append({
                    "role": "user",
                    "content": f"Observation: {observation}",
                })

        return trace

    def _parse_response(self, response: str) -> AgentStep:
        """LLM 출력을 파싱하여 Thought/Action/Final Answer를 구분한다."""
        response = response.strip()

        # Final Answer 체크
        if response.lower().startswith("final answer:"):
            return AgentStep(
                step_type=StepType.FINAL_ANSWER,
                content=response[len("final answer:"):].strip(),
            )

        # Action 파싱: Action: tool_name({"key": "value"})
        action_match = re.match(
            r'Action:\s*(\w+)\((\{.*\})\)', response, re.DOTALL
        )
        if action_match:
            tool_name = action_match.group(1)
            try:
                tool_input = json.loads(action_match.group(2))
            except json.JSONDecodeError:
                tool_input = {}
            return AgentStep(
                step_type=StepType.ACTION,
                content=response,
                tool_name=tool_name,
                tool_input=tool_input,
            )

        # 그 외는 Thought로 처리
        return AgentStep(
            step_type=StepType.THOUGHT,
            content=response,
        )

    def _execute_tool(self, tool_name: str, tool_input: dict) -> str:
        """도구를 실행하고 결과를 문자열로 반환한다."""
        if tool_name not in self.tools:
            return f"Error: Unknown tool '{tool_name}'. Available: {list(self.tools.keys())}"
        try:
            result = self.tools[tool_name](**tool_input)
            return str(result)
        except Exception as e:
            return f"Error executing {tool_name}: {type(e).__name__}: {str(e)}"

"""
프로덕션 환경의 에이전트 도구 정의.

각 도구는 입력 검증, timeout, 비용 제한을 내장하며,
실행 결과를 구조화된 형태로 반환한다.
"""
from dataclasses import dataclass
from typing import Any, Optional
import httpx
import time


@dataclass
class ToolResult:
    success: bool
    data: Any
    error: Optional[str] = None
    execution_time_ms: float = 0.0
    cost_usd: float = 0.0


class WebSearchTool:
    """웹 검색 도구.

    에이전트가 최신 정보를 조회할 때 사용한다.
    rate limit과 비용 제한을 내장한다.
    """

    def __init__(
        self,
        api_key: str,
        max_results: int = 5,
        timeout_seconds: float = 10.0,
        max_calls_per_minute: int = 10,
    ):
        self.api_key = api_key
        self.max_results = max_results
        self.timeout_seconds = timeout_seconds
        self.max_calls_per_minute = max_calls_per_minute
        self._call_timestamps: list[float] = []

    def _check_rate_limit(self) -> bool:
        now = time.time()
        self._call_timestamps = [
            ts for ts in self._call_timestamps if now - ts < 60
        ]
        return len(self._call_timestamps) < self.max_calls_per_minute

    def __call__(self, query: str) -> ToolResult:
        if not query or len(query) > 500:
            return ToolResult(
                success=False,
                data=None,
                error="Query must be 1-500 characters",
            )

        if not self._check_rate_limit():
            return ToolResult(
                success=False,
                data=None,
                error=f"Rate limit exceeded: max {self.max_calls_per_minute}/min",
            )

        start = time.monotonic()
        try:
            # 실제 검색 API 호출 (예시: Tavily, Serper 등)
            with httpx.Client(timeout=self.timeout_seconds) as client:
                response = client.get(
                    "https://api.search-provider.com/search",
                    params={"q": query, "max_results": self.max_results},
                    headers={"Authorization": f"Bearer {self.api_key}"},
                )
                response.raise_for_status()
                elapsed = (time.monotonic() - start) * 1000
                self._call_timestamps.append(time.time())

                return ToolResult(
                    success=True,
                    data=response.json(),
                    execution_time_ms=elapsed,
                    cost_usd=0.001,  # 건당 추정 비용
                )
        except httpx.TimeoutException:
            return ToolResult(
                success=False,
                data=None,
                error=f"Search timed out after {self.timeout_seconds}s",
                execution_time_ms=(time.monotonic() - start) * 1000,
            )
        except httpx.HTTPStatusError as e:
            return ToolResult(
                success=False,
                data=None,
                error=f"HTTP {e.response.status_code}: {e.response.text[:200]}",
                execution_time_ms=(time.monotonic() - start) * 1000,
            )


class CodeExecutionTool:
    """코드 실행 도구.

    에이전트가 Python 코드를 실행해 계산이나 데이터 처리를 수행한다.
    보안을 위해 허용된 모듈만 import 가능하고, 실행 시간과 메모리를 제한한다.
    """

    ALLOWED_MODULES = {"math", "statistics", "json", "re", "datetime", "collections"}

    def __init__(self, timeout_seconds: float = 5.0):
        self.timeout_seconds = timeout_seconds

    def __call__(self, code: str) -> ToolResult:
        if not code or len(code) > 5000:
            return ToolResult(
                success=False,
                data=None,
                error="Code must be 1-5000 characters",
            )

        # import 검사: 허용된 모듈만 사용 가능
        import_lines = [
            line.strip() for line in code.splitlines()
            if line.strip().startswith("import ") or line.strip().startswith("from ")
        ]
        for line in import_lines:
            module = line.split()[1].split(".")[0]
            if module not in self.ALLOWED_MODULES:
                return ToolResult(
                    success=False,
                    data=None,
                    error=f"Module '{module}' not allowed. Allowed: {self.ALLOWED_MODULES}",
                )

        start = time.monotonic()
        try:
            # 제한된 환경에서 실행
            local_vars: dict = {}
            exec(code, {"__builtins__": {}}, local_vars)  # noqa: S102

            elapsed = (time.monotonic() - start) * 1000
            # 'result' 변수가 있으면 그것을 반환
            result = local_vars.get("result", str(local_vars))

            return ToolResult(
                success=True,
                data=result,
                execution_time_ms=elapsed,
            )
        except Exception as e:
            return ToolResult(
                success=False,
                data=None,
                error=f"{type(e).__name__}: {str(e)}",
                execution_time_ms=(time.monotonic() - start) * 1000,
            )

"""
에이전트의 작업 기억(working memory) 관리.

전체 대화 이력을 유지하되, LLM에 전달할 때는
중요도 기반으로 요약/선택해서 컨텍스트 윈도우에 맞춘다.
"""
from dataclasses import dataclass, field
from typing import Optional
import hashlib


@dataclass
class MemoryEntry:
    role: str
    content: str
    step_number: int
    importance: float = 0.5  # 0.0 ~ 1.0
    token_count: int = 0
    content_hash: str = ""

    def __post_init__(self):
        if not self.content_hash:
            self.content_hash = hashlib.md5(
                self.content.encode()
            ).hexdigest()[:8]


class SlidingWindowMemory:
    """슬라이딩 윈도우 + 중요도 기반 메모리 관리.

    최근 K개 메시지는 항상 유지하고,
    그 이전의 메시지는 importance 점수에 따라 선별한다.
    """

    def __init__(
        self,
        max_tokens: int = 8192,
        recent_window: int = 6,     # 최근 N개는 항상 유지
        system_prompt_tokens: int = 500,
    ):
        self.max_tokens = max_tokens
        self.recent_window = recent_window
        self.system_prompt_tokens = system_prompt_tokens
        self.entries: list[MemoryEntry] = []

    def add(self, entry: MemoryEntry):
        # 중복 방지
        if any(e.content_hash == entry.content_hash for e in self.entries):
            return
        self.entries.append(entry)

    def get_context(self, system_message: str) -> list[dict]:
        """LLM에 전달할 메시지 리스트를 구성한다.

        1. 시스템 프롬프트는 항상 포함
        2. 최근 recent_window개는 항상 포함
        3. 나머지는 importance 순으로 예산 내에서 포함
        """
        budget = self.max_tokens - self.system_prompt_tokens
        messages = [{"role": "system", "content": system_message}]

        if not self.entries:
            return messages

        # 최근 메시지 먼저 확보
        recent = self.entries[-self.recent_window:]
        older = self.entries[:-self.recent_window] if len(self.entries) > self.recent_window else []

        recent_tokens = sum(e.token_count for e in recent)

        # 오래된 메시지 중 중요한 것을 예산 내에서 추가
        remaining_budget = budget - recent_tokens
        selected_older = sorted(older, key=lambda e: e.importance, reverse=True)

        included_older = []
        for entry in selected_older:
            if remaining_budget <= 0:
                break
            if entry.token_count <= remaining_budget:
                included_older.append(entry)
                remaining_budget -= entry.token_count

        # 시간순으로 정렬해서 메시지 구성
        included_older.sort(key=lambda e: e.step_number)
        all_entries = included_older + recent

        for entry in all_entries:
            messages.append({"role": entry.role, "content": entry.content})

        return messages

    def mark_important(self, step_number: int, importance: float = 1.0):
        """특정 단계의 중요도를 높인다.

        도구 실행 결과, 핵심 발견 등을 표시할 때 사용.
        """
        for entry in self.entries:
            if entry.step_number == step_number:
                entry.importance = importance
                break

"""
에이전트 실행의 비용과 자원 사용을 제어하는 가드레일.
"""
from dataclasses import dataclass
from typing import Optional
import time


@dataclass
class AgentBudget:
    max_llm_calls: int = 15
    max_tool_calls: int = 10
    max_total_tokens: int = 50_000
    max_cost_usd: float = 0.50
    max_wall_time_seconds: float = 120.0


@dataclass
class AgentUsage:
    llm_calls: int = 0
    tool_calls: int = 0
    total_tokens: int = 0
    total_cost_usd: float = 0.0
    start_time: float = 0.0

    def elapsed_seconds(self) -> float:
        return time.time() - self.start_time if self.start_time else 0.0


class BudgetGuard:
    """에이전트 실행 예산 감시자.

    매 단계 전에 check()를 호출하여 예산 초과 여부를 확인한다.
    초과 시 에이전트는 현재까지의 결과로 조기 종료해야 한다.
    """

    def __init__(self, budget: AgentBudget):
        self.budget = budget
        self.usage = AgentUsage()

    def start(self):
        self.usage.start_time = time.time()

    def record_llm_call(self, tokens: int, cost_usd: float):
        self.usage.llm_calls += 1
        self.usage.total_tokens += tokens
        self.usage.total_cost_usd += cost_usd

    def record_tool_call(self, cost_usd: float = 0.0):
        self.usage.tool_calls += 1
        self.usage.total_cost_usd += cost_usd

    def check(self) -> Optional[str]:
        """예산 초과 시 사유를 반환한다. 정상이면 None."""
        if self.usage.llm_calls >= self.budget.max_llm_calls:
            return f"LLM call limit reached: {self.usage.llm_calls}/{self.budget.max_llm_calls}"

        if self.usage.tool_calls >= self.budget.max_tool_calls:
            return f"Tool call limit reached: {self.usage.tool_calls}/{self.budget.max_tool_calls}"

        if self.usage.total_tokens >= self.budget.max_total_tokens:
            return f"Token limit reached: {self.usage.total_tokens}/{self.budget.max_total_tokens}"

        if self.usage.total_cost_usd >= self.budget.max_cost_usd:
            return f"Cost limit reached: ${self.usage.total_cost_usd:.3f}/${self.budget.max_cost_usd:.3f}"

        elapsed = self.usage.elapsed_seconds()
        if elapsed >= self.budget.max_wall_time_seconds:
            return f"Time limit reached: {elapsed:.1f}s/{self.budget.max_wall_time_seconds}s"

        return None

    def summary(self) -> dict:
        return {
            "llm_calls": f"{self.usage.llm_calls}/{self.budget.max_llm_calls}",
            "tool_calls": f"{self.usage.tool_calls}/{self.budget.max_tool_calls}",
            "tokens": f"{self.usage.total_tokens}/{self.budget.max_total_tokens}",
            "cost_usd": f"${self.usage.total_cost_usd:.4f}/${self.budget.max_cost_usd:.4f}",
            "elapsed_s": f"{self.usage.elapsed_seconds():.1f}/{self.budget.max_wall_time_seconds}",
        }

"""
에이전트 평가 프레임워크.

정답률 외에 효율성, 도구 사용 적절성, 추론 품질을
종합적으로 측정한다.
"""
from dataclasses import dataclass


@dataclass
class AgentEvalMetrics:
    # 정확도
    final_answer_correct: bool
    partial_credit: float          # 0.0 ~ 1.0 (부분 점수)

    # 효율성
    total_steps: int
    total_tool_calls: int
    total_tokens: int
    total_cost_usd: float
    wall_time_seconds: float

    # 도구 사용 품질
    unnecessary_tool_calls: int    # 불필요한 도구 호출 수
    failed_tool_calls: int         # 실패한 도구 호출 수
    tool_call_accuracy: float      # 올바른 도구를 올바른 입력으로 호출한 비율

    # 추론 품질
    reasoning_coherence: float     # 추론의 논리적 일관성 (0.0 ~ 1.0)
    hallucination_count: int       # 근거 없는 주장 수

    @property
    def efficiency_score(self) -> float:
        """효율성 점수: 정답에 도달하기까지 얼마나 적은 자원을 사용했는가."""
        if not self.final_answer_correct:
            return 0.0
        # 적을수록 효율적 -> 역수로 변환
        step_penalty = min(self.total_steps / 10, 1.0)
        cost_penalty = min(self.total_cost_usd / 0.10, 1.0)
        return max(0.0, 1.0 - (step_penalty + cost_penalty) / 2)

    @property
    def overall_score(self) -> float:
        """종합 점수."""
        weights = {
            "accuracy": 0.4,
            "efficiency": 0.2,
            "tool_quality": 0.2,
            "reasoning": 0.2,
        }
        accuracy = 1.0 if self.final_answer_correct else self.partial_credit
        return (
            weights["accuracy"] * accuracy
            + weights["efficiency"] * self.efficiency_score
            + weights["tool_quality"] * self.tool_call_accuracy
            + weights["reasoning"] * self.reasoning_coherence
        )