Tác giả: Đội ngũ kỹ thuật HolySheep AI — đăng lần đầu tháng 03/2026, cập nhật theo phiên bản LangGraph 0.3.x.
Sáu tuần trước, tôi ngồi nhìn dashboard Vercel hiển thị hóa đơn OpenAI tháng đó là ¥28.4 triệu (≈ $4,134) — tất cả chỉ để vận hành một workflow Agent duy nhất phục vụ 38,000 session phân tích tài liệu pháp lý. Hóa đơn tháng trước là ¥19.2 triệu. Đường cong đi lên theo cấp số nhân, và team finance bắt đầu gửi email cảnh báo màu đỏ. Đó là lúc chúng tôi quyết định viết lại playbook orchestration, và trong bài viết này tôi sẽ chia sẻ toàn bộ playbook đó — bao gồm cả những cú đốt tiền dọc đường.
Trước khi đi vào code, một điều quan trọng: bài viết này giả định bạn đang dùng LangGraph (không phải raw LangChain AgentExecutor) và bạn đang phải đối mặt với một workflow có ≥3 bước có điều kiện, có checkpoint, và cần rollback. Nếu Agent của bạn chỉ là một prompt + một tool call, bạn không cần state machine. Nhưng nếu bạn đã từng thấy if/else trong prompt bị "ăn" bởi context window, thì đây là bài viết dành cho bạn.
1. Vì sao AI Agent cần máy trạng thái?
Trong ba năm build Agent, tôi đã thấy một pattern lặp đi lặp lại: mọi Agent phức tạp đều trở thành một state machine dù bạn có muốn hay không. Sự khác biệt duy nhất là nó sẽ là một state machine tường minh (như LangGraph) hay một state machine ẩn (như chuỗi if/else trong prompt mà LLM phải tự parse).
Một state machine cho Agent có bốn thành phần cốt lõi:
- State: một TypedDict lưu messages, biến trung gian, metadata của session
- Nodes: các hàm Python nhận state, trả về partial state update
- Edges: cạnh cố định (luôn đi tiếp) hoặc conditional (rẽ nhánh theo điều kiện)
- Checkpoint: cơ chế persist state để resume, debug, và human-in-the-loop
LangGraph của LangChain cung cấp đúng bốn thành phần đó. Nhưng tôi chọn LangGraph không phải vì nó "hay", mà vì nó visualizable — team product có thể mở dashboard LangGraph Studio và thấy chính xác node nào đang được trigger cho session của user. Điều đó tiết kiệm cho chúng tôi khoảng 6 giờ debug/tuần.
2. Ba pattern tôi đã ship trong production
Trong hệ thống của chúng tôi, có ba state machine pattern xuất hiện lặp đi lặp lại:
- Sequential pipeline: Researcher → Writer → Critic. Phù hợp khi các bước có quan hệ phụ thuộc dữ liệu một chiều.
- Review loop với conditional exit: Executor ↔ Reviewer, thoát khi đạt tiêu chí hoặc vượt quá max iterations. Pattern này rất phổ biến cho code generation.
- Supervisor multi-agent: Một LLM "quản lý" chọn tool/agent con, các agent con có memory riêng. Đây là pattern Anthropic dùng trong bài báo "Building Effective Agents".
Hai pattern đầu tôi sẽ demo trong bài này. Pattern thứ ba sẽ nằm ở phần code cuối — nó phức tạp nhất nhưng lại là thứ chúng tôi tiết kiệm được nhiều tiền nhất khi chuyển sang HolySheep AI, vì nó cho phép mix model rẻ (DeepSeek) và model đắt (Claude) trong cùng một workflow.
3. Code #1: Máy trạng thái cơ bản cho Research Workflow
Đây là state machine đơn giản nhất mà vẫn hữu ích: hai node, hai cạnh cố định, một state TypedDict. Tôi dùng nó làm template cho 80% workflow nội bộ.
"""
File: research_agent.py
Mô tả: Máy trạng thái 2 node cho workflow phân tích tài liệu.
Stack: LangGraph 0.3 + HolySheep AI gateway.
"""
import os
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END, START
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage
--- Cấu hình gateway HolySheep ---
Tỷ giá ¥1=$1 giúp đội ngũ Trung Quốc không bị mất 7% phí FX như khi dùng
credit card quốc tế. Latency gateway <50ms theo benchmark p50 (xem mục 8).
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
llm = ChatOpenAI(
model="gpt-4.1",
temperature=0.2,
max_tokens=2048,
timeout=30,
)
class ResearchState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
topic: str
draft: str
final_report: str
def researcher(state: ResearchState) -> dict:
"""Node 1: tạo outline và draft đầu tiên."""
prompt = (
f"Bạn là researcher. Hãy phân tích chủ đề: {state['topic']}. "
f"Trả về outline 5 ý chính và draft ngắn 300 từ."
)
response = llm.invoke([HumanMessage(content=prompt)])
return {
"messages": [response],
"draft": response.content,
}
def writer(state: ResearchState) -> dict:
"""Node 2: biến draft thành báo cáo hoàn chỉnh."""
prompt = (
f"Viết lại draft sau thành báo cáo chuyên nghiệp 800 từ, "
f"có heading rõ ràng:\n\n{state['draft']}"
)
response = llm.invoke([HumanMessage(content=prompt)])
return {
"messages": [response],
"final_report": response.content,
}
--- Khai báo đồ thị ---
workflow = StateGraph(ResearchState)
workflow.add_node("researcher", researcher)
workflow.add_node("writer", writer)
workflow.add_edge(START, "researcher")
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", END)
app = workflow.compile()
--- Chạy thử ---
if __name__ == "__main__":
result = app.invoke({
"messages": [],
"topic": "Tác động của AI Agent đến thị trường lao động Việt Nam 2026",
"draft": "",
"final_report": "",
})
print(result["final_report"])
Điểm quan trọng: OPENAI_API_BASE phải trỏ về https://api.holysheep.ai/v1. LangChain ChatOpenAI đọc biến này và tự động tuân thủ OpenAI-compatible schema của HolySheep. Bạn không cần đổi import, không cần wrapper class, không cần shim. Toàn bộ SDK LangChain hoạt động nguyên xi — đây là lý do migration dễ đến vậy.
4. Code #2: Review loop với conditional edge + checkpoint
Pattern này là thứ tôi dùng cho mọi code-generation Agent: chạy executor, đưa output cho reviewer, nếu reviewer không duyệt thì quay lại executor với feedback. Thoát khi đạt max iterations.
"""
File: code_review_loop.py
Mô tả: Máy trạng thái có conditional edge cho vòng lặp review code.
Tính năng: persist state qua Postgres, resume khi crash.
"""
import os
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END, START
from langgraph.checkpoint.postgres import PostgresSaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
Mix 2 model trong cùng workflow: DeepSeek rẻ cho executor,
Claude Sonnet đắt hơn cho reviewer (vì reviewer cần reasoning sâu).
executor_llm = ChatOpenAI(model="deepseek/deepseek-v3.2", temperature=0.1)
reviewer_llm = ChatOpenAI(model="claude-sonnet-4.5", temperature=0)
class CodeReviewState(TypedDict):
task: str
code: str
review: str
iterations: int
approved: bool
MAX_ITER = 3
def executor(state: CodeReviewState) -> dict:
feedback = state.get("review", "Chưa có feedback.")
prompt = (
f"Task: {state['task']}\n\n"
f"Code hiện tại:\n{state.get('code', '(chưa có)')}\n\n"
f"Feedback từ reviewer:\n{feedback}\n\n"
"Hãy viết lại code cải thiện theo feedback."
)
response = executor_llm.invoke([HumanMessage(content=prompt)])
return {
"code": response.content,
"iterations": state.get("iterations", 0) + 1,
}
def reviewer(state: CodeReviewState) -> dict:
prompt = (
f"Review code Python sau cho task: {state['task']}\n\n"
f"``python\n{state['code']}\n``\n\n"
"Nếu code đạt yêu cầu, trả lời duy nhất một dòng: APPROVED\n"
"Nếu chưa đạt, liệt kê bullet các vấn đề cần sửa."
)
response = reviewer_llm.invoke([HumanMessage(content=prompt)])
content = response.content.strip()
approved = content.startswith("APPROVED")
return {
"review": content,
"approved": approved,
}
def should_continue(state: CodeReviewState) -> Literal["executor", END]:
if state.get("approved"):
return END
if state.get("iterations", 0) >= MAX_ITER:
return END
return "executor"
--- Build graph ---
workflow = StateGraph(CodeReviewState)
workflow.add_node("executor", executor)
workflow.add_node("reviewer", reviewer)
workflow.add_edge(START, "executor")
workflow.add_edge("executor", "reviewer")
workflow.add_conditional_edges(
"reviewer",
should_continue,
{"executor": "executor", END: END},
)
--- Checkpoint Postgres để resume ---
DB_URI = "postgresql://user:pass@localhost:5432/langgraph"
checkpointer = PostgresSaver.from_conn_string(DB_URI)
checkpointer.setup() # chạy 1 lần để tạo bảng
app = workflow.compile(checkpointer=checkpointer)
--- Chạy ---
config = {"configurable": {"thread_id": "session-001"}}
result = app.invoke(
{"task": "Viết hàm parse CSV an toàn chống injection", "iterations