บทนำ

ในฐานะวิศวกรที่ทำงานกับ LLM API มาหลายปี ผมเคยเจอปัญหาการ Integrate function calling ที่ทำให้ productivity ลดลงอยู่บ่อยๆ บทความนี้จะเป็นคู่มือเชิงลึกสำหรับการ implement Function Calling ด้วย Gemini 2.5 Flash ผ่าน HolySheep AI ซึ่งให้บริการด้วย latency ต่ำกว่า 50ms และราคาที่ประหยัดกว่า 85% เมื่อเทียบกับ OpenAI Function Calling คือความสามารถที่ช่วยให้ LLM สามารถเรียก external tools หรือ functions ได้ ทำให้สามารถสร้าง automation workflow ที่ซับซ้อนได้ เช่น การค้นหาข้อมูล การจัดการ database หรือการเรียก API ภายนอก

สถาปัตยกรรม Function Calling ของ Gemini

Gemini ใช้สถาปัตยกรรมที่แตกต่างจาก OpenAI อย่างมีนัยสำคัญ โดยใช้ระบบ "Tool Use" แทน "Function Calling" แบบดั้งเดิม สถาปัตยกรรมหลักประกอบด้วย: 

"""
Gemini 2.5 Flash Function Calling - Basic Implementation
"""
import httpx
import json
from typing import List, Dict, Any, Optional

class HolySheepClient:
    """Client สำหรับเชื่อมต่อกับ HolySheep AI API"""
    
    BASE_URL = "https://api.holysheep.ai/v1"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.client = httpx.Client(
            timeout=30.0,
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json"
            }
        )
    
    def chat_completions(
        self,
        model: str,
        messages: List[Dict[str, Any]],
        tools: Optional[List[Dict[str, Any]]] = None,
        tool_choice: Optional[str] = None,
        temperature: float = 0.7,
        max_tokens: int = 2048
    ) -> Dict[str, Any]:
        """
        Send chat completion request with function calling support
        
        Args:
            model: Model name (e.g., "gemini-2.0-flash-exp")
            messages: List of message objects
            tools: List of tool definitions
            tool_choice: "auto", "none", or specific function name
            temperature: Sampling temperature
            max_tokens: Maximum tokens in response
        """
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens
        }
        
        if tools:
            payload["tools"] = tools
            if tool_choice:
                payload["tool_choice"] = tool_choice
        
        response = self.client.post(
            f"{self.BASE_URL}/chat/completions",
            json=payload
        )
        response.raise_for_status()
        return response.json()


ตัวอย่างการใช้งาน

client = HolySheepClient(api_key="YOUR_HOLYSHEEP_API_KEY")


กำหนด tools

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "ดึงข้อมูลสภาพอากาศของเมืองที่ระบุ",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "ชื่อเมืองที่ต้องการทราบสภาพอากาศ"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "หน่วยอุณหภูมิ"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate_route",
            "description": "คำนวณเส้นทางจากจุดเริ่มต้นไปยังจุดหมาย",
            "parameters": {
                "type": "object",
                "properties": {
                    "origin": {"type": "string"},
                    "destination": {"type": "string"},
                    "mode": {
                        "type": "string",
                        "enum": ["driving", "walking", "transit"]
                    }
                },
                "required": ["origin", "destination"]
            }
        }
    }
]


ส่ง request แรก

messages = [
    {"role": "user", "content": "สภาพอากาศในกรุงเทพวันนี้เป็นอย่างไร และถ้าจะเดินทางจากสยามไปเยาวราชใช้เวลากี่นาที?"}
]

response = client.chat_completions(
    model="gemini-2.0-flash-exp",
    messages=messages,
    tools=tools
)

print(json.dumps(response, indent=2, ensure_ascii=False))

การจัดการ Execution Loop สำหรับ Multi-Tool Calling

ในกรณีที่ต้องการเรียกหลาย tools หรือต้องการให้ model ตัดสินใจเองว่าจะเรียก tool ใด ต้อง implement execution loop ที่รองรับการทำงานแบบ iterative 

"""
Execution Loop สำหรับ Multi-Tool Function Calling
รองรับการเรียกหลาย tools ในครั้งเดียวและการตัดสินใจอัตโนมัติ
"""
import httpx
import json
import asyncio
from typing import List, Dict, Any, Callable, Optional
from dataclasses import dataclass
from enum import Enum

class FinishReason(Enum):
    STOP = "stop"
    LENGTH = "length"
    TOOL_CALLS = "tool_calls"

@dataclass
class ToolResult:
    """ผลลัพธ์จากการ execute tool"""
    tool_call_id: str
    tool_name: str
    result: Any
    error: Optional[str] = None

class FunctionCallingExecutor:
    """Executor สำหรับจัดการ function calling loop"""
    
    def __init__(self, api_key: str, max_iterations: int = 10):
        self.api_key = api_key
        self.max_iterations = max_iterations
        self.client = httpx.Client(
            base_url="https://api.holysheep.ai/v1",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json"
            }
        )
    
    def register_tool(
        self, 
        name: str, 
        description: str, 
        parameters: Dict[str, Any],
        handler: Callable
    ):
        """Register a tool with its handler function"""
        if not hasattr(self, 'tools'):
            self.tools = []
        self.tools.append({
            "type": "function",
            "function": {
                "name": name,
                "description": description,
                "parameters": parameters
            }
        })
        if not hasattr(self, 'handlers'):
            self.handlers = {}
        self.handlers[name] = handler
    
    def _execute_tool(self, tool_call: Dict[str, Any]) -> ToolResult:
        """Execute a single tool call"""
        try:
            name = tool_call["function"]["name"]
            arguments = json.loads(tool_call["function"]["arguments"])
            
            if hasattr(self, 'handlers') and name in self.handlers:
                result = self.handlers[name](**arguments)
                return ToolResult(
                    tool_call_id=tool_call["id"],
                    tool_name=name,
                    result=result
                )
            else:
                return ToolResult(
                    tool_call_id=tool_call["id"],
                    tool_name=name,
                    result=None,
                    error=f"Handler for tool '{name}' not found"
                )
        except Exception as e:
            return ToolResult(
                tool_call_id=tool_call.get("id", ""),
                tool_name=tool_call["function"]["name"],
                result=None,
                error=str(e)
            )
    
    def _build_assistant_message(
        self, 
        response: Dict[str, Any]
    ) -> Dict[str, Any]:
        """Build assistant message from response"""
        message = {
            "role": "assistant",
            "content": response["choices"][0]["message"].get("content", "")
        }
        
        # Handle tool_calls if present
        if "tool_calls" in response["choices"][0]["message"]:
            message["tool_calls"] = response["choices"][0]["message"]["tool_calls"]
        
        return message
    
    def _build_tool_results(
        self, 
        results: List[ToolResult]
    ) -> List[Dict[str, str]]:
        """Build tool result messages for next iteration"""
        messages = []
        for result in results:
            msg = {
                "role": "tool",
                "tool_call_id": result.tool_call_id,
                "name": result.tool_name,
                "content": json.dumps(result.result, ensure_ascii=False) if result.result is not None else f"Error: {result.error}"
            }
            messages.append(msg)
        return messages
    
    def execute(self, messages: List[Dict[str, Any]]) -> Dict[str, Any]:
        """
        Execute function calling loop until completion
        
        Returns:
            Final response with all tool calls resolved
        """
        conversation = messages.copy()
        
        for iteration in range(self.max_iterations):
            # Send request
            response = self.client.post(
                "/chat/completions",
                json={
                    "model": "gemini-2.0-flash-exp",
                    "messages": conversation,
                    "tools": getattr(self, 'tools', []),
                    "temperature": 0.7,
                    "max_tokens": 2048
                }
            )
            response.raise_for_status()
            result = response.json()
            
            # Check finish reason
            finish_reason = result["choices"][0]["finish_reason"]
            
            if finish_reason == "stop":
                # No more tool calls, return final response
                return result
            
            elif finish_reason == "tool_calls":
                # Extract and execute tool calls
                assistant_msg = self._build_assistant_message(result)
                conversation.append(assistant_msg)
                
                tool_calls = assistant_msg["tool_calls"]
                
                # Execute all tools
                results = []
                for tool_call in tool_calls:
                    tool_result = self._execute_tool(tool_call)
                    results.append(tool_result)
                    print(f"✓ Executed {tool_result.tool_name}: {'Success' if not tool_result.error else tool_result.error}")
                
                # Add tool results to conversation
                tool_messages = self._build_tool_results(results)
                conversation.extend(tool_messages)
                
            else:
                # Unknown finish reason, return current response
                return result
        
        raise RuntimeError(f"Max iterations ({self.max_iterations}) exceeded")


