Bài 14: Project 4 — AI Agent tự động hóa quy trình công việc

Sau khi hoàn thành project này, bạn sẽ có:

• ✅ Agent đa bước với LangGraph 0.2.x — StateGraph, node, edge, conditional routing
• ✅ Tool calling với @tool decorator — docstring đúng để LLM chọn tool chính xác
• ✅ Human-in-the-loop (HITL) — interrupt trước node nhạy cảm, approve / reject / edit state
• ✅ Checkpointer — persist state qua restart với MemorySaver (dev) và PostgresSaver (production)
• ✅ FastAPI wrapper với endpoint /run và /approve
• ✅ Streamlit UI đơn giản cho demo
• ✅ Bộ test scenario để đánh giá accuracy của agent

Thông Tin Chung

Vì Sao Project Này Phù Hợp Cho Portfolio

Agentic AI là chủ đề được nhiều JD AI Engineer 2024–2025 đề cập. Điểm khác biệt so với RAG chatbot (Project 3) ở chỗ:

• Multi-step execution: agent tự quyết định gọi tool nào, theo thứ tự nào.
• State management: state tồn tại qua nhiều bước, có thể pause và resume.
• HITL: human có thể can thiệp trước khi agent thực hiện hành động có side-effect (gửi email, xóa file).
• Tool calling: LLM điều phối các hàm Python thực thụ, không chỉ generate text.

Tất cả các điểm trên đều là kỹ năng có thể demo cụ thể trong phỏng vấn.

Chọn 1 trong 5 biến thể phù hợp với context của bạn. Tất cả đều dùng cùng kiến trúc LangGraph.

Bài viết này dùng Email Triage Agent làm ví dụ chính. Cấu trúc code có thể áp dụng trực tiếp cho 4 biến thể còn lại — chỉ cần đổi tools và state fields.

Tools

• read_emails(folder, limit) — lấy danh sách email từ inbox
• classify_email(email) — gán category: urgent / normal / spam
• search_history(sender) — tìm email cũ từ sender để có context
• draft_reply(email_body) — soạn reply nháp 3–5 câu
• save_todo(task, due_date) — lưu task vào danh sách việc cần làm
• send_email(to, subject, body) — gửi email — chỉ chạy sau khi human approve

Workflow Graph

[START]
   │
   ▼
fetch_emails          ← đọc inbox
   │
   ▼
classify_email        ← phân loại email đầu tiên chưa xử lý
   │
   ├── "urgent"  ──▶  draft_reply  ──▶  human_review (interrupt)  ──▶  send_email  ──▶  save_action
   │
   ├── "normal"  ──▶  save_action
   │
   └── "spam"    ──▶  [END]
                             │
                      save_action ──▶ [END]

Node human_review không phải node code — đó là điểm interrupt do checkpointer tạo ra. Khi graph đến đây, nó dừng và trả quyền điều khiển về cho user (qua API).

State Fields

python -m venv .venv
source .venv/bin/activate  # Linux / macOS
# .venv\Scripts\activate   # Windows

pip install langgraph langchain langchain-openai langchain-anthropic
pip install fastapi "uvicorn[standard]" pydantic-settings
pip install python-dotenv streamlit requests

Cấu trúc thư mục gợi ý:

email-triage-agent/
├── agent/
│   ├── __init__.py
│   ├── state.py        # EmailAgentState
│   ├── tools.py        # @tool functions
│   ├── graph.py        # StateGraph builder
│   └── nodes.py        # node functions
├── api/
│   ├── __init__.py
│   └── main.py         # FastAPI app
├── ui/
│   └── app.py          # Streamlit UI
├── tests/
│   └── test_agent.py   # evaluation scenarios
├── .env
└── requirements.txt

File .env:

OPENAI_API_KEY=sk-...
# hoặc
ANTHROPIC_API_KEY=sk-ant-...

File agent/state.py:

from typing import Annotated, TypedDict, Literal
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage


class EmailAgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    emails: list[dict]                          # raw email list từ inbox
    current_email: dict | None                  # email đang process
    category: Literal["urgent", "normal", "spam"] | None
    draft_reply: str | None
    todos: list[str]
    requires_approval: bool

Lưu ý về add_messages: đây là reducer — khi một node trả về {"messages": [new_msg]}, LangGraph tự động append vào danh sách hiện có thay vì overwrite. Nếu không dùng reducer, state field sẽ bị replace hoàn toàn.

Lưu ý về context window: messages không tự trim. Nếu agent chạy nhiều vòng, list này sẽ tăng vô hạn dẫn đến token limit error. Cần có chiến lược trim — xem thêm ở phần Common Pitfalls.

File agent/tools.py:

from langchain_core.tools import tool


@tool
def read_emails(folder: str = "inbox", limit: int = 20) -> list[dict]:
    """Đọc N email mới nhất từ folder chỉ định.

    Args:
        folder: tên folder (inbox, sent, spam). Mặc định: inbox.
        limit: số lượng email tối đa. Mặc định: 20.

    Returns:
        Danh sách email, mỗi email gồm id, from, subject, body, timestamp.
    """
    # Mock data — thay bằng IMAP / Gmail API khi production
    return [
        {
            "id": "e1",
            "from": "[email protected]",
            "subject": "URGENT: Cần report Q3 trước 5pm hôm nay",
            "body": "Anh cần báo cáo Q3 trước cuối ngày. Vui lòng xác nhận.",
            "timestamp": "2026-05-27T09:00:00",
        },
        {
            "id": "e2",
            "from": "[email protected]",
            "subject": "Weekly digest — AI news",
            "body": "Here are this week's top AI stories...",
            "timestamp": "2026-05-27T08:00:00",
        },
    ]


@tool
def search_history(sender: str) -> list[dict]:
    """Tìm email cũ từ một sender để lấy context trước khi reply.

    Args:
        sender: địa chỉ email của người gửi.

    Returns:
        Danh sách email cũ (tối đa 5), sắp xếp mới nhất trước.
    """
    # Mock — thay bằng query thực từ local email store hoặc API
    return [
        {
            "id": "old1",
            "from": sender,
            "subject": "Re: Q2 report",
            "body": "Cảm ơn, đã nhận được.",
        }
    ]


@tool
def save_todo(task: str, due_date: str | None = None) -> str:
    """Lưu một task mới vào danh sách việc cần làm.

    Args:
        task: mô tả task cần làm.
        due_date: ngày hạn (ISO format, vd "2026-05-28"). Tùy chọn.

    Returns:
        Xác nhận đã lưu.
    """
    # Mock — thay bằng ghi vào database / Notion / Todoist API
    print(f"[TODO] {task} (due: {due_date or 'no deadline'})")
    return f"Saved: {task}"


@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Gửi email. CHỈ gọi tool này sau khi human đã approve draft reply.

    Args:
        to: địa chỉ email người nhận.
        subject: tiêu đề email.
        body: nội dung email.

    Returns:
        Xác nhận đã gửi.
    """
    # Mock — thay bằng SMTP / Gmail API send khi production
    print(f"[SENT] To: {to} | Subject: {subject}")
    return f"Email sent to {to}"

Docstring rất quan trọng: LLM dùng docstring (đặc biệt phần mô tả và Args) để quyết định có gọi tool này không và truyền argument gì. Nếu thiếu hoặc mơ hồ, LLM sẽ chọn sai tool hoặc bỏ qua tool.

File agent/nodes.py — định nghĩa logic từng node:

from langchain_openai import ChatOpenAI
from .state import EmailAgentState
from .tools import read_emails, save_todo, send_email

llm = ChatOpenAI(model="gpt-4o-mini")
# Bind tools để LLM biết có thể gọi những function nào
llm_with_tools = llm.bind_tools([read_emails, save_todo, send_email])


def fetch_emails(state: EmailAgentState) -> dict:
    """Node đọc email từ inbox."""
    result = read_emails.invoke({"folder": "inbox", "limit": 10})
    return {"emails": result}


def classify_email(state: EmailAgentState) -> dict:
    """Node phân loại email đầu tiên trong danh sách."""
    emails = state["emails"]
    if not emails:
        return {"current_email": None, "category": None}

    email = emails[0]
    prompt = f"""Phân loại email sau đây thành 1 trong 3 nhóm: urgent, normal, spam.

