Giới thiệu

Nói chung LangGraph là một thư viện được xây dựng dựa trên LLM và LangChain, giúp ta đơn giản hóa việc xây dựng các Agent tự động hóa các tác vụ trong thế giới thực.

Vì sao cần LangGraph?

Ta thấy rằng LangChain cho phép chúng ta xây dựng code bằng LCEL( LangChain Expression Language ). Về cơ bản LCEL xây dựng các Chain, để dễ hiểu ta hãy xét một Chain cơ bản sau:

from langchain_community.chat_models import ChatOllama
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template("{text}")
model = ChatOllama(model="llama3", temperature=0)
output_parser = StrOutputParser()

chain = prompt | model | output_parser

Có thể thấy, một Chain cơ bản gồm những thành phần cốt lõi như Prompt, ChatModel và OutputParser. Nhờ Runnable Interface, các thành phần này có thể dễ dàng “kết nối” với nhau qua toán tử |, qua đó hình thành một chuỗi thao tác tuần tự được gọi là RunnableSequence.

Tuy nhiên nhìn vào ví dụ trên khi dùng LangChain, bạn chỉ có thể tạo ra DAG (Directed Acyclic Graph) là một loại đồ thị có hướng không có chu trình, nghĩa là không hỗ trợ vòng lặp.

Ví dụ, nếu bạn xây dựng một ứng dụng dùng RAG thì có thể gặp vấn đề: Nếu bước đầu tiên lấy được dữ liệu không hữu ích, ứng dụng sẽ không quay lại để tìm thêm dữ liệu mà vẫn tiến tới bước tiếp theo. Điều này làm giảm đáng kể hiệu quả xử lý.

LangGraph mở rộng LCEL, cho phép các bước trong các Chain có thể điều chỉnh vòng lặp. Nhờ đó, các bước có thể quay lại và xử lý lại dữ liệu khi cần thiết, giúp ứng dụng hoạt động linh hoạt hơn.

Các khái niệm cơ bản trong LangGraph

Node, Edge

Để hiểu về "Graph" một cách đơn giản, chúng ta chỉ cần nắm hai yếu tố chính trong Graph:

• Node (Nút): Đại diện cho một điểm hoặc một bước trong quy trình.
• Edge (Cạnh): Kết nối giữa các nút, thể hiện mối quan hệ hoặc luồng dữ liệu.

Từ hai yếu tố này, bạn có thể "lắp ráp" một đồ thị hoàn chỉnh

Từ hình vẽ ta thấy:

Node (Node_1, Node_2): Là các điểm cơ bản trong đồ thị, đại diện cho các thành phần hoặc bước cần xử lý.

Sử dụng lệnh add_node để thêm từng nút vào đồ thị.

Edge (Cạnh): Là các đường nối có hướng giữa các node, giúp liên kết các thành phần với nhau.

Sử dụng lệnh add_edge để tạo các mối liên kết này. Khi kết hợp các nút và cạnh, chúng ta có thể xây dựng một đồ thị hoàn chỉnh để mô phỏng luồng công việc hoặc dữ liệu.

from langgraph.graph import Graph

def node1(input1: str) -> str:
    return f"Xin chào {input1}"

def node2(input2: str) -> str:
    return f"{input2}, tôi có thể giúp gì cho bạn?"

graph = Graph()
graph.add_node("node1", node1)
graph.add_node("node2", node2)
graph.add_edge("node1", "node2")
graph.set_entry_point("node1")
graph.set_finish_point("node2")

app = graph.compile()

app.invoke("Hùng")

Output: Xin chào, Hùng, tôi có thể giúp gì cho bạn?

Bạn cũng có thể thay đổi code một chút và có thể xem kết quả đầu ra của chương trình trong mỗi Node.

for output in app.stream("Hùng"):
    for key, value in output.items():
        print(key, "----", value)

output

node1 ---- Xin chào Hùng
node2 ---- Xin chào Hùng, tôi có thể giúp gì cho bạn?

