flare-agent Bài 6: Workflow Engine — Graph-based Multi-step

Series: Build Your Own AI Agent Framework trên Cloudflare Bài: 6 / 7 — @flare-agent/workflow

---

Tại sao cần Workflow Engine?

Agent Loop đơn giản (bài 5) phù hợp cho single-turn Q&A hoặc simple tool calling. Nhưng nhiều use case cần multi-step flow có cấu trúc:

• Vocabulary quiz: present word → user trả lời → evaluate → reward/hint → next word
• Onboarding: collect info → generate study plan → assign first lesson
• RAG pipeline: rephrase query → retrieve context → generate answer → validate

Agent Loop không đủ vì LLM không đảm bảo thứ tự bước, còn Workflow Engine thì deterministic.

---

Cấu trúc package

packages/workflow/
  src/
    WorkflowContext.ts   # shared state giữa steps
    Step.ts              # step definition
    Graph.ts             # DAG builder
    WorkflowEngine.ts    # executor
    index.ts

---

Workflow Context

// src/WorkflowContext.ts
import type { AgentContext } from '@flare-agent/types';

export class WorkflowContext {
  private results = new Map<string, unknown>();

  constructor(
    public readonly agentCtx: AgentContext,
    public readonly input: unknown,
    public metadata: Record<string, unknown> = {}
  ) {}

  // Lưu kết quả của 1 step
  set(stepName: string, result: unknown): void {
    this.results.set(stepName, result);
  }

  // Lấy kết quả của step trước
  get<T = unknown>(stepName: string): T | undefined {
    return this.results.get(stepName) as T;
  }

  // Lấy kết quả của step gần nhất
  getLast<T = unknown>(): T | undefined {
    const keys = [...this.results.keys()];
    if (!keys.length) return undefined;
    return this.results.get(keys[keys.length - 1]) as T;
  }

  getAll(): Record<string, unknown> {
    return Object.fromEntries(this.results);
  }
}

---

Step Definition

// src/Step.ts
import type { WorkflowContext } from './WorkflowContext';

export interface StepDefinition<TOutput = unknown> {
  name: string;
  execute(ctx: WorkflowContext): Promise<TOutput>;
  // Quyết định step tiếp theo dựa trên kết quả
  // null = kết thúc workflow
  next?(result: TOutput, ctx: WorkflowContext): string | null;
}

// Step chạy LLM call
export interface LLMStepDefinition extends StepDefinition<string> {
  prompt: string | ((ctx: WorkflowContext) => string);
}

// Step chạy logic thuần (không cần LLM)
export interface FunctionStepDefinition<TOutput>
  extends StepDefinition<TOutput> {
  execute(ctx: WorkflowContext): Promise<TOutput>;
}

---

Graph Builder

// src/Graph.ts
import type { StepDefinition } from './Step';

export class Graph {
  private steps = new Map<string, StepDefinition>();
  private startStep?: string;

  // Thêm step
  step<T>(definition: StepDefinition<T>): this {
    this.steps.set(definition.name, definition as StepDefinition);
    return this;
  }

  // Set step bắt đầu
  start(stepName: string): this {
    if (!this.steps.has(stepName)) {
      throw new Error(`Step "${stepName}" not found in graph`);
    }
    this.startStep = stepName;
    return this;
  }

  getStep(name: string): StepDefinition | undefined {
    return this.steps.get(name);
  }

  getStartStep(): string {
    if (!this.startStep) {
      // Default: step đầu tiên được thêm vào
      const first = this.steps.keys().next().value;
      if (!first) throw new Error('Graph has no steps');
      return first;
    }
    return this.startStep;
  }

  validate(): void {
    // Kiểm tra không có dangling step references
    for (const step of this.steps.values()) {
      if (step.next) {
        // next là dynamic function — không validate được statically
        // sẽ validate at runtime
      }
    }
  }
}

---

Workflow Engine

// src/WorkflowEngine.ts
import type { AgentContext } from '@flare-agent/types';
import { Graph } from './Graph';
import { WorkflowContext } from './WorkflowContext';

export interface WorkflowResult {
  output: unknown;
  steps: Array<{ name: string; result: unknown; durationMs: number }>;
  totalDurationMs: number;
}

