에이전트 도구 개발 가이드: 설계부터 최적화까지-파이썬 튜토리얼-php.cn

Agent Tool Development Guide: From Design to Optimization

1. 소개

초지능 로봇 집사(에이전트)를 조립한다고 상상해 보세요. 이 로봇에는 도라에몽의 4D 주머니처럼 작업을 완료하는 데 도움이 되는 다양한 도구가 필요합니다. 이 기사에서는 AI 집사를 더욱 유능하고 효율적으로 만들기 위한 강력한 도구를 만드는 방법을 설명합니다.

2. 두 가지 핵심 도구 디자인 패턴

2.1 동기식 도구: 즉시 응답 모드

셀프 서비스 커피 머신을 사용해 보세요.

동전을 넣고 "아메리카노" 버튼을 눌러주세요
몇 초만 기다리세요
커피가 흘러나와 마실 준비가 되었습니다

이것은 일반적인 동기식 도구 패턴입니다. 에이전트는 도구를 호출하고 즉각적인 결과를 기다립니다. 빠르고 간단합니다.

class WeatherTool(BaseTool):
    """Weather Query Tool - Synchronous Mode"""
    async def execute(self, city: str) -> dict:
        # Simple and direct like pressing a coffee machine button
        weather_data = await self.weather_api.get_current(city)
        return {
            "status": "success",
            "data": {
                "temperature": weather_data.temp,
                "humidity": weather_data.humidity,
                "description": weather_data.desc
            }
        }

로그인 후 복사

사용 사례:

빠른 쿼리: 날씨, 환율, 간단한 계산
간단한 조작: 메시지 보내기, 스위치 제어
실시간 피드백 : 인증번호 확인, 잔액조회

2.2 비동기 도구: 작업 추적 모드

배달 앱을 통해 음식을 주문한다고 상상해 보세요.

주문 후 APP에서 주문번호를 알려드립니다
언제든지 주문현황을 확인하실 수 있습니다
배송이 완료되면 APP에서 알려드립니다

이것이 비동기식 도구의 작동 방식으로, 처리 시간이 오래 걸리는 작업에 적합합니다.

class DocumentAnalysisTool(BaseTool):
    """Document Analysis Tool - Asynchronous Mode"""

    async def start_task(self, file_path: str) -> str:
        # Like placing a food delivery order, returns a task ID
        task_id = str(uuid.uuid4())
        await self.task_queue.put({
            "task_id": task_id,
            "file_path": file_path,
            "status": "processing"
        })
        return task_id

    async def get_status(self, task_id: str) -> dict:
        # Like checking food delivery status
        task = await self.task_store.get(task_id)
        return {
            "task_id": task_id,
            "status": task["status"],
            "progress": task.get("progress", 0),
            "result": task.get("result", None)
        }

로그인 후 복사

사용 사례:

시간이 많이 걸리는 작업: 대용량 파일 처리, 데이터 분석
다단계 작업: 비디오 렌더링, 보고서 생성
진행 상황 추적 필요: 모델 훈련, 일괄 처리

3. 도구 인터페이스 표준화: 범용 사양 설정

모든 전기 제품이 통합 소켓 표준을 따르는 것처럼 도구 인터페이스에도 표준화가 필요합니다. 이렇게 하면 모든 도구가 에이전트와 완벽하게 작동할 수 있습니다.

3.1 도구 설명 사양

제품 매뉴얼을 작성한다고 상상해 보세요. 사용자에게 다음 사항을 명확하게 알려야 합니다.

도구의 기능
필요한 매개변수
어떤 결과가 반환되나요?

from pydantic import BaseModel, Field

class ToolSchema(BaseModel):
    """Tool Manual Template"""
    name: str = Field(..., description="Tool name")
    description: str = Field(..., description="Tool purpose description")
    parameters: dict = Field(..., description="Required parameters")
    required: List[str] = Field(default_factory=list, description="Required parameters")

    class Config:
        schema_extra = {
            "example": {
                "name": "Weather Query",
                "description": "Query weather information for specified city",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "City name"
                        }
                    }
                },
                "required": ["city"]
            }
        }

로그인 후 복사

3.2 통합 기본 클래스

모든 전기 제품에 전원 스위치와 전원 인터페이스가 필요한 것처럼 모든 도구도 기본 사양을 따라야 합니다.

class BaseTool(ABC):
    """Base template for all tools"""

    @abstractmethod
    def get_schema(self) -> ToolSchema:
        """Tool manual"""
        pass

    def validate_input(self, params: Dict) -> Dict:
        """Parameter check, like a fuse in electrical appliances"""
        return ToolSchema(**params).dict()

    @abstractmethod
    async def execute(self, **kwargs) -> Dict:
        """Actual functionality execution"""
        pass

로그인 후 복사

4. 오류 처리: 도구의 신뢰성 향상

가전제품이 물, 충격, 과부하로부터 보호해야 하는 것처럼 도구에도 포괄적인 보호 메커니즘이 필요합니다.

4.1 오류 분류 및 처리

특급 배송을 처리한다고 상상해 보세요.

잘못된 주소 → 매개변수 오류
시스템 점검 → 서비스 일시적 중단
택배가 너무 바쁨 → 속도 제한 및 재시도 필요

class WeatherTool(BaseTool):
    """Weather Query Tool - Synchronous Mode"""
    async def execute(self, city: str) -> dict:
        # Simple and direct like pressing a coffee machine button
        weather_data = await self.weather_api.get_current(city)
        return {
            "status": "success",
            "data": {
                "temperature": weather_data.temp,
                "humidity": weather_data.humidity,
                "description": weather_data.desc
            }
        }

로그인 후 복사

4.2 재시도 메커니즘

첫 번째 시도가 실패할 때 자동으로 두 번째 배송을 준비하는 것과 같습니다.

class DocumentAnalysisTool(BaseTool):
    """Document Analysis Tool - Asynchronous Mode"""

    async def start_task(self, file_path: str) -> str:
        # Like placing a food delivery order, returns a task ID
        task_id = str(uuid.uuid4())
        await self.task_queue.put({
            "task_id": task_id,
            "file_path": file_path,
            "status": "processing"
        })
        return task_id

    async def get_status(self, task_id: str) -> dict:
        # Like checking food delivery status
        task = await self.task_store.get(task_id)
        return {
            "task_id": task_id,
            "status": task["status"],
            "progress": task.get("progress", 0),
            "result": task.get("result", None)
        }

로그인 후 복사

5. 성능 최적화: 도구 효율성 향상

5.1 캐싱 메커니즘

유명 상품을 눈에 띄는 위치에 배치하는 편의점처럼:

from pydantic import BaseModel, Field

class ToolSchema(BaseModel):
    """Tool Manual Template"""
    name: str = Field(..., description="Tool name")
    description: str = Field(..., description="Tool purpose description")
    parameters: dict = Field(..., description="Required parameters")
    required: List[str] = Field(default_factory=list, description="Required parameters")

    class Config:
        schema_extra = {
            "example": {
                "name": "Weather Query",
                "description": "Query weather information for specified city",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "City name"
                        }
                    }
                },
                "required": ["city"]
            }
        }

로그인 후 복사

5.2 동시성 제어

병원 예약제처럼 동시접속 횟수 조절:

class BaseTool(ABC):
    """Base template for all tools"""

    @abstractmethod
    def get_schema(self) -> ToolSchema:
        """Tool manual"""
        pass

    def validate_input(self, params: Dict) -> Dict:
        """Parameter check, like a fuse in electrical appliances"""
        return ToolSchema(**params).dict()

    @abstractmethod
    async def execute(self, **kwargs) -> Dict:
        """Actual functionality execution"""
        pass

로그인 후 복사

6. 테스트 및 문서화: 도구 신뢰성 보장

6.1 단위 테스트

신제품 출시 전 품질검사처럼:

class ToolError(Exception):
    """Tool error base class"""
    def __init__(self, message: str, error_code: str, retry_after: Optional[int] = None):
        self.message = message
        self.error_code = error_code
        self.retry_after = retry_after

@error_handler
async def execute(self, **kwargs):
    try:
        # Execute specific operation
        result = await self._do_work(**kwargs)
        return {"status": "success", "data": result}
    except ValidationError:
        # Parameter error, like wrong address
        return {"status": "error", "code": "INVALID_PARAMS"}
    except RateLimitError as e:
        # Need rate limiting, like courier too busy
        return {
            "status": "error", 
            "code": "RATE_LIMIT",
            "retry_after": e.retry_after
        }

로그인 후 복사

6.2 문서화 표준

자세하고 명확한 제품 설명서를 작성하는 것처럼:

class RetryableTool(BaseTool):
    @retry(
        stop=stop_after_attempt(3),  # Maximum 3 retries
        wait=wait_exponential(multiplier=1, min=4, max=10)  # Increasing wait time
    )
    async def execute_with_retry(self, **kwargs):
        return await self.execute(**kwargs)

로그인 후 복사