LangChain Agents 도구 호출 구성 상세 해설

LangChain Agents는 LLM이 외부 도구를 호출하여 복잡한 작업을 수행하는 핵심 메커니즘입니다. 이 튜토리얼에서는 프로덕션 환경에서 안정적으로 동작하는 도구 호출 아키텍처를 구성하는 방법을 깊이 있게 다룹니다. HolySheep AI의 글로벌 API 게이트웨이를 활용하면 다양한 모델의 도구 호출 성능을 통합적으로 관리할 수 있습니다.

1. LangChain Agents 아키텍처 이해

LangChain Agents는 크게 Agent, Tool, ToolKit, Executor 네 가지 컴포넌트로 구성됩니다. 저는 실제 프로덕션 환경에서 수십 개의 도구를 동시에 관리하면서 지연 시간과 비용을 최적화하는 경험을 쌓았습니다.

1.1 핵심 컴포넌트 구조

# LangChain Agents 기본 구조
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.tools import Tool
from langchain_openai import ChatOpenAI
from langchain import hub

HolySheep AI를 통한 다중 모델 통합
llm = ChatOpenAI(
    model="gpt-4o",
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",
    temperature=0.7,
    max_tokens=2048
)

도구 정의
def search_database(query: str) -> str:
    """데이터베이스 검색 도구"""
    return f"검색 결과: {query}에 대한 {len(query)}개의 레코드 발견"

Tool 객체 생성
tools = [
    Tool(
        name="database_search",
        func=search_database,
        description="사용자 질문에 따라 데이터베이스를 검색합니다. 입력: 검색 쿼리 문자열"
    )
]

프롬프트 템플릿 로드
prompt = hub.pull("hwchase17/openai-functions-agent")

Agent 생성
agent = create_openai_functions_agent(llm, tools, prompt)

Executor 생성
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

1.2 도구 호출 흐름

• 사용자 입력 → Agent가 LLM에 요청 전송
• LLM 판단 → 도구 호출 필요 여부 결정
• 도구 실행 → 정의된 함수를 호출하여 결과 반환
• 응답 통합 → 도구 결과를 최종 응답에 결합

2. HolySheep AI 다중 모델 도구 호출 구성

저는 HolySheep AI를 활용하여 다양한 모델의 도구 호출 성능을 비교하고 최적화했습니다. Claude Sonnet은 구조화된 도구 호출에서 뛰어난 정확도를 보이며, GPT-4o는 복잡한 다단계推理에 강점이 있습니다.

# HolySheep AI를 통한 다중 모델 도구 호출 비교
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain.schema import HumanMessage
import time

HolySheep AI 글로벌 게이트웨이 설정
HOLYSHEEP_BASE = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"

모델별 성능 측정
def benchmark_tool_calling(model_name: str, api_key: str, base_url: str):
    """도구 호출 성능 벤치마크"""
    
    if "claude" in model_name.lower():
        llm = ChatAnthropic(
            model_name=model_name,
            anthropic_api_key=api_key,
            base_url=base_url
        )
    else:
        llm = ChatOpenAI(
            model=model_name,
            openai_api_key=api_key,
            base_url=base_url
        )
    
    tools = [
        {
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "특정 도시의 날씨 정보를 가져옵니다",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "location": {"type": "string", "description": "도시 이름"}
                    },
                    "required": ["location"]
                }
            }
        }
    ]
    
    messages = [HumanMessage(content="서울의 날씨를 알려주세요")]
    
    start_time = time.time()
    response = llm.bind(tools=tools).invoke(messages)
    latency = (time.time() - start_time) * 1000
    
    return {
        "model": model_name,
        "latency_ms": round(latency, 2),
        "tool_calls": len(response.tool_calls) if hasattr(response, 'tool_calls') else 0
    }

벤치마크 실행
results = [
    benchmark_tool_calling("gpt-4o", HOLYSHEEP_API_KEY, HOLYSHEEP_BASE),
    benchmark_tool_calling("claude-sonnet-4-20250514", HOLYSHEEP_API_KEY, HOLYSHEEP_BASE),
]

for r in results:
    print(f"{r['model']}: {r['latency_ms']}ms, 도구 호출 {r['tool_calls']}회")

