도구 실패를 데이터로 만들기 — OpenClaw에서 에러 객체를 컨텍스트에 주입해 자가 복구 루프를 설계하는 법

# 개념적 예시 — ToolResult가 담아야 할 최소 정보
from dataclasses import dataclass
from typing import Any
 
@dataclass
class ToolResult:
    success: bool
    error: str | None       # 모델이 읽을 수 있는 자연어 에러 설명
    recoverable: bool       # True면 복구 시도, False면 즉각 에스컬레이션
    data: Any | None = None # 성공 시 반환 데이터

# 나쁜 예
ToolResult(
    success=False,
    error="HTTPError: 503",
    recoverable=True,
)
 
# 좋은 예
ToolResult(
    success=False,
    error=(
        "외부 결제 API가 일시적으로 응답 불가 상태입니다 (HTTP 503). "
        "서비스가 과부하 상태일 가능성이 높습니다. "
        "캐시된 결제 수단 목록을 대신 사용하거나, "
        "이 단계를 건너뛰고 나중에 재시도하는 방법을 고려해 보세요."
    ),
    recoverable=True,
)

# 개념적 예시 — SELF_CORRECT 전략 구현
import os
 
MAX_RETRIES = int(os.getenv("OPENCLAW_MAX_RETRIES", "3"))
 
def run_with_self_correct(agent, tool_call, context):
    for attempt in range(MAX_RETRIES):
        result = agent.execute_tool(tool_call, context)
 
        if result.success:
            return result
 
        if not result.recoverable:
            raise NonRecoverableError(result.error)
 
        context.add_observation(
            role="tool_error",
            content=(
                f"[도구 실행 실패 — 시도 {attempt + 1}/{MAX_RETRIES}]\n"
                f"도구: {tool_call.name}\n"
                f"오류: {result.error}\n"
                f"재시도 가능: {result.recoverable}\n\n"
                "위 오류를 참고해 다른 접근 방법을 시도해 주세요."
            ),
        )
 
        tool_call = agent.replan(context)
 
    return ToolResult(
        success=False,
        error=f"최대 재시도 횟수({MAX_RETRIES})를 초과했습니다.",
        recoverable=False,
    )

# 개념적 예시 — FALLBACK_TOOL 체인
TOOL_FALLBACK_CHAIN = {
    "sql_query_tool": "readonly_cache_tool",
    "payment_api_tool": "cached_payment_data_tool",
    "realtime_search_tool": "indexed_search_tool",
}
 
def run_with_fallback(agent, tool_call, context):
    result = agent.execute_tool(tool_call, context)
 
    if result.success:
        return result
 
    fallback_tool_name = TOOL_FALLBACK_CHAIN.get(tool_call.name)
    if not fallback_tool_name or not result.recoverable:
        return result
 
    context.add_observation(
        role="tool_error",
        content=(
            f"{tool_call.name} 실패. {fallback_tool_name}으로 전환합니다. "
            f"원인: {result.error}"
        ),
    )
 
    fallback_call = tool_call.with_name(fallback_tool_name)
    return agent.execute_tool(fallback_call, context)

# 개념적 예시 — 격리 영역 선언 아이디어
class ErrorBoundary:
    def __init__(self, strategy: str, **params):
        self.strategy = strategy
        self.params = params
 
enrichment_boundary = ErrorBoundary(
    strategy="skip_and_note",
    note_template="외부 데이터 보강 단계를 건너뜁니다. 이유: {error}",
)
 
cache_boundary = ErrorBoundary(
    strategy="use_cached",
    cache_key="last_successful_result",
)
 
payment_boundary = ErrorBoundary(
    strategy="ask_human",
    prompt="결제 처리 도구가 실패했습니다. 수동으로 처리하시겠습니까?",
)

# 개념적 예시 — 세션 내 자기 수정 루프
def self_correction_loop(agent, task, max_attempts=5):
    context = agent.build_context(task)
    critique_history = []
 
    for attempt in range(max_attempts):
        result = agent.execute(context)
 
        if result.success:
            return result
 
        critique = agent.generate_critique(
            task=task,
            attempt=result,
            error=result.error,
            history=critique_history,
        )
        critique_history.append(critique)
 
        context.add_reflection(
            content=f"시도 {attempt + 1} 실패 분석:\n{critique}"
        )
 
    return ToolResult(
        success=False,
        error="최대 시도 횟수 초과",
        recoverable=False,
    )

# 개념적 예시 — 정확히 동일한 도구 호출 반복 감지
from collections import deque
import hashlib
 
class LoopDetector:
    def __init__(self, window=3, threshold=3):
        self.recent_steps = deque(maxlen=window)
        self.threshold = threshold
 
    def record(self, step):
        canonical = f"{step.tool_name}:{sorted(step.args.items())}"
        step_hash = hashlib.md5(canonical.encode()).hexdigest()
        self.recent_steps.append(step_hash)
 
    def is_looping(self) -> bool:
        if len(self.recent_steps) < self.threshold:
            return False
        return len(set(self.recent_steps)) == 1
 
    def get_reflection_prompt(self) -> str:
        return (
            "현재 접근 방식이 반복적으로 실패하고 있습니다. "
            "지금까지 시도한 방법을 재검토하고, "
            "완전히 다른 도구나 전략을 선택해 주세요."
        )

# 개념적 예시 — LangGraph에서 에러 라우팅
from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal
 
class AgentState(TypedDict):
    messages: list
    error_count: int
    last_error: str | None
 
def route_on_error(state: AgentState) -> Literal["recover", "escalate", "continue"]:
    if state.get("last_error") is None:
        return "continue"
    if state["error_count"] >= 3:
        return "escalate"
    return "recover"
 
def recovery_node(state: AgentState) -> AgentState:
    error_message = {
        "role": "tool",
        "content": f"이전 시도 실패: {state['last_error']}. 다른 방법을 시도해 주세요.",
    }
    return {
        **state,
        "messages": state["messages"] + [error_message],
        "last_error": None,
    }
 
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.add_node("recover", recovery_node)
graph.add_node("escalate", escalate_node)
 
graph.add_conditional_edges(
    "tools",
    route_on_error,
    {"continue": "agent", "recover": "recover", "escalate": END},
)
graph.add_edge("recover", "agent")

# 개념적 예시 — PydanticAI의 ModelRetry 사용
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext
from pydantic_ai.exceptions import ModelRetry
 
class AnalysisResult(BaseModel):
    summary: str
    confidence: float
 
agent = Agent(
    model="claude-sonnet-5",
    result_type=AnalysisResult,
    retries=3,
)
 
@agent.result_validator
async def validate_confidence(ctx: RunContext, result: AnalysisResult) -> AnalysisResult:
    if not 0.0 <= result.confidence <= 1.0:
        raise ModelRetry(
            f"confidence는 0.0~1.0 사이여야 합니다. 받은 값: {result.confidence}"
        )
    return result