私がECサイトを運営していた頃、AIカスタマーサービスを導入しようとして壁にぶつかりました。Function Callingを活用した動的な商品検索や注文状況確認を実装したいのですが、API応答の解析とエラー処理の実装が想像以上に複雑だったのです。本記事では、私が行き着いた解決策と、HolySheep AIを活用した堅牢なFunction Calling実装について共有します。
Function Calling とは？
Function Callingは、LLMにユーザー意図を理解させ、定義した関数を呼び出すための機能です。従来のプロンプトエンジニアリングとは異なり、構造化されたJSON出力を保証し、以下のような活用シーンがあります：
- ECカスタマーサービス：商品検索在庫確認・注文ステータス取得
- RAGシステム：外部知识库检索・文書検索
- 個人開発：天气预报・计算器・日历操作
基本的なFunction Calling実装
まずはHolySheep AIを使用した基本的なFunction Callingの実装例を示します。HolySheep AIはレートが¥1=$1（公式サイト¥7.3=$1の85%節約）で、レイテンシが<50msという高速応答が特徴です。
import requests
import json
class HolySheepAIClient:
"""HolySheep AI APIクライアント - Function Calling対応"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def create_function_call_messages(self, user_message: str,
functions: list) -> list:
"""Function Calling用のメッセージを構築"""
return [
{"role": "system", "content": "あなたは有用的なアシスタントです。"},
{"role": "user", "content": user_message}
]
def call_chat_completion(self, messages: list,
functions: list) -> dict:
"""Chat Completion APIを呼び出し"""
payload = {
"model": "gpt-4.1", # $8/MTok - HolySheep価格
"messages": messages,
"functions": functions,
"function_call": "auto"
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
timeout=30
)
if response.status_code != 200:
raise APIError(
f"API Error: {response.status_code} - {response.text}"
)
return response.json()
class APIError(Exception):
"""APIエラー用カスタム例外"""
pass
使用例
client = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
functions = [
{
"name": "get_product_info",
"description": "商品の詳細情報を取得する",
"parameters": {
"type": "object",
"properties": {
"product_id": {
"type": "string",
"description": "商品ID"
}
},
"required": ["product_id"]
}
},
{
"name": "check_order_status",
"description": "注文のステータスを確認する",
"parameters": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "注文ID"
}
},
"required": ["order_id"]
}
}
]
messages = client.create_function_call_messages(
"商品IDがABC123の商品の在庫状況を教えて",
functions=functions
)
result = client.call_chat_completion(messages, functions)
print(json.dumps(result, indent=2, ensure_ascii=False))Function Calling応答の解析手法
APIからの応答を適切に解析することが重要です。HolySheep AIのAPI応答はOpenAI互換のため、同じパターンを適用できます。
import json
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
from enum import Enum
class FunctionCallStatus(Enum):
"""関数呼び出しの状態"""
SUCCESS = "success"
PARSE_ERROR = "parse_error"
VALIDATION_ERROR = "validation_error"
EXECUTION_ERROR = "execution_error"
NO_FUNCTION_CALL = "no_function_call"
@dataclass
class ParsedFunctionCall:
"""解析された関数呼び出し結果"""
function_name: str
arguments: Dict[str, Any]
status: FunctionCallStatus
raw_response: Dict[str, Any]
error_message: Optional[str] = None
class FunctionCallParser:
"""Function Calling応答パーサー"""
def __init__(self):
self.execution_registry: Dict[str, callable] = {}
def register_function(self, name: str, func: callable):
"""実行可能な関数を登録"""
self.execution_registry[name] = func
def parse_response(self, response: Dict[str, Any]) -> ParsedFunctionCall:
"""API応答を解析して関数呼び出し情報を抽出"""
try:
choices = response.get("choices", [])
if not choices:
return ParsedFunctionCall(
function_name="",
arguments={},
status=FunctionCallStatus.PARSE_ERROR,
raw_response=response,
error_message="No choices in response"
)
message = choices[0].get("message", {})
# function_callがある場合
if "function_call" in message:
fc = message["function_call"]
function_name = fc.get("name", "")
arguments_str = fc.get("arguments", "{}")
try:
arguments = json.loads(arguments_str)
except json.JSONDecodeError as e:
return ParsedFunctionCall(
function_name=function_name,
arguments={},
status=FunctionCallStatus.PARSE_ERROR,
raw_response=response,
error_message=f"Arguments parse error: {str(e)}"
)
return ParsedFunctionCall(
function_name=function_name,
arguments=arguments,
status=FunctionCallStatus.SUCCESS,
raw_response=response
)
# function_callがない場合
return ParsedFunctionCall(
function_name="",
arguments={},
status=FunctionCallStatus.NO_FUNCTION_CALL,
raw_response=response,
error_message="No function call in response"
)
except Exception as e:
return ParsedFunctionCall(
function_name="",
arguments={},
status=FunctionCallStatus.PARSE_ERROR,
raw_response=response,
error_message=f"Unexpected error: {str(e)}"
)
def execute_function(self, parsed: ParsedFunctionCall) -> Dict[str, Any]:
"""登録された関数を実行"""
if parsed.status != FunctionCallStatus.SUCCESS:
return {
"success": False,
"error": parsed.error_message
}
func = self.execution_registry.get(parsed.function_name)
if not func:
return {
"success": False,
"error": f"Function '{parsed.function_name}' not registered"
}
try:
result = func(**parsed.arguments)
return {"success": True, "result": result}
except TypeError as e:
return {
"success": False,
"error": f"Invalid arguments: {str(e)}",
"provided_args": parsed.arguments
}
except Exception as e:
return {
"success": False,
"error": f"Execution error: {str(e)}"
}
関数の登録と使用例
def get_product_info(product_id: str) -> Dict[str, Any]:
"""商品情報を取得（モック）"""
return {
"product_id": product_id,
"name": "Sample Product",
"price": 2980,
"stock": 42
}
def check_order_status(order_id: str) -> Dict[str, Any]:
"""注文ステータスを確認（モック）"""
return {
"order_id": order_id,
"status": "shipped",
"estimated_delivery": "2026-01-20"
}
パーサーを初期化
parser = FunctionCallParser()
parser.register_function("get_product_info", get_product_info)
parser.register_function("check_order_status", check_order_status)
応答を解析
sample_response = {
"choices": [{
"message": {
"role": "assistant",
"content": None,
"function_call": {
"name": "get_product_info",
"arguments": "{\"product_id\": \"ABC123\"}"
}
}
}]
}
parsed = parser.parse_response(sample_response)
print(f"Function: {parsed.function_name}")
print(f"Arguments: {parsed.arguments}")
print(f"Status: {parsed.status.value}")実践的なECシステムへの統合例
私が実際に運営していたECサイトでの実装例を示します。在庫確認・注文追跡・カート操作を一つの会話で処理できるシステムです。
import asyncio
import aiohttp
from typing import Optional, List, Dict, Any
from datetime import datetime
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ECAIFunctionHandler:
"""ECサイト用AIファンクションハンドラー"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.session: Optional[aiohttp.ClientSession] = None
# 利用可能な関数定義
self.functions = [
{
"name": "search_products",
"description": "キーワードで商品を検索する",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "検索キーワード"},
"category": {"type": "string", "description": "カテゴリ"},
"max_price": {"type": "number", "description": "最大価格"}
},
"required": ["query"]
}
},
{
"name": "get_cart",
"description": "カートの内容を取得する",
"parameters": {"type": "object", "properties": {}}
},
{
"name": "add_to_cart",
"description": "商品をカートに追加する",
"parameters": {
"type": "object",
"properties": {
"product_id": {"type": "string"},
"quantity": {"type": "integer", "minimum": 1}
},
"required": ["product_id", "quantity"]
}
},
{
"name": "place_order",
"description": "注文を確定する",
"parameters": {
"type": "object",
"properties": {
"payment_method": {"type": "string", "enum": ["credit_card", "wechat", "alipay"]}
},
"required": ["payment_method"]
}
}
]
async def __aenter__(self):
self.session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def process_message(self, user_message: str,
conversation_history: List[Dict]) -> Dict[str, Any]:
"""メッセージを送信し、Function Callingを処理"""
messages = [{"role": "system",
"content": "あなたはECサイトのAIコンシェルジュです。"}]
messages.extend(conversation_history)
messages.append({"role": "user", "content": user_message})
payload = {
"model": "gpt-4.1",
"messages": messages,
"functions": self.functions,
"temperature": 0.7
}
async with self.session.post(
f"{self.base_url}/chat/completions",
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"API Error: {response.status} - {error_text}")
raise Exception(f"API Error: {response.status}")
result = await response.json()
return self._handle_function_calls(result, messages)
def _handle_function_calls(self, response: Dict,
messages: List[Dict]) -> Dict[str, Any]:
"""関数呼び出しを処理し、結果を追加"""
choice = response.get("choices", [{}])[0]
message = choice.get("message", {})
if "function_call" not in message:
return {
"type": "text",
"content": message.get("content", "")
}
fc = message.get("function_call")
func_name = fc.get("name")
args = json.loads(fc.get("arguments", "{}"))
# 関数を実行
func_result = self._execute_function(func_name, args)
# 関数結果を会話履歴に追加
messages.append({
"role": "assistant",
"content": None,
"function_call": {
"name": func_name,
"arguments": fc.get("arguments")
}
})
messages.append({
"role": "function",
"name": func_name,
"content": json.dumps(func_result, ensure_ascii=False)
})
# 関数結果を解釈して最終応答を生成
return {
"type": "function_result",
"function": func_name,
"result": func_result,
"messages": messages
}
def _execute_function(self, name: str, args: Dict) -> Any:
"""関数の実際の実行"""
# モック実装 - 実際のECシステムではDBアクセス等を行う
function_map = {
"search_products": self._search_products,
"get_cart": self._get_cart,
"add_to_cart": self._add_to_cart,
"place_order": self._place_order
}
func = function_map.get(name)
if not func:
return {"error": f"Unknown function: {name}"}
return func(args)
def _search_products(self, args: Dict) -> Dict:
"""商品検索（モック）"""
return {
"products": [
{"id": "P001", "name": "ワイヤレスヘッドフォン", "price": 4980},
{"id": "P002", "name": "Bluetoothスピーカー", "price": 2980}
],
"total_count": 2
}
def _get_cart(self, args: Dict) -> Dict:
"""カート取得（モック）"""
return {
"items": [
{"product_id": "P001", "name": "ワイヤレスヘッドフォン",
"quantity": 1, "price": 4980}
],
"total": 4980
}
def _add_to_cart(self, args: Dict) -> Dict:
"""カート追加（モック）"""
return {
"success": True,
"message": f"商品ID {args['product_id']} を数量 {args['quantity']} で追加しました"
}
def _place_order(self, args: Dict) -> Dict:
"""注文確定（モック）"""
return {
"success": True,
"order_id": f"ORD{datetime.now().strftime('%Y%m%d%H%M%S')}",
"payment_method": args.get("payment_method"),
"total_amount": 4980
}
使用例
async def main():
async with ECAIFunctionHandler(api_key="YOUR_HOLYSHEEP_API_KEY") as handler:
conversation = []
# 商品検索
result1 = await handler.process_message(
"ワイヤレスイヤホンを探しています", conversation
)
print(f"Search Result: {result1}")
# 商品をカートに追加
if result1.get("type") == "function_result":
conversation = result1.get("messages", [])
result2 = await handler.process_message(
"1番目の商品をカートに追加して", conversation
)
print(f"Add to Cart: {result2}")
asyncio.run(main())
料金比較とHolySheepのコスト優位性
Function Callingを含むAI API運用において、料金体系は重要な選択基準です。以下の表は2026年現在の主要モデルの料金比較です：
- GPT-4.1: $8/MTok（入力）、DeepSeek V3.2: $0.42/MTok — HolySheepなら最安クラス
- Claude Sonnet 4.5: $15/MTok — 高精度が必要でないタスクには過剰
- Gemini 2.5 Flash: $2.50/MTok — コストパフォーマンス良好
HolySheep AIは¥1=$1のレートの提供により、公式サイト比85%のコスト削減を実現します。WeChat PayやAlipayにも対応しているため、中国市場向けのサービス開発にも最適です。
よくあるエラーと対処法
エラー1: Function arguments のJSONパースエラー
# 問題: argumentsが不正なJSON形式
"arguments": "{\"product_id\": }" ← カンマが不足
解決法: パース前にJSONの妥当性を検証
import json
def safe_parse_arguments(arguments_str: str) -> Optional[Dict]:
"""安全なJSONパース"""
try:
return json.loads(arguments_str)
except json.JSONDecodeError as e:
logger.warning(f"Invalid JSON: {arguments_str[:100]}")
# 自動修復を試みる
return try_fix_and_parse(arguments_str)
def try_fix_and_parse(broken_json: str) -> Optional[Dict]:
"""不完全なJSONを修復試行"""
import re
# よくあるミスを自動修復
fixed = broken_json
# 末尾のコンマを削除: {"a": 1, } → {"a": 1}
fixed = re.sub(r',(\s*[}\]])', r'\1', fixed)
# シングルクォートをダブルクォートに
fixed = re.sub(r"'([^']*)'", r'"\1"', fixed)
try:
return json.loads(fixed)
except:
return None
使用
result = safe_parse_arguments('{"product_id": }')
if result is None:
raise ValueError("Arguments parsing failed after repair attempts")エラー2: 関数が登録されていない
# 問題: LLMが未登録の関数を呼び出そうとする
Function 'unknown_function' not registered
解決法: フォールバック機構と、未登録関数の検出
class RobustFunctionRegistry:
"""堅牢な関数レジストリ"""
def __init__(self):
self.functions: Dict[str, callable] = {}
self.similarity_threshold = 0.7
def register(self, name: str, func: callable, aliases: List[str] = None):
"""関数を登録（エイリアス付き）"""
self.functions[name] = func
if aliases:
for alias in aliases:
self.functions[alias] = func
def call(self, name: str, args: Dict) -> Dict:
"""関数を呼び出し（類似関数提案付き）"""
# 完全一致
if name in self.functions:
return self._safe_execute(name, args)
# 類似関数を検索
suggestions = self._find_similar_functions(name)
if suggestions:
suggestion_names = ", ".join(suggestions)
return {
"error": f"Unknown function: '{name}'",
"suggestions": suggestions,
"message": f"Did you mean: {suggestion_names}?"
}
return {
"error": f"Function '{name}' not found",
"available_functions": list(self.functions.keys())
}
def _find_similar_functions(self, name: str) -> List[str]:
"""Levenshtein距離で類似関数を検索"""
from difflib import SequenceMatcher
matches = []
for func_name in self.functions.keys():
ratio = SequenceMatcher(None, name.lower(),
func_name.lower()).ratio()
if ratio >= self.similarity_threshold:
matches.append(func_name)
return sorted(matches, key=lambda x:
SequenceMatcher(None, name, x).ratio(),
reverse=True)[:3]
def _safe_execute(self, name: str, args: Dict) -> Dict:
"""安全な関数実行"""
try:
result = self.functions[name](**args)
return {"success": True, "result": result}
except Exception as e:
return {"success": False, "error": str(e)}
使用例
registry = RobustFunctionRegistry()
registry.register("get_product", lambda pid: {"id": pid},
aliases=["find_product", "product_info"])
result = registry.call("get_produt", {"pid": "ABC"})
出力: {'error': "Unknown function: 'get_produt'",
'suggestions': ['get_product'], ...}
エラー3: APIタイムアウトとリトライ処理
# 問題: ネットワーク不安定导致的タイムアウト
HTTPSConnectionPool: Read timed out
解決法: 指数バックオフ付きリトライ機構
import asyncio
import random
from tenacity import (
retry, stop_after_attempt, wait_exponential,
retry_if_exception_type
)
class HolySheepRetryClient:
"""リトライ機能付きHolySheep APIクライアント"""
def __init__(self, api_key: str, max_retries: int = 3):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.max_retries = max_retries
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10),
retry=retry_if_exception_type((aiohttp.ClientError, asyncio.TimeoutError))
)
async def call_with_retry(self, payload: Dict) -> Dict:
"""指数バックオフでリトライ"""
async with aiohttp.ClientSession() as session:
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
async with session.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=60)
) as response:
if response.status == 429:
# レート制限時の специальная обработка
retry_after = response.headers.get('Retry-After', '5')
logger.warning(f"Rate limited. Waiting {retry_after}s")
await asyncio.sleep(int(retry_after))
raise aiohttp.ClientError("Rate limited")
if response.status == 500:
# サーバーエラーの場合はリトライ
raise aiohttp.ClientError(f"Server error: {response.status}")
return await response.json()
async def chat_with_fallback(self, messages: List[Dict],
functions: List[Dict]) -> Dict:
"""プライマリ失敗時はフォールバックモデルを使用"""
# GPT-4.1で試行
try:
return await self.call_with_retry({
"model": "gpt-4.1",
"messages": messages,
"functions": functions
})
except Exception as e:
logger.warning(f"GPT-4.1 failed: {e}")
# Gemini 2.5 Flashでフォールバック
try:
return await self.call_with_retry({
"model": "gemini-2.5-flash",
"messages": messages,
"functions": functions
})
except Exception as e2:
logger.error(f"All models failed: {e2}")
return {"error": "All API calls failed",
"original_error": str(e2)}
使用
async def main():
client = HolySheepRetryClient("YOUR_HOLYSHEEP_API_KEY")
result = await client.chat_with_fallback(
messages=[{"role": "user",
"content": "商品の在庫を確認"}],
functions=[{
"name": "check_stock",
"parameters": {
"type": "object",
"properties": {
"product_id": {"type": "string"}
}
}
}]
)
print(result)まとめ
Function Callingの実装において、応答解析とエラー処理は非常に重要です。本記事で紹介したパターンを活用することで、堅牢なAI API統合が実現できます。
- 構造化されたパースクラスで応答を安全に解析
- 関数レジストリの類似関数提案でUXを向上
- 指数バックオフとフォールバックで可用性を確保
HolySheep AIなら、レート¥1=$1の低コストでFunction Callingを含むすべてのOpenAI互換APIを利用できます。<50msのレイテンシとWeChat Pay/Alipay対応も備えており、ECサービスやRAGシステムの構築に最適です。
👉 HolySheep AI に登録して無料クレジットを獲得