이미지 로딩 중...
AI Generated
2025. 11. 8. · 2 Views
AI 에이전트 커스텀 Tool 완벽 가이드
AI 에이전트에 나만의 기능을 추가하는 커스텀 Tool 만들기를 배웁니다. 실제 프로젝트에서 사용할 수 있는 Tool 설계부터 구현, 최적화까지 단계별로 알아봅니다.
목차
- Tool 기본 개념
- BaseTool 클래스
- StructuredTool
- Tool 입력 스키마
- Tool 실행 로직
- Tool 설명 작성
- 비동기 Tool
- Tool 에러 핸들링
- Tool 체이닝
1. Tool 기본 개념
시작하며
여러분이 AI 에이전트를 만들 때 "이 에이전트가 우리 회사 데이터베이스를 조회할 수 있으면 좋겠는데..."라고 생각해본 적 있나요? 혹은 "외부 API를 호출해서 실시간 정보를 가져왔으면..."이라는 요구사항을 받아본 적이 있으신가요?
이런 문제는 실제 개발 현장에서 매우 자주 발생합니다. 기본 LLM은 학습된 데이터로만 답변할 수 있고, 실시간 정보나 외부 시스템과의 연동이 불가능합니다.
단순한 챗봇에서 실제 업무를 처리하는 에이전트로 발전시키려면 이 한계를 극복해야 합니다. 바로 이럴 때 필요한 것이 Tool입니다.
Tool은 AI 에이전트가 외부 세계와 상호작용할 수 있게 해주는 인터페이스로, 마치 사람의 손과 발처럼 실제 작업을 수행할 수 있게 만들어줍니다.
개요
간단히 말해서, Tool은 AI 에이전트가 호출할 수 있는 함수입니다. LLM이 판단하고 결정하면, Tool이 실제 행동을 실행합니다.
왜 이 개념이 필요한지 실무 관점에서 생각해보면, LLM은 텍스트 생성에는 뛰어나지만 실제 시스템과의 통신, 데이터 조회, 파일 처리 등은 직접 할 수 없습니다. 예를 들어, 고객 문의에 답변하는 AI 에이전트가 실제 주문 상태를 확인하려면 데이터베이스를 조회하는 Tool이 필요합니다.
기존에는 LLM 출력을 파싱해서 별도 로직으로 처리했다면, 이제는 에이전트가 자동으로 적절한 Tool을 선택하고 실행할 수 있습니다. Tool의 핵심 특징은 세 가지입니다: 명확한 입력 스키마로 안전성을 보장하고, 자연어 설명으로 LLM이 언제 사용할지 판단하게 하며, 결과를 구조화된 형태로 반환합니다.
이러한 특징들이 AI 에이전트를 단순한 대화형 봇에서 실제 업무를 처리하는 자동화 시스템으로 진화시킵니다.
코드 예제
from langchain.tools import Tool
# 간단한 계산기 Tool 정의
def calculator(expression: str) -> str:
"""수식을 계산하는 함수"""
try:
# eval 사용 시 보안 주의! 실무에서는 ast.literal_eval 사용
result = eval(expression)
return f"계산 결과: {result}"
except Exception as e:
return f"계산 오류: {str(e)}"
# Tool 객체 생성
calculator_tool = Tool(
name="Calculator", # Tool 이름
func=calculator, # 실행할 함수
description="수학 계산을 수행합니다. 입력: '2 + 3' 형식의 수식" # LLM이 읽을 설명
)
# 사용 예시
result = calculator_tool.run("10 * 5")
print(result) # 출력: 계산 결과: 50
설명
이것이 하는 일: Tool은 LLM이 판단한 작업을 실제로 실행하는 실행 계층입니다. LLM은 "뭘 해야 하는지" 결정하고, Tool은 "어떻게 하는지" 구현합니다.
첫 번째로, Tool 정의 부분을 보면 일반 Python 함수를 작성합니다. 여기서는 calculator 함수가 문자열로 수식을 받아서 계산 결과를 문자열로 반환합니다.
함수의 docstring은 나중에 Tool 설명으로 활용될 수 있습니다. 왜 이렇게 하는지 이유는, LLM과의 인터페이스는 기본적으로 텍스트 기반이기 때문에 입출력을 문자열로 처리하는 것이 가장 안전하기 때문입니다.
그 다음으로, Tool 객체를 생성할 때 세 가지 핵심 정보를 제공합니다. name은 Tool의 고유 식별자로 LLM이 이 Tool을 참조할 때 사용하고, func는 실제 실행될 함수를 연결하며, description은 LLM이 이 Tool을 언제 사용해야 하는지 판단하는 가이드가 됩니다.
내부에서는 LangChain이 이 정보를 구조화하여 LLM에게 전달하고, LLM은 사용자 요청을 분석해서 적절한 Tool을 선택합니다. 마지막으로, calculator_tool.run() 메서드를 호출하면 연결된 함수가 실행되고 결과가 반환됩니다.
실제 에이전트 환경에서는 LLM이 자동으로 이 메서드를 호출하여 결과를 받아 다음 추론에 활용합니다. 여러분이 이 코드를 사용하면 기존 Python 함수를 거의 수정 없이 AI 에이전트의 Tool로 변환할 수 있습니다.
데이터베이스 조회, API 호출, 파일 처리 등 어떤 작업이든 함수로 만들 수 있다면 Tool로 만들 수 있고, 복잡한 비즈니스 로직도 LLM의 자연어 이해 능력과 결합할 수 있습니다.
실전 팁
💡 Tool 이름은 명확하고 설명적으로 작성하세요. "tool1", "func" 같은 이름 대신 "SearchDatabase", "SendEmail" 처럼 기능을 명확히 표현해야 LLM이 올바르게 선택합니다.
💡 description은 "무엇을", "언제", "어떻게" 사용하는지 모두 포함하세요. "데이터베이스 검색" 보다는 "고객 ID로 주문 내역을 검색합니다. 입력: 고객ID(문자열), 출력: 주문 목록(JSON)"이 훨씬 효과적입니다.
💡 Tool 함수 내부에서는 반드시 예외 처리를 하세요. Tool이 예외를 발생시키면 전체 에이전트 실행이 중단될 수 있으므로, try-except로 감싸서 에러 메시지를 문자열로 반환하는 것이 안전합니다.
💡 eval() 사용은 보안 위험이 큽니다. 실무에서는 ast.literal_eval()을 사용하거나, 수식 파서 라이브러리를 활용하세요.
💡 Tool 실행 시간이 길다면 타임아웃을 설정하세요. 외부 API 호출이나 데이터베이스 쿼리는 예상보다 오래 걸릴 수 있으므로, timeout 파라미터나 asyncio.wait_for()로 제한을 두는 것이 좋습니다.
2. BaseTool 클래스
시작하며
여러분이 복잡한 Tool을 만들 때 "입력 검증은 어떻게 하지?", "여러 개의 파라미터를 받으려면?", "비동기로 실행하고 싶은데..." 같은 고민을 해본 적 있나요? 이런 문제는 Tool이 복잡해질수록 더 심각해집니다.
간단한 Tool 클래스는 단일 문자열 입력만 받을 수 있고, 타입 검증이나 복잡한 입력 구조를 다루기 어렵습니다. 실무에서는 대부분 여러 파라미터를 받고, 각각의 타입과 유효성을 검증해야 하는 상황이 많습니다.
바로 이럴 때 필요한 것이 BaseTool 클래스입니다. BaseTool을 상속받으면 Pydantic 기반의 강력한 입력 검증, 비동기 실행, 에러 핸들링 등 프로덕션 레벨의 Tool을 만들 수 있습니다.
개요
간단히 말해서, BaseTool은 커스텀 Tool을 만들기 위한 추상 기본 클래스입니다. 여러분의 Tool 클래스가 BaseTool을 상속받으면 모든 필수 기능이 자동으로 제공됩니다.
왜 이 개념이 필요한지 실무 관점에서 설명하면, 단순 함수 기반 Tool은 확장성이 떨어집니다. 예를 들어, 이메일 발송 Tool을 만든다면 수신자, 제목, 본문, 첨부파일 등 여러 파라미터가 필요하고, 각각의 형식도 검증해야 합니다.
함수 하나로는 이런 복잡성을 관리하기 어렵습니다. 기존에는 함수 내부에서 수동으로 파라미터를 파싱하고 검증했다면, 이제는 Pydantic 모델로 자동 검증하고 명확한 타입 힌트를 제공할 수 있습니다.
BaseTool의 핵심 특징은: 클래스 기반 구조로 상태를 유지할 수 있고, args_schema로 복잡한 입력 스키마를 정의하며, _run과 _arun 메서드로 동기/비동기를 모두 지원합니다. 이러한 특징들이 단순한 함수 Tool을 엔터프라이즈급 컴포넌트로 발전시킵니다.
코드 예제
from langchain.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Optional
# 입력 스키마 정의
class EmailInput(BaseModel):
to: str = Field(description="수신자 이메일 주소")
subject: str = Field(description="이메일 제목")
body: str = Field(description="이메일 본문")
cc: Optional[str] = Field(None, description="참조 이메일 주소 (선택)")
# BaseTool 상속받아 커스텀 Tool 생성
class EmailTool(BaseTool):
name = "send_email"
description = "이메일을 발송합니다. 업무 알림이나 고객 응대에 사용하세요."
args_schema = EmailInput # 입력 스키마 연결
def _run(self, to: str, subject: str, body: str, cc: Optional[str] = None) -> str:
"""동기 실행 메서드"""
# 실제로는 SMTP 서버 연결 등의 로직
result = f"이메일 발송 완료\n받는이: {to}\n제목: {subject}"
if cc:
result += f"\n참조: {cc}"
return result
async def _arun(self, to: str, subject: str, body: str, cc: Optional[str] = None) -> str:
"""비동기 실행 메서드 (선택)"""
# 비동기 이메일 발송 로직
return self._run(to, subject, body, cc)
# 사용
email_tool = EmailTool()
result = email_tool.run({"to": "user@example.com", "subject": "안녕하세요", "body": "테스트 메일입니다"})
설명
이것이 하는 일: BaseTool은 Tool의 모든 라이프사이클을 관리하는 추상 클래스입니다. 입력 검증부터 실행, 에러 처리까지 표준화된 방식을 제공합니다.
첫 번째로, EmailInput 클래스를 보면 Pydantic의 BaseModel을 상속받아 입력 스키마를 정의합니다. 각 필드는 타입과 설명을 가지며, Field()의 description은 LLM이 파라미터 용도를 이해하는 데 사용됩니다.
Optional로 선택적 파라미터도 표현할 수 있습니다. 왜 이렇게 하는지 이유는, Pydantic이 자동으로 타입 검증과 변환을 수행하여 잘못된 입력이 Tool 내부로 들어오는 것을 사전에 방지하기 때문입니다.
그 다음으로, EmailTool 클래스는 BaseTool을 상속받고 필수 속성들을 정의합니다. name과 description은 클래스 변수로 선언하고, args_schema에 앞서 정의한 EmailInput을 연결합니다.
_run 메서드는 실제 Tool 로직을 구현하는 곳으로, 메서드 시그니처가 EmailInput의 필드와 정확히 일치해야 합니다. LangChain은 내부적으로 스키마를 파싱하여 LLM에게 제공하고, LLM이 생성한 파라미터를 검증한 후 _run 메서드에 전달합니다.
마지막으로, Tool 인스턴스를 생성하고 run() 메서드를 호출하면 딕셔너리 형태의 입력이 자동으로 파싱되고 검증됩니다. 타입이 맞지 않거나 필수 필드가 누락되면 Pydantic이 자동으로 ValidationError를 발생시켜 잘못된 실행을 방지합니다.
여러분이 이 코드를 사용하면 타입 안정성이 보장된 Tool을 만들 수 있습니다. IDE의 자동완성과 타입 체크가 작동하여 개발 생산성이 높아지고, 런타임 에러가 줄어들며, Tool의 사용법이 스키마로 명확하게 문서화되어 유지보수가 쉬워집니다.
또한 _arun을 구현하면 비동기 환경에서도 동일한 Tool을 사용할 수 있어 성능 최적화가 가능합니다.
실전 팁
💡 _run 메서드의 파라미터 이름과 타입은 args_schema의 필드와 정확히 일치해야 합니다. 불일치하면 런타임 에러가 발생할 수 있습니다.
💡 args_schema를 정의하지 않으면 LangChain이 _run 메서드의 시그니처를 자동으로 파싱하지만, 명시적으로 정의하는 것이 더 안전하고 LLM에게 더 좋은 힌트를 제공합니다.
💡 _arun을 구현하지 않으면 기본적으로 _run을 동기적으로 호출합니다. 성능이 중요하지 않다면 _run만 구현해도 충분합니다.
💡 클래스 기반 Tool의 장점은 __init__에서 설정을 받을 수 있다는 점입니다. 예를 들어 API 키, 데이터베이스 연결 등을 생성 시점에 주입할 수 있습니다.
💡 return_direct=True 속성을 설정하면 Tool 결과를 중간 추론 없이 바로 사용자에게 반환합니다. 최종 답변 생성 Tool에 유용합니다.
3. StructuredTool
시작하며
여러분이 "BaseTool은 너무 장황한데, 더 간단하게 Tool을 만들 수 없을까?"라고 생각해본 적 있나요? 혹은 "이미 작성된 함수가 있는데, 이걸 Tool로 바꾸려면 전체를 다시 작성해야 하나?"라는 고민을 해본 적이 있으신가요?
이런 문제는 기존 코드베이스를 AI 에이전트와 통합할 때 자주 발생합니다. 이미 잘 작동하는 유틸리티 함수들이 있지만, 이를 BaseTool 클래스로 래핑하려면 많은 보일러플레이트 코드가 필요합니다.
특히 빠른 프로토타이핑이나 간단한 Tool이 필요할 때는 과도한 작업이 됩니다. 바로 이럴 때 필요한 것이 StructuredTool입니다.
StructuredTool.from_function()을 사용하면 기존 함수를 거의 수정 없이 즉시 Tool로 변환할 수 있습니다.
개요
간단히 말해서, StructuredTool은 일반 함수를 Tool로 변환해주는 팩토리 클래스입니다. 복잡한 클래스 정의 없이 데코레이터나 팩토리 메서드로 간편하게 Tool을 생성합니다.
왜 이 개념이 필요한지 실무 관점에서 보면, 모든 Tool이 클래스 기반일 필요는 없습니다. 예를 들어, 날씨 조회, 환율 계산, 텍스트 변환 같은 단순 기능은 함수만으로도 충분하고, 기존 유틸리티 함수를 재사용하고 싶을 때 클래스로 다시 작성하는 것은 비효율적입니다.
기존에는 BaseTool 클래스를 만들고 _run 메서드를 정의했다면, 이제는 함수 하나만 작성하고 from_function()으로 즉시 Tool화할 수 있습니다. StructuredTool의 핵심 특징은: 기존 함수를 그대로 사용할 수 있고, 타입 힌트에서 자동으로 스키마를 추출하며, 데코레이터 방식으로도 사용 가능합니다.
이러한 특징들이 개발 속도를 크게 향상시키고 코드 재사용성을 높입니다.
코드 예제
from langchain.tools import StructuredTool
from pydantic import BaseModel, Field
# 입력 스키마 정의
class WeatherInput(BaseModel):
city: str = Field(description="날씨를 조회할 도시 이름")
unit: str = Field(default="celsius", description="온도 단위 (celsius 또는 fahrenheit)")
# 일반 함수 작성
def get_weather(city: str, unit: str = "celsius") -> str:
"""특정 도시의 현재 날씨를 조회합니다."""
# 실제로는 API 호출
temp = 22 if unit == "celsius" else 72
return f"{city}의 현재 날씨: {temp}°{'C' if unit == 'celsius' else 'F'}, 맑음"
# StructuredTool로 변환
weather_tool = StructuredTool.from_function(
func=get_weather,
name="WeatherChecker",
description="도시 이름으로 현재 날씨를 조회합니다. 실시간 기상 정보를 제공합니다.",
args_schema=WeatherInput, # 스키마 명시
return_direct=False # 결과를 에이전트에게 반환
)
# 사용
result = weather_tool.run({"city": "서울", "unit": "celsius"})
print(result)
설명
이것이 하는 일: StructuredTool은 함수와 메타데이터를 받아서 내부적으로 BaseTool을 상속한 클래스를 동적으로 생성합니다. 개발자는 간단한 인터페이스만 사용하면 됩니다.
첫 번째로, WeatherInput 스키마와 get_weather 함수를 일반적인 Python 코드로 작성합니다. 함수의 파라미터는 스키마의 필드와 이름이 일치해야 하고, 타입 힌트를 명확히 작성하면 LangChain이 자동으로 파싱할 수 있습니다.
docstring은 Tool 설명의 일부로 활용될 수 있습니다. 왜 이렇게 하는지 이유는, 기존 코드 스타일을 유지하면서도 Tool로 변환할 수 있어 코드베이스의 일관성이 유지되기 때문입니다.
그 다음으로, StructuredTool.from_function() 메서드를 호출할 때 여러 설정을 제공합니다. func에는 실행할 함수를, name과 description은 Tool의 메타데이터를, args_schema는 입력 검증 스키마를 전달합니다.
from_function()은 내부적으로 이 정보들을 조합하여 완전한 BaseTool 객체를 생성하고 반환합니다. args_schema를 생략하면 함수의 타입 힌트에서 자동으로 추론하지만, 명시적으로 제공하는 것이 LLM에게 더 나은 컨텍스트를 제공합니다.
마지막으로, 생성된 weather_tool은 일반 Tool과 동일하게 run() 메서드로 실행됩니다. 딕셔너리 입력이 스키마에 따라 검증되고, 함수에 전달되며, 결과가 문자열로 반환됩니다.
return_direct 옵션으로 결과 처리 방식도 제어할 수 있습니다. 여러분이 이 코드를 사용하면 프로토타이핑 속도가 크게 빨라집니다.
10분 안에 여러 Tool을 만들어 테스트할 수 있고, 기존 유틸리티 라이브러리를 에이전트에 즉시 연결할 수 있으며, 나중에 필요하면 BaseTool 클래스로 리팩토링하기도 쉽습니다. 또한 함수 단위로 유닛 테스트를 작성할 수 있어 테스트도 간편합니다.
실전 팁
💡 args_schema를 생략하면 타입 힌트에서 자동 추론되지만, Field()의 description이 없어 LLM이 파라미터 용도를 정확히 파악하지 못할 수 있습니다. 가능하면 명시적으로 제공하세요.
💡 함수가 복잡한 객체를 반환하면 에이전트가 처리하기 어려울 수 있습니다. 항상 문자열이나 JSON 형태로 직렬화 가능한 결과를 반환하세요.
💡 비동기 함수도 coroutine_func 파라미터로 전달할 수 있습니다. async def로 작성된 함수는 coroutine_func에, 일반 함수는 func에 전달하세요.
💡 handle_tool_error=True 옵션을 설정하면 Tool 실행 중 예외가 발생해도 에이전트가 계속 실행됩니다. 프로덕션 환경에서 권장됩니다.
💡 여러 함수를 빠르게 Tool로 변환할 때는 리스트 컴프리헨션을 활용하세요: tools = [StructuredTool.from_function(func=f, ...) for f in my_functions]
4. Tool 입력 스키마
시작하며
여러분이 Tool을 만들 때 "LLM이 잘못된 파라미터를 전달하면 어떻게 하지?", "날짜 형식이나 이메일 주소 같은 특정 포맷을 강제할 수 없을까?"라는 고민을 해본 적 있나요? 이런 문제는 Tool이 외부 API나 데이터베이스와 통신할 때 치명적입니다.
잘못된 형식의 입력이 그대로 전달되면 시스템 오류, 데이터 손상, 심지어 보안 취약점까지 발생할 수 있습니다. 단순 문자열 검증으로는 복잡한 비즈니스 규칙을 표현하기 어렵고, 에러 메시지도 불명확합니다.
바로 이럴 때 필요한 것이 Pydantic 기반의 입력 스키마입니다. Pydantic의 강력한 검증 기능을 활용하면 타입 검사, 포맷 검증, 범위 제한, 커스텀 밸리데이터까지 모두 구현할 수 있습니다.
개요
간단히 말해서, 입력 스키마는 Tool이 받을 수 있는 파라미터의 "계약서"입니다. Pydantic BaseModel을 상속받아 각 필드의 타입, 제약조건, 설명을 정의합니다.
왜 이 개념이 필요한지 실무 관점에서 보면, AI는 완벽하지 않습니다. LLM이 생성한 파라미터가 항상 올바른 형식이라고 가정할 수 없습니다.
예를 들어, 결제 Tool에 음수 금액이 전달되거나, 이메일 Tool에 잘못된 주소 형식이 들어오면 실제 시스템에 문제가 발생합니다. 기존에는 함수 내부에서 if문으로 일일이 검증했다면, 이제는 스키마 선언만으로 자동 검증되고 명확한 에러 메시지를 받을 수 있습니다.
입력 스키마의 핵심 특징은: Pydantic의 모든 검증 기능을 사용할 수 있고, Field()로 메타데이터와 제약조건을 추가하며, LLM에게 파라미터 사용법을 명확히 전달합니다. 이러한 특징들이 Tool의 신뢰성과 안정성을 크게 향상시킵니다.
코드 예제
from pydantic import BaseModel, Field, EmailStr, validator, conint, constr
from typing import Optional
from datetime import datetime
class PaymentInput(BaseModel):
"""결제 처리 Tool의 입력 스키마"""
# 금액: 양수만 허용, 범위 제한
amount: conint(gt=0, le=1000000) = Field(
description="결제 금액 (1원 ~ 1,000,000원)"
)
# 이메일: 자동 포맷 검증
customer_email: EmailStr = Field(
description="고객 이메일 주소"
)
# 카드 번호: 정규식 검증
card_number: constr(regex=r'^\d{4}-\d{4}-\d{4}-\d{4}$') = Field(
description="카드 번호 (0000-0000-0000-0000 형식)"
)
# 선택적 파라미터
note: Optional[str] = Field(
None,
max_length=200,
description="결제 메모 (최대 200자)"
)
# 커스텀 검증: 과거 날짜 확인
@validator('card_number')
def validate_card(cls, v):
"""카드 번호 유효성 추가 검증"""
# 실제로는 Luhn 알고리즘 등 사용
if v.startswith('0000'):
raise ValueError('유효하지 않은 카드 번호입니다')
return v
# Tool 생성 시 스키마 적용
from langchain.tools import StructuredTool
def process_payment(amount: int, customer_email: str, card_number: str, note: Optional[str] = None) -> str:
return f"결제 완료: {amount}원, {customer_email}"
payment_tool = StructuredTool.from_function(
func=process_payment,
name="ProcessPayment",
description="고객 결제를 처리합니다",
args_schema=PaymentInput
)
설명
이것이 하는 일: 입력 스키마는 Tool 실행 전에 파라미터를 검증하는 게이트키퍼 역할을 합니다. Pydantic이 자동으로 타입 변환과 검증을 수행하여 잘못된 데이터가 Tool 로직에 도달하지 못하게 막습니다.
첫 번째로, PaymentInput 클래스의 각 필드를 보면 다양한 검증 방법을 사용합니다. conint(gt=0, le=1000000)는 정수 범위를 제한하고, EmailStr은 이메일 형식을 자동 검증하며, constr(regex=...)는 정규식 패턴 매칭을 수행합니다.
Field()의 description은 LLM에게 전달되어 올바른 값을 생성하도록 가이드합니다. 왜 이렇게 하는지 이유는, 선언적 방식으로 검증 규칙을 표현하면 코드가 읽기 쉽고, 재사용 가능하며, 자동으로 문서화되기 때문입니다.
그 다음으로, @validator 데코레이터로 커스텀 검증 로직을 추가할 수 있습니다. validate_card 메서드는 카드 번호의 추가 규칙을 검사하고, 조건을 만족하지 않으면 ValueError를 발생시킵니다.
Pydantic은 이 예외를 캐치하여 사용자 친화적인 ValidationError로 변환하고, 어떤 필드에서 어떤 문제가 발생했는지 명확히 알려줍니다. 내부적으로 모든 validator는 필드 값이 설정되기 전에 실행되어 잘못된 데이터의 유입을 원천 차단합니다.
마지막으로, 이 스키마를 args_schema에 연결하면 Tool 실행 시 자동으로 검증이 수행됩니다. LLM이 생성한 파라미터가 딕셔너리 형태로 전달되면, Pydantic이 이를 PaymentInput 객체로 변환하며 모든 제약조건을 확인합니다.
검증 실패 시 구체적인 에러 메시지와 함께 실행이 중단되어 시스템을 보호합니다. 여러분이 이 코드를 사용하면 방어적 프로그래밍이 자동화됩니다.
Tool 함수 내부에서는 입력이 이미 검증되었다고 신뢰할 수 있어 비즈니스 로직에만 집중할 수 있고, 에러가 발생하면 정확한 원인을 파악할 수 있으며, 스키마 자체가 API 문서 역할을 하여 유지보수가 쉬워집니다. 또한 LLM에게 명확한 가이드를 제공하여 잘못된 파라미터 생성 확률을 줄입니다.
실전 팁
💡 Field()의 description은 가능한 구체적으로 작성하세요. "이메일"보다는 "고객의 주문 확인 이메일을 받을 주소"처럼 맥락을 포함하면 LLM이 더 정확한 값을 생성합니다.
💡 conint, constr 같은 제약 타입은 런타임 검증뿐 아니라 IDE 타입 힌트에도 도움이 됩니다. 적극 활용하세요.
💡 복잡한 검증 로직은 @validator보다 @root_validator를 사용하세요. 여러 필드 간의 관계를 검증할 때 유용합니다 (예: 시작일이 종료일보다 빠른지 확인).
💡 환경변수나 설정 파일의 값은 Field(default_factory=...)로 주입할 수 있습니다. 예: api_key: str = Field(default_factory=lambda: os.getenv("API_KEY"))
💡 Optional을 남발하면 검증이 약해집니다. 정말 선택적인 필드만 Optional로 표시하고, 필수 필드는 명확히 요구하세요.
5. Tool 실행 로직
시작하며
여러분이 Tool 스키마를 완벽하게 정의했다면, 이제 "실제로 무슨 일을 할지" 구현할 차례입니다. "API를 어떻게 호출하지?", "데이터베이스 연결은 어디서 관리하지?", "에러가 발생하면 어떻게 알려주지?"라는 질문들이 생깁니다.
이런 고민은 Tool의 핵심인 실행 로직을 작성할 때 반드시 고려해야 합니다. 스키마는 입력만 검증할 뿐, 실제 비즈니스 로직은 _run 메서드에서 구현됩니다.
여기서 외부 시스템 통신, 데이터 변환, 에러 처리 등 모든 실제 작업이 일어납니다. 바로 이럴 때 필요한 것이 체계적인 _run 메서드 구현 패턴입니다.
명확한 구조로 작성하면 디버깅이 쉽고, 에러 핸들링이 견고하며, 유지보수가 편리한 Tool을 만들 수 있습니다.
개요
간단히 말해서, _run 메서드는 Tool의 "두뇌"입니다. 검증된 입력을 받아 실제 작업을 수행하고, 결과를 에이전트가 이해할 수 있는 형태로 반환합니다.
왜 이 개념이 필요한지 실무 관점에서 보면, 스키마만으로는 아무 일도 일어나지 않습니다. 예를 들어, 데이터베이스 조회 Tool이라면 _run에서 실제 쿼리를 실행하고, API 호출 Tool이라면 HTTP 요청을 보내고, 파일 처리 Tool이라면 파일 I/O를 수행해야 합니다.
기존에는 모든 로직을 한 함수에 섞어서 작성했다면, 이제는 입력 검증은 스키마가, 실행은 _run이 담당하여 관심사가 분리됩니다. _run 메서드의 핵심 특징은: 검증된 입력만 받아 안전하게 실행하고, 외부 리소스 관리와 에러 처리를 책임지며, 결과를 문자열로 직렬화하여 반환합니다.
이러한 특징들이 Tool을 안정적이고 예측 가능한 컴포넌트로 만듭니다.
코드 예제
from langchain.tools import BaseTool
from pydantic import BaseModel, Field
import requests
from typing import Optional
class WeatherAPIInput(BaseModel):
city: str = Field(description="도시 이름")
units: str = Field(default="metric", description="온도 단위")
class WeatherAPITool(BaseTool):
name = "weather_api"
description = "실시간 날씨 정보를 조회합니다"
args_schema = WeatherAPIInput
# 생성자에서 설정 주입
api_key: str
base_url: str = "https://api.weather.example.com"
def _run(self, city: str, units: str = "metric") -> str:
"""실제 실행 로직"""
try:
# 1. 요청 준비
params = {
"q": city,
"units": units,
"appid": self.api_key
}
# 2. 외부 API 호출
response = requests.get(
f"{self.base_url}/weather",
params=params,
timeout=10 # 타임아웃 설정
)
# 3. 응답 검증
response.raise_for_status()
data = response.json()
# 4. 결과 포맷팅
temp = data["main"]["temp"]
desc = data["weather"][0]["description"]
result = f"{city} 날씨: {temp}°{'C' if units=='metric' else 'F'}, {desc}"
return result
except requests.Timeout:
return f"오류: {city} 날씨 조회 시간 초과"
except requests.HTTPError as e:
return f"오류: API 요청 실패 ({e.response.status_code})"
except KeyError:
return "오류: API 응답 형식이 올바르지 않습니다"
except Exception as e:
return f"알 수 없는 오류: {str(e)}"
# 사용
tool = WeatherAPITool(api_key="your-api-key")
result = tool.run({"city": "서울"})
설명
이것이 하는 일: _run 메서드는 검증된 파라미터를 받아 비즈니스 로직을 실행하고, 성공/실패 여부와 상관없이 항상 문자열 결과를 반환해야 합니다. 첫 번째로, 메서드 시그니처를 보면 스키마의 필드와 정확히 일치합니다.
city와 units를 파라미터로 받고, 기본값도 스키마와 동일하게 설정합니다. 클래스 속성으로 api_key와 base_url을 정의하여 생성 시점에 설정을 주입받습니다.
왜 이렇게 하는지 이유는, 환경에 따라 달라지는 설정(API 키, URL 등)을 하드코딩하지 않고 런타임에 제공할 수 있기 때문입니다. 이를 통해 개발/스테이징/프로덕션 환경에서 동일한 Tool 클래스를 다른 설정으로 사용할 수 있습니다.
그 다음으로, try 블록 내부에서 실제 작업을 4단계로 나눕니다. 첫째, API 요청 파라미터를 준비하고, 둘째, requests로 HTTP 호출을 수행하며, 셋째, 응답 상태와 데이터를 검증하고, 넷째, 에이전트가 이해할 수 있는 자연어 형식으로 결과를 포맷팅합니다.
timeout 파라미터로 무한 대기를 방지하고, raise_for_status()로 HTTP 에러를 예외로 변환합니다. 내부적으로 각 단계가 실패하면 해당하는 except 블록으로 점프하여 적절한 에러 메시지를 반환합니다.
마지막으로, except 블록들은 예상 가능한 모든 에러를 처리합니다. Timeout은 네트워크 지연, HTTPError는 API 오류, KeyError는 응답 형식 변경을 의미하며, 각각에 맞는 사용자 친화적인 메시지를 반환합니다.
마지막 Exception은 예상치 못한 모든 에러를 캐치하여 Tool이 예외를 발생시켜 에이전트 전체를 중단시키는 것을 방지합니다. 여러분이 이 코드를 사용하면 프로덕션급 Tool을 작성할 수 있습니다.
모든 에러 케이스가 처리되어 에이전트가 중단 없이 계속 실행되고, 명확한 에러 메시지로 디버깅이 쉬우며, 설정 주입 패턴으로 테스트와 배포가 편리해집니다. 또한 단계별로 나뉜 로직은 나중에 수정하거나 확장하기도 쉽습니다.
실전 팁
💡 _run은 반드시 문자열을 반환해야 합니다. 딕셔너리나 객체를 반환하고 싶다면 json.dumps()로 직렬화하세요.
💡 외부 API 호출 시 항상 timeout을 설정하세요. 기본값은 무한 대기이므로 에이전트가 멈출 수 있습니다.
💡 에러 메시지는 LLM이 읽고 다음 행동을 결정하는 데 사용됩니다. "에러 발생"보다는 "API 키가 만료되었습니다. 관리자에게 문의하세요" 같이 구체적으로 작성하세요.
💡 민감한 정보(API 키, 비밀번호)는 로그에 출력하지 마세요. 에러 메시지에도 포함하지 않도록 주의하세요.
💡 무거운 초기화 작업(DB 연결, 파일 로드)은 __init__에서 한 번만 수행하고, _run에서는 재사용하세요. 매번 재연결하면 성능이 떨어집니다.
6. Tool 설명 작성
시작하며
여러분이 완벽한 Tool을 만들었는데 "LLM이 이 Tool을 전혀 사용하지 않네?", "엉뚱한 상황에서 이 Tool을 호출하는데?"라는 경험을 해본 적 있나요? 이런 문제는 Tool의 기능은 완벽해도 설명이 부족할 때 발생합니다.
LLM은 코드를 읽을 수 없고, 오직 여러분이 제공한 텍스트 설명만으로 Tool의 용도를 판단합니다. 설명이 모호하거나 불완전하면 LLM은 잘못된 판단을 내리게 됩니다.
바로 이럴 때 필요한 것이 효과적인 Tool 설명 작성 기법입니다. 명확하고 구체적인 설명은 LLM이 올바른 상황에서 올바른 Tool을 선택하게 만듭니다.
개요
간단히 말해서, Tool 설명은 LLM에게 "이 Tool을 언제, 어떻게 사용해야 하는지" 알려주는 사용 설명서입니다. description 필드와 args_schema의 Field 설명이 핵심입니다.
왜 이 개념이 필요한지 실무 관점에서 보면, 에이전트의 성공은 Tool 선택의 정확도에 달려 있습니다. 예를 들어, "데이터 조회" Tool과 "데이터 수정" Tool이 있을 때, 사용자가 "재고를 확인해줘"라고 하면 조회 Tool을 사용해야 합니다.
하지만 설명이 불명확하면 LLM이 수정 Tool을 선택할 수도 있습니다. 기존에는 간단히 "데이터베이스 Tool"이라고만 작성했다면, 이제는 언제, 무엇을, 어떻게 하는지 모두 포함하여 LLM의 이해를 돕습니다.
Tool 설명의 핵심 특징은: 목적과 사용 시점을 명확히 하고, 입력/출력 형식을 구체적으로 설명하며, 제약사항과 주의사항을 포함합니다. 이러한 특징들이 LLM의 Tool 선택 정확도를 크게 향상시킵니다.
코드 예제
from langchain.tools import BaseTool
from pydantic import BaseModel, Field
class DatabaseQueryInput(BaseModel):
table: str = Field(
description="조회할 테이블 이름. 사용 가능한 테이블: users, orders, products"
)
query: str = Field(
description="SQL WHERE 절 (예: 'age > 20', 'status = active'). SELECT나 UPDATE문은 불가능합니다."
)
limit: int = Field(
default=10,
description="반환할 최대 레코드 수 (1-100). 기본값 10"
)
class DatabaseQueryTool(BaseTool):
name = "database_query"
# ✅ 좋은 예시: 명확하고 구체적인 설명
description = (
"데이터베이스에서 정보를 조회할 때 사용합니다. "
"이 Tool은 읽기 전용이며, 데이터를 수정하거나 삭제할 수 없습니다. "
"사용자가 '확인', '조회', '찾아줘' 같은 요청을 하면 이 Tool을 사용하세요. "
"결과는 JSON 배열 형태로 반환됩니다. "
"예시 사용 시점: '최근 주문 10건 보여줘', '활성 사용자 수는?'"
)
args_schema = DatabaseQueryInput
def _run(self, table: str, query: str, limit: int = 10) -> str:
# 실제 DB 조회 로직
return f"조회 완료: {table} 테이블에서 {limit}건 조회"
# ❌ 나쁜 예시: 모호한 설명
class BadTool(BaseTool):
name = "db_tool"
description = "데이터베이스 작업" # 너무 짧고 모호함
def _run(self, input: str) -> str:
return "완료"
# 사용 예시
good_tool = DatabaseQueryTool()
# LLM이 이 설명을 읽고 적절한 상황에서 선택
설명
이것이 하는 일: Tool 설명은 LLM의 프롬프트에 포함되어 "사용 가능한 도구 목록"으로 제시됩니다. LLM은 사용자 요청과 각 Tool의 설명을 비교하여 가장 적합한 Tool을 선택합니다.
첫 번째로, description 필드를 작성할 때는 여러 요소를 포함해야 합니다. "무엇을 하는가" (데이터베이스 조회), "언제 사용하는가" (사용자가 '확인', '조회' 요청 시), "제약사항은 무엇인가" (읽기 전용), "출력 형식은" (JSON 배열) 등을 모두 명시합니다.
왜 이렇게 하는지 이유는, LLM이 맥락을 고려하여 정확한 판단을 내리려면 충분한 정보가 필요하기 때문입니다. "데이터베이스 Tool"이라는 설명만으로는 읽기인지 쓰기인지, 어떤 상황에서 사용하는지 알 수 없습니다.
그 다음으로, args_schema의 각 Field 설명도 매우 중요합니다. table 필드의 설명에서는 사용 가능한 테이블 목록을 제시하고, query 필드에서는 형식 예시와 제약사항을 명시하며, limit 필드에서는 범위와 기본값을 설명합니다.
내부적으로 LangChain은 이 정보들을 구조화하여 LLM에게 전달하고, LLM은 이를 참고하여 올바른 형식의 파라미터를 생성합니다. "조회할 테이블"보다 "조회할 테이블 이름.
사용 가능: users, orders, products"가 훨씬 더 구체적인 가이드를 제공합니다. 마지막으로, 좋은 설명과 나쁜 설명을 비교해보면 차이가 명확합니다.
BadTool의 "데이터베이스 작업"은 너무 모호하여 LLM이 언제 사용해야 할지, 무엇을 할 수 있는지 전혀 알 수 없습니다. 반면 DatabaseQueryTool의 설명은 읽기 전용임을 명시하고, 사용 시점의 예시까지 제공하여 LLM의 올바른 선택을 유도합니다.
여러분이 이 코드를 사용하면 에이전트의 Tool 선택 정확도가 크게 향상됩니다. LLM이 잘못된 Tool을 선택하는 빈도가 줄어들고, 사용자 의도를 더 정확히 파악하여 실행하며, 디버깅 시 어떤 Tool이 왜 선택되었는지 이해하기 쉬워집니다.
또한 새로운 팀원이 코드를 볼 때도 각 Tool의 역할을 빠르게 파악할 수 있습니다.
실전 팁
💡 description에 "언제 사용하는가"를 꼭 포함하세요. "사용자가 ~를 요청하면", "~가 필요할 때" 같은 표현이 LLM의 판단에 큰 도움이 됩니다.
💡 유사한 기능의 Tool이 여러 개 있다면, 차이점을 명확히 설명하세요. "이 Tool은 읽기 전용입니다. 쓰기는 update_database Tool을 사용하세요" 같은 식으로 구분합니다.
💡 Field 설명에 예시를 포함하세요. "날짜 형식"보다 "날짜 (예: 2024-01-15)"가 LLM에게 더 명확한 가이드를 제공합니다.
💡 너무 긴 설명은 오히려 혼란을 줄 수 있습니다. 핵심만 간결하게, 하지만 충분한 정보를 제공하는 균형을 찾으세요. 3-5문장이 적당합니다.
💡 Tool 설명을 수정한 후에는 실제 에이전트 실행을 테스트하세요. 때로는 의도와 다르게 LLM이 해석할 수 있으므로, A/B 테스트로 어떤 설명이 더 효과적인지 확인하세요.
7. 비동기 Tool
시작하며
여러분이 여러 Tool을 동시에 실행하거나, I/O 대기 시간이 긴 작업을 처리할 때 "Tool이 하나씩만 실행되어 너무 느린데?", "API 응답을 기다리는 동안 다른 작업도 할 수 있지 않을까?"라는 생각을 해본 적 있나요? 이런 문제는 동기 방식으로 Tool을 실행할 때 필연적으로 발생합니다.
하나의 Tool이 외부 API를 호출하고 응답을 기다리는 동안 전체 에이전트가 멈춰 있습니다. 여러 Tool을 순차적으로 실행하면 대기 시간이 누적되어 사용자 경험이 나빠집니다.
바로 이럴 때 필요한 것이 비동기 Tool입니다. _arun 메서드를 구현하면 여러 Tool이 동시에 실행되고, I/O 대기 중에도 다른 작업을 처리할 수 있어 전체 응답 속도가 크게 빨라집니다.
개요
간단히 말해서, 비동기 Tool은 async/await 패턴으로 실행되는 Tool입니다. _run 대신 _arun 메서드를 구현하거나 두 가지를 모두 제공할 수 있습니다.
왜 이 개념이 필요한지 실무 관점에서 보면, 현대 웹 애플리케이션은 대부분 비동기 방식으로 작동합니다. 예를 들어, FastAPI나 asyncio 기반 에이전트에서는 동기 Tool이 전체 이벤트 루프를 블로킹하여 성능 저하를 일으킵니다.
또한 여러 외부 API를 동시에 호출해야 할 때 비동기가 필수입니다. 기존에는 각 Tool이 순차적으로 실행되어 총 시간이 모든 Tool 실행 시간의 합이었다면, 이제는 병렬 실행으로 가장 느린 Tool의 시간만큼만 소요됩니다.
비동기 Tool의 핵심 특징은: asyncio와 호환되어 이벤트 루프를 블로킹하지 않고, 여러 Tool을 동시 실행할 수 있으며, 비동기 라이브러리(aiohttp, asyncpg 등)를 활용할 수 있습니다. 이러한 특징들이 에이전트의 성능을 수배~수십배 향상시킵니다.
코드 예제
from langchain.tools import BaseTool
from pydantic import BaseModel, Field
import aiohttp
import asyncio
from typing import Optional
class AsyncWeatherInput(BaseModel):
city: str = Field(description="도시 이름")
class AsyncWeatherTool(BaseTool):
name = "async_weather"
description = "비동기로 날씨를 조회합니다"
args_schema = AsyncWeatherInput
api_key: str
# 동기 버전 (fallback)
def _run(self, city: str) -> str:
"""동기 환경에서 사용"""
return f"{city} 날씨 조회 (동기 모드)"
# 비동기 버전 (주요 구현)
async def _arun(self, city: str) -> str:
"""비동기 환경에서 사용"""
try:
# aiohttp로 비동기 HTTP 요청
async with aiohttp.ClientSession() as session:
async with session.get(
"https://api.weather.example.com/weather",
params={"q": city, "appid": self.api_key},
timeout=aiohttp.ClientTimeout(total=10)
) as response:
data = await response.json()
temp = data["main"]["temp"]
return f"{city}: {temp}°C"
except asyncio.TimeoutError:
return f"{city} 날씨 조회 시간 초과"
except Exception as e:
return f"오류: {str(e)}"
# 여러 Tool 동시 실행 예시
async def run_multiple_tools():
tool = AsyncWeatherTool(api_key="key")
# 여러 도시 날씨를 동시에 조회
tasks = [
tool._arun("서울"),
tool._arun("부산"),
tool._arun("대구")
]
# 병렬 실행 (총 시간 ≈ 가장 느린 하나의 시간)
results = await asyncio.gather(*tasks)
return results
설명
이것이 하는 일: 비동기 Tool은 asyncio 이벤트 루프에서 실행되며, I/O 작업 중에는 제어권을 반환하여 다른 코루틴이 실행될 수 있게 합니다. 이를 통해 동시성을 극대화합니다.
첫 번째로, AsyncWeatherTool 클래스를 보면 _run과 _arun 두 메서드를 모두 구현합니다. _run은 동기 환경에서의 폴백으로, 간단한 구현만 제공하고, _arun이 실제 비동기 로직을 담당합니다.
async def로 선언된 _arun 내부에서는 await 키워드로 비동기 작업을 기다립니다. 왜 이렇게 하는지 이유는, LangChain 에이전트는 환경에 따라 자동으로 _run 또는 _arun을 선택하므로, 양쪽을 모두 제공하면 어떤 환경에서도 작동하기 때문입니다.
그 다음으로, aiohttp를 사용한 비동기 HTTP 요청을 보면, async with로 세션과 응답 객체를 관리합니다. session.get()은 비동기 함수이므로 await을 사용하고, response.json()도 비동기이므로 await이 필요합니다.
timeout 파라미터는 asyncio.TimeoutError를 발생시켜 무한 대기를 방지합니다. 내부적으로 aiohttp는 소켓 I/O를 비동기로 처리하여, 응답을 기다리는 동안 이벤트 루프가 다른 태스크를 실행할 수 있게 합니다.
마지막으로, run_multiple_tools 함수는 여러 Tool을 동시 실행하는 패턴을 보여줍니다. _arun을 여러 번 호출하여 코루틴 객체 리스트를 만들고, asyncio.gather()로 모두 동시에 실행합니다.
만약 각 도시 조회에 3초씩 걸린다면, 동기 방식은 9초가 소요되지만 비동기 방식은 약 3초만 걸립니다. 여러분이 이 코드를 사용하면 에이전트의 응답 속도가 크게 개선됩니다.
여러 외부 API를 호출하는 경우 병렬 처리로 시간을 절약하고, FastAPI 같은 비동기 프레임워크와의 통합이 자연스러우며, 서버 리소스를 효율적으로 활용하여 더 많은 동시 사용자를 처리할 수 있습니다. 또한 asyncpg, motor 같은 비동기 데이터베이스 클라이언트를 활용할 수 있어 전체 스택을 비동기로 구성할 수 있습니다.
실전 팁
💡 _arun만 구현하고 _run을 생략하면 동기 환경에서 사용할 수 없습니다. 간단하게라도 _run을 제공하거나, raise NotImplementedError로 명시적으로 차단하세요.
💡 aiohttp, httpx 같은 비동기 HTTP 클라이언트를 사용하세요. requests는 동기 라이브러리이므로 _arun에서 사용하면 이벤트 루프를 블로킹합니다.
💡 asyncio.gather()에 return_exceptions=True를 추가하면 일부 Tool이 실패해도 나머지는 계속 실행됩니다. 에러 핸들링이 더 유연해집니다.
💡 비동기 Tool에서 time.sleep()을 사용하지 마세요. 대신 await asyncio.sleep()을 사용해야 이벤트 루프를 블로킹하지 않습니다.
💡 프로덕션 환경에서는 세션 재사용을 고려하세요. 매번 aiohttp.ClientSession()을 생성하는 대신, 클래스 속성으로 세션을 유지하면 성능이 향상됩니다.
8. Tool 에러 핸들링
시작하며
여러분이 Tool을 프로덕션에 배포했을 때 "API 키가 만료되면 어떻게 하지?", "네트워크 오류가 발생하면 에이전트가 멈추나?", "사용자에게 어떤 메시지를 보여줘야 할까?"라는 걱정을 해본 적 있나요? 이런 문제는 실제 운영 환경에서 반드시 발생합니다.
외부 시스템은 항상 불안정하고, 사용자 입력은 예측 불가능하며, 네트워크는 언제든 끊길 수 있습니다. Tool이 예외를 발생시키면 전체 에이전트가 중단되어 사용자 경험이 크게 나빠집니다.
바로 이럴 때 필요한 것이 체계적인 에러 핸들링입니다. Tool 레벨과 에이전트 레벨 모두에서 에러를 처리하면 장애 상황에서도 우아하게 복구하고 사용자에게 유용한 피드백을 제공할 수 있습니다.
개요
간단히 말해서, Tool 에러 핸들링은 예외를 캐치하여 사용자 친화적인 메시지로 변환하고, 에이전트가 계속 실행되도록 보장하는 방어 메커니즘입니다. 왜 이 개념이 필요한지 실무 관점에서 보면, 에러는 정상적인 흐름의 일부입니다.
예를 들어, 결제 Tool에서 잔액 부족은 에러가 아니라 처리해야 할 비즈니스 케이스입니다. 반면 네트워크 타임아웃은 재시도가 필요한 일시적 오류이고, API 키 만료는 관리자 개입이 필요한 치명적 오류입니다.
기존에는 모든 예외를 동일하게 처리했다면, 이제는 에러 유형별로 다른 전략을 사용하여 복구 가능성을 높입니다. 에러 핸들링의 핵심 특징은: 예외를 문자열 메시지로 변환하여 에이전트 중단을 방지하고, 에러 유형별로 다른 메시지를 제공하며, 로깅과 모니터링을 통합합니다.
이러한 특징들이 에이전트를 안정적이고 운영 가능한 시스템으로 만듭니다.
코드 예제
from langchain.tools import BaseTool
from pydantic import BaseModel, Field
import requests
import logging
from typing import Optional
# 로깅 설정
logger = logging.getLogger(__name__)
class PaymentInput(BaseModel):
amount: int = Field(description="결제 금액")
card_number: str = Field(description="카드 번호")
class PaymentTool(BaseTool):
name = "process_payment"
description = "결제를 처리합니다"
args_schema = PaymentInput
handle_tool_error = True # 에러를 예외 대신 문자열로 반환
api_key: str
def _run(self, amount: int, card_number: str) -> str:
try:
# 비즈니스 로직 검증
if amount <= 0:
return "❌ 결제 실패: 금액은 0보다 커야 합니다"
# API 호출
response = requests.post(
"https://api.payment.example.com/charge",
json={"amount": amount, "card": card_number},
headers={"Authorization": f"Bearer {self.api_key}"},
timeout=15
)
# HTTP 상태 코드별 처리
if response.status_code == 200:
return f"✅ 결제 완료: {amount}원"
elif response.status_code == 402:
# 비즈니스 에러 (잔액 부족 등)
error_msg = response.json().get("message", "잔액 부족")
return f"❌ 결제 실패: {error_msg}"
elif response.status_code == 401:
# 인증 오류
logger.error("Payment API 인증 실패")
return "❌ 시스템 오류: 관리자에게 문의하세요 (오류 코드: AUTH)"
else:
response.raise_for_status()
except requests.Timeout:
logger.warning(f"Payment timeout: amount={amount}")
return "⚠️ 결제 처리 중입니다. 잠시 후 다시 확인해주세요"
except requests.ConnectionError:
logger.error("Payment API 연결 실패")
return "❌ 네트워크 오류: 인터넷 연결을 확인해주세요"
except requests.HTTPError as e:
logger.error(f"Payment HTTP error: {e}")
return f"❌ 결제 시스템 오류: 잠시 후 다시 시도해주세요"
except Exception as e:
# 예상치 못한 모든 에러
logger.exception("Unexpected payment error")
return f"❌ 알 수 없는 오류가 발생했습니다. 고객센터에 문의하세요"
# handle_tool_error를 커스텀 함수로 사용
def custom_error_handler(error: Exception) -> str:
"""Tool 에러를 처리하는 커스텀 핸들러"""
logger.error(f"Tool error: {type(error).__name__}: {str(error)}")
return f"Tool 실행 중 오류가 발생했습니다: {str(error)[:100]}"
class SafeTool(BaseTool):
name = "safe_tool"
description = "커스텀 에러 핸들러를 사용하는 Tool"
handle_tool_error = custom_error_handler # 함수 전달
def _run(self, input: str) -> str:
# 의도적으로 예외 발생
raise ValueError("테스트 에러")
설명
이것이 하는 일: 에러 핸들링은 예외가 에이전트 외부로 전파되는 것을 막고, LLM이 이해할 수 있는 형태로 변환하여 다음 행동을 결정하게 합니다. 첫 번째로, handle_tool_error=True 옵션을 설정하면 _run 메서드에서 발생한 모든 예외를 LangChain이 자동으로 캐치하여 문자열로 변환합니다.
하지만 더 나은 방법은 _run 내부에서 try-except로 직접 처리하는 것입니다. 왜냐하면 에러 유형별로 다른 메시지를 제공하고, 로깅을 추가하며, 일부는 재시도하고 일부는 즉시 실패시키는 등 세밀한 제어가 가능하기 때문입니다.
자동 처리는 폴백으로만 사용하세요. 그 다음으로, PaymentTool의 에러 처리를 단계별로 보면, 먼저 비즈니스 로직 검증(금액 > 0)을 수행하고, HTTP 상태 코드별로 분기 처리하며, 각 예외 유형마다 다른 except 블록을 제공합니다.
402는 비즈니스 에러이므로 사용자에게 구체적인 이유를 알려주고, 401은 시스템 오류이므로 내부 정보를 숨기고 관리자 개입을 요청하며, Timeout은 재시도 가능한 일시적 오류이므로 "잠시 후 다시 확인"을 안내합니다. 내부적으로 logger를 사용하여 모든 에러를 기록하므로, 나중에 모니터링 시스템에서 패턴을 분석할 수 있습니다.
마지막으로, custom_error_handler 함수는 handle_tool_error에 커스텀 로직을 제공하는 방법을 보여줍니다. True 대신 함수를 전달하면, 예외 객체를 받아 원하는 형태의 메시지를 반환할 수 있습니다.
이 방식은 모든 Tool에 공통 에러 처리 로직을 적용할 때 유용합니다. 여러분이 이 코드를 사용하면 프로덕션 환경의 안정성이 크게 향상됩니다.
예상 가능한 에러는 우아하게 처리되어 사용자 경험이 좋아지고, 로그를 통해 문제를 빠르게 진단할 수 있으며, 일시적 오류와 치명적 오류를 구분하여 적절히 대응할 수 있습니다. 또한 에러 메시지에 이모지(✅, ❌, ⚠️)를 사용하면 LLM이 결과를 더 쉽게 파악하고 사용자에게도 시각적으로 명확합니다.
실전 팁
💡 에러 메시지는 LLM이 읽는다는 것을 기억하세요. "Error 500"보다 "결제 시스템이 일시적으로 사용 불가능합니다. 5분 후 재시도하세요"가 LLM에게 더 유용한 정보입니다.
💡 민감한 정보는 에러 메시지에 포함하지 마세요. API 키, 내부 경로, 스택 트레이스 등은 로그에만 기록하고 사용자 메시지에는 제외하세요.
💡 재시도 로직은 Tool 내부에 구현할 수 있습니다. tenacity 라이브러리의 @retry 데코레이터를 _run에 적용하면 자동 재시도가 가능합니다.
💡 에러율을 모니터링하세요. 특정 Tool의 에러율이 높다면 외부 API 문제, 설정 오류, 또는 설계 결함일 수 있습니다. Sentry, DataDog 같은 도구와 통합하세요.
💡 사용자 입력 검증은 Pydantic 스키마에서, 비즈니스 로직 검증은 _run 초반에, 외부 시스템 에러는 try-except로 처리하는 계층적 접근을 사용하세요.
9. Tool 체이닝
시작하며
여러분이 복잡한 작업을 처리할 때 "이 작업은 여러 Tool을 순서대로 실행해야 하는데...", "Tool A의 결과를 Tool B의 입력으로 넘기려면?", "사용자가 한 번에 여러 단계를 요청하면 어떻게 하지?"라는 고민을 해본 적 있나요? 이런 문제는 단일 Tool로는 해결할 수 없는 복잡한 워크플로우에서 발생합니다.
예를 들어, "최근 주문을 조회하고, 배송 상태를 확인하고, 고객에게 이메일을 보내라"는 요청은 세 개의 Tool을 연속으로 실행해야 합니다. 각 Tool의 출력이 다음 Tool의 입력이 되는 파이프라인을 구성해야 합니다.
바로 이럴 때 필요한 것이 Tool 체이닝입니다. 여러 Tool을 조합하여 복잡한 작업을 자동화하고, LLM이 각 단계를 추론하여 최종 목표에 도달하게 만듭니다.
개요
간단히 말해서, Tool 체이닝은 여러 Tool의 출력과 입력을 연결하여 복잡한 워크플로우를 구성하는 패턴입니다. 에이전트가 자동으로 Tool 순서를 결정하거나, 명시적인 체인을 만들 수 있습니다.
왜 이 개념이 필요한지 실무 관점에서 보면, 실제 비즈니스 프로세스는 단일 동작으로 끝나지 않습니다. 예를 들어, 고객 지원 에이전트는 "문제 확인 → 데이터베이스 조회 → 해결책 적용 → 결과 통보"의 과정을 거칩니다.
각 단계를 별도 Tool로 만들면 재사용성이 높아지고, 조합 방식으로 다양한 워크플로우를 구성할 수 있습니다. 기존에는 하나의 거대한 Tool에 모든 로직을 넣었다면, 이제는 작은 Tool들을 만들고 LLM이나 체인으로 조합하여 유연성을 높입니다.
Tool 체이닝의 핵심 특징은: 에이전트가 자동으로 Tool 순서를 결정하거나, LangChain Expression Language(LCEL)로 명시적 파이프라인을 구성하며, 중간 결과를 자동으로 전달합니다. 이러한 특징들이 복잡한 자동화를 가능하게 합니다.
코드 예제
from langchain.tools import StructuredTool
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
# 1. 개별 Tool 정의
def search_orders(customer_id: str) -> str:
"""고객 ID로 최근 주문을 조회합니다"""
# DB 조회 로직
return f"주문 ID: ORDER-123, 상품: 노트북, 상태: 배송중"
def check_shipping(order_id: str) -> str:
"""주문 ID로 배송 상태를 확인합니다"""
# 배송 API 호출
return f"{order_id} 배송 상태: 서울 허브 도착, 예상 도착: 내일"
def send_notification(customer_id: str, message: str) -> str:
"""고객에게 알림을 발송합니다"""
# 이메일/SMS 발송
return f"{customer_id}에게 알림 발송 완료: {message}"
# 2. Tool 객체 생성
tools = [
StructuredTool.from_function(
func=search_orders,
name="SearchOrders",
description="고객의 최근 주문을 조회합니다. 주문 내역이 필요할 때 사용하세요"
),
StructuredTool.from_function(
func=check_shipping,
name="CheckShipping",
description="주문 ID로 배송 상태를 확인합니다. 배송 추적이 필요할 때 사용하세요"
),
StructuredTool.from_function(
func=send_notification,
name="SendNotification",
description="고객에게 알림 메시지를 보냅니다. 최종 결과를 전달할 때 사용하세요"
),
]
# 3. 에이전트 생성 (자동 체이닝)
llm = ChatOpenAI(model="gpt-4", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", "당신은 고객 지원 에이전트입니다. 주문 조회와 배송 확인을 도와줍니다."),
("user", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
# 4. 실행 - LLM이 자동으로 Tool 순서 결정
result = agent_executor.invoke({
"input": "고객 ID CUST-456의 주문을 조회하고, 배송 상태를 확인한 후, 결과를 고객에게 알려주세요"
})
# LLM이 자동으로 다음 순서로 실행:
# 1. SearchOrders(customer_id="CUST-456")
# 2. CheckShipping(order_id="ORDER-123") # 1번 결과에서 추출
# 3. SendNotification(customer_id="CUST-456", message="배송 상태...") # 2번 결과 포함
설명
이것이 하는 일: Tool 체이닝은 LLM이 사용자 요청을 분석하여 필요한 Tool들을 순서대로 선택하고, 각 Tool의 출력을 다음 Tool의 입력으로 전달하는 자동화된 파이프라인을 만듭니다. 첫 번째로, 세 개의 Tool을 정의할 때 각각 단일 책임 원칙을 따릅니다.
search_orders는 조회만, check_shipping은 확인만, send_notification은 발송만 담당합니다. 각 Tool의 description은 "언제 사용하는가"를 명확히 하여 LLM이 올바른 순서를 결정하도록 돕습니다.
왜 이렇게 하는지 이유는, 작은 Tool들은 재사용성이 높고, 테스트가 쉬우며, 다양한 조합으로 다른 워크플로우를 구성할 수 있기 때문입니다. 그 다음으로, AgentExecutor를 생성할 때 tools 리스트를 전달하면, LLM은 각 Tool의 이름과 설명을 프롬프트로 받습니다.
사용자가 "주문 조회 → 배송 확인 → 알림"을 요청하면, LLM은 이를 세 단계로 분해하고 각 단계에 맞는 Tool을 선택합니다. 내부적으로 ReAct 패턴이 작동하여: 1) 사용자 요청 분석 (Reasoning), 2) Tool 선택 및 실행 (Action), 3) 결과 확인 (Observation), 4) 다음 행동 결정 (Reasoning) 순서를 반복합니다.
마지막으로, agent_executor.invoke()를 호출하면 전체 체인이 실행됩니다. verbose=True 옵션으로 각 단계를 확인할 수 있습니다.
LLM은 첫 번째 Tool(SearchOrders)의 결과에서 "주문 ID: ORDER-123"을 파싱하여 두 번째 Tool(CheckShipping)의 파라미터로 사용하고, 마지막 Tool(SendNotification)에는 이전 결과들을 요약하여 전달합니다. 에이전트 스크래치패드에 모든 중간 결과가 누적되어 LLM이 컨텍스트를 유지합니다.
여러분이 이 코드를 사용하면 복잡한 업무 자동화가 가능합니다. 단순 반복 작업을 에이전트에게 맡기고, 새로운 Tool을 추가하면 기존 워크플로우도 자동으로 확장되며, 사용자는 자연어로 복잡한 요청을 할 수 있어 UX가 크게 개선됩니다.
또한 각 Tool의 성공/실패를 추적하여 어느 단계에서 문제가 발생했는지 쉽게 파악할 수 있습니다.
실전 팁
💡 Tool이 너무 많으면 LLM이 선택을 혼란스러워할 수 있습니다. 10개 이하로 유지하거나, 태스크별로 Tool 그룹을 나누세요.
💡 각 Tool의 description에 의존성을 명시하세요. "이 Tool은 SearchOrders의 결과가 필요합니다" 같은 힌트가 도움이 됩니다.
💡 verbose=True로 실행 과정을 확인하면 LLM이 어떤 순서로 Tool을 선택하는지 이해할 수 있습니다. 예상과 다르면 description을 조정하세요.
💡 max_iterations 파라미터로 무한 루프를 방지하세요. AgentExecutor(max_iterations=10)로 최대 반복 횟수를 제한할 수 있습니다.
💡 중간 결과를 명확한 형식으로 반환하세요. "주문 ID: ORDER-123" 처럼 파싱하기 쉬운 형식이 다음 Tool에 전달될 때 에러가 줄어듭니다. JSON 형태도 좋은 선택입니다.