Để vẽ graph hãy sử dụng code sau:

from IPython.display import display, Image
display(Image(app.get_graph().draw_mermaid_png()))

conditional_edges

Khi đọc đến đây, chúng ta có thể tự hỏi: Việc sử dụng phương pháp này khác gì so với cách làm truyền thống (ví dụ: viết một đoạn logic LLM thông thường hoặc sử dụng LCEL)? Phải chăng chỉ là thực thi tuần tự, nhưng được gói trong một cái “vỏ Graph” mà thôi?

Thực tế, ví dụ trên đúng là như vậy. Nhưng hãy tiếp tục theo dõi: Chúng ta sẽ xem một kiểu mô hình khác - mô hình phân nhánh.

Trong graph, kiểu phân nhánh này được gọi là “điều kiện cạnh” (conditional edge), và được thực hiện bằng cách sử dụng add_conditional_edges

Code sẽ trông như này:

from langgraph.graph import Graph, END

def node_entry(input: str) -> str:
    return f"{input}"

# The function decides the next direction
def decide_which(input: str) -> str:
    if "Xin chào" in input:
        return "greeting"
    elif "ngày sinh" in input:
        return "birth"
    return "end"

def node_greeting(input: str) -> str:
    return f"Nội dung đã nhắc đến : {input}\nXin chào bạn!"

def node_birth(input: str) -> str:
    return f"Nội dung đã nhắc đến : {input}\nCâu trả lời: Tôi sinh ngày 10/1"

graph = Graph()
graph.add_node("entry", node_entry)
graph.add_node("node_greeting", node_greeting)
graph.add_node("node_birth", node_birth)

graph.set_entry_point("entry")

graph.add_conditional_edges(
    source="entry",
    path=decide_which,
    path_map={
        "greeting": "node_greeting",
        "birth": "node_birth",
        "end": END,
    },
)

graph.add_edge("node_greeting", END)
graph.add_edge("node_birth", END)

app = graph.compile()
app.invoke("ngày sinh của bạn là bn?")

ouput

Nội dung đã nhắc đến : ngày sinh của bạn là bn?
Câu trả lời: Tôi sinh ngày 10/1

Giờ cùng hãy thử viết code với hình dưới nhé.

Lưu ý: node_greeting và node_birth đều có Edge trỏ ngược lại.

State, StateGraph

Bằng cách kết hợp Nodes và Edges, chúng ta có thể tạo các quy trình công việc phức tạp, thậm chí có thể lặp đi lặp lại và thay đổi State (trạng thái) theo thời gian. Tuy nhiên, sức mạnh thực sự nằm ở cách LangGraph quản lý State đó.

Chúng ta cùng quay lại với ví dụ bên trên. Đẻ thực hiện được code theo hình, ta chỉ cần thay đổi đích đến của 2 cạnh

graph.add_edge("node_greeting", END)
graph.add_edge("node_birth", END)

thành:

graph.add_edge("node_greeting", "entry")
graph.add_edge("node_birth", "entry")

Nhưng khi thực hiện app.invoke("ngày sinh của bạn là bn?") chúng ta bắt gặp lỗi Recursion limit of 25 reached without hitting a stop condition. Điều này xảy ra vì đã xuất hiện vòng lặp vô hạn. Node entry không nhận biết được liệu câu hỏi đã được trả lời tại node_greeting hay node_birth hay chưa dẫn đến agent trao đổi qua lại với nhau mà không có điểm dừng. Chúng ta cần một biến trạng thái để lưu lại thông tin giữa các node, và đó là công dụng của Sate. Hãy thử ứng dụng State để giải quyết bài toán này.

Code tham khảo như sau:

from langgraph.graph import END, StateGraph
from typing import TypedDict

class OverallState(TypedDict):
    input: str
    answer: str

def node_entry(state: OverallState) -> OverallState:
    return state

