Agentic AI: 스키마 검증된 도구 실행 및 결정론적 캐싱

I’m happy to translate the article for you, but I’ll need the full text you’d like translated. Could you please paste the content (excluding the source line you already provided) here? Once I have it, I’ll keep the source link unchanged at the top and translate the rest into Korean while preserving the original formatting and technical terms.

개요

에이전시 AI 시스템이 실패하는 이유는 모델이 추론을 못해서가 아니라 도구 실행이 관리되지 않기 때문입니다.

에이전트가 계획, 재시도, 자기 비판, 협업 등을 허용받으면 도구 호출이 급격히 증가합니다. 엄격한 제어가 없으면 인프라 장애, 예측할 수 없는 비용 증가, 비결정적 행동이 발생합니다.

이 문서는 에이전시 AI 시스템의 도구 실행 레이어를 다음 두 가지 명시적이고 독립적인 메커니즘을 사용해 설계하는 방법을 설명합니다:

• 계약 기반 도구 실행
• 결정론적 도구 결과 캐싱

각 메커니즘은 서로 다른 유형의 프로덕션 실패를 해결하며 별도로 구현되어야 합니다.

실제 프로덕션 시나리오

배경

SRE 팀을 위한 Incident Analysis Agent를 구축하고 있습니다.

에이전트가 수행하는 작업

• 서비스 로그 가져오기
• 오류 패턴 분석
• 신뢰도가 낮을 경우 로그 재수집
• 두 번째 에이전트(비평가)가 결과를 검증하도록 허용

도구 특성

Tool name: fetch_service_logs

Backend: Elasticsearch / Loki / Splunk

Latency: 300–800 ms

• 속도 제한
• 실행당 비용이 많이 듦

이는 일반적인 실무 에이전트 작업 부하입니다.

Source: …

Part I: Contract‑Driven Tool Execution in Agentic AI Systems

계약이 없는 경우의 문제

LLM이 도구 인자를 직접 출력하면 런타임은 다음과 같은 입력을 받게 됩니다:

{"service": "auth", "window": "24 hours"}
{"service": "Auth Service", "window": "yesterday"}
{"service": ["auth"], "window": 24}
{"service": "", "window": "24h"}

왜 이런 일이 발생하는가

• LLM은 자연어로 추론한다
• LLM은 인자를 바꿔 말한다
• LLM은 타입‑안전 시스템이 아니다

프로덕션에서 깨지는 것들

• 잘못된 Elasticsearch 쿼리
• 전체 인덱스 스캔
• 쿼리‑빌더 충돌
• 조용한 데이터 손상
• 재시도 루프가 실패를 증폭시킴

모델이 항상 유효한 입력을 만든다고 기대하는 것은 시스템 설계가 아닙니다.

계약‑기반 도구 실행이 의미하는 바

계약‑기반 실행은 다음을 의미합니다:

• 런타임이 도구 인터페이스를 소유한다
• 모델은 그 인터페이스에 맞춰야 한다
• 잘못된 입력은 인프라에 절대 도달하지 않는다

이는 프로덕션 API에서 사용되는 경계 강제와 동일합니다.

Step 1: 엄격한 도구 계약 정의

from pydantic import BaseModel, Field, field_validator
import re
from typing import List

class FetchServiceLogsInput(BaseModel):
    service: str = Field(
        ...,
        description="Kubernetes service name, lowercase, no spaces"
    )
    window: str = Field(
        ...,
        description="Time window format: 5m, 1h, 24h"
    )

    @field_validator("service")
    @classmethod
    def validate_service(cls, value: str) -> str:
        if not value:
            raise ValueError("service cannot be empty")
        if not re.fullmatch(r"[a-z0-9\-]+", value):
            raise ValueError("service must be lowercase alphanumeric with dashes")
        return value

    @field_validator("window")
    @classmethod
    def validate_window(cls, value: str) -> str:
        if not re.fullmatch(r"\d+(m|h)", value):
            raise ValueError("window must be like 5m, 1h, 24h")
        return value

class FetchServiceLogsOutput(BaseModel):
    logs: List[str]

이러한 검증이 방지하는 것

이 게이트를 통과하지 않으면 인프라에 도달하지 못합니다.

Step 2: 실제 도구 구현

def fetch_service_logs(service: str, window: str) -> list[str]:
    print(f"QUERY logs for service={service}, window={window}")
    return [
        f"[ERROR] timeout detected in {service}",
        f"[WARN] retry triggered in {service}",
    ]

Step 3: 런타임이 소유하는 도구 레지스트리

TOOLS = {
    "fetch_service_logs": {
        "version": "v1",
        "input_model": FetchServiceLogsInput,
        "output_model": FetchServiceLogsOutput,
        "handler": fetch_service_logs,
        "cache_ttl": 3600,   # seconds
    }
}

에이전트는 도구를 임의로 만들거나, 스키마를 우회하거나, 버전을 바꿀 수 없습니다.

Step 4: 계약‑기반 실행 경계

def execute_tool_contract(tool_name: str, raw_args: dict):
    tool = TOOLS[tool_name]

    # 계약에 따라 입력 검증
    args = tool["input_model"](**raw_args)

    # 정제된 딕셔너리로 핸들러 호출
    raw_result = tool["handler"](**args.model_dump())

    # 결과를 출력 모델에 래핑
    return tool["output_model"](logs=raw_result)

