Mastra Agent 安全与审批

Guardrails、审批机制、最佳实践

🛡️ Guardrails (安全防护)

Guardrails 通过 Processors 实现安全控制，保护 AI 系统免受恶意输入和不当输出的影响。

Input Guardrails

1. Unicode 规范化

import { UnicodeNormalizer } from '@mastra/core/processors';

const agent = new Agent({
  id: 'normalized-agent',
  inputProcessors: [
    new UnicodeNormalizer({
      stripControlChars: true,
      collapseWhitespace: true
    })
  ]
});

2. 提示词注入检测

import { PromptInjectionDetector } from '@mastra/core/processors';

const secureAgent = new Agent({
  id: 'secure-agent',
  inputProcessors: [
    new PromptInjectionDetector({
      model: 'openai/gpt-4o-mini',
      threshold: 0.8,
      strategy: 'rewrite', // 'block' | 'rewrite'
      detectionTypes: ['injection', 'jailbreak', 'system-override']
    })
  ]
});

策略选择:

策略	行为
block	检测到威胁时拒绝请求
rewrite	使用 LLM 重写输入以移除威胁

3. 语言检测与翻译

import { LanguageDetector } from '@mastra/core/processors';

const multilingualAgent = new Agent({
  id: 'multilingual-agent',
  inputProcessors: [
    new LanguageDetector({
      model: 'openai/gpt-4o-mini',
      targetLanguages: ['English', 'en'],
      strategy: 'translate',
      threshold: 0.8
    })
  ]
});

Output Guardrails

1. 内容审核

import { ContentModerator } from '@mastra/core/processors';

const moderatedAgent = new Agent({
  id: 'moderated-agent',
  outputProcessors: [
    new ContentModerator({
      model: 'openai/gpt-4o-mini',
      categories: ['hate', 'violence', 'sexual', 'self-harm'],
      threshold: 0.7,
      action: 'block' // 'block' | 'flag' | 'redact'
    })
  ]
});

2. PII 检测与脱敏

import { PIIDetector } from '@mastra/core/processors';

const piiAgent = new Agent({
  id: 'pii-agent',
  outputProcessors: [
    new PIIDetector({
      types: ['email', 'phone', 'ssn', 'credit-card'],
      action: 'redact', // 'redact' | 'block' | 'flag'
      redactWith: '[REDACTED]'
    })
  ]
});

3. 输出验证

import { OutputValidator } from '@mastra/core/processors';

const validatedAgent = new Agent({
  id: 'validated-agent',
  outputProcessors: [
    new OutputValidator({
      schema: z.object({
        answer: z.string().min(10),
        confidence: z.number().min(0).max(1)
      }),
      onValidationError: 'retry' // 'retry' | 'throw' | 'default'
    })
  ]
});

---

✅ 审批机制

1. 工具调用审批

import { Agent } from '@mastra/core/agent';

const agent = new Agent({
  id: 'approval-agent',
  model: 'openai/gpt-4o',
  tools: {
    deleteFile: {
      description: 'Delete a file',
      parameters: z.object({ path: z.string() }),
      execute: async ({ path }) => {
        // 请求人工审批
        const approved = await requestApproval({
          action: 'delete_file',
          details: { path },
          reason: 'Destructive operation'
        });

        if (!approved) {
          throw new Error('Operation rejected by user');
        }

        return await fs.unlink(path);
      }
    }
  }
});

2. 批量审批

class ApprovalManager {
  private pendingApprovals: Map<string, ApprovalRequest> = new Map();

  async requestApproval(request: ApprovalRequest): Promise<boolean> {
    const id = generateId();
    this.pendingApprovals.set(id, request);

    // 通知用户
    await this.notifyUser(id, request);

    // 等待响应
    return await this.waitForResponse(id);
  }

  async approve(id: string) {
    const request = this.pendingApprovals.get(id);
    if (request) {
      request.resolve(true);
      this.pendingApprovals.delete(id);
    }
  }

  async reject(id: string) {
    const request = this.pendingApprovals.get(id);
    if (request) {
      request.resolve(false);
      this.pendingApprovals.delete(id);
    }
  }
}

3. 条件审批

const agent = new Agent({
  id: 'conditional-approval-agent',
  tools: {
    executeCommand: {
      description: 'Execute shell command',
      parameters: z.object({ command: z.string() }),
      execute: async ({ command }) => {
        // 危险命令需要审批
        const isDangerous = ['rm', 'sudo', 'chmod'].some(cmd =>
          command.includes(cmd)
        );

        if (isDangerous) {
          const approved = await requestApproval({
            action: 'execute_command',
            details: { command },
            reason: 'Potentially dangerous command'
          });

          if (!approved) {
            throw new Error('Command rejected');
          }
        }

        return await exec(command);
      }
    }
  }
});

---

🔒 最佳实践

1. 分层防护

const productionAgent = new Agent({
  id: 'production-agent',
  // Input 层
  inputProcessors: [
    new UnicodeNormalizer(),
    new PromptInjectionDetector({ strategy: 'block' }),
    new LanguageDetector({ targetLanguages: ['en'] })
  ],
  // Output 层
  outputProcessors: [
    new PIIDetector({ action: 'redact' }),
    new ContentModerator({ action: 'block' }),
    new OutputValidator({ schema: outputSchema })
  ]
});

2. 审计日志

class AuditLogger {
  async log(event: AuditEvent) {
    await db.auditLogs.create({
      timestamp: new Date(),
      agentId: event.agentId,
      action: event.action,
      input: event.input,
      output: event.output,
      userId: event.userId,
      approved: event.approved
    });
  }
}

const agent = new Agent({
  id: 'audited-agent',
  preProcessors: [
    async ({ messages, context }) => {
      await auditLogger.log({
        action: 'agent_input',
        input: messages,
        userId: context.userId
      });
      return { messages, context };
    }
  ],
  postProcessors: [
    async ({ text, context }) => {
      await auditLogger.log({
        action: 'agent_output',
        output: text,
        userId: context.userId
      });
      return { text, context };
    }
  ]
});

3. 速率限制

class RateLimiter {
  private requests: Map<string, number[]> = new Map();

  async checkLimit(userId: string, limit: number, window: number): Promise<boolean> {
    const now = Date.now();
    const userRequests = this.requests.get(userId) || [];

    // 清理过期请求
    const validRequests = userRequests.filter(t => now - t < window);

    if (validRequests.length >= limit) {
      return false;
    }

    validRequests.push(now);
    this.requests.set(userId, validRequests);
    return true;
  }
}

const agent = new Agent({
  id: 'rate-limited-agent',
  preProcessors: [
    async ({ messages, context }) => {
      const allowed = await rateLimiter.checkLimit(
        context.userId,
        100, // 100 requests
        60 * 1000 // per minute
      );

      if (!allowed) {
        throw new Error('Rate limit exceeded');
      }

      return { messages, context };
    }
  ]
});

4. 工具权限控制

class ToolPermissionManager {
  private permissions: Map<string, Set<string>> = new Map();

  async checkPermission(userId: string, toolName: string): Promise<boolean> {
    const userTools = this.permissions.get(userId);
    return userTools?.has(toolName) || false;
  }
}

const agent = new Agent({
  id: 'permission-controlled-agent',
  tools: {
    deleteFile: {
      description: 'Delete a file',
      parameters: z.object({ path: z.string() }),
      execute: async ({ path }, context) => {
        const hasPermission = await permissionManager.checkPermission(
          context.userId,
          'deleteFile'
        );

        if (!hasPermission) {
          throw new Error('Permission denied');
        }

        return await fs.unlink(path);
      }
    }
  }
});

5. 错误处理

const agent = new Agent({
  id: 'error-handling-agent',
  postProcessors: [
    async ({ text, context }) => {
      try {
        // 验证输出
        await validateOutput(text);
        return { text, context };
      } catch (error) {
        // 记录错误
        await logger.error('Output validation failed', { error, text });

        // 返回安全的默认响应
        return {
          text: 'I apologize, but I cannot provide that information.',
          context
        };
      }
    }
  ]
});

---

🔗 相关资源

• Mastra Documentation
• Agent API Reference
• Processors Reference
• Mastra Examples

---

本指南基于 Mastra v1 文档编写，持续更新以跟踪最新特性。