ตัวอย่างการใช้งาน

executor = FunctionCallingExecutor(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    max_iterations=10
)


Register tools

def get_weather(city: str, unit: str = "celsius") -> Dict[str, Any]:
    """Simulate weather API call"""
    return {
        "city": city,
        "temperature": 32 if unit == "celsius" else 89,
        "condition": " partly cloudy",
        "humidity": 75,
        "wind": "10 km/h"
    }

def calculate_route(origin: str, destination: str, mode: str = "driving") -> Dict[str, Any]:
    """Simulate routing API call"""
    return {
        "origin": origin,
        "destination": destination,
        "mode": mode,
        "duration_minutes": 25 if mode == "driving" else 45,
        "distance_km": 8.5
    }

executor.register_tool(
    name="get_weather",
    description="ดึงข้อมูลสภาพอากาศของเมืองที่ระบุ",
    parameters={
        "type": "object",
        "properties": {
            "city": {"type": "string"},
            "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
        },
        "required": ["city"]
    },
    handler=get_weather
)

executor.register_tool(
    name="calculate_route",
    description="คำนวณเส้นทางจากจุดเริ่มต้นไปยังจุดหมาย",
    parameters={
        "type": "object",
        "properties": {
            "origin": {"type": "string"},
            "destination": {"type": "string"},
            "mode": {"type": "string", "enum": ["driving", "walking", "transit"]}
        },
        "required": ["origin", "destination"]
    },
    handler=calculate_route
)


Execute

messages = [
    {"role": "user", "content": "บอกสภาพอากาศในกรุงเทพ และคำนวณเส้นทางจากสยามไปเยาวราชแบบเดินทางด้วยรถยนต์"}
]

result = executor.execute(messages)
print("Final Response:")
print(json.dumps(result, indent=2, ensure_ascii=False))

การควบคุม Concurrency และ Rate Limiting

สำหรับ production system ที่ต้องรองรับ request จำนวนมาก การจัดการ concurrency อย่างเหมาะสมเป็นสิ่งจำเป็น ต่อไปนี้คือ pattern ที่ผมใช้ในงานจริง 

"""
Concurrent Function Calling Executor with Rate Limiting
รองรับการประมวลผลแบบ parallel พร้อม rate limit control
"""
import httpx
import json
import asyncio
import time
from typing import List, Dict, Any, Callable, Optional
from dataclasses import dataclass, field
from collections import deque
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed

@dataclass
class RateLimiter:
    """Token bucket rate limiter implementation"""
    rate: float  # requests per second
    capacity: int
    tokens: float = field(init=False)
    last_update: float = field(init=False)
    lock: threading.Lock = field(default_factory=threading.Lock)
    
    def __post_init__(self):
        self.tokens = float(self.capacity)
        self.last_update = time.time()
    
    def acquire(self, tokens_needed: int = 1) -> float:
        """
        Acquire tokens, return wait time if necessary
        """
        with self.lock:
            now = time.time()
            elapsed = now - self.last_update
            self.tokens = min(
                self.capacity,
                self.tokens + elapsed * self.rate
            )
            self.last_update = now
            
            if self.tokens >= tokens_needed:
                self.tokens -= tokens_needed
                return 0.0
            else:
                wait_time = (tokens_needed - self.tokens) / self.rate
                return max(0, wait_time)

class AsyncFunctionCallingExecutor:
    """Async executor สำหรับ high-throughput function calling"""
    
    def

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง