LangChain Agent 개발: ToolCalling과 Function Schema 완벽 가이드

AI 에이전트가 자율적으로 태스크를 수행하려면 외부 도구(Tools) 호출이 필수입니다. 이번 튜토리얼에서는 LangChain에서 ToolCalling을 구성하고 Function Schema를 설정하는 방법을 심층적으로 다룹니다. 특히 HolySheep AI를 활용한 고성능·저비용 에이전트 구축 전략을 실제 마이그레이션 사례와 함께 살펴보겠습니다.

실제 마이그레이션 사례: 서울의 AI 스타트업

서울 강남구에 위치한 AI 스타트업 'A社'는 고객 상담 자동화 에이전트를 개발 중이었습니다. 기존 에이전트는 GPT-4 API에 직접 연결되어 있었으며, 도구 호출 시 평균 420ms의 지연 시간과 월 $4,200의 높은 비용이 문제였습니다.

비즈니스 맥락

A社의 상담 에이전트는 일평균 50,000건의 사용자 질의에 응답해야 합니다. 주요 기능은:

• 주문 조회 및 배송 추적 API 연동
• 반품/환불 정책 기반 자동 답변
• 다국어 실시간 번역

기존架构에서는 각 도구 호출마다 전체 컨텍스트를 다시 전송해야 했고, 이는 비용과 지연의 주요 원인이었습니다.

HolySheep AI 선택 이유

A社가 HolySheep AI로 마이그레이션한 핵심 이유는:

• 단일 API 키로 다중 모델 통합: 도구 실행에는 DeepSeek V3.2($0.42/MTok), 응답 생성에는 Claude Sonnet 4.5($4.5/MTok) 혼용
• 420ms → 180ms 지연 감소: 최적화된 라우팅과 캐싱
• 월 비용 $4,200 → $680: 83.8% 비용 절감
• 국내 결제 지원: 해외 신용카드 없이 원화 결제 가능

마이그레이션 단계

1단계: base_url 교체

# 기존 코드 (OpenAI 직접 연결)
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4-turbo",
    api_key="sk-...",  # OpenAI API 키
    base_url="https://api.openai.com/v1"
)

마이그레이션 후 (HolySheep AI)
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4-turbo",
    api_key="YOUR_HOLYSHEEP_API_KEY",  # HolySheep API 키
    base_url="https://api.holysheep.ai/v1"  # HolySheep 엔드포인트
)

2단계: 카나리아 배포

import os
from langchain_core.tools import tool
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI

