ARCH · TOOLS

Tool Use 고급

⚡ Parallel Tool Calling 🔗 Tool Chaining 📦 결과 캐싱 & 샌드박스

병렬 도구 호출, 도구 체이닝, 동적 도구 선택, 결과 캐싱, 안전한 코드 실행 샌드박스까지 프로덕션 Tool Use 패턴을 마스터합니다.

Parallel Tool Calling

Claude를 포함한 최신 LLM은 단일 응답에서 여러 도구를 동시에 요청할 수 있습니다. 독립적인 정보 조회를 병렬화해 지연을 크게 줄입니다.

❌ 순차 실행 (3.2초) LLM 추론 search() weather() stocks() 최종 답변 ✅ 병렬 실행 (1.1초) LLM 추론 3개 동시 요청 search() weather() stocks() 결과 통합 → 최종 답변 3배 빠름 동일한 결과, 더 낮은 지연 그림 1. 순차 vs 병렬 도구 실행 — 독립 도구는 병렬화로 지연 최소화
pythonparallel_tools.py — 병렬 도구 호출 처리
import asyncio
from langchain_core.messages import ToolMessage
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-opus-4-7")

async def parallel_tool_node(state) -> dict:
    """LLM이 요청한 모든 도구를 병렬로 실행"""
    last_message = state["messages"][-1]
    tool_calls = last_message.tool_calls

    if not tool_calls:
        return {}

    # 모든 도구 호출을 동시에 실행
    tasks = [
        execute_tool_async(call["name"], call["args"], call["id"])
        for call in tool_calls
    ]
    tool_results = await asyncio.gather(*tasks, return_exceptions=True)

    tool_messages = []
    for call, result in zip(tool_calls, tool_results):
        if isinstance(result, Exception):
            content = f"Error: {str(result)}"
        else:
            content = str(result)

        tool_messages.append(ToolMessage(
            tool_call_id=call["id"],
            name=call["name"],
            content=content
        ))

    return {"messages": tool_messages}

async def execute_tool_async(name: str, args: dict, call_id: str):
    """도구 이름으로 실제 함수 디스패치"""
    tool_map = {
        "search":  search_tool,
        "weather": weather_tool,
        "stocks":  stocks_tool,
    }
    fn = tool_map.get(name)
    if not fn:
        raise ValueError(f"Unknown tool: {name}")
    return await fn(**args)

도구 결과 캐싱

동일한 도구 호출이 반복될 때 이전 결과를 재사용합니다. API 비용 절감과 속도 향상을 동시에 달성합니다.

pythontool_cache.py — TTL 기반 도구 결과 캐시
import hashlib, json, time
from functools import wraps

class ToolCache:
    def __init__(self, default_ttl: int = 300):
        self._cache: dict[str, tuple[any, float]] = {}  # key → (result, expires_at)
        self.default_ttl = default_ttl

    def _make_key(self, tool_name: str, args: dict) -> str:
        payload = json.dumps({"tool": tool_name, "args": args}, sort_keys=True)
        return hashlib.sha256(payload.encode()).hexdigest()[:16]

    def get(self, tool_name: str, args: dict):
        key = self._make_key(tool_name, args)
        if key in self._cache:
            result, expires_at = self._cache[key]
            if time.time() < expires_at:
                return result
            del self._cache[key]
        return None

    def set(self, tool_name: str, args: dict, result, ttl: int = None):
        key = self._make_key(tool_name, args)
        self._cache[key] = (result, time.time() + (ttl or self.default_ttl))

cache = ToolCache()

# TTL은 도구별로 다르게 설정
TOOL_TTL = {
    "search":        300,   # 5분 — 검색 결과
    "weather":       600,   # 10분 — 날씨
    "stock_price":   30,    # 30초 — 주가 (실시간)
    "db_query":      0,     # 캐시 없음 — 동적 데이터
    "wikipedia":     86400, # 24시간 — 정적 지식
}

async def cached_tool_call(tool_name: str, args: dict):
    ttl = TOOL_TTL.get(tool_name, 300)

    if ttl > 0:
        cached = cache.get(tool_name, args)
        if cached is not None:
            return cached

    result = await execute_tool_async(tool_name, args, "")

    if ttl > 0:
        cache.set(tool_name, args, result, ttl)

    return result

동적 도구 선택

모든 도구를 LLM에 제공하면 컨텍스트 윈도우가 낭비됩니다. 요청 내용에 따라 관련 도구만 선택적으로 주입합니다.

pythondynamic_tools.py — 관련 도구만 동적으로 주입
from langchain_core.tools import BaseTool
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

# 도구를 벡터 스토어에 등록
def build_tool_index(tools: list[BaseTool]) -> Chroma:
    from langchain_core.documents import Document
    docs = [
        Document(
            page_content=f"{t.name}: {t.description}",
            metadata={"tool_name": t.name}
        )
        for t in tools
    ]
    return Chroma.from_documents(docs, embedding=OpenAIEmbeddings())

tool_index = build_tool_index(ALL_TOOLS)
tool_map = {t.name: t for t in ALL_TOOLS}

def select_tools(query: str, k: int = 5) -> list[BaseTool]:
    """쿼리와 가장 관련된 도구 k개 선택"""
    results = tool_index.similarity_search(query, k=k)
    return [
        tool_map[doc.metadata["tool_name"]]
        for doc in results
        if doc.metadata["tool_name"] in tool_map
    ]

# 노드에서 사용
async def adaptive_agent_node(state) -> dict:
    query = state["messages"][-1].content

    # 이 요청에 맞는 도구만 선택
    relevant_tools = select_tools(query, k=5)
    llm_with_tools = llm.bind_tools(relevant_tools)

    response = await llm_with_tools.ainvoke(state["messages"])
    return {"messages": [response]}

코드 실행 샌드박스

pythonsandbox.py — 격리된 Python 코드 실행
import subprocess, tempfile, os

async def safe_python_exec(code: str, timeout: int = 10) -> dict:
    """격리된 프로세스에서 Python 코드 실행"""
    with tempfile.NamedTemporaryFile(
        suffix=".py", mode="w", delete=False
    ) as f:
        f.write(code)
        tmp_path = f.name

    try:
        result = subprocess.run(
            ["python", tmp_path],
            capture_output=True,
            text=True,
            timeout=timeout,
            # 보안: 네트워크/파일 시스템 제한
            env={
                "PATH": "/usr/bin",
                "PYTHONPATH": "",
            }
        )
        return {
            "stdout": result.stdout[:5000],  # 최대 5000자
            "stderr": result.stderr[:1000],
            "returncode": result.returncode,
            "success": result.returncode == 0
        }
    except subprocess.TimeoutExpired:
        return {"error": f"Timeout ({timeout}s) 초과", "success": False}
    finally:
        os.unlink(tmp_path)

# 도구로 등록
from langchain_core.tools import tool

@tool
async def python_repl(code: str) -> str:
    """Python 코드를 안전한 샌드박스에서 실행합니다.

    Args:
        code: 실행할 Python 코드
    """
    result = await safe_python_exec(code)
    if result["success"]:
        return result["stdout"] or "(출력 없음)"
    return f"오류:\n{result.get('stderr', result.get('error', ''))}"
↑ 목차