LangGraph: สร้าง Agent ด้วย Tool Calling และ Bounded Execution เพื่อตรวจสอบโค้ดอัตโนมัติ

การปรับแต่งและแก้ไขโค้ดโดยอัตโนมัตินั้น จำเป็นต้องใช้กระบวนการที่ซับซ้อนกว่า prompt chain แบบเส้นตรงทั่วไป เมื่อ agent ต้องเข้ามาแก้ไขโค้ดใน repository ระบบจะต้องอ่านไฟล์, ดำเนินการแก้ไข, รันชุดทดสอบ, ตรวจสอบผลลัพธ์ และแก้ไขข้อผิดพลาดด้วยตนเองเมื่อพบ regression อย่างไรก็ตาม รูปแบบ agent ที่ทำงานเป็นรอบวนซ้ำ (cyclic agent patterns)—ซึ่งมักพัฒนาบนสถาปัตยกรรม ReAct (Reasoning and Acting)—อาจก่อให้เกิดความเสี่ยงในการดำเนินงานหากไม่มีการควบคุมอย่างเหมาะสม เพราะ agent ที่เจอปัญหา dependency เสียหายหรือการทดสอบล้มเหลวซ้ำๆ อาจหลุดเข้าไปในลูปการทำงานที่ไม่มีวันสิ้นสุดและสิ้นเปลืองค่าใช้จ่ายมหาศาล

คู่มือนี้ได้นำแนวคิดจาก สถาปัตยกรรม ReAct สำหรับ LangGraph ที่จัดทำโดย Google AI มาปรับใช้เพื่อสร้าง coding agent ที่มีขอบเขตการทำงาน (bounded execution) อย่างรัดกุม โดยระบบจะดำเนินการแก้ไขโค้ด ตรวจสอบความถูกต้องด้วยชุดทดสอบอัตโนมัติ และจำกัดจำนวนรอบการทำงานอย่างเคร่งครัด

---

สถาปัตยกรรมหลัก: State, Nodes และ Conditional Edges

LangGraph จำลองขั้นตอนการทำงานของ agent ในรูปแบบ directed graph ซึ่งต่างจาก directed acyclic graph (DAG) แบบคงที่ทั่วไป เนื่องจาก LangGraph รองรับการทำงานแบบวนรอบ (cycles) ได้ตั้งแต่ต้น ซึ่งเป็นหัวใจสำคัญของกระบวนการคิดและให้เหตุผลซ้ำของ agent

          [Entrypoint]
               │
               ▼
        ┌─────────────┐
        │  call_model │◄────────────┐
        └──────┬──────┘             │
               │                    │
     [should_continue]              │
        /             \             │
   (has tools)     (done / max)     │
      /                 \           │
     ▼                   ▼          │
┌───────────┐         ┌─────┐       │
│ call_tool │         │ END │       │
└─────┬─────┘         └─────┘       │
      │                             │
      └─────────────────────────────┘

โครงสร้างของ graph ประกอบด้วย 3 องค์ประกอบหลักทางสถาปัตยกรรม ได้แก่:

1. State: โครงสร้างข้อมูลแบบ typed ที่รวมศูนย์ไว้สำหรับจัดเก็บประวัติข้อความ (message history), รัน metadata และตัวนับขั้นตอน (step counters) อย่างชัดเจน
2. Nodes: ฟังก์ชันภาษา Python ทั่วไปที่รับ state ปัจจุบันเข้ามา ดำเนินการประมวลผลแยกเฉพาะส่วน (เช่น การเรียกใช้โมเดลหรือการเรียกใช้เครื่องมือ) และส่งคืนค่า attribute ของ state ที่อัปเดตแล้ว
3. Conditional Edges: ฟังก์ชันตัดสินใจที่คอยตรวจสอบ state (เช่น การตรวจว่ามีการเรียกใช้เครื่องมือหรือไม่ หรือจำนวนรอบเกินขีดจำกัดที่ตั้งไว้หรือไม่) เพื่อเลือกว่าจะส่งต่อไปยังตัวรันเครื่องมือ หรือสิ้นสุดการทำงานที่ END

---

ข้อกำหนดเบื้องต้นและ Dependencies

ตรวจสอบให้แน่ใจว่าสภาพแวดล้อมการพัฒนาของคุณติดตั้งแพ็กเกจสำหรับการจัดการ graph และตัวขับเคลื่อนโมเดลที่จำเป็นเรียบร้อยแล้ว:

pip install langgraph langchain-core langchain-openai

สำหรับการเชื่อมต่อ agent เข้ากับผู้ให้บริการ inference ผ่าน OpenAI-compatible gateway ให้กำหนดค่า environment variables ให้พร้อม ทั้งนี้ การใช้งาน OneMux จะช่วยให้การจัดการ API key จากศูนย์กลางและการควบคุมงบประมาณระหว่างหลายโมเดลเป็นเรื่องง่ายขึ้น:

export ONEMUX_API_KEY="your_onemux_api_key"
export ONEMUX_BASE_URL="https://api.onemux.net/v1"

(หมายเหตุ: ควรตรวจสอบเสมอว่าโมเดลเป้าหมายที่คุณเลือก รองรับพารามิเตอร์การเรียกใช้ structured tool-calling ผ่าน OpenAI-compatible endpoints)

---

ขั้นตอนที่ 1: การกำหนด Bounded Agent State

ในเวิร์กโฟลว์การสนทนาทั่วไป MessagesState จะทำหน้าที่จัดการการต่อท้ายประวัติการสนทนา แต่สำหรับงานด้านการเขียนโค้ด เราจะกำหนด AgentState ขึ้นมาอย่างชัดเจนโดยใช้ TypedDict ของ Python เพื่อติดตามลำดับข้อความควบคู่ไปกับตัวนับ (step_count) และขีดจำกัดสูงสุด (max_steps)

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    step_count: int
    max_steps: int

ตัว reducer อย่าง add_messages จะช่วยให้แน่ใจว่ารายการข้อความใหม่จะถูกต่อท้ายประวัติการสนทนาอย่างถูกต้อง โดยไม่เขียนทับข้อความหรือผลการสังเกตก่อนหน้า

---

ขั้นตอนที่ 2: การพัฒนาเครื่องมือสำหรับระบบไฟล์และการตรวจสอบ

Agent จำเป็นต้องมีเครื่องมือสำหรับอ่านไฟล์ต้นฉบับ, ดำเนินการแก้ไข และรันชุดทดสอบของ repository เราจะรวมการทำงานเหล่านี้ไว้ภายใต้ decorator @tool ของ LangChain

import subprocess
from langchain_core.tools import tool

@tool
def read_file(filepath: str) -> str:
    """Reads contents of a file from the repository."""
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            return f.read()
    except Exception as e:
        return f"Error reading {filepath}: {str(e)}"

@tool
def write_file(filepath: str, content: str) -> str:
    """Overwrites a file with modified source code."""
    try:
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(content)
        return f"Successfully updated {filepath}"
    except Exception as e:
        return f"Error writing {filepath}: {str(e)}"

@tool
def run_test_suite(test_target: str) -> str:
    """Runs pytest on a specific file or test target."""
    try:
        result = subprocess.run(
            ["pytest", test_target],
            capture_output=True,
            text=True,
            timeout=30
        )
        return f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}\nExit Code: {result.returncode}"
    except subprocess.TimeoutExpired:
        return "Error: Test execution timed out after 30 seconds."
    except Exception as e:
        return f"Error executing tests: {str(e)}"

tools = [read_file, write_file, run_test_suite]
tools_by_name = {t.name: t for t in tools}

---

ขั้นตอนที่ 3: การตั้งค่าโมเดลผ่าน Unified API

เราจะกำหนดค่าเริ่มต้นของโมเดลด้วย ChatOpenAI โดยชี้ไปยัง proxy endpoint กลางของผู้ให้บริการ สามารถศึกษารายละเอียดเพิ่มเติมได้ที่ OneMux Docs และ Quickstart Guide เพื่อตรวจสอบความเข้ากันได้ของพารามิเตอร์สำหรับการเรียกใช้ function calling

import os
from langchain_openai import ChatOpenAI

api_key = os.getenv("ONEMUX_API_KEY")
base_url = os.getenv("ONEMUX_BASE_URL", "https://api.onemux.net/v1")

llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.0,
    api_key=api_key,
    base_url=base_url
)

# Bind tools to the model instance
model = llm.bind_tools(tools)

---

ขั้นตอนที่ 4: การสร้าง Graph Nodes และระบบป้องกันลูป (Loop Guards)

Nodes จะทำหน้าที่ประมวลผลลอจิกและเพิ่มจำนวนรอบการทำงาน หาก agent ทำงานวนซ้ำไปเรื่อยๆ โดยไม่สามารถแก้ปัญหาการทดสอบที่ผิดพลาดได้ ลอจิกของ edge จะตรวจพบเงื่อนไข step_count >= max_steps และเปลี่ยนเส้นทางการทำงานไปยังจุดสิ้นสุดของ graph ทันที

from langchain_core.messages import ToolMessage, AIMessage
from langchain_core.runnables import RunnableConfig

def call_model(state: AgentState, config: RunnableConfig):
    """Calls the model and increments the iteration step count."""
    response = model.invoke(state["messages"], config)
    current_steps = state.get("step_count", 0) + 1
    return {
        "messages": [response],
        "step_count": current_steps
    }