HolySheep AI 연결
llm = ChatOpenAI(
    model="gpt-4-turbo",
    api_key=os.environ.get("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1",
    timeout=30,
    max_retries=3
)

Traffic Splitting: 10% 카나리아 → HolySheep
def get_llm(variant="control"):
    if variant == "treatment":
        return ChatOpenAI(
            model="gpt-4-turbo",
            api_key=os.environ.get("HOLYSHEEP_API_KEY"),
            base_url="https://api.holysheep.ai/v1"
        )
    return ChatOpenAI(
        model="gpt-4-turbo",
        api_key=os.environ.get("OPENAI_API_KEY"),
        base_url="https://api.openai.com/v1"
    )

A/B 테스트 모니터링
import random
variant = "treatment" if random.random() < 0.1 else "control"
llm = get_llm(variant)

마이그레이션 후 30일 실측치

지표	마이그레이션 전	마이그레이션 후	개선율
평균 응답 지연	420ms	180ms	57% 감소
월간 API 비용	$4,200	$680	83.8% 절감
도구 호출 성공률	94.2%	99.1%	4.9% 향상
Token 소비량	850M tokens/월	720M tokens/월	15.3% 감소

ToolCalling과 Function Schema의 핵심 개념

ToolCalling이란?

ToolCalling은 LLM이 사용자의 질의에 응답하기 위해 외부 함수를 호출하는 메커니즘입니다. LangChain에서는 @tool 데코레이터를 사용하여 도구를 정의하고, Function Schema를 통해 LLM이 도구의 입력 형식을 이해하도록 합니다.

from langchain_core.tools import tool
from pydantic import BaseModel, Field
from typing import Optional

1. 도구 정의 with Pydantic Input Schema
@tool
class OrderSearchInput(BaseModel):
    order_id: str = Field(description="조회할 주문 ID (예: ORD-20240115-001)")
    include_items: bool = Field(default=True, description="주문 상품 포함 여부")

@tool
class ShippingTrackInput(BaseModel):
    tracking_number: str = Field(description="운송장 번호")
    carrier: Optional[str] = Field(default="auto", description="택배사 (auto/cj/lotte/post)")

@tool
def get_order_status(order_id: str, include_items: bool = True) -> dict:
    """
    주문 ID로 주문 상태 및 상세 정보를 조회합니다.
    
    Args:
        order_id: 조회할 주문 ID
        include_items: 주문 상품 목록 포함 여부
    
    Returns:
        주문 상태, 배송 정보, 상품 목록을 포함한 딕셔너리
    """
    # 실제 API 호출 로직
    return {
        "order_id": order_id,
        "status": "배송 중",
        "estimated_delivery": "2024-01-18",
        "items": [
            {"name": "스마트폰 케이스", "qty": 1, "price": 15000},
            {"name": "보호필름", "qty": 2, "price": 8000}
        ] if include_items else []
    }

@tool
def track_shipment(tracking_number: str, carrier: str = "auto") -> dict:
    """
    운송장 번호로 배송 추적을 수행합니다.
    
    Args:
        tracking_number: 운송장 번호
        carrier: 택배사 코드
    
    Returns:
        배송 진행 상황 리스트
    """
    # 실제 추적 API 연동
    return {
        "tracking": tracking_number,
        "carrier": carrier,
        "history": [
            {"status": "집화 완료", "location": "서울 강남 hub", "time": "2024-01-15 14:30"},
            {"status": "배송 중", "location": "서울 송파 depot", "time": "2024-01-16 09:15"},
            {"status": "배달 준비", "location": "부산 해운대 delivery", "time": "2024-01-17 08:00"}
        ]
    }

2. 도구 목록 생성
tools = [get_order_status, track_shipment]

LangChain Agent with Function Schema

Function Schema는 도구의 이름, 설명, 파라미터를 LLM이 이해할 수 있는 구조화된 형식으로 정의합니다. HolySheep AI를 사용할 때 올바른 Schema 설정이 중요합니다.

import os
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

HolySheep AI API 연결
llm = ChatOpenAI(
    model="gpt-4-turbo",
    api_key=os.environ.get("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1",
    temperature=0.3,  # 도구 호출은 낮은 temperature 권장
    streaming=True    # 스트리밍 응답 지원
)

시스템 프롬프트 설정
system_prompt = """당신은 한국어 고객 상담 에이전트입니다.

**Guidelines:**
1. 먼저 고객의 주문 ID 또는 운송장 번호를 확인하세요
2. order_id가 있으면 get_order_status를 먼저 호출하세요
3. tracking_number가 있으면 track_shipment를 호출하세요
4. 모든 응답은 한국어로 작성하세요
5. 민감한 정보(가격, 수량)는 숫자와 함께 단위를 명시하세요

**도구 사용 규칙:**
- 필수 파라미터 누락 시 고객에게 추가 정보를 요청하세요
- 여러 도구를 연속으로 호출할 수 있습니다
- 도구 호출 결과를 사용자에게 자연스럽게 번역하여 전달하세요"""

프롬프트 템플릿 구성
prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    MessagesPlaceholder(variable_name="chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")
])

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

AgentExecutor 생성
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5,
    early_stopping_method="force",
    handle_parsing_errors=True
)

실행 예제
def handle_customer_query(query: str, chat_history: list = None):
    """고객 질의 처리"""
    result = agent_executor.invoke({
        "input": query,
        "chat_history": chat_history or []
    })
    return result["output"]

테스트
if __name__ == "__main__":
    response = handle_customer_query(
        "안녕하세요, 주문번호 ORD-20240115-001 상태 확인해주세요."
    )
    print(response)

Function Schema 자동 생성 vs 수동 정의

자동 생성 (Recommended)

# Pydantic 모델 기반 자동 Schema 생성
from langchain_core.tools import tool
from pydantic import BaseModel, Field
from typing import List, Optional
import json

class Product(BaseModel):
    product_id: str
    name: str
    quantity: int
    price: int

class RefundRequest(BaseModel):
    order_id: str = Field(description="환불 요청할 주문 ID")
    reason: str = Field(description="환불 사유 ( defective/wrong_item/delivery_delay/customer_request )")
    items: Optional[List[str]] = Field(default=None, description="부분 환불 시 상품 ID 리스트")

@tool
def process_refund(request: RefundRequest) -> dict:
    """환불 요청을 처리합니다."""
    return {
        "refund_id": f"REF-{request.order_id}",
        "status": "approved",
        "estimated_days": 3
    }

생성된 Function Schema 확인
print("=== 자동 생성된 Function Schema ===")
for t in [process_refund]:
    schema = t.get_json_schema()
    print(f"Tool Name: {schema['name']}")
    print(f"Description: {schema.get('description', 'N/A')}")
    print(f"Parameters: {json.dumps(schema['parameters'], indent=2, ensure_ascii=False)}")
    print("---")

수동 정의 (고급)

from langchain_core.tools import StructuredTool

수동 Function Schema 정의
manual_schema = {
    "name": "calculate_shipping",
    "description": "상품 무게와 배송 지역을 기반으로 배송비를 계산합니다",
    "parameters": {
        "type": "object",
        "properties": {
            "weight_kg": {
                "type": "number",
                "description": "상품 총 중량 (kg)",
                "minimum": 0.1,
                "maximum": 30.0
            },
            "region": {
                "type": "string",
                "description": "배송 지역 코드",
                "enum": ["SEOUL", "GYEONGGI", "GANGWON", "CHUNGBUK", "CHUNGNAM", 
                        "JEONBUK", "JEONNAM", "GYEONGBUK", "GYEONGNAM", "JEJU"]
            },
            "express": {
                "type": "boolean",
                "description": "당일/익일 배송 여부",
                "default": False
            }
        },
        "required": ["weight_kg", "region"]
    }
}

def calculate_shipping_internal(weight_kg: float, region: str, express: bool = False) -> dict:
    """실제 배송비 계산 로직"""
    base_rate = {
        "SEOUL": 2500, "GYEONGGI": 3000, "GANGWON": 4000,
        "CHUNGBUK": 4000, "CHUNGNAM": 4500, "JEONBUK": 4500,
        "JEONNAM": 5000, "GYEONGBUK": 4500, "GYEONGNAM": 5000, "JEJU": 6000
    }
    
    rate = base_rate.get(region, 5000)
    if express:
        rate += 3000
    
    # 무게 추가 요금 (5kg 이상)
    if weight_kg > 5:
        rate += int((weight_kg - 5) * 500)
    
    return {
        "weight": weight_kg,
        "region": region,
        "express": express,
        "shipping_fee": rate,
        "currency": "KRW",
        "estimated_days": 1 if express else 3
    }

StructuredTool로 감싸기
shipping_tool = StructuredTool.from_function(
    func=calculate_shipping_internal,
    name=manual_schema["name"],
    description=manual_schema["description"],
    args_schema=manual_schema["parameters"]
)

HolySheep AI에서 다중 모델 활용

HolySheep AI의 단일 API 키로 여러 모델을 활용하면 비용과 성능을 최적화할 수 있습니다. 저는 도구 실행에는 경제적인 모델을, 최종 응답 생성에는 고성능 모델을 사용하는 구성 패턴을 권장합니다.

import os
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

HolySheep AI 다중 모델 설정
class MultiModelAgent:
    """도구 실행용/응답 생성용 분리 에이전트"""
    
    def __init__(self):
        # 도구 실행용: DeepSeek V3.2 (비용 효율적)
        self.tool_llm = ChatOpenAI(
            model="deepseek-chat",  # HolySheep에서 deepseek-chat 사용
            api_key=os.environ.get("HOLYSHEEP_API_KEY"),
            base_url="https://api.holysheep.ai/v1",
            temperature=0.1
        )
        
        # 응답 생성용: Claude Sonnet 4.5 (고품질)
        self.response_llm = ChatOpenAI(
            model="claude-3-5-sonnet-20241022",  # HolySheep에서 anthropic 모델명 사용
            api_key=os.environ.get("HOLYSHEEP_API_KEY"),
            base_url="https://api.holysheep.ai/v1",
            temperature=0.7
        )
    
    def process_with_tools(self, user_input: str, tools: list) -> str:
        """도구 호출을 포함한 처리"""
        # 1단계: 도구 호출 판단 (비용 효율적 모델)
        tool_agent = create_openai_functions_agent(
            self.tool_llm, 
            tools, 
            self._get_tool_prompt()
        )
        
        tool_executor = AgentExecutor(
            agent=tool_agent,
            tools=tools,
            verbose=False,
            max_iterations=3
        )
        
        # 2단계: 도구 실행
        tool_result = tool_executor.invoke({"input": user_input})
        
        # 3단계: 최종 응답 생성 (고품질 모델)
        response_prompt = f"""사용자 질문: {user_input}

도구 실행 결과:
{tool_result.get('output', '도구 결과 없음')}

위 정보를 바탕으로 자연스럽고 친절한 한국어 응답을 작성해주세요.
응답 형식:
- 핵심 정보를 먼저 전달
- 필요시 단계별 안내 제공
- 추가 도움이 필요하면 제안"""
        
        final_response = self.response_llm.invoke(response_prompt)
        return final_response.content
    
    def _get_tool_prompt(self) -> ChatPromptTemplate:
        return ChatPromptTemplate.from_messages([
            ("system", "도구만 사용하여 질문에 답하세요. 도구로 답할 수 없으면 '도구를 사용할 수 없습니다'라고 응답하세요."),
            ("human", "{input}")
        ])