From: {email['from']}
Subject: {email['subject']}
Body: {email['body'][:500]}

Trả về đúng 1 từ: urgent, normal, hoặc spam. Không giải thích thêm."""

    response = llm.invoke(prompt)
    category = response.content.strip().lower()
    # Fallback nếu LLM trả về text không mong đợi
    if category not in ("urgent", "normal", "spam"):
        category = "normal"

    return {"current_email": email, "category": category}


def draft_reply_node(state: EmailAgentState) -> dict:
    """Node soạn draft reply cho email urgent."""
    email = state["current_email"]
    history = state.get("messages", [])

    prompt = f"""Soạn một reply email ngắn gọn (3–5 câu) cho email sau. Giọng văn chuyên nghiệp, lịch sự.

Email gốc:
From: {email['from']}
Subject: {email['subject']}
Body: {email['body']}"""

    response = llm.invoke(prompt)
    return {
        "draft_reply": response.content,
        "requires_approval": True,
    }


def save_action(state: EmailAgentState) -> dict:
    """Node lưu task follow-up vào todo list."""
    email = state["current_email"]
    if not email:
        return {}
    task = f"Follow up: [{email['subject']}] from {email['from']}"
    save_todo.invoke({"task": task})
    return {"todos": state.get("todos", []) + [task]}

File agent/graph.py — lắp ráp graph:

from langgraph.graph import StateGraph, START, END
from .state import EmailAgentState
from .nodes import fetch_emails, classify_email, draft_reply_node, save_action


def route_after_classify(state: EmailAgentState) -> str:
    """Conditional routing dựa trên category."""
    category = state.get("category")
    if category == "urgent":
        return "draft_reply"
    elif category == "normal":
        return "save_action"
    else:
        # spam hoặc None
        return END


def build_graph():
    builder = StateGraph(EmailAgentState)

    # Thêm nodes
    builder.add_node("fetch_emails", fetch_emails)
    builder.add_node("classify_email", classify_email)
    builder.add_node("draft_reply", draft_reply_node)
    builder.add_node("save_action", save_action)

    # Thêm edges
    builder.add_edge(START, "fetch_emails")
    builder.add_edge("fetch_emails", "classify_email")
    builder.add_conditional_edges("classify_email", route_after_classify)
    builder.add_edge("draft_reply", "save_action")
    builder.add_edge("save_action", END)

    return builder

Lưu ý: chưa có .compile() ở đây — compile sẽ được thực hiện ở bước 5 để thêm checkpointer và interrupt_before.

HITL trong LangGraph 0.2.x hoạt động qua 2 thành phần:

• checkpointer: lưu state tại mỗi bước để có thể resume sau khi pause.
• interrupt_before: danh sách node names — graph sẽ dừng trước khi vào node đó.

from langgraph.checkpoint.memory import MemorySaver
from .graph import build_graph

checkpointer = MemorySaver()  # in-memory, dùng cho dev / testing
builder = build_graph()

graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["save_action"],  # dừng trước save_action để user review draft
)

Quy trình chạy với HITL:

config = {"configurable": {"thread_id": "session-001"}}

# --- Lần chạy 1: graph sẽ dừng trước "save_action" ---
state_after_run1 = graph.invoke(
    {"emails": [], "todos": [], "requires_approval": False},
    config=config,
)

# Lúc này graph đã pause, state được persist trong checkpointer
current_state = graph.get_state(config)
print("Draft reply:", current_state.values.get("draft_reply"))
print("Pending nodes:", current_state.next)
# Output: Pending nodes: ('save_action',)

# --- User review và quyết định ---
# Option A: approve và resume nguyên state
graph.invoke(None, config=config)

# Option B: user sửa draft_reply trước khi approve
graph.update_state(config, {"draft_reply": "Kính gửi, tôi đã nhận được yêu cầu..."})
graph.invoke(None, config=config)

# Option C: cancel — không resume, kết thúc thread
# (chỉ cần không gọi invoke nữa)

graph.invoke(None, config=config): truyền None làm input nghĩa là "tiếp tục từ điểm bị interrupt, dùng state hiện tại".

File api/main.py:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from agent.graph import build_graph
from langgraph.checkpoint.memory import MemorySaver

app = FastAPI(title="Email Triage Agent API")

checkpointer = MemorySaver()
builder = build_graph()
graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["save_action"],
)


class RunRequest(BaseModel):
    thread_id: str


class ApproveRequest(BaseModel):
    thread_id: str
    approved: bool
    edited_reply: str | None = None


@app.post("/run")
async def run_agent(req: RunRequest):
    """Khởi động agent cho một thread. Trả về state sau khi pause."""
    config = {"configurable": {"thread_id": req.thread_id}}
    try:
        await graph.ainvoke(
            {"emails": [], "todos": [], "requires_approval": False},
            config=config,
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

    state = graph.get_state(config)
    return {
        "status": "paused" if state.next else "complete",
        "pending_nodes": list(state.next),
        "draft_reply": state.values.get("draft_reply"),
        "category": state.values.get("category"),
        "current_email": state.values.get("current_email"),
    }


@app.post("/approve")
async def approve(req: ApproveRequest):
    """Human approve / reject. Nếu approve, resume graph."""
    config = {"configurable": {"thread_id": req.thread_id}}
    state = graph.get_state(config)

    if not state.next:
        raise HTTPException(status_code=400, detail="No pending action for this thread.")

    if not req.approved:
        return {"status": "cancelled"}

    # Nếu user sửa draft_reply, cập nhật state trước khi resume
    if req.edited_reply is not None:
        graph.update_state(config, {"draft_reply": req.edited_reply})

    await graph.ainvoke(None, config=config)
    return {"status": "complete"}


@app.get("/state/{thread_id}")
async def get_state(thread_id: str):
    """Xem state hiện tại của một thread."""
    config = {"configurable": {"thread_id": thread_id}}
    state = graph.get_state(config)
    if state is None:
        raise HTTPException(status_code=404, detail="Thread not found.")
    return {
        "values": state.values,
        "next": list(state.next),
    }

Chạy server:

uvicorn api.main:app --reload --port 8000

File ui/app.py:

import streamlit as st
import requests

API_BASE = "http://localhost:8000"

st.title("Email Triage Agent")

thread_id = st.text_input("Thread ID", value="user-001")

if st.button("Run Agent"):
    with st.spinner("Agent đang xử lý..."):
        resp = requests.post(f"{API_BASE}/run", json={"thread_id": thread_id})
    if resp.ok:
        st.session_state["run_result"] = resp.json()
    else:
        st.error(f"Lỗi: {resp.text}")

if "run_result" in st.session_state:
    result = st.session_state["run_result"]
    st.subheader("Kết quả")
    st.json({k: v for k, v in result.items() if k != "draft_reply"})

    if result.get("status") == "paused":
        st.warning(f"Chờ approval: node {result.get('pending_nodes')}")
        draft = st.text_area(
            "Draft Reply (có thể chỉnh sửa trước khi gửi)",
            value=result.get("draft_reply", ""),
            key="edited_reply",
        )
        col1, col2 = st.columns(2)
        if col1.button("Approve & Send"):
            resp2 = requests.post(f"{API_BASE}/approve", json={
                "thread_id": thread_id,
                "approved": True,
                "edited_reply": st.session_state["edited_reply"],
            })
            if resp2.ok:
                st.success("Đã gửi email.")
                del st.session_state["run_result"]
            else:
                st.error(resp2.text)
        if col2.button("Cancel"):
            requests.post(f"{API_BASE}/approve", json={
                "thread_id": thread_id,
                "approved": False,
            })
            st.info("Đã hủy.")
            del st.session_state["run_result"]
    else:
        st.success("Agent đã hoàn thành.")

Chạy UI:

streamlit run ui/app.py

MemorySaver lưu state trong RAM — mất toàn bộ khi server restart. Để persist qua restart, dùng PostgresSaver:

from langgraph.checkpoint.postgres import PostgresSaver
import os

conn_string = os.environ["DATABASE_URL"]
# vd: "postgresql://user:pass@host:5432/dbname"

checkpointer = PostgresSaver.from_conn_string(conn_string)
checkpointer.setup()  # tạo table cần thiết, an toàn khi chạy nhiều lần

graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["save_action"],
)

Cài thêm dependency:

pip install langgraph-checkpoint-postgres psycopg2-binary

Với PostgresSaver, mỗi thread_id có thể resume độc lập, kể cả khi server đã restart hoặc scale lên nhiều instance.

Agent không có loss function rõ ràng như model ML — cần tự xây evaluation framework.

Test Scenario Format

# tests/test_agent.py
import pytest
from agent.graph import build_graph
from langgraph.checkpoint.memory import MemorySaver

SCENARIOS = [
    {
        "name": "urgent_email_from_boss",
        "input_email": {
            "id": "t1",
            "from": "[email protected]",
            "subject": "URGENT: Server down",
            "body": "Production server is down. Fix immediately.",
        },
        "expected_category": "urgent",
        "expected_tools_called": ["draft_reply", "save_action"],
    },
    {
        "name": "spam_newsletter",
        "input_email": {
            "id": "t2",
            "from": "[email protected]",
            "subject": "50% off sale today only!!!",
            "body": "Click here to buy now...",
        },
        "expected_category": "spam",
        "expected_tools_called": [],
    },
    # ... thêm 20-50 scenario
]


@pytest.mark.parametrize("scenario", SCENARIOS, ids=[s["name"] for s in SCENARIOS])
def test_classification(scenario):
    checkpointer = MemorySaver()
    builder = build_graph()
    graph = builder.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": scenario["name"]}}
    result = graph.invoke(
        {"emails": [scenario["input_email"]], "todos": [], "requires_approval": False},
        config=config,
    )
    state = graph.get_state(config)
    assert state.values.get("category") == scenario["expected_category"]

Metrics Cần Track

Checklist trước khi deploy:

• Không hardcode API key — dùng env var hoặc secret manager.
• Thêm rate limiting trên FastAPI endpoint /run — tránh spam LLM call.
• Log error về Sentry hoặc tương đương — không debug bằng print().
• README có demo URL và hướng dẫn chạy local.

Sau khi single-agent hoạt động ổn định, có thể mở rộng thành multi-agent để show thêm kỹ năng:

• Researcher Agent: nhận email về chủ đề kỹ thuật → tìm kiếm web → tóm tắt.
• Writer Agent: nhận kết quả từ Researcher → soạn reply chi tiết.
• Supervisor Agent: điều phối Researcher và Writer bằng langgraph-supervisor.

# Ví dụ cấu trúc multi-agent với langgraph-supervisor
# pip install langgraph-supervisor

from langgraph_supervisor import create_supervisor
from langchain_openai import ChatOpenAI

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

# researcher_agent và writer_agent là các compiled graph riêng
supervisor = create_supervisor(
    agents=[researcher_agent, writer_agent],
    llm=supervisor_llm,
    prompt="Bạn là supervisor điều phối 2 agent..."
).compile()

Multi-agent không cần thiết cho MVP — thêm khi đã có single-agent hoạt động và muốn show thêm chiều sâu technical trong phỏng vấn.

Project hoàn chỉnh cần đạt:

• Classification accuracy ≥ 85% trên test set 20–50 scenario.
• HITL flow hoạt động đúng: approve gửi email, reject cancel, edit sửa draft trước khi gửi.
• State persist qua server restart (PostgresSaver).
• Demo URL hoạt động — recruiter có thể tự test.
• README có diagram workflow, hướng dẫn chạy local và link demo.

✅ State schema dùng TypedDict + add_messages reducer cho messages
✅ Tools cần docstring đầy đủ — LLM dựa vào đó để routing
✅ add_conditional_edges để route dựa trên state value
✅ HITL = checkpointer + interrupt_before — thiếu 1 trong 2 thì không hoạt động
✅ graph.update_state() cho phép human sửa state trước khi resume
✅ MemorySaver cho dev, PostgresSaver cho production
✅ Evaluation: xây test scenario thủ công, đo classification accuracy và tool call correctness
✅ Deploy: Render (backend) + Streamlit Cloud (UI) + env var cho secrets

Bài 15: Project 5 (optional) — Multi-modal AI Search Engine