LangGraph: Xây dựng Agent với Tool Calling và Bounded Execution để xác thực mã nguồn

Việc tự động sửa đổi mã nguồn đòi hỏi nhiều hơn một chuỗi prompt tuyến tính thông thường. Khi một agent can thiệp vào một repository, nó cần đọc các tệp, áp dụng thay đổi, chạy bộ kiểm thử (test suite), phân tích kết quả đầu ra và tự sửa lỗi nếu phát sinh hồi quy (regression). Tuy nhiên, các mô hình agent lặp—thường dựa trên mô hình ReAct (Reasoning and Acting)—sẽ tiềm ẩn rủi ro vận hành nếu không được kiểm soát. Một agent gặp lỗi phụ thuộc (dependency) hoặc test hỏng kéo dài có thể rơi vào các vòng lặp thực thi vô tận và tốn kém chi phí.

Dựa trên kiến trúc ReAct cho LangGraph do Google AI công bố, hướng dẫn này điều chỉnh các cơ chế state-graph cốt lõi thành một coding agent chuyên dụng, có giới hạn thực thi rõ ràng. Quá trình triển khai sẽ thực hiện sửa đổi mã nguồn, xác thực thay đổi bằng bộ test tự động và áp dụng ngân sách số lần lặp nghiêm ngặt.

---

Kiến trúc cốt lõi: State, Node và Conditional Edge

LangGraph mô hình hóa quy trình làm việc của agent dưới dạng đồ thị có hướng. Không giống như các đồ thị có hướng không chu trình (DAG) tĩnh, LangGraph hỗ trợ trực tiếp các chu trình (cycle)—yếu tố then chốt cho quá trình suy luận lặp của agent.

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

Đồ thị hoạt động dựa trên ba thành phần kiến trúc chính:

1. State: Cấu trúc dữ liệu có kiểu định sẵn tập trung, lưu trữ lịch sử tin nhắn, metadata của lượt chạy và bộ đếm bước rõ ràng.
2. Nodes: Các hàm Python nhận state hiện tại, thực thi các tính toán độc lập (gọi mô hình hoặc thực thi công cụ) và trả về các thuộc tính state đã cập nhật.
3. Conditional Edges: Các hàm quyết định kiểm tra state (chẳng hạn như có yêu cầu gọi công cụ hay không, hoặc đã vượt quá ngưỡng lặp tối đa chưa) để chuyển tiếp đến công cụ chạy (tool runner) hoặc kết thúc tại END.

---

Điều kiện tiên quyết và các thư viện phụ thuộc

Đảm bảo môi trường phát triển của bạn đã cài đặt đầy đủ các gói điều phối đồ thị và driver mô hình cần thiết:

pip install langgraph langchain-core langchain-openai

Để kết nối agent với các nhà cung cấp mô hình suy luận thông qua một gateway tương thích OpenAI, hãy cấu hình các biến môi trường của bạn. Sử dụng OneMux giúp đơn giản hóa việc quản lý API key tập trung và kiểm soát chi phí trên nhiều mô hình:

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

(Lưu ý: Luôn kiểm tra xem mô hình mục tiêu bạn chọn có hỗ trợ các tham số gọi công cụ có cấu trúc qua endpoint tương thích OpenAI hay không).

---

Bước 1: Định nghĩa State có giới hạn cho Agent

Trong các quy trình hội thoại tiêu chuẩn, MessagesState sẽ xử lý việc nối thêm lịch sử chat. Đối với các tác vụ lập trình, chúng ta định nghĩa rõ ràng AgentState bằng TypedDict của Python để theo dõi chuỗi tin nhắn cùng với một bộ đếm cụ thể (step_count) và ngưỡng tối đa (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 đảm bảo danh sách tin nhắn mới được nối thêm vào lịch sử hội thoại một cách tuần tự thay vì ghi đè lên các quan sát trước đó.

---

Bước 2: Triển khai các công cụ tương tác File System và kiểm thử

Agent cần các công cụ để kiểm tra tệp nguồn, cập nhật nội dung và chạy test của repository. Chúng ta đóng gói các thao tác này bằng decorator @tool của 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}

---

Bước 3: Thiết lập mô hình qua API hợp nhất

Chúng ta khởi tạo mô hình bằng ChatOpenAI, trỏ đến endpoint proxy hợp nhất của nhà cung cấp. Tham khảo OneMux Docs và Hướng dẫn Quickstart để đảm bảo tính tương thích của tham số khi gọi hàm (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)

---

Bước 4: Xây dựng các Graph Node và chốt chặn vòng lặp

Các node sẽ thực thi logic và tăng biến đếm giới hạn. Nếu agent lặp lại nhiều lần mà vẫn không sửa được test lỗi, logic tại edge sẽ phát hiện điều kiện step_count >= max_steps và chuyển hướng thẳng đến bước hoàn tất đồ thị.

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"

---

Bước 5: Ghép nối và biên dịch Graph

Kết hợp các thành phần lại với nhau bằng StateGraph và tiến hành biên dịch quy trình:

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()

---

Bước 6: Thực thi chỉnh sửa mã nguồn kèm xác thực bằng kiểm thử

Để chạy quy trình, hãy xác định tác vụ lập trình cùng với giới hạn thực thi nghiêm ngặt:

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.

---

Giới hạn và giải pháp củng cố khi triển khai Production

Khi triển khai các agent sửa đổi mã nguồn trên môi trường production, hãy cân nhắc các biện pháp bảo vệ sau:

• Môi trường thực thi Sandbox: Việc chạy các lệnh shell tùy ý như pytest trực tiếp trên máy cục bộ tiềm ẩn rủi ro phá vỡ môi trường hoặc thực thi script không an toàn. Hãy cô lập quá trình chạy test bên trong các container Docker dùng một lần (ephemeral) hoặc các sandbox bảo mật.
• Giới hạn Token của State: Đầu ra công cụ quá dài—chẳng hạn như stack trace nhiều trang—có thể nhanh chóng làm tràn context window. Cần chủ động cắt ngắn (truncate) thông báo kết quả test trước khi ghi vào lịch sử state.
• Kiểm soát chi phí mô hình: Các vòng lặp ReAct sẽ nhân số lượng token tiêu thụ lên nhiều lần. Hãy theo dõi mức sử dụng qua bảng điều khiển OneMux Pricing và đánh giá các mô hình suy luận chi phí thấp hơn tại OneMux Models để tối ưu chi phí trên từng tác vụ.

---

Nguồn tham khảo

• Google AI for Developers: ReAct agent from scratch with Gemini and LangGraph (Cập nhật ngày 2 tháng 9 năm 2026).