xFlow vs Mastra

// Mastra — minimal workflow + agent.
import { Mastra } from "@mastra/core"
import { Agent } from "@mastra/core/agent"
import { createWorkflow, createStep } from "@mastra/core/workflows"
import { z } from "zod"
import { openai } from "@ai-sdk/openai"

const triage = new Agent({
  name: "triage",
  model: openai("gpt-4o"),
  instructions: "Classify the issue.",
  tools: { /* ... */ },
})

const classify = createStep({
  id: "classify",
  inputSchema:  z.object({ issue: z.string() }),
  outputSchema: z.object({ label: z.enum(["bug", "feature", "support"]) }),
  execute: async ({ inputData }) => {
    const r = await triage.generate([{ role: "user", content: inputData.issue }])
    return { label: r.text as "bug" | "feature" | "support" }
  },
})

const handle = createStep({ /* ... */ })

const wf = createWorkflow({
  id: "issue.triage",
  inputSchema:  z.object({ issue: z.string() }),
  outputSchema: z.object({ ticketId: z.string() }),
})
  .then(classify)
  .branch([
    [async ({ inputData }) => inputData.label === "bug",     bugStep],
    [async ({ inputData }) => inputData.label === "feature", featureStep],
    [async ({ inputData }) => inputData.label === "support", supportStep],
  ])
  .commit()

new Mastra({ agents: { triage }, workflows: { issueTriage: wf } })

// Mastra workflow composition primitives.

createWorkflow({ id, inputSchema, outputSchema })
  .then(step)                                  // sequence
  .parallel([step1, step2, step3])             // concurrent fan-out
  .branch([[condFn, stepA], [condFn, stepB]])  // exclusive routing
  .dountil(step, condFn)                       // loop until cond
  .dowhile(step, condFn)                       // loop while cond
  .foreach(step, { items: "$.list" })          // iterate over array
  .map(step)                                   // transform input → output
  .sleep(durationMs)                           // pause
  .waitForEvent(eventName)                     // pause for external signal
  .commit()                                    // finalize the workflow

// Steps are async functions:
createStep({
  id, inputSchema, outputSchema,
  execute: async ({ inputData, getStepResult, suspend, resume }) => { ... }
})

// Mastra agent runtime — Agent class.

const agent = new Agent({
  name:         "support-bot",
  description:  "answers product questions",
  instructions: "You are a helpful assistant...",
  model:        openai("gpt-4o"),
  tools:        { search, lookupOrder, escalate },
  memory:       new Memory({ storage: pg, vector: pg, embedder: openai("text-embedding-3-small") }),
  evals:        { /* metric registrations */ },
  voice:        new OpenAIVoice({ /* ... */ }),
})

// Three calling shapes:
const r  = await agent.generate(messages, { runId, threadId })
const s  = await agent.stream(messages,   { runId, threadId })
const so = await agent.streamObject(...)

// Tools are local LangChain-style functions (or AI SDK 'tool()'):
const search = createTool({
  id:           "search",
  description:  "search docs",
  inputSchema:  z.object({ q: z.string() }),
  outputSchema: z.object({ results: z.array(z.string()) }),
  execute:      async ({ context: { q } }) => fetchDocs(q),
})

// All tool execution is in-process. No id@version, no placement,
// no claim model, no signing, no multi-writer.

// Mastra memory — thread-based.

new Memory({
  storage:  pg,                   // libsql / pg / upstash / d1 / ...
  vector:   pg,                   // pgvector / pinecone / qdrant / ...
  embedder: openai("text-embedding-3-small"),
  options: {
    lastMessages: 20,
    semanticRecall: { topK: 5, messageRange: 4 },
    workingMemory: { enabled: true, template: "..." },
  },
})

// Threads = conversation containers; messages stored verbatim.
// Working memory = an evolving JSON document the agent maintains.
// Semantic recall = vector lookup over message history.

// Memory is per-agent, not per-flow. Workflow runs and agent threads
// are different persistence concepts in Mastra.

// Mastra suspend/resume — built-in HITL.

const approve = createStep({
  id: "approve",
  inputSchema:  z.object({ amount: z.number() }),
  outputSchema: z.object({ approved: z.boolean() }),
  resumeSchema: z.object({ decision: z.enum(["yes", "no"]) }),
  execute: async ({ inputData, suspend, resumeData }) => {
    if (!resumeData) {
      await suspend({ ask: "approve charge?", amount: inputData.amount })
      return { approved: false }              // unreached on suspend
    }
    return { approved: resumeData.decision === "yes" }
  },
})

// Resume from outside:
await mastra.getWorkflow("checkout").createRun({ runId }).resume({
  step: approve,
  resumeData: { decision: "yes" },
})

// State is persisted to the storage layer between suspend and resume.