2.1 HolySheep AI 모델별 비용 및 성능 비교

3. 고급 도구 정의 및 바인딩 패턴

3.1 Typed Tool 정의

저는 프로덕션 환경에서 타입 안전성을 확보하기 위해 Pydantic 모델을 활용한 도구 정의를 권장합니다. 이렇게 하면 런타임 에러를 크게 줄일 수 있습니다.

# Pydantic 기반 타입 안전한 도구 정의
from pydantic import BaseModel, Field
from langchain.tools import StructuredTool
from typing import Optional, List

class SearchInput(BaseModel):
    """검색 도구 입력 스키마"""
    query: str = Field(description="검색할 쿼리 문자열")
    max_results: int = Field(default=10, description="최대 결과 수", ge=1, le=100)
    include_filters: Optional[List[str]] = Field(default=None, description="필터 조건")

class DatabaseSearchTool:
    """프로덕션 레벨 데이터베이스 검색 도구"""
    
    def __init__(self, connection_string: str, timeout: int = 30):
        self.connection_string = connection_string
        self.timeout = timeout
        self._cache = {}
    
    def search(self, query: str, max_results: int = 10, include_filters: Optional[List[str]] = None) -> dict:
        """실제 검색 로직"""
        cache_key = f"{query}:{max_results}:{tuple(include_filters or [])}"
        
        if cache_key in self._cache:
            return {"cached": True, **self._cache[cache_key]}
        
        # 실제 DB 검색 시뮬레이션
        results = {
            "query": query,
            "total_found": max_results,
            "results": [{"id": i, "score": 1.0 - (i * 0.05)} for i in range(min(max_results, 20))],
            "filters_applied": include_filters or []
        }
        
        self._cache[cache_key] = results
        return results

StructuredTool로 변환
search_tool = StructuredTool.from_function(
    func=DatabaseSearchTool(connection_string="postgresql://...").search,
    name="database_search",
    description="데이터베이스에서 관련 정보를 검색합니다. 복잡한 쿼리와 필터링을 지원합니다.",
    args_schema=SearchInput,
    return_direct=False
)

3.2 비동기 도구 호출

대규모 프로덕션 환경에서는 동시 도구 호출이 필수적입니다. 저는 asyncio와 함께 비동기 도구를 구현하여 처리량을 크게 향상시켰습니다.

# 비동기 도구 호출 구성
import asyncio
from langchain.tools import tool
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI

@tool
async def async_search(query: str, use_cache: bool = True) -> str:
    """비동기 검색 도구"""
    await asyncio.sleep(0.1)  # 실제 API 지연 시뮬레이션
    return f"비동기 검색 결과: {query}"

@tool  
async def async_calculate(expression: str) -> str:
    """비동기 계산 도구"""
    await asyncio.sleep(0.05)
    result = eval(expression)  # 프로덕션에서는 eval 대신安全的 파서 사용
    return str(result)

@tool
async def async_fetch_url(url: str) -> str:
    """비동기 URL 가져오기"""
    await asyncio.sleep(0.2)
    return f"가져온 콘텐츠: {url[:50]}..."

HolySheep AI를 사용한 비동기 Agent 설정
async_llm = ChatOpenAI(
    model="gpt-4o",
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",
    streaming=True,
    max_tokens=2048
)

동시 도구 실행 테스트
async def test_concurrent_tools():
    tools = [async_search, async_calculate, async_fetch_url]
    agent = create_openai_functions_agent(async_llm, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools, verbose=False)
    
    start = asyncio.get_event_loop().time()
    
    # 순차 실행 vs 동시 실행 비교
    task = executor.ainvoke({
        "input": "1) 'AI 트렌드'를 검색하고 2) 2+2를 계산하고 3) https://example.com의 내용을 가져와"
    })
    
    result = await asyncio.wait_for(task, timeout=30.0)
    elapsed = (asyncio.get_event_loop().time() - start) * 1000
    
    print(f"동시 도구 호출 완료: {elapsed:.2f}ms")
    return result

asyncio.run(test_concurrent_tools())

4. 동시성 제어 및 성능 최적화

4.1 Rate Limiting 구현

HolySheep AI를 통해 여러 모델을 사용할 때, 각 모델의 rate limit을 준수하면서 최적의 처리량을 달성하는 것이 중요합니다.

# 토큰 기반 Rate Limiter 구현
import asyncio
import time
from collections import defaultdict
from dataclasses import dataclass
from typing import Dict, Optional

@dataclass
class RateLimitConfig:
    """Rate Limit 설정"""
    requests_per_minute: int
    tokens_per_minute: int
    max_concurrent: int = 5

class TokenBucketRateLimiter:
    """토큰 버킷 기반 Rate Limiter"""
    
    def __init__(self, config: RateLimitConfig):
        self.config = config
        self.tokens = config.tokens_per_minute
        self.last_refill = time.time()
        self.semaphore = asyncio.Semaphore(config.max_concurrent)
        self.request_timestamps: list = []
        self._lock = asyncio.Lock()
    
    async def acquire(self, tokens_needed: int = 1000) -> bool:
        """토큰 획득"""
        async with self._lock:
            self._refill()
            
            while self.tokens < tokens_needed:
                await asyncio.sleep(0.1)
                self._refill()
            
            self.tokens -= tokens_needed
            self.request_timestamps.append(time.time())
            self._clean_old_requests()
            
            return True
    
    def _refill(self):
        """토큰 재충전"""
        now = time.time()
        elapsed = now - self.last_refill
        refill_amount = elapsed * (self.config.tokens_per_minute / 60)
        self.tokens = min(self.config.tokens_per_minute, self.tokens + refill_amount)
        self.last_refill = now
    
    def _clean_old_requests(self):
        """1분 이전 요청 제거"""
        cutoff = time.time() - 60
        self.request_timestamps = [t for t in self.request_timestamps if t > cutoff]
    
    @property
    def current_rpm(self) -> int:
        return len(self.request_timestamps)
    
    @property
    def available_tokens(self) -> int:
        self._refill()
        return int(self.tokens)

HolySheep AI 모델별 Rate Limit 설정
MODEL_LIMITS: Dict[str, RateLimitConfig] = {
    "gpt-4o": RateLimitConfig(requests_per_minute=500, tokens_per_minute=150000, max_concurrent=10),
    "claude-sonnet-4-20250514": RateLimitConfig(requests_per_minute=400, tokens_per_minute=120000, max_concurrent=8),
    "gemini-2.5-flash": RateLimitConfig(requests_per_minute=1000, tokens_per_minute=500000, max_concurrent=20),
    "deepseek-chat": RateLimitConfig(requests_per_minute=2000, tokens_per_minute=1000000, max_concurrent=50),
}

Rate Limiter 미들웨어
class HolySheepRateLimiter:
    """HolySheep AI API Rate Limiter 미들웨어"""
    
    def __init__(self):
        self.limiters: Dict[str, TokenBucketRateLimiter] = {}
        self._init_limiters()
    
    def _init_limiters(self):
        for model, config in MODEL_LIMITS.items():
            self.limiters[model] = TokenBucketRateLimiter(config)
    
    async def call_with_limit(self, model: str, tokens: int, call_func):
        """Rate Limit 적용 함수 호출"""
        limiter = self.limiters.get(model)
        if not limiter:
            raise ValueError(f"Unknown model: {model}")
        
        await limiter.acquire(tokens)
        return await call_func()

사용 예시
rate_limiter = HolySheepRateLimiter()

async def optimized_api_call():
    # DeepSeek V3.2는 높은 동시성 허용 → 비용 효율적
    result = await rate_limiter.call_with_limit(
        "deepseek-chat",
        tokens=1500,
        call_func=lambda: llm.invoke("테스트 입력")
    )
    return result

4.2 응답 캐싱 전략

# Redis 기반 도구 응답 캐싱
import hashlib
import json
import redis
from typing import Any, Optional
from datetime import timedelta