export class WorkflowEngine {
  constructor(
    private graph: Graph,
    private maxSteps = 20 // tránh infinite loop
  ) {}

  async run(
    input: unknown,
    agentCtx: AgentContext
  ): Promise<WorkflowResult> {
    const ctx = new WorkflowContext(agentCtx, input);
    const stepLog: WorkflowResult['steps'] = [];
    const startTime = Date.now();

    let currentStepName: string | null = this.graph.getStartStep();
    let stepCount = 0;

    while (currentStepName !== null) {
      if (stepCount >= this.maxSteps) {
        throw new Error(`Workflow exceeded max steps (${this.maxSteps})`);
      }

      const step = this.graph.getStep(currentStepName);
      if (!step) {
        throw new Error(`Step "${currentStepName}" not found`);
      }

      // Execute step
      const stepStart = Date.now();
      const result = await step.execute(ctx);
      const durationMs = Date.now() - stepStart;

      // Lưu kết quả
      ctx.set(currentStepName, result);
      stepLog.push({ name: currentStepName, result, durationMs });
      stepCount++;

      // Quyết định step tiếp theo
      if (step.next) {
        currentStepName = step.next(result, ctx);
      } else {
        currentStepName = null; // step cuối, kết thúc
      }
    }

    return {
      output: ctx.getLast(),
      steps: stepLog,
      totalDurationMs: Date.now() - startTime,
    };
  }
}

---

Ví dụ thực tế — Vocabulary Quiz Flow

import { Graph, WorkflowEngine } from '@flare-agent/workflow';

const quizGraph = new Graph()
  .step({
    name: 'fetch-word',
    execute: async (ctx) => {
      const db = ctx.agentCtx.env.DB as D1Database;
      const userId = ctx.agentCtx.userId!;
      // Lấy từ chưa học
      return db.prepare(
        `SELECT * FROM vocabulary
         WHERE id NOT IN (
           SELECT word_id FROM progress WHERE user_id = ?
         ) LIMIT 1`
      ).bind(userId).first();
    },
    next: (word) => word ? 'present-word' : 'no-more-words',
  })
  .step({
    name: 'present-word',
    execute: async (ctx) => {
      const word = ctx.get<any>('fetch-word');
      // Trả về word data — frontend hiển thị
      return { word: word.word, hint: word.definition };
    },
    next: () => 'evaluate-answer',
  })
  .step({
    name: 'evaluate-answer',
    execute: async (ctx) => {
      const word = ctx.get<any>('fetch-word');
      const userAnswer = ctx.input as string;
      const correct = userAnswer.toLowerCase() === word.word.toLowerCase();
      return { correct, word: word.word };
    },
    next: (result) => result.correct ? 'save-progress' : 'give-hint',
  })
  .step({
    name: 'give-hint',
    execute: async (ctx) => {
      const word = ctx.get<any>('fetch-word');
      return { hint: word.example_sentence };
    },
    next: () => null, // kết thúc — chờ user thử lại
  })
  .step({
    name: 'save-progress',
    execute: async (ctx) => {
      const word = ctx.get<any>('fetch-word');
      const db = ctx.agentCtx.env.DB as D1Database;
      await db.prepare(
        'INSERT INTO progress (user_id, word_id, status) VALUES (?, ?, ?)'
      ).bind(ctx.agentCtx.userId, word.id, 'learned').run();
      return { saved: true, nextStep: 'fetch-word' };
    },
    next: () => 'fetch-word', // loop — lấy từ tiếp theo
  })
  .step({
    name: 'no-more-words',
    execute: async () => ({ message: 'Bạn đã học hết tất cả từ vựng! 🎉' }),
    // no next → kết thúc
  })
  .start('fetch-word');

// Chạy workflow
const engine = new WorkflowEngine(quizGraph);
const result = await engine.run(userAnswer, agentCtx);

---

Checklist

• Tạo packages/workflow/
• Test quiz flow với mock D1
• Verify conditional branching hoạt động đúng
• Test max steps protection

---

Bài cuối: Bài 7 — Áp dụng vào edu-ai-platform: End-to-end