Local SE Agent Masterclass

# 1.1 ไฟล์ memory/__init__.py
"""
Package บริหารจัดการความจำและ Context Window ของ Agent
"""
from memory.context import SlidingWindowContext

__all__ = ["SlidingWindowContext"]

# -------------------------------------------------------------

# 1.2 ไฟล์ memory/context.py
from typing import List, Dict, Any

class SlidingWindowContext:
    """
    ตัวจัดการ Context Window แบบ Sliding Window พร้อมกลไก Anchor Pinning
    สำหรับควบคุมปริมาณ Token ไม่ให้บวมพองบน Apple Silicon Mac 16GB
    """
    def __init__(self, keep_last_turns: int = 6):
        """
        :param keep_last_turns: จำนวนข้อความล่าสุดที่ต้องการรักษาไว้ (ค่าเริ่มต้น: 6 ข้อความ)
        """
        self.keep_last_turns = keep_last_turns

    def trim(self, messages: List[Dict[str, str]]) -> List[Dict[str, str]]:
        """
        ตัดทอนประวัติข้อความให้กระชับ โดยยังคงรักษา System Prompt บรรทัดแรกไว้เสมอ 100%
        
        :param messages: ลิสต์ของข้อความทั้งหมด เช่น [{'role': 'system', ...}, {'role': 'user', ...}]
        :return: ลิสต์ข้อความที่ถูกตัดทอนแล้ว ปลอดภัยต่อ Context Window
        """
        # หากประวัติข้อความยังมีขนาดไม่เกินงบประมาณ (System Prompt 1 อัน + จำนวนรอบที่กำหนด) ไม่ต้องตัด
        if len(messages) <= self.keep_last_turns + 1:
            return messages

        # 1. [Anchor Pinning]: แยก System Prompt ตัวแรกสุดออกมาเสมอ
        system_prompt = messages[0]

        # 2. [Sliding Window]: ดึงเฉพาะ N ข้อความล่าสุดที่อยู่ท้ายลิสต์
        recent_messages = messages[-self.keep_last_turns:]

        # 3. ตรวจสอบความสมบูรณ์ของประวัติ: ถ้าข้อความแรกของ window ลอยมาเป็น tool observation
        # เพื่อป้องกันไม่ให้โมเดลสับสนว่า observation นี้เป็นของ tool อะไร เราจะคงคู่บริบทไว้
        pruned_result = [system_prompt] + recent_messages
        return pruned_result

    def estimate_tokens(self, messages: List[Dict[str, str]]) -> int:
        """
        คำนวณประเมินจำนวน Token คร่าวๆ (heuristic: 1 token ~ 4 ตัวอักษรภาษาอังกฤษ หรือ 1.5 ตัวอักษรไทย)
        ช่วยให้ประเมินการใช้หน่วยความจำได้โดยไม่ต้องลงไลบรารี tokenizer เพิ่ม
        """
        total_chars = sum(len(msg.get("content", "")) for msg in messages)
        # ประมาณการแบบปลอดภัย (Conservative Estimate)
        return int(total_chars / 3.2)

import json
from typing import List, Dict, Any, Optional
from core.client import OllamaLLMClient
from core.parser import StructuredOutputParser, JSONDecodeError
from core.dispatcher import ToolDispatcher, ToolExecutionError
from memory.context import SlidingWindowContext

class MaxIterationsReachedError(Exception):
    """Exception เมื่อลูปทำงานเกินขีดจำกัดสูงสุดที่ตั้งไว้ (Guardrail)"""
    pass

class ReActAgentEngine:
    """
    แกนกลางควบคุมลูป ReAct (Reason ➔ Act ➔ Observe) แบบอัตโนมัติ
    [อัปเกรด Day 8]: มี SlidingWindowContext ควบคุมไม่ให้ Context บวมพอง
    """
    def __init__(
        self,
        client: OllamaLLMClient,
        dispatcher: ToolDispatcher,
        max_iterations: int = 6,
        memory_window: int = 6
    ):
        self.client = client
        self.dispatcher = dispatcher
        self.max_iterations = max_iterations
        # ติดตั้ง Memory Manager ประจำตัว Engine
        self.memory = SlidingWindowContext(keep_last_turns=memory_window)
        self.messages: List[Dict[str, str]] = []

    def run(self, user_prompt: str, system_prompt: str) -> str:
        """
        เริ่มลูป ReAct อัตโนมัติ: ควบคุมงบประมาณ Token ทุกรอบก่อนส่งเข้าโมเดล
        """
        self.messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ]

        for step in range(1, self.max_iterations + 1):
            # [จุดสำคัญ Day 8]: รูดหน้าต่างตัดทอนข้อความก่อนยิงเรียก Ollama ทุกครั้ง
            active_context = self.memory.trim(self.messages)
            est_tokens = self.memory.estimate_tokens(active_context)

            print(f"\n🔄 [Step {step}/{self.max_iterations}] Agent ประมวลผล (Messages: {len(active_context)} | ~{est_tokens} Tokens)...")

            # ยิงข้อความที่ผ่านการ Trim แล้วเข้า Ollama
            raw_response = self.client.chat(active_context, temperature=0.1, json_mode=True)

            try:
                decision = StructuredOutputParser.parse(raw_response)
            except JSONDecodeError as err:
                print(f"⚠️ โมเดลส่ง JSON ไม่สมบูรณ์: {err}")
                self.messages.append({"role": "assistant", "content": raw_response})
                self.messages.append({
                    "role": "user",
                    "content": "Error: Your previous response was not valid JSON. Please respond strictly in the defined JSON schema."
                })
                continue

            thought = decision.get("thought", "ไม่มี Thought ระบุ")
            print(f"💡 Thought: {thought}")

            # 1. ตรวจสอบเงื่อนไขจบงาน
            if "final_answer" in decision:
                final_answer = decision["final_answer"]
                print(f"🎯 Final Answer: {final_answer}")
                return final_answer

            # 2. หากโมเดลต้องการเรียก Tool
            if "tool" in decision:
                tool_name = decision.get("tool")
                arguments = decision.get("arguments", {})

                print(f"🛠️ Action: เรียกใช้ Tool '{tool_name}' ด้วยค่า: {arguments}")

                try:
                    observation = self.dispatcher.dispatch(tool_name, arguments)
                except ToolExecutionError as err:
                    observation = f"Tool Execution Error: {str(err)}"
                except Exception as err:
                    observation = f"System Error while executing tool: {str(err)}"

                obs_summary = str(observation).strip()
                if len(obs_summary) > 200:
                    print(f"👁️ Observation: {obs_summary[:200]}... [ตัดทอน]")
                else:
                    print(f"👁️ Observation:\n{obs_summary}")

                # บันทึกประวัติรอบนี้เข้า self.messages สะสมไว้
                self.messages.append({"role": "assistant", "content": raw_response})
                self.messages.append({
                    "role": "user",
                    "content": f"Observation from {tool_name}:\n{observation}"
                })
            else:
                self.messages.append({"role": "assistant", "content": raw_response})
                self.messages.append({
                    "role": "user",
                    "content": "Invalid State: You must provide either a 'tool' or a 'final_answer'."
                })

        raise MaxIterationsReachedError(
            f"Agent reached maximum iterations limit ({self.max_iterations}) without completing the task."
        )

from memory.context import SlidingWindowContext

def main():
    print("🚀 เริ่มทดสอบ Day 8: Sliding Window Memory Management บน Mac M3...")

    # ตั้งค่างบประมาณ: เก็บ System Prompt + ข้อความ 6 อันล่าสุด
    memory_manager = SlidingWindowContext(keep_last_turns=6)

    # 1. จำลองประวัติบทสนทนาขนาดยาว 13 ข้อความ (System 1 + สนทนาไปมา 12 ข้อความ)
    simulated_messages = [
        {"role": "system", "content": "You are a Senior Autonomous SE Agent."}, # Index 0: Anchor
        {"role": "user", "content": "รอบที่ 1: ช่วยตรวจไฟล์ A หน่อย"},
        {"role": "assistant", "content": "รอบที่ 1: กำลังอ่านไฟล์ A..."},
        {"role": "user", "content": "รอบที่ 2: ช่วยตรวจไฟล์ B หน่อย"},
        {"role": "assistant", "content": "รอบที่ 2: กำลังอ่านไฟล์ B..."},
        {"role": "user", "content": "รอบที่ 3: ช่วยรันเทสรอบที่ 1"},
        {"role": "assistant", "content": "รอบที่ 3: เทสล้มเหลวที่จุด X"},
        {"role": "user", "content": "รอบที่ 4: ช่วยแก้โค้ดที่จุด X"},
        {"role": "assistant", "content": "รอบที่ 4: แก้ไขโค้ดเรียบร้อย"},
        {"role": "user", "content": "รอบที่ 5: ช่วยรันเทสรอบที่ 2"},
        {"role": "assistant", "content": "รอบที่ 5: เทสผ่านเรียบร้อย"},
        {"role": "user", "content": "รอบที่ 6: ช่วยสรุปผลลัพธ์ให้หน่อย"},
        {"role": "assistant", "content": "รอบที่ 6: ผลลัพธ์สมบูรณ์ 100%"}
    ]

    print(f"📊 จำนวนข้อความดั้งเดิมก่อนตัด: {len(simulated_messages)} ข้อความ")
    print(f"📈 ประมาณการ Token ก่อนตัด: ~{memory_manager.estimate_tokens(simulated_messages)} Tokens")

    # 2. สั่ง Trim ข้อความ
    trimmed_messages = memory_manager.trim(simulated_messages)

    print("\n" + "=" * 60)
    print(f"✂️ ผลลัพธ์หลังผ่าน SlidingWindowContext.trim(): {len(trimmed_messages)} ข้อความ")
    print(f"📉 ประมาณการ Token หลังตัด: ~{memory_manager.estimate_tokens(trimmed_messages)} Tokens (ประหยัดแรม Mac M3)")
    print("=" * 60)

    # 3. ตรวจสอบว่า Anchor ยังอยู่ครบไหม
    assert trimmed_messages[0]["role"] == "system", "❌ ผิดพลาด: System Prompt หลุดหาย!"
    assert trimmed_messages[0]["content"] == "You are a Senior Autonomous SE Agent.", "❌ เนื้อหา System Prompt เพี้ยน!"
    assert len(trimmed_messages) == 7, f"❌ คาดหวัง 7 ข้อความ (System 1 + 6 ล่าสุด) แต่ได้ {len(trimmed_messages)}"

    print("✅ ตรวจสอบข้อความที่ถูกเก็บไว้ใน Context Window:")
    for idx, msg in enumerate(trimmed_messages):
        prefix = "📌 [ANCHOR]" if idx == 0 else f"🪟 [WINDOW {idx}]"
        print(f"  {prefix} ({msg['role']}): {msg['content'][:50]}")

    print("\n🎉 ผ่านการทดสอบ 100%! ข้อความเก่ารอบที่ 1-3 ถูก Evict ทิ้งอย่างสะอาดเอี่ยม ป้องกัน Token ระเบิดแรมแน่นอน!")

if __name__ == "__main__":
    main()