class ToolResponseCache:
    """도구 호출 결과 캐싱"""
    
    def __init__(self, redis_url: str = "redis://localhost:6379", ttl: int = 3600):
        self.redis = redis.from_url(redis_url)
        self.ttl = ttl
    
    def _make_key(self, tool_name: str, arguments: dict) -> str:
        """캐시 키 생성"""
        args_str = json.dumps(arguments, sort_keys=True)
        hash_obj = hashlib.sha256(f"{tool_name}:{args_str}".encode())
        return f"tool_cache:{tool_name}:{hash_obj.hexdigest()[:16]}"
    
    def get(self, tool_name: str, arguments: dict) -> Optional[Any]:
        """캐시 조회"""
        key = self._make_key(tool_name, arguments)
        cached = self.redis.get(key)
        if cached:
            return json.loads(cached)
        return None
    
    def set(self, tool_name: str, arguments: dict, result: Any) -> bool:
        """캐시 저장"""
        key = self._make_key(tool_name, arguments)
        return self.redis.setex(
            key,
            timedelta(seconds=self.ttl),
            json.dumps(result)
        )
    
    def invalidate(self, tool_name: str, arguments: Optional[dict] = None):
        """캐시 무효화"""
        if arguments:
            key = self._make_key(tool_name, arguments)
            self.redis.delete(key)
        else:
            pattern = f"tool_cache:{tool_name}:*"
            for key in self.redis.scan_iter(pattern):
                self.redis.delete(key)

캐싱이 적용된 도구 래퍼
def cached_tool(tool_func, cache: ToolResponseCache):
    """도구에 캐싱 기능 추가"""
    async def wrapper(**kwargs):
        cached_result = cache.get(tool_func.name, kwargs)
        if cached_result:
            return cached_result
        
        result = await tool_func(**kwargs)
        cache.set(tool_func.name, kwargs, result)
        return result
    
    return wrapper

5. 비용 최적화 전략

저는 HolySheep AI를 통해 월 $2,000 이상의 API 비용을 약 40% 절감한 경험이 있습니다. 핵심은 모델 선택과 프롬프트 최적화의 균형입니다.

5.1 모델 라우팅 전략

# 지능형 모델 라우팅
from enum import Enum
from dataclasses import dataclass
from typing import Callable

class TaskComplexity(Enum):
    SIMPLE = "simple"
    MODERATE = "moderate"
    COMPLEX = "complex"

@dataclass
class ModelRoutingConfig:
    """모델 라우팅 설정"""
    complexity_threshold: int = 100  # 토큰 수 기준
    
    # HolySheep AI 모델별 가격
    MODEL_PRICES = {
        "gpt-4o": 7.50,        # $/MTok (프로모션 적용)
        "gpt-4o-mini": 0.60,
        "claude-sonnet-4-20250514": 4.50,  # HolySheep 특별가
        "gemini-2.5-flash": 2.50,
        "deepseek-chat": 0.42,
    }

class IntelligentRouter:
    """지능형 모델 라우터"""
    
    def __init__(self, config: ModelRoutingConfig = None):
        self.config = config or ModelRoutingConfig()
    
    def estimate_complexity(self, input_text: str, tool_count: int) -> TaskComplexity:
        """작업 복잡도 추정"""
        base_tokens = len(input_text.split()) * 1.3  # 대략적 토큰 수
        
        complexity_score = base_tokens + (tool_count * 50)
        
        if complexity_score < self.config.complexity_threshold:
            return TaskComplexity.SIMPLE
        elif complexity_score < self.config.complexity_threshold * 3:
            return TaskComplexity.MODERATE
        return TaskComplexity.COMPLEX
    
    def select_model(self, complexity: TaskComplexity, has_tools: bool) -> str:
        """최적 모델 선택"""
        
        if complexity == TaskComplexity.SIMPLE:
            # 고속·저비용 모델
            if has_tools:
                return "gemini-2.5-flash"  # 도구 호출 최적화
            return "deepseek-chat"  # 최저 비용
        
        elif complexity == TaskComplexity.MODERATE:
            # 균형 잡힌 모델
            return "gpt-4o-mini"  # 비용 효율적
        
        else:
            # 고품질 모델
            if has_tools:
                return "claude-sonnet-4-20250514"  # 도구 호출 정확도 최고
            return "gpt-4o"
    
    def calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        """비용 계산 (입력+출력 2:1 비율 가정)"""
        price = self.config.MODEL_PRICES.get(model, 7.50)
        total_tokens = input_tokens + (output_tokens * 2)
        return (total_tokens / 1_000_000) * price
    
    def route_and_execute(self, input_text: str, tool_count: int, has_tools: bool) -> dict:
        """라우팅 실행"""
        complexity = self.estimate_complexity(input_text, tool_count)
        selected_model = self.select_model(complexity, has_tools)
        
        estimated_tokens = int(len(input_text.split()) * 1.3) + 500
        estimated_cost = self.calculate_cost(selected_model, estimated_tokens, 200)
        
        return {
            "complexity": complexity.value,
            "selected_model": selected_model,
            "estimated_cost_usd": round(estimated_cost, 4),
            "savings_vs_baseline": round(
                self.calculate_cost("gpt-4o", estimated_tokens, 200) - estimated_cost, 4
            )
        }

사용 예시
router = IntelligentRouter()

test_cases = [
    ("오늘 날씨 알려줘", 1, False),
    ("데이터베이스에서 최근 7일 주문 내역을 검색하고 분석해줘", 2, True),
    ("최신 AI 연구 동향을 조사하고 보고서를 작성해줘", 5, True),
]

for text, tools, has_tools in test_cases:
    result = router.route_and_execute(text, tools, has_tools)
    print(f"입력: '{text[:20]}...'")
    print(f"  → 모델: {result['selected_model']}, 예상 비용: ${result['estimated_cost_usd']}, 절감: ${result['savings_vs_baseline']}")

6. 자주 발생하는 오류와 해결책

오류 1: Tool Call Argument 파싱 실패

# 오류 메시지: "Failed to parse arguments for tool X: missing required field Y"

원인: Pydantic 스키마와 실제 호출 인자가 불일치
해결: 도구 스키마 정의를 엄격하게 검증

from pydantic import ValidationError

def validate_tool_schema(tool_schema: dict) -> bool:
    """도구 스키마 검증"""
    required_fields = ["name", "description", "parameters"]
    
    for field in required_fields:
        if field not in tool_schema:
            raise ValueError(f"Missing required field: {field}")
    
    params = tool_schema.get("parameters", {})
    if params.get("type") != "object":
        raise ValueError("Parameters must be of type 'object'")
    
    if "required" not in params:
        # required 필드가 없으면 모든 속성은 선택사항으로 처리
        params["required"] = []
    
    return True

올바른 스키마 정의 예시
CORRECT_TOOL_SCHEMA = {
    "name": "calculate",
    "description": "수학적 계산을 수행합니다",
    "parameters": {
        "type": "object",
        "properties": {
            "expression": {
                "type": "string",
                "description": "계산할 수학 표현식"
            }
        },
        "required": ["expression"]  # 반드시 포함
    }
}

try:
    validate_tool_schema(CORRECT_TOOL_SCHEMA)
    print("✓ 스키마 검증 통과")
except ValueError as e:
    print(f"✗ 스키마 오류: {e}")

# 오류 메시지: "Rate limit exceeded for model X. Retry after Y seconds"

원인: 요청 빈도가 모델 제한을 초과
해결:了指 수 제한 및 지수 백오프 구현

import asyncio
from functools import wraps
import time

class RateLimitError(Exception):
    """Rate Limit 초과 예외"""
    def __init__(self, retry_after: float):
        self.retry_after = retry_after
        super().__init__(f"Rate limit exceeded. Retry after {retry_after} seconds")

