Local SE Agent Masterclass

from typing import Callable, Dict, Any

class ToolExecutionError(Exception):
    """Custom exception สำหรับข้อผิดพลาดระหว่างรัน Tool จริง"""
    pass

class ToolDispatcher:
    """
    ศูนย์จดทะเบียนฟังก์ชันและตัวกลางสั่งรันคำสั่งจาก LLM (Function Dispatcher)
    """
    def __init__(self):
        # เก็บฟังก์ชัน Python จริงในรูป: {"tool_name": callable_function}
        self.registry: Dict[str, Callable[..., Any]] = {}

    def register(self, name: str, func: Callable[..., Any]) -> None:
        """
        จดทะเบียนฟังก์ชันเข้าสู่ระบบ
        :param name: ชื่อ Tool ที่โมเดลจะใช้อ้างอิง (เช่น 'get_system_info')
        :param func: ฟังก์ชัน Python จริง
        """
        self.registry[name] = func

    def dispatch(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
        """
        เรียกใช้ฟังก์ชันตามที่โมเดลขอมา พร้อมระบบดักจับ Error
        :param tool_name: ชื่อฟังก์ชันที่ขอรัน
        :param arguments: พารามิเตอร์ที่แกะมาจาก JSON ของโมเดล
        """
        if tool_name not in self.registry:
            raise ToolExecutionError(f"ไม่พบ Tool ชื่อ '{tool_name}' ในระบบ Registry")

        func = self.registry[tool_name]
        try:
            # Unpack dictionary arguments เข้าไปในฟังก์ชัน Python จริง
            return func(**arguments)
        except TypeError as err:
            raise ToolExecutionError(
                f"โมเดลส่งพารามิเตอร์ให้ฟังก์ชัน '{tool_name}' ไม่ถูกต้อง: {str(err)}"
            )
        except Exception as err:
            raise ToolExecutionError(
                f"เกิดข้อผิดพลาดขณะรันฟังก์ชัน '{tool_name}': {str(err)}"
            )

from core.client import OllamaLLMClient
from core.parser import StructuredOutputParser
from core.dispatcher import ToolDispatcher, ToolExecutionError
import platform

# 1. ประกาศฟังก์ชันตัวอย่างที่โปรแกรมเตรียมไว้ให้ Agent ใช้งาน
def get_system_specs() -> str:
    """ดึงข้อมูลสถาปัตยกรรมของเครื่อง Mac"""
    return f"OS: {platform.system()} | Machine: {platform.machine()} | Architecture: Apple Silicon"

def calculate_metric(a: float, b: float, operation: str) -> float:
    """คำนวณตัวเลขทางคณิตศาสตร์อย่างแม่นยำ"""
    if operation == "multiply":
        return a * b
    elif operation == "add":
        return a + b
    elif operation == "divide":
        return a / b if b != 0 else 0.0
    return 0.0

def main():
    print("🚀 เริ่มทดสอบ Day 3: Tool Calling & Function Dispatcher บน Mac M3...")
    
    # 2. ติดตั้ง Dispatcher และจดทะเบียนฟังก์ชัน
    dispatcher = ToolDispatcher()
    dispatcher.register("get_system_specs", get_system_specs)
    dispatcher.register("calculate_metric", calculate_metric)

    client = OllamaLLMClient(model_name="qwen2.5-coder:7b")
    if not client.check_health():
        print("❌ ข้อผิดพลาด: ไม่พบ Ollama Daemon หรือโมเดล qwen2.5-coder:7b")
        return

    # 3. กำหนด System Prompt บอกว่าเรามี Tool อะไรบ้าง และบังคับตอบเป็น JSON สั่งงาน
    system_prompt = (
        "You are an AI Agent with access to tools. If you need external data or calculations,\n"
        "respond ONLY in JSON format to request tool execution.\n\n"
        "Available Tools:\n"
        "1. get_system_specs(): No parameters. Returns OS and hardware information.\n"
        "2. calculate_metric(a: number, b: number, operation: string): Operations are 'add', 'multiply', 'divide'.\n\n"
        "JSON Response Schema:\n"
        "{\n"
        '  "tool": string,\n'
        '  "arguments": object\n'
        "}"
    )

    user_query = "ช่วยคำนวณพื้นที่รูปสี่เหลี่ยมที่มีความกว้าง 14.5 เมตร และความยาว 32.0 เมตร ให้หน่อย"
    print(f"💬 User Prompt: {user_query}")

    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_query}
    ]

    print("\n📡 กำลังส่ง Prompt ให้โมเดลเลือก Tool...")
    raw_response = client.chat(messages, temperature=0.1, json_mode=True)
    print(f"📥 โมเดลตอบกลับมาเป็น JSON:\n{raw_response}")

    # 4. แปลงคำตอบและส่งให้ Dispatcher รันฟังก์ชันจริง
    try:
        parsed_action = StructuredOutputParser.parse(raw_response)
        tool_name = parsed_action.get("tool")
        arguments = parsed_action.get("arguments", {})

        print(f"\n⚙️ Dispatcher กำลังประมวลผล...")
        print(f"  👉 สั่งรันฟังก์ชัน: {tool_name}")
        print(f"  👉 ป้อนพารามิเตอร์: {arguments}")

        # รันคำสั่งจริงใน Python
        result = dispatcher.dispatch(tool_name, arguments)
        print(f"\n🎯 ผลลัพธ์จากการคำนวณจริงในเครื่อง Mac: {result}")
        print(f"🎉 สำเร็จ 100%! เราได้ติดแขนติดขาให้ LLM สั่งงานโค้ดภายนอกได้แล้ว!")

    except ToolExecutionError as err:
        print(f"❌ Dispatcher Error: {err}")
    except Exception as err:
        print(f"❌ Error: {err}")

if __name__ == "__main__":
    main()