계약 강제 실행 흐름

Agent emits tool call
        ↓
Raw arguments (untrusted)
        ↓
Schema validation
   ┌───────────────┐
   │ Invalid       │ → reject and re‑plan
   └───────────────┘
          ↓
       Valid
          ↓
Tool executes
          ↓
Infrastructure queried safely

Part II: 에이전트형 AI 시스템에서 결정적 캐싱

계약이 추가된 후의 문제

완벽한 검증을 수행하더라도 에이전트는 작업을 반복합니다:

execute_tool_contract(
    "fetch_service_logs",
    {"service": "auth-service", "window": "24h"}
)

execute_tool_contract(
    "fetch_service_logs",
    {"window": "24h", "service": "auth-service"}
)

같은 의도, 같은 백엔드가 두 번 실행됩니다.

왜 단순 캐싱이 실패하는가

{"service": "auth-service", "window": "24h"}
{"window": "24h", "service": "auth-service"}

다른 문자열 → 서로 다른 캐시 키가 생성됩니다. 의미적으로는 동일하지만.

에이전트형 시스템은 원시 문자열 일치가 아니라 의미적 동등성을 요구합니다.

결정적 캐싱에 필요한 인프라

1. 정규화(Canonicalisation) – 들어오는 인수를 결정적이고 정렬된 형태(예: 정렬된 JSON)로 변환합니다.
2. 해시 기반 캐시 키 – 정규화된 페이로드와 툴 버전을 결합해 안정적인 해시(SHA‑256)를 계산합니다.
3. 결과 저장 – 해시와 TTL과 함께 출력 모델(또는 직렬화된 형태)을 영구 저장합니다.
4. 캐시 조회 래퍼 – 핸들러를 호출하기 전에 캐시를 확인합니다. 히트 시 저장된 결과를 반환하고, 미스 시 실행 후 저장합니다.

간단한 구현 예시:

import json, hashlib, time
from collections import defaultdict

# Simple in‑memory cache for illustration
_CACHE = defaultdict(dict)   # {tool_name: {hash: (timestamp, result)}}

def _canonicalise(args: dict) -> str:
    """Return a deterministic JSON string with sorted keys."""
    return json.dumps(args, sort_keys=True, separators=(",", ":"))

def _hash_payload(tool_name: str, payload: str) -> str:
    return hashlib.sha256(f"{tool_name}:{payload}".encode()).hexdigest()

def execute_with_cache(tool_name: str, raw_args: dict):
    tool = TOOLS[tool_name]

    # 1️⃣ Validate input
    args = tool["input_model"](**raw_args)

    # 2️⃣ Canonicalise & hash
    payload = _canonicalise(args.model_dump())
    key = _hash_payload(tool_name, payload)

    # 3️⃣ Cache lookup
    entry = _CACHE[tool_name].get(key)
    if entry:
        ts, cached_result = entry
        # (Cache hit logic would go here)
        return cached_result

    # 4️⃣ Execute and store
    raw_result = tool["handler"](**args.model_dump())
    validated = tool["output_model"](logs=raw_result)
    _CACHE[tool_name][key] = (time.time(), validated)
    return validated

예시 정규화 형태

fetch_service_logs|auth-service|24h|v1

단계 2: 캐시 설정 (Redis 예시)

import redis
import hashlib
import json

redis_client = redis.Redis(host="localhost", port=6379)

def cache_key(canonical: str) -> str:
    return hashlib.sha256(canonical.encode()).hexdigest()

단계 3: 캐시된 툴 실행

def execute_tool_cached(tool_name: str, raw_args: dict):
    tool = TOOLS[tool_name]

    args = tool["input_model"](**raw_args)

    canonical = json.dumps(
        {
            "tool": tool_name,
            "version": tool["version"],
            "args": args.model_dump(),
        },
        sort_keys=True,
        separators=(",", ":")
    )
    key = cache_key(canonical)

    cached = redis_client.get(key)
    if cached:
        print("CACHE HIT — skipping infra call")
        return tool["output_model"](**json.loads(cached))

    print("CACHE MISS — executing tool")
    raw_result = tool["handler"](**args.model_dump())
    validated = tool["output_model"](logs=raw_result)

    redis_client.setex(
        key,
        tool["cache_ttl"],
        validated.model_dump_json()
    )
    return validated

결정적 캐싱을 위한 실행 흐름

Validated tool request
        ↓
Canonicalization
        ↓
Hash generation
        ↓
Redis lookup
   ┌───────────────┐
   │ Cache HIT     │ → return cached result
   └───────────────┘
          ↓
       Cache MISS
          ↓
Execute expensive tool
          ↓
Validate output
          ↓
Sto

```markdown
TTL과 함께 결과 재생성
          ↓
결과 반환

책임 분리

최종 요약

Agentic AI 시스템은 도구 실행을 백엔드 인프라처럼 설계하고, LLM 부수 효과로 취급하지 않을 때 프로덕션에 준비됩니다.

• 계약은 실행을 안전하게 합니다.

• 캐싱은 실행을 확장 가능하게 합니다.

• 둘 중 하나라도 건너뛰면 실패가 보장됩니다.