Local SE Agent Masterclass

from pathlib import Path
from typing import Optional

class FileAccessError(Exception):
    """Custom exception เมื่อเกิดข้อผิดพลาดในการเข้าถึงไฟล์หรือละเมิดความปลอดภัย"""
    pass

def read_file(filepath: str, workspace_root: Optional[str] = None, max_lines: int = 500) -> str:
    """
    เครื่องมือสำหรับเปิดอ่านเนื้อหาไฟล์ในโปรเจกต์อย่างปลอดภัย (Safe File System Reader)
    
    :param filepath: ที่อยู่ของไฟล์สัมพัทธ์ (เช่น 'sample_code/PaymentService.java')
    :param workspace_root: โฟลเดอร์รากที่อนุญาตให้อ่าน (ค่าเริ่มต้นคือโฟลเดอร์ปัจจุบันที่รันโปรแกรม)
    :param max_lines: จำนวนบรรทัดสูงสุดที่อนุญาตให้อ่าน ป้องกัน Token ล้น Context Window
    :return: เนื้อหาของไฟล์ในรูปข้อความสตริง
    """
    try:
        # 1. กำหนด Workspace Root Directory และแปลงเป็น Absolute Path
        root = Path(workspace_root).resolve() if workspace_root else Path.cwd().resolve()
        
        # 2. นำ filepath เป้าหมายมาต่อกับ root แล้วแปลงเป็น Absolute Path
        target_path = (root / filepath).resolve()
        
        # 3. [Security Guardrail]: ป้องกัน Path Traversal Attack
        # ตรวจสอบว่า target_path ต้องอยู่ภายใต้โฟลเดอร์ root เท่านั้น
        if not str(target_path).startswith(str(root)):
            raise FileAccessError(
                f"Security Violation: ไม่อนุญาตให้อ่านไฟล์นอกโฟลเดอร์โปรเจกต์ '{filepath}'"
            )
        
        # 4. ตรวจสอบว่ามีไฟล์อยู่จริงหรือไม่
        if not target_path.exists():
            raise FileAccessError(f"ไม่พบไฟล์เป้าหมายที่พิกัด: '{filepath}'")
            
        if not target_path.is_file():
            raise FileAccessError(f"พิกัด '{filepath}' เป็นโฟลเดอร์ ไม่ใช่ไฟล์ข้อความ")
            
        # 5. เปิดอ่านไฟล์ด้วยการระบุ encoding='utf-8' อย่างชัดเจน
        with open(target_path, "r", encoding="utf-8", errors="replace") as file:
            lines = []
            for idx, line in enumerate(file):
                if idx >= max_lines:
                    lines.append(f"\n[...ข้อความถูกตัดทอน เนื่องจากไฟล์มีความยาวเกิน {max_lines} บรรทัด...]")
                    break
                lines.append(line)
                
            return "".join(lines)
            
    except FileAccessError:
        raise
    except Exception as err:
        raise FileAccessError(f"เกิดข้อผิดพลาดขณะอ่านไฟล์ '{filepath}': {str(err)}")

package com.enterprise.payment;

public class PaymentService {
    private final double TAX_RATE = 0.07;

    public double calculateTotalWithTax(double amount) {
        if (amount <= 0) {
            throw new IllegalArgumentException("จำนวนเงินต้องมากกว่า 0");
        }
        return amount + (amount * TAX_RATE);
    }

    public boolean processTransaction(String accountId, double amount) {
        // จำลองการตัดเงินผ่าน Gateway
        return amount > 0 && accountId != null && !accountId.isEmpty();
    }
}

from core.client import OllamaLLMClient
from core.parser import StructuredOutputParser
from core.dispatcher import ToolDispatcher
from core.tools.file_reader import read_file, FileAccessError
import json

def main():
    print("🚀 เริ่มทดสอบ Day 4: File System Reader Tool บน Mac M3...")
    
    # 1. ติดตั้ง Dispatcher และจดทะเบียน File Reader เข้าสู่ระบบ
    dispatcher = ToolDispatcher()
    dispatcher.register("read_file", read_file)
    
    client = OllamaLLMClient(model_name="qwen2.5-coder:7b")
    if not client.check_health():
        print("❌ ข้อผิดพลาด: ไม่พบ Ollama Daemon หรือโมเดล qwen2.5-coder:7b")
        return

    # 2. System Prompt ระบุความสามารถของ Tool อ่านไฟล์ พร้อม JSON Schema
    system_prompt = (
        "You are an Autonomous Software Engineering Agent.\n"
        "You have access to tools to inspect local source code.\n"
        "When you need to read a file, respond ONLY with a JSON action object.\n\n"
        "Available Tools:\n"
        "1. read_file(filepath: string): Reads file content safely from the workspace.\n\n"
        "JSON Response Schema:\n"
        "{\n"
        '  "tool": "read_file",\n'
        '  "arguments": {\n'
        '    "filepath": "path/to/file"\n'
        '  }\n'
        "}"
    )

    user_query = "ช่วยตรวจสอบโค้ดในไฟล์ sample_code/PaymentService.java ให้หน่อยว่ามีเมธอดอะไรบ้าง"
    print(f"💬 User Prompt: {user_query}")

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

    # 3. ยิง Prompt รอบที่ 1: ให้โมเดลขอเรียก Tool
    print("\n📡 กำลังส่งโจทย์ให้โมเดลประมวลผล...")
    raw_response = client.chat(messages, temperature=0.1, json_mode=True)
    print(f"📥 โมเดลตัดสินใจส่ง JSON Action:\n{raw_response}")

    try:
        action = StructuredOutputParser.parse(raw_response)
        tool_name = action.get("tool")
        arguments = action.get("arguments", {})

        print(f"\n⚙️ Dispatcher กำลังสั่งรัน Tool '{tool_name}'...")
        # 4. Dispatcher รันฟังก์ชัน read_file จริงบนเครื่อง Mac
        file_content = dispatcher.dispatch(tool_name, arguments)
        print(f"✅ อ่านไฟล์สำเร็จ! (ความยาวโค้ด: {len(file_content)} ตัวอักษร)")

        # 5. ส่งผลลัพธ์จากไฟล์ (Observation) กลับไปให้โมเดลสรุปคำตอบรอบที่ 2
        messages.append({"role": "assistant", "content": raw_response})
        messages.append({
            "role": "user",
            "content": f"Observation from read_file:\n```java\n{file_content}\n```\nโปรดสรุปว่าในคลาสนี้มีเมธอดชื่ออะไรบ้างและทำหน้าที่อะไร"
        })

        print("\n🧠 โมเดลกำลังอ่านโค้ดจริงที่เปิดได้ และสังเคราะห์คำตอบ...")
        final_answer = client.chat(messages, temperature=0.2, json_mode=False)
        print("\n🎯 สรุปผลลัพธ์จาก AI Agent:")
        print("--------------------------------------------------")
        print(final_answer)
        print("--------------------------------------------------")
        print("🎉 สำเร็จ 100%! Agent ของคุณมีดวงตาเปิดอ่านซอร์สโค้ดจริงในเครื่องได้แล้ว!")

    except FileAccessError as err:
        print(f"❌ File Access Guardrail Blocked: {err}")
    except Exception as err:
        print(f"❌ Error: {err}")

if __name__ == "__main__":
    main()