🔗 LangGraph Cơ bản

LangGraph là framework từ LangChain team để build complex, stateful agent workflows. Bài này sẽ giúp bạn master các concepts cốt lõi của LangGraph.

Tại sao cần LangGraph?

Vấn đề với simple agents

Agents đơn giản gặp khó khăn với:

Complex multi-step workflows
Conditional logic và branching
State management across steps
Error handling và retries
Human-in-the-loop

LangGraph giải quyết những vấn đề này bằng cách model agents như graphs:

Diagram

graph TD
    Start([Start]) --> A[Gather Info]
    A --> B{Enough Info?}
    B -->|Yes| C[Generate Response]
    B -->|No| D[Ask Clarifying Question]
    D --> A
    C --> End([End])

Core Concepts

1. State

State là "bộ nhớ" của graph, được truyền qua các nodes:

Python

1from typing import TypedDict, Annotated
2import operator
3
4class AgentState(TypedDict):
5    # Messages accumulate với operator.add
6    messages: Annotated[list, operator.add]
7    
8    # Simple values được overwrite
9    current_step: str
10    user_query: str
11    
12    # Optional fields
13    search_results: list | None
14    final_answer: str | None

2. Nodes

Nodes là functions xử lý state:

Python

1def gather_info_node(state: AgentState):
2    """Node thu thập thông tin"""
3    messages = state["messages"]
4    query = state["user_query"]
5    
6    # Do something
7    response = llm.invoke(messages)
8    
9    # Return state updates
10    return {
11        "messages": [response],
12        "current_step": "gathered"
13    }
14
15def generate_response_node(state: AgentState):
16    """Node tạo response cuối cùng"""
17    context = state.get("search_results", [])
18    
19    response = llm.invoke([
20        SystemMessage("Tạo response dựa trên context"),
21        HumanMessage(f"Context: {context}")
22    ])
23    
24    return {
25        "final_answer": response.content,
26        "messages": [response]
27    }

3. Edges

Edges connect nodes và define flow:

Python

1from langgraph.graph import StateGraph, END
2
3# Create graph
4workflow = StateGraph(AgentState)
5
6# Add nodes
7workflow.add_node("gather", gather_info_node)
8workflow.add_node("respond", generate_response_node)
9
10# Simple edge: gather → respond
11workflow.add_edge("gather", "respond")
12
13# End edge: respond → END
14workflow.add_edge("respond", END)
15
16# Set start
17workflow.set_entry_point("gather")

4. Conditional Edges

Branching logic dựa trên state:

Python

1def should_continue(state: AgentState) -> str:
2    """Quyết định bước tiếp theo"""
3    messages = state["messages"]
4    last_message = messages[-1]
5    
6    # Nếu có tool call, đi đến action node
7    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
8        return "action"
9    
10    # Không có tool call, kết thúc
11    return "end"
12
13# Add conditional edge
14workflow.add_conditional_edges(
15    "reasoning",  # From node
16    should_continue,  # Router function
17    {
18        "action": "action_node",  # If returns "action"
19        "end": END  # If returns "end"
20    }
21)

Build một Graph hoàn chỉnh

Ví dụ: Research Agent

Python

1from langgraph.graph import StateGraph, END
2from langchain_openai import ChatOpenAI
3from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
4from typing import TypedDict, Annotated, Literal
5import operator
6
7# 1. Define State
8class ResearchState(TypedDict):
9    messages: Annotated[list, operator.add]
10    topic: str
11    search_results: list
12    outline: str
13    draft: str
14    final_report: str
15
16# 2. Define Nodes
17llm = ChatOpenAI(model="gpt-4o")
18
19def search_node(state: ResearchState):
20    """Tìm kiếm thông tin về topic"""
21    topic = state["topic"]
22    
23    # Giả lập search
24    results = [
25        f"Result 1 about {topic}",
26        f"Result 2 about {topic}",
27        f"Result 3 about {topic}"
28    ]
29    
30    return {
31        "search_results": results,
32        "messages": [AIMessage(content=f"Found {len(results)} results")]
33    }
34
35def outline_node(state: ResearchState):
36    """Tạo outline từ search results"""
37    results = state["search_results"]
38    topic = state["topic"]
39    
40    response = llm.invoke([
41        SystemMessage("Tạo outline cho research report"),
42        HumanMessage(f"Topic: {topic}\nResults: {results}")
43    ])
44    
45    return {
46        "outline": response.content,
47        "messages": [response]
48    }
49
50def draft_node(state: ResearchState):
51    """Viết draft dựa trên outline"""
52    outline = state["outline"]
53    results = state["search_results"]
54    
55    response = llm.invoke([
56        SystemMessage("Viết draft report dựa trên outline"),
57        HumanMessage(f"Outline:\n{outline}\n\nSources:\n{results}")
58    ])
59    
60    return {
61        "draft": response.content,
62        "messages": [response]
63    }
64
65def review_node(state: ResearchState):
66    """Review và finalize report"""
67    draft = state["draft"]
68    
69    response = llm.invoke([
70        SystemMessage("Review và polish report này"),
71        HumanMessage(draft)
72    ])
73    
74    return {
75        "final_report": response.content,
76        "messages": [response]
77    }
78
79# 3. Build Graph
80workflow = StateGraph(ResearchState)
81
82workflow.add_node("search", search_node)
83workflow.add_node("outline", outline_node)
84workflow.add_node("draft", draft_node)
85workflow.add_node("review", review_node)
86
87# Linear flow: search → outline → draft → review → END
88workflow.add_edge("search", "outline")
89workflow.add_edge("outline", "draft")
90workflow.add_edge("draft", "review")
91workflow.add_edge("review", END)
92
93workflow.set_entry_point("search")
94
95# 4. Compile
96graph = workflow.compile()
97
98# 5. Run
99result = graph.invoke({
100    "topic": "AI Agents in 2024",
101    "messages": [],
102    "search_results": [],
103    "outline": "",
104    "draft": "",
105    "final_report": ""
106})
107
108print(result["final_report"])

