Xây dựng AI Agent với LangGraph: Giải bài toán Workflow phức tạp và tự sửa lỗi

Artificial Intelligence tutorial - IT technology blog
Artificial Intelligence tutorial - IT technology blog

Tại sao LangChain là chưa đủ và LangGraph xuất hiện đúng lúc?

Nếu từng xây dựng chatbot với LangChain, bạn sẽ thấy nó hoạt động rất tốt cho các tác vụ tuần tự. Tuy nhiên, khi đối mặt với bài toán cần AI tự sửa code hay lặp lại quy trình kiểm thử, các “chuỗi” (chains) truyền thống bắt đầu lộ rõ nhược điểm. Các chain này thường đi theo đường thẳng (DAG), nghĩa là chỉ đi từ A đến B rồi kết thúc mà không thể quay lại bước trước để sửa sai.

AI Agent trong thực tế cần khả năng suy nghĩ, thử nghiệm và lặp lại (loops). Đó là lý do mình chuyển sang dùng LangGraph. Thư viện này cho phép chúng ta tạo ra các workflow dạng biểu đồ có vòng lặp (cycles). Nhờ đó, việc kiểm soát trạng thái (state) của Agent trở nên chặt chẽ và minh bạch hơn hẳn.

Mình đã áp dụng LangGraph để xử lý các pipeline dữ liệu lớn tại công ty. Kết quả cho thấy hệ thống hoạt động ổn định hơn, cho phép lập trình viên can thiệp trực tiếp vào từng bước nhảy của AI. Đây là điều mà các Agent mặc định trước đây rất khó thực hiện một cách mượt mà.

Chuẩn bị môi trường cài đặt

Trước tiên, hãy cài đặt các thư viện cần thiết. Mình khuyên bạn nên dùng môi trường ảo (venv) để quản lý version tốt hơn, tránh xung đột với các dự án cũ.

pip install -U langgraph langchain_openai langchain_core

Để debug hiệu quả, bạn nên đăng ký LangSmith. Công cụ này giúp theo dõi chi tiết từng token và thời gian phản hồi của LLM. Hãy cấu hình các biến môi trường như sau:

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY="your-api-key"
export OPENAI_API_KEY="sk-..."

Cấu hình AI Agent với State và Node

Trong LangGraph, bạn chỉ cần tập trung vào ba khái niệm: State (Trạng thái), Nodes (Nút xử lý) và Edges (Luồng di chuyển).

1. Định nghĩa State (Trạng thái)

State là nơi lưu trữ toàn bộ dữ liệu mà Agent đang xử lý. Mỗi Node khi chạy xong sẽ cập nhật thông tin vào đây. Sử dụng TypedDict giúp cấu trúc dữ liệu luôn rõ ràng và dễ bảo trì.

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    # add_messages giúp cộng dồn tin nhắn mới vào lịch sử thay vì ghi đè
    messages: Annotated[list, add_messages]

2. Xây dựng các Nodes (Xử lý logic)

Mỗi Node thực chất là một hàm Python nhận vào State hiện tại và trả về kết quả cập nhật. Dưới đây là cách tạo một node chatbot cơ bản sử dụng model GPT-4o:

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

def chatbot_node(state: AgentState):
    return {"messages": [llm.invoke(state["messages"])]}

3. Thiết lập biểu đồ và điều hướng (Edges)

Đây là lúc bạn vẽ sơ đồ hoạt động cho Agent. Kinh nghiệm của mình là hãy phác thảo luồng đi ra giấy trước khi gõ code. Điều này giúp bạn tránh bị rối khi workflow có quá nhiều nhánh rẽ.

from langgraph.graph import StateGraph, START, END

# Khởi tạo graph
workflow = StateGraph(AgentState)

# Thêm node
workflow.add_node("chatbot", chatbot_node)

# Nối luồng: Bắt đầu -> chatbot -> Kết thúc
workflow.add_edge(START, "chatbot")
workflow.add_edge("chatbot", END)

app = workflow.compile()

Xây dựng Workflow đa bước có điều kiện (Conditional Edges)

Một Agent thực thụ phải biết khi nào cần dùng công cụ hỗ trợ. Ví dụ, nếu AI gặp câu hỏi về giá chứng khoán, nó cần tự gọi tool tra cứu thay vì trả lời bừa bãi. Chúng ta sẽ dùng hàm điều kiện để điều hướng luồng đi.

def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

workflow.add_conditional_edges(
    "chatbot",
    should_continue,
    {
        "tools": "tools_node",
        END: END
    }
)

Cấu trúc này tạo ra một vòng lặp thông minh: Hỏi LLM -> Gọi Tool -> Cập nhật State -> Hỏi lại LLM. Quy trình này lặp lại cho đến khi Agent tìm ra câu trả lời chính xác nhất cho người dùng.

Kiểm soát chi phí và Monitoring thực tế

Khi đưa vào vận hành, việc giám sát Agent là nhiệm vụ bắt buộc. Mình từng chứng kiến một con agent bị lặp vô tận, tiêu tốn hơn 50 USD tiền API chỉ trong 15 phút do thiếu cơ chế ngắt sớm.

Để ngăn chặn rủi ro này, bạn phải luôn thiết lập recursion_limit khi gọi Agent:

config = {"configurable": {"thread_id": "1"}, "recursion_limit": 10}
for event in app.stream({"messages": [("user", "Giá Bitcoin hiện tại bao nhiêu?")]}, config):
    for value in event.values():
        print("Phản hồi từ Agent:", value["messages"][-1].content)

Đừng quên tận dụng LangSmith để theo dõi trực quan sơ đồ chạy. Mỗi khi thay đổi logic biểu đồ, hãy kiểm tra xem các Edge có nhảy đúng như thiết kế hay không. Việc xây dựng Agent bằng LangGraph có thể khó tiếp cận lúc đầu, nhưng sự linh hoạt mà nó mang lại cho các bài toán automation phức tạp là hoàn toàn xứng đáng.

Share: