10. Sustainable Execution - Thực thi bền bỉ và Phục hồi lỗi

Trong môi trường production, lỗi là điều không thể tránh khỏi: API của LLM có thể bị timeout, server database có thể bị sập giữa chừng, hoặc kết nối internet bị gián đoạn. Sustainable Execution (Thực thi bền bỉ) là khả năng của LangGraph giúp Agent tiếp tục công việc ngay tại nơi nó bị dừng lại mà không phải làm lại từ đầu.

I. Durable Execution là gì?

Durable Execution là một kỹ thuật lưu lại tiến trình tại các điểm mấu chốt. Nếu một quy trình bị gián đoạn, nó có thể "tỉnh dậy" sau đó (thậm chí là nhiều ngày sau) và chạy tiếp từ trạng thái đã lưu.

Để thực thi bền bỉ một cách an toàn, chúng ta cần hai yếu tố:

• Retry Policy: Tự động thử lại khi gặp lỗi tạm thời.
• Tính lũy đẳng (Idempotency): Đảm bảo một hành động (như thanh toán) không bị lặp lại gây hậu quả xấu khi Node phải chạy lại. LangGraph cung cấp hàm task() để giải quyết vấn đề này.

II. Retry Policy: Tuyến phòng thủ đầu tiên

Bạn có thể cấu hình chính sách thử lại trực tiếp khi thêm Node vào Graph. Nếu Node ném ra một ngoại lệ, LangGraph sẽ tự động thực thi lại Node đó theo cấu hình:

.addNode("payment", paymentNode, {
  retry_policy: {
    max_attempts: 2,          // Thử lại tối đa 2 lần
    backoff_factor: 2,       // Thời gian chờ tăng dần giữa các lần thử
  }
})

III. Sử dụng task() để bảo vệ Side-effects

Nếu trong một Node bạn thực hiện gọi API thanh toán rồi mới cập nhật Database. Nếu bước cập nhật DB bị lỗi, khi Node chạy lại, API thanh toán sẽ bị gọi lần nữa.

Để tránh việc "trừ tiền khách hàng 2 lần", chúng ta bọc hành động thanh toán trong hàm task(). LangGraph sẽ lưu kết quả của task vào checkpoint. Khi Node chạy lại, LangGraph sẽ lấy kết quả đã lưu thay vì thực thi lại hành động đó.

IV. Mã nguồn: 10-sustainable-execution.ts

Ví dụ dưới đây minh họa một Agent thanh toán. Chúng ta sẽ giả lập một lỗi Database để quan sát cách Agent phục hồi mà không gọi lại hàm thanh toán.

// path: 10-sustainable-execution.ts
import {
  END,
  MemorySaver,
  START,
  StateGraph,
  task,
} from '@langchain/langgraph';
import { generateImage } from '@workspace/util-langchain';
import z from 'zod';

const State = z.object({
  orderId: z.string(),
  paymentStatus: z.string().optional(),
  dbUpdated: z.boolean().default(false), // Track DB update state
});

// 1. Idempotent Payment Task
// LangGraph ensures this runs ONLY once per thread
const chargeCreditCard = task('charge_card', async (orderId: string) => {
  console.log(`💳 Charging credit card for order: ${orderId}`);
  // Simulate external payment API call
  return 'SUCCESS_TRANSACTION_ID_123';
});

// 2. Main Payment Node
async function paymentNode(state: typeof State.Shape) {
  console.log('\n--- PAYMENT NODE START ---');

  // If node retries, this result is restored from checkpoint
  const receipt = await chargeCreditCard(state.orderId);

  console.log('✅ Payment completed with receipt:', receipt);

  // Simulate DB failure on first attempt
  if (!state.dbUpdated) {
    console.log('❌ Database update failed (simulated crash)');
    throw new Error('Database connection lost!');
  }

  console.log('✅ Database updated successfully');

  return {
    paymentStatus: `Paid with receipt: ${receipt}`,
    dbUpdated: true,
  };
}

// 3. Graph Setup with Retry Policy
const checkpointer = new MemorySaver();

const workflow = new StateGraph(State)
  .addNode('payment', paymentNode, {
    retry_policy: {
      max_attempts: 2, // Retry node up to 2 times
    },
  })
  .addEdge(START, 'payment')
  .addEdge('payment', END)
  .compile({ checkpointer });

// 4. Execution
async function run() {
  await generateImage(
    workflow,
    'graph-ignore/scripts-10-sustainable-execution.jpg',
  );

  const config = { configurable: { thread_id: 'order_001' } };

  try {
    console.log('\n🚀 First Run (Expect Failure + Retry)');
    await workflow.invoke(
      {
        orderId: 'PRO_123',
        dbUpdated: false, // Force failure
      },
      config,
    );
  } catch (err) {
    console.log('⚠️ Workflow failed after all retries');
  }

  console.log('\n🧠 Restoring State From Checkpoint...');
  const snapshot = await workflow.getState(config);
  console.log('Checkpoint State:', snapshot.values);

  console.log('\n🚀 Second Run (Resume + Succeed)');
  // We simulate fixing the DB issue by passing dbUpdated: true
  await workflow.invoke(
    {
      ...snapshot.values,
      dbUpdated: true,
    },
    config,
  );

  const finalState = await workflow.getState(config);
  console.log('\n🎉 FINAL STATE:', finalState.values);
}

run();

V. Giải thích cơ chế hoạt động

1. Lần chạy 1: Node payment được kích hoạt. Task charge_card thực hiện thanh toán thành công. Tuy nhiên, logic DB bị lỗi (throw Error). retry_policy kích hoạt và chạy lại Node lần 2 nhưng vẫn lỗi. Quy trình dừng lại và lưu checkpoint.
2. Điểm mấu chốt: Mặc dù Node payment chạy 2 lần, nhưng log 💳 Charging credit card... chỉ xuất hiện duy nhất 1 lần. Đó là nhờ hàm task() đã ghi nhớ kết quả.
3. Lần chạy 2: Khi chúng ta gọi lại invoke với dữ liệu đã sửa, Node payment chạy lại. LangGraph nạp kết quả thanh toán từ bộ nhớ và đi thẳng tới bước cập nhật DB thành công.

Sơ đồ hoạt động:

VI. Tổng kết

• Sustainable Execution giúp Agent không bị "chết" hoàn toàn khi gặp lỗi hạ tầng.
• Retry Policy xử lý các lỗi API tạm thời một cách tự động.
• task() đảm bảo các hành động quan trọng (Side-effects) không bị thực thi lặp lại, giữ cho hệ thống luôn ở trạng thái an toàn.

Trong bài học tới, chúng ta sẽ học về Streaming, cách để người dùng có thể quan sát tiến trình của Agent theo thời gian thực thay vì phải chờ đợi toàn bộ quy trình bền bỉ này kết thúc!

👉 Bài tiếp theo: 11. Streaming - Tương tác thời gian thực