Visualize Graph

Python

1# Tạo diagram của graph
2from IPython.display import Image, display
3
4display(Image(graph.get_graph().draw_mermaid_png()))

Output:

Diagram

graph TD
    __start__ --> search
    search --> outline
    outline --> draft
    draft --> review
    review --> __end__

Checkpointing (Persistence)

Lưu state để resume later:

Python

1from langgraph.checkpoint.memory import MemorySaver
2
3# Memory checkpointer
4checkpointer = MemorySaver()
5
6# Compile với checkpointer
7graph = workflow.compile(checkpointer=checkpointer)
8
9# Run với thread_id để track
10config = {"configurable": {"thread_id": "research-123"}}
11result = graph.invoke(initial_state, config=config)
12
13# Resume later với same thread_id
14state = graph.get_state(config)

Human-in-the-Loop

Pause để chờ human input:

Python

1from langgraph.graph import StateGraph, END
2from langgraph.checkpoint.memory import MemorySaver
3
4def human_review_node(state):
5    """Node chờ human review"""
6    draft = state["draft"]
7    
8    # Node này chỉ format output
9    # Human sẽ review và provide feedback
10    return {
11        "awaiting_review": True,
12        "messages": [AIMessage(f"Draft ready for review:\n{draft}")]
13    }
14
15# Add interrupt before human_review
16graph = workflow.compile(
17    checkpointer=MemorySaver(),
18    interrupt_before=["human_review"]  # Pause trước node này
19)
20
21# Run until interrupt
22result = graph.invoke(initial_state, config)
23# Graph pauses at human_review
24
25# Human provides feedback
26graph.update_state(config, {
27    "human_feedback": "Please add more details about X"
28})
29
30# Continue from interrupt
31result = graph.invoke(None, config)  # None = continue from checkpoint

Streaming

Stream outputs real-time:

Python

1# Stream all events
2for event in graph.stream(initial_state):
3    print(event)
4
5# Stream specific events
6for event in graph.stream(initial_state, stream_mode="values"):
7    print(f"Current state: {event}")
8
9# Stream với updates mode
10for node_name, output in graph.stream(initial_state, stream_mode="updates"):
11    print(f"{node_name}: {output}")

Bài tập thực hành

Hands-on Exercise

Build một Customer Support Graph:

Nodes:
- classify: Phân loại query (billing/technical/general)
- billing_handler: Xử lý billing questions
- technical_handler: Xử lý technical issues
- general_handler: Xử lý general queries
- respond: Tạo final response
Flow:
- classify → (conditional) → handler nodes → respond → END
Test với các queries:
- "Tôi muốn hủy subscription"
- "App không hoạt động"
- "Giờ làm việc của công ty"

Python

1# Starter code
2from langgraph.graph import StateGraph, END
3from typing import TypedDict, Annotated, Literal
4import operator
5
6class SupportState(TypedDict):
7    messages: Annotated[list, operator.add]
8    query: str
9    category: Literal["billing", "technical", "general"] | None
10    response: str | None
11
12def classify_node(state: SupportState):
13    # TODO: Implement classification
14    pass
15
16def route_by_category(state: SupportState) -> str:
17    # TODO: Return category name
18    pass
19
20# TODO: Build the graph

Trong bài tiếp theo, chúng ta sẽ build First Agent - một agent hoàn chỉnh sử dụng tools với LangGraph.

LangGraph Cơ bản

🔗 LangGraph Cơ bản

Tại sao cần LangGraph?

Core Concepts

1. State

2. Nodes

3. Edges

4. Conditional Edges

Build một Graph hoàn chỉnh

Ví dụ: Research Agent

Visualize Graph

Checkpointing (Persistence)

Human-in-the-Loop

Streaming

Bài tập thực hành

Tiếp theo

Tài liệu tham khảo