LangGraph - Nâng cấp "Tư duy vòng lặp" (Self-Correction) cho Jarvis

[AI Hunter]

Anh em code Agent bằng LangChain cũ thường gặp một vấn đề: Code chạy từ A đến Z, nếu giữa đường gặp lỗi hoặc kết quả không ưng ý, nó không biết quay đầu. Nó cứ thế đi tiếp và trả về kết quả rác.

Ví dụ: Bảo Jarvis viết code Python.

1. Nó viết code.
2. Code đó chạy bị lỗi Syntax.
3. Nó vẫn trả về cho bạn đoạn code lỗi đó.

Giải pháp: Chúng ta cần một vòng lặp (Loop).

• Jarvis viết code.
• Jarvis tự chạy thử (Execute).
• Nếu lỗi -> Jarvis tự đọc lỗi -> Tự sửa lại code -> Chạy lại.
• Đến khi nào hết lỗi -> Mới trả về cho User.

Đó chính là sức mạnh của LangGraph - Thư viện xây dựng State Machine (Máy trạng thái) cho AI.

1. Tư duy: StateGraph (Đồ thị trạng thái)​

Thay vì code if/else lằng nhằng, chúng ta sẽ định nghĩa một quy trình gồm các Node (Bước xử lý) và Edge (Đường đi).

• State: Bộ nhớ chung của quy trình (Lưu câu hỏi, lịch sử, kết quả tool...).
• Node: Các hành động (Ví dụ: Agent Node để suy nghĩ, Tool Node để search Google).
• Edge: Điều kiện chuyển tiếp (Ví dụ: Nếu Tool trả về kết quả rỗng -> Quay lại Agent để nghĩ cách khác).

2. Cài đặt​

pip install langgraph langchain_openai langchain_core

3. Code thực chiến: Agent biết tự sửa sai​

Chúng ta sẽ build một Agent đơn giản: Có khả năng search thông tin. Nhưng nếu search không ra, nó sẽ tự biết search lại bằng từ khóa khác (tối đa 3 lần).

import operator
from typing import Annotated, TypedDict, Union
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END

# 1. Định nghĩa State (Bộ nhớ của Graph)
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]
    loop_count: int # Đếm số lần lặp để tránh loop vô tận

# 2. Định nghĩa Tools (Công cụ)
@tool
def check_weather(city: str):
    """Tra cứu thời tiết của một thành phố."""
    if "hanoi" in city.lower():
        return "Hà Nội hôm nay mưa to."
    return "Không tìm thấy dữ liệu."

tools = [check_weather]
llm = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)

# 3. Định nghĩa Nodes (Các bước xử lý)

# Node 1: Agent suy nghĩ
def agent_node(state: AgentState):
    messages = state['messages']
    response = llm.invoke(messages)
    return {"messages": [response], "loop_count": state.get("loop_count", 0)}

# Node 2: Thực thi Tool
def tool_node(state: AgentState):
    last_message = state['messages'][-1]
    tool_calls = last_message.tool_calls
   
    results = []
    for t in tool_calls:
        # Giả lập gọi tool (Ở production sẽ gọi hàm thật)
        if t['name'] == 'check_weather':
            res = check_weather.invoke(t['args'])
            results.append(HumanMessage(content=f"Tool Output: {res}", name=t['name']))
           
    return {"messages": results, "loop_count": state['loop_count'] + 1}

# 4. Định nghĩa Logic điều hướng (Conditional Edge)
def should_continue(state: AgentState):
    last_message = state['messages'][-1]
   
    # Nếu AI không gọi tool nữa -> Kết thúc
    if not last_message.tool_calls:
        return "end"
   
    # Nếu lặp quá 3 lần mà vẫn chưa xong -> Buộc dừng
    if state['loop_count'] >= 3:
        return "end"
       
    # Nếu AI muốn gọi tool -> Sang node Tool
    return "continue"

# 5. Xây dựng Graph (Vẽ sơ đồ)
workflow = StateGraph(AgentState)

workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)

workflow.set_entry_point("agent") # Bắt đầu từ Agent

# Thêm đường đi
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "tools",
        "end": END
    }
)

# Tool chạy xong thì quay lại Agent để đọc kết quả
workflow.add_edge("tools", "agent")

# Compile thành ứng dụng chạy được
app = workflow.compile()

# 6. Chạy thử
inputs = {"messages": [HumanMessage(content="Thời tiết Sài Gòn thế nào?")], "loop_count": 0}

for output in app.stream(inputs):
    for key, value in output.items():
        print(f"Node '{key}' vừa chạy xong.")
        print("---")

4. Phân tích luồng chạy​

Khi bạn hỏi: "Thời tiết Sài Gòn thế nào?"

1. Agent Node: Gọi tool check_weather('Sài Gòn').
2. Tool Node: Tool trả về "Không tìm thấy dữ liệu" (Vì trong code mình chỉ cài logic cho Hà Nội).
3. Quay lại Agent Node: Agent đọc thấy tool bảo không tìm thấy.
   - Nếu là Agent cũ (Linear): Nó sẽ trả lời luôn "Tôi không biết".
   - Nếu là Agent LangGraph: Nó có thể tự suy luận: "À, hay là mình thử tìm 'TP.HCM' xem sao?".
4. Agent Node (Lần 2): Gọi tool check_weather('TP.HCM').

Đây chính là khả năng Self-Correction (Tự sửa). Bạn có thể cài đặt logic để Agent thử đi thử lại các phương án khác nhau cho đến khi thành công.

5. Ứng dụng thực tế: Code Assistant​

Ứng dụng mạnh nhất của LangGraph là làm Coding Agent:

1. Generate: Viết code.
2. Test: Chạy Unit Test.
3. Fix: Nếu Test fail -> Đọc lỗi -> Quay lại bước Generate sửa code.
4. Loop: Lặp lại đến khi Test pass thì thôi.

Jarvis của bạn sẽ trở thành một lập trình viên thực thụ, không chỉ biết viết code mà còn biết đảm bảo code chạy được!

Tổng kết​

LangGraph hơi khó nuốt lúc đầu vì tư duy Graph khác với tư duy tuần tự. Nhưng một khi đã nắm vững, bạn có thể xây dựng những quy trình nghiệp vụ cực kỳ phức tạp (ví dụ: Quy trình phê duyệt văn bản, Quy trình xử lý khiếu nại khách hàng...).