def call_tool(state: AgentState):
    """Executes tool calls generated by the most recent model response."""
    last_message = state["messages"][-1]
    outputs = []
    
    for tool_call in getattr(last_message, "tool_calls", []):
        tool_inst = tools_by_name.get(tool_call["name"])
        if tool_inst:
            result = tool_inst.invoke(tool_call["args"])
        else:
            result = f"Error: Tool {tool_call['name']} not found."
            
        outputs.append(
            ToolMessage(
                content=str(result),
                name=tool_call["name"],
                tool_call_id=tool_call["id"]
            )
        )
    return {"messages": outputs}

def should_continue(state: AgentState):
    """Evaluates tool requests and halts execution if boundaries are crossed."""
    messages = state["messages"]
    last_message = messages[-1]
    step_count = state.get("step_count", 0)
    max_steps = state.get("max_steps", 5)

    # Boundary check: Halt if loop budget exceeded
    if step_count >= max_steps:
        return "end"

    # Continue if the model requested another tool call
    if getattr(last_message, "tool_calls", None):
        return "continue"
        
    return "end"

---

ขั้นตอนที่ 5: การประกอบและ Compile Graph

นำองค์ประกอบต่างๆ มาประกอบเข้าด้วยกันโดยใช้ StateGraph และทำการ compile ขั้นตอนการทำงาน:

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)

# Register Nodes
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tool)

# Set Entrypoint
workflow.set_entry_point("agent")

# Define Conditional and Static Edges
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "tools",
        "end": END
    }
)
workflow.add_edge("tools", "agent")

# Compile Graph
app = workflow.compile()

---

ขั้นตอนที่ 6: การแก้ไขโค้ดพร้อมตรวจสอบด้วยชุดทดสอบ

สำหรับการรันเวิร์กโฟลว์ ให้กำหนดงานเขียนโค้ดพร้อมทั้งตั้งขอบเขตการทำงานที่เข้มงวด:

initial_input = {
    "messages": [(
        "user",
        "Fix the bug in utils/math_lib.py where divide_numbers fails on zero values. "
        "Verify your fix by running the test suite on tests/test_math_lib.py. "
        "Only stop once all tests pass."
    )],
    "step_count": 0,
    "max_steps": 6
}

for event in app.stream(initial_input, stream_mode="values"):
    latest = event["messages"][-1]
    if isinstance(latest, AIMessage) and latest.tool_calls:
        print(f"[Action] Invoking tools: {[t['name'] for t in latest.tool_calls]}")
    elif isinstance(latest, ToolMessage):
        print(f"[Observation] {latest.name} -> {latest.content[:100]}...")
    else:
        print(f"[Agent] {latest.content}")

[Action] Invoking tools: ['read_file']
[Observation] read_file -> def divide_numbers(a, b):
    return a / b...
[Action] Invoking tools: ['write_file']
[Observation] write_file -> Successfully updated utils/math_lib.py...
[Action] Invoking tools: ['run_test_suite']
[Observation] run_test_suite -> STDOUT:
============================= test session starts ==============================
... 1 passed in 0.02s...
[Agent] The divide_numbers function has been updated to handle zero division safely, and all tests in tests/test_math_lib.py pass.

---

ข้อจำกัดและการเสริมความปลอดภัยสำหรับ Production

เมื่อต้องการนำ agent สำหรับแก้ไขโค้ดไปติดตั้งใช้งานบนสภาพแวดล้อมจริง (Production) ควรพิจารณามาตรการป้องกันความปลอดภัยดังต่อไปนี้:

• Sandboxing Execution: การรันคำสั่งเชลล์ใดๆ เช่น pytest บนเครื่องโดยตรงอาจเสี่ยงต่อความเสียหายของสภาพแวดล้อม หรือการรันสคริปต์ที่ไม่ปลอดภัย จึงควรแยกการรันการทดสอบไว้ใน Docker container แบบชั่วคราวหรือ sandbox ที่มีความปลอดภัยสูง
• State Token Limits: ผลลัพธ์จากการเรียกใช้เครื่องมือที่มีขนาดยาว—เช่น stack trace หลายหน้า—อาจทำให้ context window เต็มอย่างรวดเร็ว จึงควรตัดทอน (truncate) ข้อความผลการทดสอบก่อนบันทึกลงในประวัติของ state
• Managing Model Costs: ลูปการทำงานแบบ ReAct ที่มีการวนซ้ำจะเพิ่มการใช้โทเค็นอย่างทวีคูณ คุณสามารถติดตามการใช้งานผ่านแดชบอร์ด OneMux Pricing และประเมินการเลือกใช้โมเดลการให้เหตุผลที่คุ้มค่ากว่าได้จาก OneMux Models เพื่อควบคุมต้นทุนต่อภารกิจให้เหมาะสมที่สุด

---

แหล่งที่มาอ้างอิง

• Google AI for Developers: ReAct agent from scratch with Gemini and LangGraph (อัปเดตเมื่อ 2 กันยายน 2026)