import { Mastra, Agent } from '@mastra/core';
import { anthropic } from '@ai-sdk/anthropic';
import { queryOrderTool, initiateRefundTool, queryShippingTool } from './tools';
 
// 电商租户的客服 agent
const ecommerceAgent = new Agent({
  name: 'ecommerce-customer-service',
  // system prompt 会在每次对话开始时注入
  instructions: `你是电商平台的智能客服助手。你的职责是帮助用户查询订单状态、处理退款申请和跟踪物流信息。
 
处理原则：
- 查询订单时，必须先确认订单号格式（纯数字，6-12位）
- 退款申请超过1000元时，告知用户需要人工审核，预计24小时内处理
- 物流查询结果中，如果显示"已签收"，询问用户是否收到商品
- 用户表达不满时，先表达理解，再给出解决方案
 
禁止的行为：
- 不能承诺具体的退款时间（退款时间由财务系统决定）
- 不能透露其他用户的订单信息`,
  model: anthropic('claude-sonnet-4-5'),
  tools: {
    queryOrder: queryOrderTool,
    initiateRefund: initiateRefundTool,
    queryShipping: queryShippingTool,
  },
});

// 非流式调用，适合后台批处理
const response = await ecommerceAgent.generate([
  { role: 'user', content: '我的订单 #12345 在哪里？' },
]);
console.log(response.text);
 
// 流式调用，适合实时对话界面
const stream = await ecommerceAgent.stream([
  { role: 'user', content: '我的订单 #12345 在哪里？' },
]);
for await (const chunk of stream.textStream) {
  process.stdout.write(chunk);
}

import { createTool } from '@mastra/core/tools';
import { z } from 'zod';
 
export const queryOrderTool = createTool({
  id: 'query-order',
  description: '根据订单ID查询订单状态和基本信息。输入订单ID，返回状态、金额、收货地址。',
  inputSchema: z.object({
    orderId: z.string().describe('订单ID，纯数字格式，例如：12345'),
  }),
  execute: async ({ context }) => {
    const { orderId } = context;
    // 实际项目中这里调用内部订单系统 API
    // 本示例用 mock 数据模拟
    const mockOrders: Record<string, object> = {
      '12345': {
        orderId: '12345',
        status: 'shipped',           // 已发货
        amount: 299.00,
        createdAt: '2024-03-10',
        carrier: 'SF Express',
        trackingNumber: 'SF7890123456',
        address: '北京市朝阳区某某街道',
      },
      '67890': {
        orderId: '67890',
        status: 'processing',        // 处理中
        amount: 1299.00,
        createdAt: '2024-03-12',
        carrier: null,
        trackingNumber: null,
        address: '上海市浦东新区某某路',
      },
    };
 
    const order = mockOrders[orderId];
    if (!order) {
      return { error: `订单 #${orderId} 不存在` };
    }
    return order;
  },
});

import { Memory } from '@mastra/memory';
import { PgVector, PostgresStore } from '@mastra/pg';
 
// 使用 PostgreSQL 持久化记忆（生产必须）
const memory = new Memory({
  // 存储后端：对话历史存 PostgreSQL
  storage: new PostgresStore({
    connectionString: process.env.DATABASE_URL!,
  }),
  // 向量后端：语义检索用 pgvector
  vector: new PgVector(process.env.DATABASE_URL!),
  options: {
    // 保留最近 20 条消息（超出的部分通过语义检索补充）
    lastMessages: 20,
    // 语义检索：召回最相关的 5 条历史消息
    semanticRecall: {
      topK: 5,
      messageRange: { before: 2, after: 1 },
    },
  },
});

// 生成隔离的 threadId
function buildThreadId(tenantId: string, userId: string, sessionId: string): string {
  return `${tenantId}:${userId}:${sessionId}`;
}
 
// 调用时传入 threadId
const response = await ecommerceAgent.generate(
  [{ role: 'user', content: userMessage }],
  {
    resourceId: userId,           // Mastra 用于标识用户
    threadId: buildThreadId(tenantId, userId, sessionId),
  }
);

// 错误：只用 sessionId，不同租户可能碰撞
const threadId = sessionId;
 
// 正确：包含 tenantId，彻底隔离
const threadId = `${tenantId}:${userId}:${sessionId}`;

import { createWorkflow, createStep } from '@mastra/core';
import { z } from 'zod';
 
