Giáo án LangGraph 101: Vì sao cần LangGraph và những khái niệm cơ bản
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:
- 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. - Đ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
All Rights Reserved