async def with_exponential_backoff(
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0
):
    """지수 백오프 데코레이터"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            last_exception = None
            
            for attempt in range(max_retries):
                try:
                    return await func(*args, **kwargs)
                
                except RateLimitError as e:
                    last_exception = e
                    # HolySheep AI 권장 지수 백오프
                    delay = min(base_delay * (2 ** attempt), max_delay)
                    jitter = delay * 0.1 * (time.time() % 1)
                    
                    print(f"Rate limit hit. Waiting {delay + jitter:.2f}s (attempt {attempt + 1}/{max_retries})")
                    await asyncio.sleep(delay + jitter)
                
                except Exception as e:
                    # 다른 오류는 즉시 발생
                    raise
            
            raise last_exception  # 최대 재시도 횟수 초과
        
        return wrapper
    return decorator

사용 예시
@with_exponential_backoff(max_retries=5, base_delay=2.0)
async def call_holysheep_api(model: str, prompt: str):
    """HolySheep AI API 호출"""
    # Rate limit 감지 로직
    response = await api_request(model, prompt)
    
    if response.status == 429:
        raise RateLimitError(retry_after=response.headers.get("Retry-After", 60))
    
    return response

# 오류 메시지: "Tool output must be a string. Received: dict"

원인: StructuredTool이 반환값을 문자열로 기대
해결: 반환값을 명시적으로 문자열로 변환

from typing import Union
import json

def normalize_tool_output(output: Union[str, dict, list, object]) -> str:
    """도구 출력 정규화"""
    if isinstance(output, str):
        return output
    
    if isinstance(output, (dict, list)):
        return json.dumps(output, ensure_ascii=False, indent=2)
    
    # 기타 객체: __str__ 또는 __repr__ 사용
    return str(output)

도구 정의 시 적용
def safe_tool(func):
    """출력을 자동으로 정규화하는 도구 래퍼"""
    from langchain.tools import Tool
    
    def wrapper(input_str: str) -> str:
        try:
            result = func(input_str)
            return normalize_tool_output(result)
        except Exception as e:
            return f"오류 발생: {str(e)}"
    
    return Tool(
        name=func.__name__,
        func=wrapper,
        description=func.__doc__ or "도구 설명 없음"
    )

올바른 사용 예시
@safe_tool
def search_products(query: str) -> dict:
    """제품 검색 도구"""
    results = [{"id": 1, "name": "제품 A"}, {"id": 2, "name": "제품 B"}]
    return {"query": query, "results": results}

실행 결과는 자동으로 문자열로 변환됨
output = search_products("노트북")
print(type(output))  # 
print(output)  # '{"query": "노트북", "results": [...]}'

# 오류 메시지: "Maximum context length exceeded"

원인: 대화 기록 또는 도구 결과가 컨텍스트를 초과
해결: 메시지 압축 및 중요 정보 추출

from langchain.schema import SystemMessage, HumanMessage, AIMessage
from langchain.prompts import MessagesPlaceholder

def truncate_messages(messages: list, max_tokens: int = 3000) -> list:
    """메시지 목록 트렁케이션"""
    total_tokens = sum(len(m.content) // 4 for m in messages)  # 대략적 토큰 계산
    
    if total_tokens <= max_tokens:
        return messages
    
    # 시스템 메시지는 유지,古い 메시지부터 제거
    system_msg = None
    truncated = []
    
    for msg in messages:
        if isinstance(msg, SystemMessage):
            system_msg = msg
        else:
            truncated.append(msg)
    
    # 토큰 제한에 맞춰 제거
    current_tokens = sum(len(m.content) // 4 for m in truncated)
    result = [system_msg] if system_msg else []
    
    for msg in reversed(truncated):
        if current_tokens <= max_tokens:
            result.insert(1 if system_msg else 0, msg)
            break
        current_tokens -= len(msg.content) // 4
    else:
        # 가장最近の 메시지 유지
        result.extend(truncated[-2:])
    
    return result

def summarize_old_messages(messages: list, llm) -> list:
    """이전 대화 요약"""
    if len(messages) <= 4:
        return messages
    
    # 최근 4개 메시지 제외한 것들 요약
    old_messages = messages[:-4]
    summary_prompt = "다음 대화를 100단어 이내로 요약하세요:"
    
    for msg in old_messages:
        if isinstance(msg, (HumanMessage, AIMessage)):
            summary_prompt += f"\n{msg.type}: {msg.content[:200]}"
    
    summary = llm.invoke(summary_prompt)
    
    return [
        SystemMessage(content=f"이전 대화 요약: {summary.content}")
    ] + messages[-4:]

최대 컨텍스트를 활용한 Agent 설정
MAX_CONTEXT_TOKENS = 120000  # Claude 3.5 Sonnet 기준

prompt = hub.pull("hwchase17/openai-functions-agent")
prompt.messages = truncate_messages(prompt.messages, MAX_CONTEXT_TOKENS)