# The function decides the next direction
def decide_which(state: OverallState) -> str:
    if state["answer"] != "":
        return "end"
    if "Xin chào" in state["input"]:
        return "greeting"
    elif "ngày sinh" in state["input"]:
        return "birth"
    return "end"


def node_greeting(state: OverallState) -> OverallState:
    return OverallState(
        input=state["input"],
        answer=f"Nội dung đã nhắc đến : {state['input']}\nXin chào bạn!",
    )

def node_birth(state: OverallState) -> OverallState:

    return OverallState(
        input=state["input"],
        answer=f"Nội dung đã nhắc đến : {state['input']}\nCâu trả lời: Tôi sinh ngày 10/1",
    )

graph = StateGraph(OverallState)
graph.add_node("entry", node_entry)
graph.add_node("node_greeting", node_greeting)
graph.add_node("node_birth", node_birth)

graph.set_entry_point("entry")

graph.add_conditional_edges(
    source="entry",
    path=decide_which,
    path_map={
        "greeting": "node_greeting",
        "birth": "node_birth",
        "end": END,
    },
)

graph.add_edge("node_greeting", "entry")
graph.add_edge("node_birth", "entry")

app = graph.compile()

app.invoke({"input": "ngày sinh của bạn là bn?", "answer": ""})

Output :

{
'input': 'ngày sinh của bạn là bn?',
'answer': 'Nội dung đã nhắc đến : ngày sinh của bạn là bn?\nCâu trả lời: Tôi sinh ngày 10/1'
}

Chúng ta sẽ khởi tạo một State để chứa thông tin về câu input và câu trả lời. Dựa vào thông tin của State, tại node entry ta dễ dàng xác định câu hỏi đã được trả lời hay chưa và từ đó kết thúc vòng lặp. Class StateGraph cùng được dùng để thay thế Graph. Thật dễ dàng phải không ^^

Reducers

Langgraph cho phép sau chúng ta sau mỗi node chỉ cần return một dict những gì cần update tại state mà không nhất thiết phải return toàn bộ State. Với ví dụ trên ta hoàn toàn có thể thay thế phần return dài dòng như này

def node_greeting(state: OverallState) -> OverallState:
    return OverallState(
        input=state["input"],
        answer=f"Nội dung đã nhắc đến : {state['input']}\nXin chào bạn!",
    )

bằng như này:

def node_greeting(state: OverallState) -> dict:
    return {"answer":f"Nội dung đã nhắc đến : {state['input']}\nXin chào bạn!"}

Nhất là đối với cách Graph lớn, State của chúng ta càng phức phạp thì cách thực hiện như này giúp code trở lên tinh gọn hơn, dễ kiểm soát hơn.

Messages trong State của Graph

Tại sao sử dụng Messages?

Hầu hết các bên cung cấp LLM hiện tại đều chấp nhận một danh sách các tin nhắn làm đầu vào. Đặc biệt, ChatModel của LangChain chấp nhận một danh sách các đối tượng Message. Các tin nhắn này có nhiều dạng khác nhau như HumanMessage hoặc AIMessage. vd:

chat = ChatOpenAI(model="gpt-4o")
messages = [
    HumanMessage(content="Xin chào! Bạn có thể giúp tôi không?"),
    AIMessage(content="Chào bạn! Rất vui được giúp đỡ. Bạn cần hỗ trợ về vấn đề gì?"),
    HumanMessage(content="Tôi muốn biết thời tiết hôm nay ở Hà Nội như thế nào.")
]
response = chat.invoke(messages)

Trong nhiều trường hợp, việc lưu trữ lịch sử cuộc trò chuyện trước đó dưới dạng danh sách các tin nhắn trong State của graph là rất hữu ích.

Cách dùng Messages trong Graph

LangGraph hỗ trợ cập nhật messages trong state, khi sử dụng thêm hàm reducers add_messages, các Messages sẽ tự động append vào list messages sau mỗi lần update. Khi không dùng, các Messages sẽ được ghi đè, và state cuối sẽ chỉ lưu Messages của node cuối cùng. Hãy thử một ví dụ dưới đây:

from typing import Annotated, List
from typing_extensions import TypedDict
from langchain_core.messages import AnyMessage, AIMessage
from langgraph.graph.message import add_messages
from langgraph.graph import StateGraph, END

class State(TypedDict):
    messages: Annotated[List[AnyMessage], add_messages]

def node_1(state: State) -> dict:
    output_messages = [AIMessage(content="message từ node 1")]
    return {"messages": output_messages}

def node_2(state: State) -> dict:
    output_messages = [AIMessage(content="message từ node 2")]
    return {"messages": output_messages}

graph_builder = StateGraph(State)

graph_builder.add_node("node_1", node_1)
graph_builder.add_node("node_2", node_2)

graph_builder.set_entry_point("node_1")
graph_builder.add_edge("node_1", "node_2")
graph_builder.add_edge("node_2", END)

app = graph_builder.compile()

initial_input = {"messages": [HumanMessage(content="Xin chào!")]}
final_state = app.invoke(initial_input)

for i, msg in enumerate(final_state['messages']):
    print(f"Tin nhắn {i+1}: '{msg.content}'")

bạn có thể thay messages: Annotated[List[AnyMessage], add_messages] bằng messages: List[AnyMessage] để thấy sự khác biệt

MessagesState

Vì việc có một danh sách các tin nhắn trong State của bạn rất phổ biến, nên có một State dựng sẵn gọi là MessagesState giúp dễ dàng sử dụng tin nhắn. Thông thường mọi người thường kế thừa State này và thêm các trường khác.

from langgraph.graph import MessagesState

class MyCustomState(MessagesState):
    # messages: Annotated[list[AnyMessage], add_messages] đã được định nghĩa trong MessagesState
    user_name: str
    some_data: dict

Send

Trong các đồ thị (graphs) thông thường, các nút và cạnh (edges) thường được xác định trước và tất cả hoạt động trên cùng một State xuyên suốt graph. Tuy nhiên, có những trường hợp một nút có thể tạo ra một danh sách các mục và bạn muốn áp dụng một nút khác (một agent khác) cho từng mục đó. Ví dụ, hãy nhìn vào hình minh họa, mình có một node generate_topics tạo ra các mục subject, và bây giờ mình muốn gửi từng subject này vào agent generate_joke sau đó tổng hợp lại và gửi đến node best_joke thì đây là lúc ta cần sử dụng đến Send.

Send giải quyết vấn đề này bằng cách cho phép một cạnh điều kiện (conditional edge) trả về một danh sách các đối tượng Send. Mỗi đối tượng Send chỉ định:

• Tên của nút đích (node name).
• Trạng thái (state) cụ thể sẽ được gửi đến nút đó.

Điều này cho phép bạn "phân nhánh" (map) một tác vụ ra nhiều lần với các đầu vào khác nhau và sau đó có thể "gom" (reduce) kết quả lại. Việc tách ra như này, cũng giúp hàm generate_joke được thực hiện song song, giúp tối ưu hóa thời gian so với sử dụng vòng lặp hoặc phải thực hiện lập trình bất đồng bộ ^^

Code tham khảo cho ví dụ trên như sau:

import operator
from typing import Annotated
from typing_extensions import TypedDict

from langgraph.types import Send
from langgraph.graph import END, StateGraph, START


class OverallState(TypedDict):
    topic: str
    subjects: list
    jokes: Annotated[list, operator.add]
    best_selected_joke: str

class JokeState(TypedDict):
    subject: str

def generate_topics(state: OverallState):
    return {"subjects": ["lions", "elephants", "penguins"]}