사용 예시
if __name__ == "__main__":
    agent = MultiModelAgent()
    
    # 테스트 쿼리
    result = agent.process_with_tools(
        "ORD-20240115-001 주문의 배송 상황을 추적해주세요.",
        tools=[get_order_status, track_shipment]
    )
    print(result)

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

오류 1: ToolSchema 파싱 실패 - "malformed tool call"

# ❌ 오류 발생 코드
@tool
def search_products(query: str, category: str = None):
    """상품 검색 도구"""
    return [{"name": "sample", "price": 1000}]

문제: category가 Optional임에도 required로 인식됨
LLM이 category 없이 호출 시 파싱 오류 발생

✅ 해결 방법 1: 명시적 Optional 사용
from typing import Optional

@tool
def search_products(query: str, category: Optional[str] = None):
    """상품 검색 도구
    
    Args:
        query: 검색어 (필수)
        category: 카테고리 필터 (선택, 예: electronics/clothing/food)
    """
    return [{"name": "sample", "price": 1000}]

✅ 해결 방법 2: Pydantic BaseModel 사용 (권장)
from pydantic import BaseModel, Field
from typing import Optional

class SearchProductsInput(BaseModel):
    query: str = Field(description="검색어 (필수)")
    category: Optional[str] = Field(default=None, description="카테고리 필터")

@tool(args_schema=SearchProductsInput)
def search_products(query: str, category: Optional[str] = None):
    """상품 검색 도구"""
    return [{"name": "sample", "price": 1000}]

✅ 해결 방법 3: base_url 인증 오류 시 확인
HolySheep AI에서 제공하는 정확한 base_url 사용
llm = ChatOpenAI(
    model="gpt-4-turbo",
    api_key="YOUR_HOLYSHEEP_API_KEY",  # HolySheep에서 발급받은 키
    base_url="https://api.holysheep.ai/v1"  # 반드시 /v1 포함
)

오류 2: Rate Limit 초과 - "rate limit exceeded"

# ❌ 오류 발생: 동시 요청过多
async def process_batch_queries(queries: list):
    tasks = [agent_executor.ainvoke({"input": q}) for q in queries]
    results = await asyncio.gather(*tasks)  # 동시 100개 → Rate Limit

✅ 해결 방법 1: Rate Limiter 구현
import asyncio
from datetime import datetime, timedelta

class RateLimiter:
    """HolySheep AI Rate Limit 핸들러"""
    def __init__(self, max_calls: int = 100, window_seconds: int = 60):
        self.max_calls = max_calls
        self.window = timedelta(seconds=window_seconds)
        self.calls = []
    
    async def acquire(self):
        now = datetime.now()
        # 윈도우 밖 요청 제거
        self.calls = [t for t in self.calls if now - t < self.window]
        
        if len(self.calls) >= self.max_calls:
            sleep_time = (self.calls[0] + self.window - now).total_seconds()
            if sleep_time > 0:
                await asyncio.sleep(sleep_time)
        
        self.calls.append(now)

✅ 해결 방법 2: semaphore로 동시성 제어
import asyncio

async def process_batch_queries_safe(queries: list, max_concurrent: int = 10):
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def limited_process(q):
        async with semaphore:
            await rate_limiter.acquire()  # Rate Limit 체크
            return await agent_executor.ainvoke({"input": q})
    
    tasks = [limited_process(q) for q in queries]
    return await asyncio.gather(*tasks)

✅ 해결 방법 3: HolySheep AI 대시보드에서 Rate Limit 확인
https://www.holysheep.ai/dashboard 에서 현재 플랜의 제한 확인
필요시 플랜 업그레이드 또는 Rate Limit 조정 요청

오류 3: Context Window 초과 - "maximum context length exceeded"

# ❌ 오류 발생: 긴 대화 히스토리 누적
대화마다 전체 히스토리 전송 → 컨텍스트 초과

✅ 해결 방법 1: Summarizing Memory 사용
from langchain.memory import ConversationSummaryMemory
from langchain.agents import AgentExecutor, create_openai_functions_agent

memory = ConversationSummaryMemory(
    llm=ChatOpenAI(
        model="gpt-3.5-turbo",  # 요약용으로economical 모델
        api_key=os.environ.get("HOLYSHEEP_API_KEY"),
        base_url="https://api.holysheep.ai/v1"
    ),
    memory_key="chat_history",
    return_messages=True
)

