私はLangChainを用いたエンタープライズAIアプリケーション開発で500以上のカスタムツールを実装してきた経験があります。本稿では、LangChainにおける自作ツールの開発から登録、そして本番環境での最適な運用まで、包括的に解説します。HolySheep AIは私のチームにとって今すぐ登録最もコスト効率の良いAPIプロバイダーであり、本記事すべてのコードはHolySheep AI環境を前提として書かれています。
LangChainツールのアーキテクチャ概要
LangChainのツールシステムは、ツール呼び出しエージェントが外部システムと対話するための標準化されたインターフェースを提供します。基本的なツール定義から、非同期処理対応、パフォーマンス監視組み込みのハイブリッド設計まで、段階的に見ていきましょう。
自作ツールの基本構造
LangChainにおけるツールは、BaseToolクラスを継承して定義します。最小構成ではname、description、funcの3要素が必要です。 HolySheep AIの<50msレイテンシを活かした高速ツール応答を実現するため、ツール設計段階から非同期パターンを採用することを強く推奨します。
"""
LangChain 自作ツールの基本構造
HolySheep AI API compatible implementation
"""
from langchain.tools import BaseTool
from langchain.pydantic_v1 import Field, BaseModel
from typing import Optional, Type
import asyncio
import time
from dataclasses import dataclass
ツール入力スキーマの定義
class WeatherInput(BaseModel):
city: str = Field(description="查询対象都市名")
country: str = Field(default="JP", description="国コード（ISO 3166-1 alpha-2）")
@dataclass
class ToolExecutionResult:
"""ツール実行結果のラッパー"""
success: bool
result: Optional[dict] = None
error: Optional[str] = None
execution_time_ms: float = 0.0
class WeatherTool(BaseTool):
"""天気情報取得ツール - 基本的なツール実装例"""
name: str = "get_weather"
description: str = "指定された都市の現在天気を取得します。入力: city（都市名）、country（国コード）"
args_schema: Type[BaseModel] = WeatherInput
def _run(self, city: str, country: str = "JP") -> dict:
"""同期実行メソッド"""
start_time = time.perf_counter()
# 実際のAPI呼び出しはここに実装
# HolySheep AI経由で外部APIを呼び出す例
result = {
"city": city,
"country": country,
"temperature": 22.5,
"condition": "晴れ",
"humidity": 65,
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
}
execution_time = (time.perf_counter() - start_time) * 1000
return {
"data": result,
"execution_time_ms": round(execution_time, 2)
}
async def _arun(self, city: str, country: str = "JP") -> dict:
"""非同期実行メソッド - 本番環境では必須"""
start_time = time.perf_counter()
# HolySheep AI APIとの非同期通信
await asyncio.sleep(0.01) # 実際のAPI呼び出しを模擬
result = {
"city": city,
"country": country,
"temperature": 22.5,
"condition": "晴れ",
"humidity": 65,
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
}
execution_time = (time.perf_counter() - start_time) * 1000
return {
"data": result,
"execution_time_ms": round(execution_time, 2)
}
ツールインスタンス化
weather_tool = WeatherTool()HolySheep AIと統合した高度なツールクラス設計
私のプロジェクトでは、ツールの再利用性と監視機能を最大化するため、ベースクラスを自作して継承する設計を採用しています。HolySheep AIのAPIキーを環境変数で管理し、レート制限内での効率的なAPI呼び出しを保証します。
"""
HolySheep AI統合 高機能ツールベースクラス
レート制限対応・再試行機構・コスト追跡付き
"""
import os
import asyncio
import aiohttp
import time
import hashlib
from abc import ABC, abstractmethod
from typing import Any, Optional, Dict, List
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from collections import defaultdict
from threading import Lock
import json
@dataclass
class RateLimiter:
"""トークンベースレートリミッター"""
requests_per_minute: int = 60
tokens_per_minute: int = 100000
_request_timestamps: List[float] = field(default_factory=list)
_token_usage: List[tuple] = field(default_factory=list) # (timestamp, tokens)
_lock: Lock = field(default_factory=Lock)
def __post_init__(self):
self._lock = Lock()
async def acquire(self, estimated_tokens: int = 1000) -> float:
"""レート制限内で待機し、待機時間を返す"""
wait_time = 0.0
with self._lock:
now = time.time()
cutoff = now - 60
# 1分以内のリクエスト数をチェック
self._request_timestamps = [t for t in self._request_timestamps if t > cutoff]
if len(self._request_timestamps) >= self.requests_per_minute:
wait_time = 60 - (now - self._request_timestamps[0])
# 1分以内のトークン使用量をチェック
self._token_usage = [(t, tok) for t, tok in self._token_usage if t > cutoff]
current_tokens = sum(tok for _, tok in self._token_usage)
if current_tokens + estimated_tokens > self.tokens_per_minute:
if self._token_usage:
oldest = self._token_usage[0][0]
wait_time = max(wait_time, 60 - (now - oldest))
if wait_time > 0:
self._request_timestamps.append(now + wait_time)
else:
self._request_timestamps.append(now)
self._token_usage.append((now, estimated_tokens))
if wait_time > 0:
await asyncio.sleep(wait_time)
return wait_time
class HolySheepAIClient:
"""HolySheep AI APIクライアント - カスタムツール向け"""
def __init__(
self,
api_key: Optional[str] = None,
base_url: str = "https://api.holysheep.ai/v1",
rate_limiter: Optional[RateLimiter] = None
):
self.api_key = api_key or os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
self.base_url = base_url
self.rate_limiter = rate_limiter or RateLimiter()
self._cost_tracker: Dict[str, List[dict]] = defaultdict(list)
self._session: Optional[aiohttp.ClientSession] = None
async def _get_session(self) -> aiohttp.ClientSession:
if self._session is None or self._session.closed:
self._session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
return self._session
async def chat_completion(
self,
messages: List[Dict[str, str]],
model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: int = 1000
) -> Dict[str, Any]:
"""
HolyShe