# Here we generate a joke, given a subject
def generate_joke(state: JokeState):
    import time
    time.sleep(3)  # Simulate a delay in joke generation
    joke_map = {
        "lions": "Why don't lions like fast food? Because they can't catch it!",
        "elephants": "Why don't elephants use computers? They're afraid of the mouse!",
        "penguins": (
            "Why don’t penguins like talking to strangers at parties? "
            "Because they find it hard to break the ice."
        ),
    }
    return {"jokes": [joke_map[state["subject"]]]}

def continue_to_jokes(state: OverallState):
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]


# Here we will judge the best joke
def best_joke(state: OverallState):
    return {"best_selected_joke": "penguins"}


# Construct the graph: here we put everything together to construct our graph
graph = StateGraph(OverallState)
graph.add_node("generate_topics", generate_topics)
graph.add_node("generate_joke", generate_joke)
graph.add_node("best_joke", best_joke)
graph.add_edge(START, "generate_topics")
graph.add_conditional_edges("generate_topics", continue_to_jokes, ["generate_joke"])
graph.add_edge("generate_joke", "best_joke")
graph.add_edge("best_joke", END)
app = graph.compile()

for step in app.stream({"topic": "animals"}):
    print(step)

Command

Đôi khi, việc kết hợp điều hướng và cập nhật state tại cùng một hàm Node khá hữu ích. LangGraph cung cấp một cách để làm điều đó bằng cách trả về một đối tượng Command từ các hàm Node.

Command cho phép bạn:

1. Update State: Tương tự như cách một Node thông thường trả về một dict để cập nhật State.
2. Điều hướng (goto): Chỉ định Node tiếp theo sẽ được thực thi.

Lưu ý: Khi trả về Command trong các hàm Node của bạn, bạn phải thêm chú thích kiểu trả về (return type annotation) với danh sách các tên Node mà Node đó có thể điều hướng đến, ví dụ: Command[Literal["other_node", "another_node_option"]]. Điều này cần thiết để render graph và cho LangGraph biết rằng Node hiện tại có thể điều hướng đến các Node đó.

def my_node(state: State) -> Command[Literal["other_node"]]:
    return Command(
        # state update
        update={"foo": "bar"},
        # control flow
        goto="my_other_node"
    )

Với Command , bạn cũng có thể thực hiện điều khiển luống graph theo ý muốn (giống hệt với conditional edges):

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    if state["foo"] == "bar":
        return Command(update={"foo": "baz"}, goto="my_other_node")

Khi nào nên sử dụng Command thay vì conditional_edges?

• Sử dụng Command khi bạn cần cả cập nhật State của graph và định tuyến đến một Node khác trong cùng một bước logic của Node đó. Ví dụ, khi triển khai việc chuyển giao giữa nhiều agent (multi-agent handoffs) nơi việc định tuyến đến một agent khác và truyền một số thông tin cho agent đó là quan trọng.
• Sử dụng conditional_edges để định tuyến giữa các Node một cách có điều kiện mà không cần cập nhật State tại thời điểm quyết định định tuyến đó (State đã được cập nhật bởi Node nguồn trước đó).

Trực quan hóa graph

Để hiển thị được giao diện của graph studio như trên, dễ debug và trực quan hóa được luồng chạy giữa các agent, bạn chỉ cần làm như sau:

Cài đặt thư viện:

pip install -U "langgraph-cli[inmem]"

Tạo 1 file langgraph.json có nội dung như sau:

{
  "dependencies": ["."],
  "graphs": {
    "agent": "đường dẫn tới file chứa biến app"
  },
  "env": ".env"
}

Biến app là biến được gán bởi graph.compile() vd:

app = graph.compile()

Cuối cùng run hệ thống bằng lệnh

langgraph dev

---

Kết

Hy vọng bài viết này giúp bạn hiểu hơn về LangGraph, nắm bắt được các công cụ để xây dựng xây dựng một hệ thống Agent cho riêng mình. Nếu có gì thắc mắc hoặc cần trao đổi, hãy mạnh dạn để lại comment nhé ^^.

Nguồn tham khảo

https://langchain-ai.github.io/langgraph/