agent = create_openai_functions_agent(tool_llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,  # 자동 요약
    verbose=True
)

✅ 해결 방법 2: 토큰Budget 기반 필터링
from langchain_core.messages import HumanMessage, AIMessage

def trim_messages(messages: list, max_tokens: int = 8000) -> list:
    """토큰Budget 기반으로 메시지 트리밍"""
    # 최근 메시지부터 유지
    trimmed = []
    total_tokens = 0
    
    for msg in reversed(messages):
        msg_tokens = estimate_tokens(msg)
        if total_tokens + msg_tokens <= max_tokens:
            trimmed.insert(0, msg)
            total_tokens += msg_tokens
        else:
            break
    
    return trimmed

✅ 해결 방법 3: HolySheep AI의 긴 컨텍스트 모델 활용
DeepSeek Chat은 긴 컨텍스트에 최적화되어 있음
long_context_llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.environ.get("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"
)
DeepSeek V3.2는 128K 컨텍스트 윈도우 지원

오류 4: Model Not Found - "model not found or not supported"

# ❌ 오류 발생: HolySheep AI에서 지원하지 않는 모델명 사용
llm = ChatOpenAI(
    model="gpt-4.1",  # 잘못된 모델명
    api_key=os.environ.get("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"
)

✅ 해결: HolySheep AI 지원 모델명 확인 및 사용
HolySheep AI에서 지원하는 모델 목록:
SUPPORTED_MODELS = {
    # OpenAI Compatible
    "gpt-4-turbo": "gpt-4-turbo",
    "gpt-4o": "gpt-4o",
    "gpt-4o-mini": "gpt-4o-mini",
    "gpt-3.5-turbo": "gpt-3.5-turbo",
    
    # Claude Compatible (Anthropic)
    "claude-3-5-sonnet-20241022": "claude-3-5-sonnet-20241022",
    "claude-3-opus-20240229": "claude-3-opus-20240229",
    "claude-3-haiku-20240307": "claude-3-haiku-20240307",
    
    # DeepSeek
    "deepseek-chat": "deepseek-chat",
    "deepseek-coder": "deepseek-coder",
    
    # Gemini
    "gemini-1.5-pro": "gemini-1.5-pro",
    "gemini-1.5-flash": "gemini-1.5-flash",
}

def get_holysheep_model(model_name: str) -> ChatOpenAI:
    """HolySheep AI 호환 모델명 반환"""
    normalized = model_name.lower().strip()
    
    if normalized not in SUPPORTED_MODELS:
        available = ", ".join(SUPPORTED_MODELS.keys())
        raise ValueError(f"지원하지 않는 모델: {model_name}. 사용 가능한 모델: {available}")
    
    return ChatOpenAI(
        model=SUPPORTED_MODELS[normalized],
        api_key=os.environ.get("HOLYSHEEP_API_KEY"),
        base_url="https://api.holysheep.ai/v1"
    )

사용
llm = get_holysheep_model("gpt-4-turbo")

성능 최적화 팁

저의 실제 프로젝트 경험에서 적용한 성능 최적화 전략은 다음과 같습니다:

• 도구 우선순위 설정:高频 도구를 앞에 배치하여 LLM의 잘못된 도구 선택 방지
• 도구 설명 최적화: 명확하고 구체적인 description으로 의도 파악 정확도 향상
• Streaming 활용: 긴 응답의 UX 개선과 perceived latency 감소
• 모델 분리: 도구 실행에는 DeepSeek, 최종 응답에는 Claude 혼용

결론

LangChain에서 ToolCalling과 Function Schema를 올바르게 구성하면 신뢰성 있고 확장 가능한 AI 에이전트를 구축할 수 있습니다. HolySheep AI를 활용하면:

• 다중 모델을 단일 API 키로 통합 관리
• 기존 OpenAI/Anthropic 코드와 호환되는 migration
• 최대 83%의 비용 절감과 57%의 지연 감소
• 국내 결제 지원으로 개발 편의성 향상

도구 호출의 핵심은 명확한 Schema 정의와 적절한 에러 핸들링입니다. 위에서 소개한 패턴들을 바탕으로 고성능 AI 에이전트를 구축해 보세요.

HolySheep AI의 다양한 모델定价과 기능은 공식 웹사이트에서 확인할 수 있습니다. 무료 크레딧 제공으니 먼저 테스트해 보시는 것을 권장합니다.

👉 HolySheep AI 가입하고 무료 크레딧 받기