// 步骤1：意图分类
const classifyIntentStep = createStep({
  id: 'classify-intent',
  inputSchema: z.object({ message: z.string() }),
  outputSchema: z.object({
    intent: z.enum(['query_order', 'refund', 'shipping', 'other']),
    orderId: z.string().optional(),
  }),
  execute: async ({ inputData }) => {
    // 实际项目中调用 LLM 做意图分类
    // 这里用简单规则模拟
    const message = inputData.message;
    const orderMatch = message.match(/#(\d+)/);
    const orderId = orderMatch ? orderMatch[1] : undefined;
 
    if (message.includes('在哪') || message.includes('物流') || message.includes('快递')) {
      return { intent: 'shipping', orderId };
    }
    if (message.includes('退款') || message.includes('退货')) {
      return { intent: 'refund', orderId };
    }
    if (orderId) {
      return { intent: 'query_order', orderId };
    }
    return { intent: 'other' };
  },
});
 
// 步骤2：执行订单查询
const executeOrderQueryStep = createStep({
  id: 'execute-order-query',
  inputSchema: z.object({
    intent: z.string(),
    orderId: z.string().optional(),
  }),
  outputSchema: z.object({ result: z.any() }),
  execute: async ({ inputData }) => {
    if (!inputData.orderId) {
      return { result: { error: '未识别到订单号' } };
    }
    // 调用订单查询逻辑
    return { result: { orderId: inputData.orderId, status: 'shipped' } };
  },
});
 
// 组合成 workflow
const orderQueryWorkflow = createWorkflow({
  id: 'order-query',
  inputSchema: z.object({ message: z.string() }),
  outputSchema: z.object({ result: z.any() }),
})
  .then(classifyIntentStep)
  .then(executeOrderQueryStep)
  .commit();

import { Agent } from '@mastra/core';
import { generateObject } from 'ai';
import { anthropic } from '@ai-sdk/anthropic';
import { z } from 'zod';
 
// 子任务的数据结构
interface TaskPlan {
  id: string;
  description: string;    // 对 Executor 的自然语言说明
  tool?: string;          // 对应的 tool id（可选，有些 task 不需要 tool）
  params?: Record<string, unknown>;  // tool 的入参（Planner 预填，Executor 可覆盖）
  dependsOn?: string[];   // 依赖的其他 task id（有依赖的 task 不能并行）
  type: 'query' | 'mutation';        // 只读 or 写操作
}
 
interface TaskResult {
  taskId: string;
  success: boolean;
  data?: unknown;
  error?: string;
}
 
interface SessionContext {
  tenantId: string;
  userId: string;
  sessionId: string;
}
 
interface AgentResponse {
  text: string;
  tasks: TaskResult[];
  plannerTokens: number;
  reflectorTokens: number;
}
 
// Planner：把用户消息分解为子任务列表
async function runPlanner(
  userMessage: string,
  availableTools: string[],
): Promise<{ plans: TaskPlan[]; usage: { inputTokens: number; outputTokens: number } }> {
  const result = await generateObject({
    model: anthropic('claude-sonnet-4-5'),
    system: `你是任务规划器。根据用户消息，把任务分解为有序的子任务列表。
可用工具：${availableTools.join(', ')}
 
规则：
- 每个子任务对应一个具体操作
- 如果子任务 B 依赖子任务 A 的结果，在 dependsOn 中声明
- 只读操作标记 type: query，写操作标记 type: mutation
- 任务越细粒度越好，但不要过度拆分（不超过5个子任务）`,
    prompt: userMessage,
    schema: z.object({
      plans: z.array(
        z.object({
          id: z.string(),
          description: z.string(),
          tool: z.string().optional(),
          params: z.record(z.unknown()).optional(),
          dependsOn: z.array(z.string()).optional(),
          type: z.enum(['query', 'mutation']),
        })
      ),
    }),
  });
 
  return {
    plans: result.object.plans,
    usage: {
      inputTokens: result.usage.promptTokens,
      outputTokens: result.usage.completionTokens,
    },
  };
}
 
// Executor：执行单个 task
async function executeTask(
  task: TaskPlan,
  agent: Agent,
  previousResults: Map<string, TaskResult>,
): Promise<TaskResult> {
  try {
    // 把依赖 task 的结果注入到当前 task 的 prompt 中
    const dependencyContext = task.dependsOn
      ?.map((depId) => {
        const dep = previousResults.get(depId);
        return dep ? `${depId} 的结果：${JSON.stringify(dep.data)}` : '';
      })
      .filter(Boolean)
      .join('\n');
 
    const prompt = dependencyContext
      ? `${task.description}\n\n相关上下文：\n${dependencyContext}`
      : task.description;
 
    const response = await agent.generate([{ role: 'user', content: prompt }]);
 
    return {
      taskId: task.id,
      success: true,
      data: response.text,
    };
  } catch (err) {
    return {
      taskId: task.id,
      success: false,
      error: err instanceof Error ? err.message : String(err),
    };
  }
}
 
// Reflector：验证执行结果是否满足用户意图
async function runReflector(
  userMessage: string,
  plans: TaskPlan[],
  results: TaskResult[],
): Promise<{ satisfied: boolean; feedback: string; usage: { inputTokens: number; outputTokens: number } }> {
  const resultSummary = results
    .map((r) => `- ${r.taskId}: ${r.success ? '成功' : '失败'} | ${JSON.stringify(r.data || r.error)}`)
    .join('\n');
 
  const reflectResult = await generateObject({
    model: anthropic('claude-sonnet-4-5'),
    system: '你是结果验证器。判断任务执行结果是否完整满足用户的原始意图。',
    prompt: `用户原始消息：${userMessage}
 
执行计划：${plans.map((p) => p.description).join('; ')}
 
执行结果：
${resultSummary}
 
请判断结果是否满足用户意图，并给出简短说明。`,
    schema: z.object({
      satisfied: z.boolean().describe('结果是否完整满足用户意图'),
      feedback: z.string().describe('说明原因，如果不满足请指出缺少什么'),
    }),
  });
 
  return {
    satisfied: reflectResult.object.satisfied,
    feedback: reflectResult.object.feedback,
    usage: {
      inputTokens: reflectResult.usage.promptTokens,
      outputTokens: reflectResult.usage.completionTokens,
    },
  };
}
 
// 主函数：串联三层
export async function runPlannerExecutorReflector(
  agent: Agent,
  userMessage: string,
  _context: SessionContext,
  availableTools: string[],
): Promise<AgentResponse> {
  // 第一层：Planner 分解任务
  const { plans, usage: plannerUsage } = await runPlanner(userMessage, availableTools);
 
  // 判断是否跳过 Reflector（简单查询任务）
  const shouldSkipReflector =
    plans.length === 1 && plans[0].type === 'query';
 
  // 第二层：Executor 并行执行无依赖的任务，串行执行有依赖的任务
  const results: TaskResult[] = [];
  const resultMap = new Map<string, TaskResult>();
 
  // 拓扑排序：按依赖关系分层
  const executed = new Set<string>();
  let iterations = 0;
  const maxIterations = plans.length + 1;
 
  while (executed.size < plans.length && iterations < maxIterations) {
    iterations++;
    // 找出所有依赖已满足、尚未执行的 task
    const readyTasks = plans.filter((task) => {
      if (executed.has(task.id)) return false;
      if (!task.dependsOn || task.dependsOn.length === 0) return true;
      return task.dependsOn.every((dep) => executed.has(dep));
    });
 
    if (readyTasks.length === 0) break; // 防止循环依赖死锁
 
    // 并行执行这一批 ready 的 tasks
    const batchResults = await Promise.all(
      readyTasks.map((task) => executeTask(task, agent, resultMap))
    );
 
    for (const result of batchResults) {
      results.push(result);
      resultMap.set(result.taskId, result);
      executed.add(result.taskId);
    }
  }
 
  // 第三层：Reflector 验证结果（简单查询跳过）
  let reflectorUsage = { inputTokens: 0, outputTokens: 0 };
  if (!shouldSkipReflector) {
    const reflection = await runReflector(userMessage, plans, results);
    reflectorUsage = reflection.usage;
 
    if (!reflection.satisfied) {
      // 结果不满足时，这里可以触发重新规划
      // 生产中需要设置最大重试次数（通常 2 次），防止无限循环
      console.warn(`Reflector 判定结果不满足：${reflection.feedback}`);
    }
  }
 
  // 汇总所有 task 结果，生成最终回复
  const successResults = results.filter((r) => r.success).map((r) => r.data);
  const finalText = successResults.join('\n');
 
  return {
    text: finalText,
    tasks: results,
    plannerTokens: plannerUsage.inputTokens + plannerUsage.outputTokens,
    reflectorTokens: reflectorUsage.inputTokens + reflectorUsage.outputTokens,
  };
}

租户A（电商）普通客服对话
  → Mastra Agent
  → 查订单 / 查物流 / 发起退款（金额 ≤ 1000元）

租户A 退款审批流程（金额 > 1000元）
  → LangGraph.js StateGraph
  → 节点1：提取退款信息
  → 节点2：调用退款预审 API
  → 节点3：interrupt，等待人工审批（通过飞书机器人推送审批消息）
  → 节点4：执行退款，更新订单状态

租户B（SaaS）普通客服对话
  → Mastra Agent
  → 查账单 / 创建工单 / 文档问答

租户C（金融机构）合规报告生成
  → LangGraph.js StateGraph + Temporal
  → 可能涉及 50+ 次 LLM call，执行时间 3-5 分钟
  → 每个节点的输入/输出写入审计日志

import { StateGraph, Annotation, interrupt, Command } from '@langchain/langgraph';
import { PostgresSaver } from '@langchain/langgraph-checkpoint-postgres';
 
// 定义 workflow 的共享状态
const RefundState = Annotation.Root({
  orderId: Annotation<string>(),
  amount: Annotation<number>(),
  reason: Annotation<string>(),
  approvalStatus: Annotation<'pending' | 'approved' | 'rejected'>({
    default: () => 'pending',
    reducer: (_, next) => next,
  }),
  refundId: Annotation<string | null>({
    default: () => null,
    reducer: (_, next) => next,
  }),
});
 
// 节点1：验证退款信息
async function validateRefund(state: typeof RefundState.State) {
  // 调用内部 API 验证订单和退款金额
  console.log(`验证退款：订单 ${state.orderId}，金额 ${state.amount} 元`);
  return state; // 验证通过，状态不变
}
 
// 节点2：等待人工审批（interrupt 节点）
async function waitForApproval(state: typeof RefundState.State) {
  // interrupt 会暂停 workflow，把当前状态存入 checkpoint
  // 外部系统（飞书审批机器人）收到通知后，调用 graph.invoke 传入审批结果继续执行
  const decision = interrupt({
    message: `请审批退款申请`,
    orderId: state.orderId,
    amount: state.amount,
    reason: state.reason,
  });
 
  // decision 是审批结果：{ approved: true } 或 { approved: false, reason: '...' }
  return {
    approvalStatus: (decision as { approved: boolean }).approved ? 'approved' : 'rejected',
  };
}
 
// 节点3：执行退款
async function executeRefund(state: typeof RefundState.State) {
  if (state.approvalStatus !== 'approved') {
    console.log('退款被拒绝，流程结束');
    return state;
  }
 
  const refundId = `RF${Date.now()}`;
  // 调用退款 API
  console.log(`执行退款：${refundId}`);
  return { refundId };
}
 
// 构建 workflow
const builder = new StateGraph(RefundState)
  .addNode('validate', validateRefund)
  .addNode('waitForApproval', waitForApproval)
  .addNode('executeRefund', executeRefund)
  .addEdge('__start__', 'validate')
  .addEdge('validate', 'waitForApproval')
  .addEdge('waitForApproval', 'executeRefund')
  .addEdge('executeRefund', '__end__');
 
// checkpointer：把每个节点的执行状态持久化到 PostgreSQL
// 进程重启后从最后一个完成的节点继续，而不是从头开始
const checkpointer = PostgresSaver.fromConnString(process.env.DATABASE_URL!);
await checkpointer.setup();
 
const refundWorkflow = builder.compile({ checkpointer });
 
// 第一次调用：执行到 interrupt 节点时暂停
const threadConfig = { configurable: { thread_id: 'refund-12345-001' } };
const result1 = await refundWorkflow.invoke(
  { orderId: '67890', amount: 1299, reason: '商品质量问题' },
  threadConfig
);
// result1.__interrupt__ 非空，说明 workflow 在等待审批
 
// 人工审批完成后，传入结果继续执行
const result2 = await refundWorkflow.invoke(
  new Command({ resume: { approved: true } }),
  threadConfig // 相同的 thread_id，从上次暂停的地方继续
);

import { createTool } from '@mastra/core/tools';
import { z } from 'zod';
 
// ---- 订单查询工具 ----
export const queryOrderTool = createTool({
  id: 'query-order',
  description: '根据订单ID查询订单状态、金额、商品列表和收货地址。用户询问订单情况时调用此工具。',
  inputSchema: z.object({
    orderId: z.string().describe('订单ID，6-12位纯数字'),
  }),
  execute: async ({ context }) => {
    const { orderId } = context;
 
    // Mock 数据：模拟真实订单系统的返回格式
    const orders: Record<string, object> = {
      '12345': {
        orderId: '12345',
        status: 'shipped',
        statusLabel: '已发货',
        amount: 299.00,
        items: [{ name: '无线耳机', qty: 1, price: 299.00 }],
        createdAt: '2024-03-10T14:30:00Z',
        paidAt: '2024-03-10T14:31:00Z',
        carrier: 'SF Express',
        trackingNumber: 'SF7890123456',
        shippingAddress: {
          name: '张三',
          phone: '138****8888',
          address: '北京市朝阳区某某街道 18 号',
        },
      },
      '67890': {
        orderId: '67890',
        status: 'processing',
        statusLabel: '备货中',
        amount: 1299.00,
        items: [{ name: '机械键盘', qty: 1, price: 1299.00 }],
        createdAt: '2024-03-12T09:00:00Z',
        paidAt: '2024-03-12T09:01:00Z',
        carrier: null,
        trackingNumber: null,
        shippingAddress: {
          name: '李四',
          phone: '139****9999',
          address: '上海市浦东新区某某路 88 号',
        },
      },
    };
 
    const order = orders[orderId];
    if (!order) {
      return { found: false, error: `订单 #${orderId} 不存在，请确认订单号是否正确` };
    }
    return { found: true, order };
  },
});
 
// ---- 物流查询工具 ----
export const queryShippingTool = createTool({
  id: 'query-shipping',
  description: '根据快递单号查询物流轨迹和预计送达时间。需要先调用 query-order 获取 trackingNumber。',
  inputSchema: z.object({
    trackingNumber: z.string().describe('快递单号，例如：SF7890123456'),
    carrier: z.string().describe('快递公司名称，例如：SF Express'),
  }),
  execute: async ({ context }) => {
    const { trackingNumber } = context;
 
    // Mock 数据：模拟物流系统的返回格式
    const shippingData: Record<string, object> = {
      'SF7890123456': {
        trackingNumber: 'SF7890123456',
        status: 'in_transit',
        statusLabel: '运输中',
        estimatedDelivery: '2024-03-13',
        currentLocation: '北京转运中心',
        timeline: [
          { time: '2024-03-12 08:30', location: '深圳发货仓库', event: '包裹已揽收' },
          { time: '2024-03-12 15:00', location: '深圳机场', event: '已发出，飞机运输中' },
          { time: '2024-03-12 22:00', location: '北京首都机场', event: '已到达目的地转运中心' },
          { time: '2024-03-13 06:00', location: '北京转运中心', event: '分拣中' },
        ],
      },
    };
 
    const shipping = shippingData[trackingNumber];
    if (!shipping) {
      return { found: false, error: `快递单号 ${trackingNumber} 暂无轨迹信息，可能尚未揽收` };
    }
    return { found: true, shipping };
  },
});
 
// ---- 退款发起工具 ----
export const initiateRefundTool = createTool({
  id: 'initiate-refund',
  description: '为指定订单发起退款申请。超过1000元的退款需要人工审批，此工具只负责提交申请。',
  inputSchema: z.object({
    orderId: z.string().describe('订单ID'),
    reason: z.string().describe('退款原因，例如：商品质量问题、不想要了、发错货'),
    amount: z.number().describe('退款金额，单位：元。不能超过订单金额'),
  }),
  execute: async ({ context }) => {
    const { orderId, reason, amount } = context;
 
    // 超过1000元需要人工审批
    const requiresApproval = amount > 1000;
 
    // Mock 退款申请逻辑
    const refundId = `RF${Date.now()}`;
 
    return {
      refundId,
      orderId,
      amount,
      reason,
      status: requiresApproval ? 'pending_approval' : 'processing',
      statusLabel: requiresApproval ? '待人工审批' : '处理中',
      message: requiresApproval
        ? `退款申请 ${refundId} 已提交。由于金额超过1000元，需要人工审核，预计24小时内完成审核并通知您。`
        : `退款申请 ${refundId} 已提交，预计3-5个工作日到账。`,
      estimatedTime: requiresApproval ? '24小时内审核' : '3-5个工作日',
    };
  },
});

import { Mastra, Agent } from '@mastra/core';
import { anthropic } from '@ai-sdk/anthropic';
import { Memory } from '@mastra/memory';
import { queryOrderTool, initiateRefundTool, queryShippingTool } from './tools';
 
// 构建电商客服 agent
// tenantId 用于隔离不同租户的配置和数据
export function createEcommerceAgent(tenantId: string): Agent {
  return new Agent({
    name: `ecommerce-customer-service-${tenantId}`,
    instructions: `你是一个电商平台的智能客服助手，服务于平台的所有用户。
 
你的能力范围：
- 查询订单状态和详情（工具：query-order）
- 查询物流轨迹（工具：query-shipping，需要先有快递单号）
- 发起退款申请（工具：initiate-refund）
 
处理订单查询的标准流程：
1. 从用户消息中提取订单号（#后面的数字，或直接提到的数字串）
2. 调用 query-order 获取订单信息
3. 如果订单状态是 shipped，主动调用 query-shipping 获取物流详情
4. 整合两个工具的结果，给出完整回复
 
处理退款的标准流程：
1. 确认订单号和退款原因
2. 调用 query-order 获取订单金额（确保退款金额不超过实际支付金额）
3. 调用 initiate-refund 提交申请
4. 根据金额告知用户是否需要人工审批
 
重要约束：
- 退款金额超过1000元时，明确告知需要人工审批，不要给出确定的退款时间
- 不要向用户透露系统内部错误信息，统一回复"系统繁忙，请稍后重试"
- 涉及隐私的订单信息（完整手机号、完整地址）不要完整输出，保持脱敏格式
 
回复风格：
- 简洁直接，不要过多寒暄
- 数字用中文表达习惯（3-5个工作日，而不是"3~5 business days"）
- 如果需要多步操作，在回复末尾说明已经完成了什么、还需要等待什么`,
 
    model: anthropic('claude-sonnet-4-5'),
    tools: {
      queryOrder: queryOrderTool,
      queryShipping: queryShippingTool,
      initiateRefund: initiateRefundTool,
    },
  });
}
 
// Mastra 实例：全局单例，管理所有 agent 和服务
export function createMastraInstance(): Mastra {
  return new Mastra({
    agents: {
      ecommerceAgent: createEcommerceAgent('tenant-ecommerce'),
    },
  });
}

// 模拟一次完整对话
async function demoConversation() {
  const mastra = createMastraInstance();
  const agent = mastra.getAgent('ecommerceAgent');
 
  const tenantId = 'tenant-ecommerce';
  const userId = 'user-001';
  const sessionId = 'session-abc123';
 
  // 构建隔离的 threadId
  const threadId = `${tenantId}:${userId}:${sessionId}`;
 
  console.log('用户：我的订单 #12345 什么时候到？\n');
 
  // 第一轮：查询订单状态
  const response = await agent.generate(
    [{ role: 'user', content: '我的订单 #12345 什么时候到？' }],
    {
      resourceId: userId,
      threadId,
    }
  );
 
  // Agent 内部执行过程（通过工具调用日志可以看到）：
  // 1. 识别意图：查询订单，提取订单号 12345
  // 2. 调用 queryOrder({ orderId: '12345' })
  //    → 返回：status: 'shipped', trackingNumber: 'SF7890123456'
  // 3. 发现状态是 shipped，自动调用 queryShipping({ trackingNumber: 'SF7890123456', carrier: 'SF Express' })
  //    → 返回：in_transit，预计明天到达，当前位置北京转运中心
  // 4. 整合两个工具的结果，生成回复
 
  console.log('助手：', response.text);
  // 输出示例：
  // 您的订单 #12345（无线耳机）已发货，快递单号 SF7890123456（顺丰速运）。
  // 目前包裹在北京转运中心，预计明天（3月13日）送达。
  // 如有疑问可拨打顺丰客服 400-811-1111。
}

import Fastify from 'fastify';
import { streamText } from 'ai';
import { anthropic } from '@ai-sdk/anthropic';
import { createMastraInstance } from './agent';
 
const fastify = Fastify({ logger: true });
 
// 流式聊天端点
fastify.post('/api/chat', async (request, reply) => {
  const { message, sessionId, tenantId, userId } = request.body as {
    message: string;
    sessionId: string;
    tenantId: string;
    userId: string;
  };
 
  // 设置 SSE 响应头
  reply.raw.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive',
    // CORS 头（如果前端和后端不同域）
    'Access-Control-Allow-Origin': '*',
  });
 
  const threadId = `${tenantId}:${userId}:${sessionId}`;
  const mastra = createMastraInstance();
  const agent = mastra.getAgent('ecommerceAgent');
 
  try {
    // 获取对话历史（实际项目中从 Memory 读取）
    const history: Array<{ role: 'user' | 'assistant'; content: string }> = [];
 
    // 使用 Vercel AI SDK 的 streamText
    // Mastra agent 的 stream 方法内部也是基于 streamText
    const result = await streamText({
      model: anthropic('claude-sonnet-4-5'),
      system: agent.instructions,
      messages: [
        ...history,
        { role: 'user', content: message },
      ],
      tools: {
        // 把 Mastra 工具转换为 Vercel AI SDK 格式
        // 实际项目中通过 agent.stream() 直接处理，这里展示底层机制
        queryOrder: {
          description: queryOrderTool.description,
          parameters: queryOrderTool.inputSchema,
          execute: async (args) => {
            return queryOrderTool.execute({ context: args, threadId } as Parameters<typeof queryOrderTool.execute>[0]);
          },
        },
      },
      onChunk: ({ chunk }) => {
        // 每个 chunk 立即推送给客户端
        if (chunk.type === 'text-delta') {
          // SSE 格式：data: <json>\n\n
          reply.raw.write(`data: ${JSON.stringify({ type: 'text', content: chunk.textDelta })}\n\n`);
        } else if (chunk.type === 'tool-call') {
          // 通知前端：正在调用工具（可用于显示"查询中..."提示）
          reply.raw.write(`data: ${JSON.stringify({ type: 'tool-call', toolName: chunk.toolName })}\n\n`);
        }
      },
      onFinish: ({ usage }) => {
        // 流结束时，推送 token 用量（用于计费和监控）
        // 注意：流式模式下 usage 在最后一个 chunk 之后才能拿到
        reply.raw.write(`data: ${JSON.stringify({ type: 'usage', usage })}\n\n`);
        reply.raw.write('data: [DONE]\n\n');
        reply.raw.end();
      },
    });
 
    // 等待流结束（onFinish 里已经处理了 end，这里只是等待 promise resolve）
    await result.text;
  } catch (err) {
    // 错误时推送错误事件，然后关闭连接
    const errorMessage = err instanceof Error ? err.message : '未知错误';
    reply.raw.write(`data: ${JSON.stringify({ type: 'error', message: errorMessage })}\n\n`);
    reply.raw.end();
  }
});

// 前端：消费 SSE 流
async function chat(message: string) {
  const response = await fetch('/api/chat', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      message,
      sessionId: 'session-001',
      tenantId: 'tenant-ecommerce',
      userId: 'user-001',
    }),
  });
 
  const reader = response.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = '';
 
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
 
    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split('\n\n');
    buffer = lines.pop() || '';   // 最后一段可能不完整，保留到下次
 
    for (const line of lines) {
      if (!line.startsWith('data: ')) continue;
      const data = line.slice(6);
      if (data === '[DONE]') return;
 
      const event = JSON.parse(data);
      if (event.type === 'text') {
        process.stdout.write(event.content);  // 逐字追加到界面
      } else if (event.type === 'tool-call') {
        console.log(`\n[正在${event.toolName === 'queryOrder' ? '查询订单' : '查询物流'}...]`);
      } else if (event.type === 'usage') {
        console.log(`\n[Token 用量 - 输入: ${event.usage.promptTokens}, 输出: ${event.usage.completionTokens}]`);
      }
    }
  }
}

const result = await streamText({
  model: anthropic('claude-sonnet-4-5'),
  messages: [...],
  onStepFinish: ({ usage, stepType }) => {
    // 每一步（包括工具调用的中间步骤）都会触发
    // 累加 usage 得到完整 token 消耗
    totalInputTokens += usage.promptTokens;
    totalOutputTokens += usage.completionTokens;
  },
  onFinish: ({ usage }) => {
    // 这里的 usage 只包含最后一步
    // 用 totalInputTokens / totalOutputTokens 才是完整数据
  },
});