提示注入攻击如今已成为大模型在生产部署中面临的常见风险。攻击者通过构造恶意输入，有可能绕过系统预设的防护机制，迫使模型输出非预期内容，甚至泄露敏感数据。ERNIE-4.5-0.3B-PT 作为一款高性能的文本生成模型，在真实应用场景中同样需要应对此类威胁。

本文将重点介绍如何为 ERNIE-4.5-0.3B-PT 搭建一个具备安全防护能力的部署环境。从输入过滤到输出检查，再到访问控制，整个方案包含多个可落地的防护措施，帮助你逐步构建多层次的防御体系。即便你之前对安全部署接触不多，按照本文步骤也能顺利实施。

1. 基础环境配置与模型快速部署

1.1 系统依赖与运行环境

请确保系统满足以下最低要求：Python 3.8 或更高版本、至少 8GB 内存（推荐 16GB），以及充足磁盘空间。操作系统推荐使用 Ubuntu 20.04 及以上版本。

# 创建独立的 Python 虚拟环境
python -m venv ernie-prod-env
source ernie-prod-env/bin/activate

# 安装必要的第三方库
pip install torch transformers vllm
pip install fastapi uvicorn   # 用于搭建 API 服务

1.2 快速启动模型服务

使用 vLLM 可以快速部署 ERNIE-4.5-0.3B-PT，在提升推理效率的同时，也能利用其自带的部分安全特性：

from vllm import LLM, SamplingParams

# 加载模型
model_llm = LLM(
    model="baidu/ERNIE-4.5-0.3B-PT",
    trust_remote_code=True,
    max_model_len=2048,
    enable_prefix_caching=True
)

# 配置生成参数
generate_params = SamplingParams(
    temperature=0.7,
    top_p=0.9,
    max_tokens=512
)

2. 核心安全防护措施

2.1 输入数据校验与清洗

输入过滤是第一道屏障，能有效阻挡大部分恶意尝试。我们需要对用户提交的文本进行多重规则校验：

import re
from typing import List

class InputSanitizer:
    def __init__(self):
        # 定义需拦截的危险正则
        self.hazard_patterns = [
            r"ignore.*previous|忘记.*之前",   # 试图绕过历史指令
            r"system.*prompt|系统.*提示",      # 针对系统提示词
            r"password|token|key|密码|密钥",    # 敏感关键词
            r" bool:
        """判断输入文本是否包含危险内容"""
        # 限制输入长度
        if len(text) > 1000:
            return False

        # 逐条匹配危险模式
        for pattern in self.hazard_patterns:
            if re.search(pattern, text, re.IGNORECASE):
                return False
        return True

    def clean_text(self, text: str) -> str:
        """清洗并截断输入文本"""
        # 删除 HTML 标签
        cleaned = re.sub(r'<[^>]*>', '', text)
        # 移除不可见控制字符
        cleaned = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', cleaned)
        return cleaned[:1000]   # 最终长度限制

2.2 输出内容安全检测

即便输入通过了检查，仍需对模型输出进行实时监控，防止生成不当内容：

class OutputSafetyChecker:
    def __init__(self):
        self.sensitive_keywords = [
            "暴力", "仇恨言论", "歧视",
            "敏感政治", "违法内容", "隐私信息"
        ]

    def evaluate_output(self, generated_text: str) -> dict:
        """评估生成文本的安全性"""
        result = {
            "is_safe": True,
            "warnings": [],
            "risk_score": 0.0
        }

        # 逐一检查敏感词
        for keyword in self.sensitive_keywords:
            if keyword in generated_text:
                result["is_safe"] = False
                result["warnings"].append(f"检测到敏感内容：{keyword}")

        # 检查是否有个人信息泄露风险
        if self.has_pii(generated_text):
            result["is_safe"] = False
            result["warnings"].append("疑似包含个人身份信息")
        return result

    def has_pii(self, text: str) -> bool:
        """使用正则检测常见 PII 模式"""
        pii_regexes = [
            r'\d{18}|\d{17}X',    # 身份证号
            r'1[3-9]\d{9}',        # 手机号码
            r'\w+@\w+\.\w+',       # 邮箱地址
        ]
        return any(re.search(pattern, text) for pattern in pii_regexes)

2.3 基于角色的访问控制

为不同用户分配差异化权限，控制可使用的功能和资源上限：

from enum import Enum

class Role(Enum):
    VISITOR = 1      # 访客，仅基础问答
    NORMAL = 2       # 普通用户，可使用高级功能
    ADMIN = 3        # 管理员，拥有完整权限

class PermissionGuard:
    def __init__(self):
        self.role_limits = {
            Role.VISITOR: {"max_len": 500, "daily_calls": 100},
            Role.NORMAL:  {"max_len": 1000, "daily_calls": 1000},
            Role.ADMIN:   {"max_len": 2000, "daily_calls": 10000}
        }

    def has_permission(self, user_role: Role, text: str) -> bool:
        """根据角色校验输入长度是否达标"""
        limit = self.role_limits[user_role]
        return len(text) <= limit["max_len"]

3. 集成完整的安全部署方案

3.1 安全中间件组合

将以上各组件集成到一个统一的中间件中，方便在 API 层统一进行管控：

from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel

api_app = FastAPI(title="ERNIE 安全防护 API")

class SecurityPipeline:
    def __init__(self):
        self.input_filter = InputSanitizer()
        self.output_checker = OutputSafetyChecker()
        self.access_guard = PermissionGuard()

    async def handle_incoming(self, request: Request):
        """处理进入的请求：权限校验、输入清洗"""
        payload = await request.json()
        user_text = payload.get("text", "")
        user_role = payload.get("role", Role.VISITOR)

        # 1. 权限检查
        if not self.access_guard.has_permission(user_role, user_text):
            raise HTTPException(status_code=403, detail="权限不足")

        # 2. 输入安全性检测
        if not self.input_filter.is_text_safe(user_text):
            raise HTTPException(status_code=400, detail="输入包含不安全内容")

        # 3. 输入清洗
        return self.input_filter.clean_text(user_text)

    async def handle_outgoing(self, output: str):
        """对模型输出进行安全过滤"""
        safety_result = self.output_checker.evaluate_output(output)
        if not safety_result["is_safe"]:
            return "抱歉，无法生成该内容。请尝试其他问题。"
        return output

# 实例化安全管道
security_pipeline = SecurityPipeline()

@api_app.post("/generate")
async def generate_response(request: Request):
    try:
        # 处理输入
        safe_input = await security_pipeline.handle_incoming(request)

        # 调用模型推理
        outputs = model_llm.generate([safe_input], generate_params)
        result_text = outputs[0].outputs[0].text

        # 处理输出
        final_output = await security_pipeline.handle_outgoing(result_text)

        return {"result": final_output, "status": "ok"}

    except HTTPException as http_err:
        raise http_err
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"服务内部错误：{str(e)}")

3.2 安全日志与监控

建立完善的日志系统有助于追踪异常行为：

import logging
from datetime import datetime

class SecurityAuditLogger:
    def __init__(self):
        logging.basicConfig(
            filename=f'audit_{datetime.now().strftime("%Y%m%d")}.log',
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )

    def record_event(self, event_type: str, metadata: dict):
        """记录安全相关事件"""
        log_content = {
            "timestamp": datetime.now().isoformat(),
            "event_type": event_type,
            "details": metadata
        }
        logging.info(f"Security Log: {log_content}")

    def log_rejection(self, raw_input: str, cause: str):
        """记录被拒绝的请求"""
        self.record_event("request_rejected", {
            "truncated_input": raw_input[:100],
            "reason": cause
        })

4. 高级防御与性能优化

4.1 上下文隔离

防止用户输入干扰系统预设指令：

class ContextSeparator:
    def __init__(self, system_instruction: str):
        self.system_instruction = system_instruction

    def compose_prompt(self, user_message: str) -> str:
        """使用显式分隔符区分系统提示与用户输入"""
        return f"""{self.system_instruction}

[用户输入开始]
{user_message}
[用户输入结束]

[助手回答开始]
"""

4.2 频率限制与反滥用机制

通过速率控制防范 DDoS 和资源滥用：

from collections import defaultdict
import time

class RateController:
    def __init__(self, limit: int = 100, window: int = 3600):
        self.request_limit = limit
        self.time_window = window
        self.user_requests = defaultdict(list)

    def allow_request(self, user_id: str) -> bool:
        """判断当前请求是否允许通过"""
        now = time.time()
        records = [t for t in self.user_requests[user_id] if now - t < self.time_window]

        if len(records) >= self.request_limit:
            return False

        records.append(now)
        self.user_requests[user_id] = records
        return True

5. 实践建议

在真实部署中，建议根据业务场景对安全策略进行调优：

• 避免过度拦截：建立白名单和人工复核机制，对疑似误杀请求进行二次确认。
• 兼顾性能：对安全检查开启异步执行，对可信用户适度降低检查频次。
• 持续更新：关注最新的安全研究报告，定期更新拦截规则库。

通过实施以上输入校验、输出监控、权限分级、速率控制等多项措施，可以为 ERNIE-4.5-0.3B-PT 构建起一套有效的防护体系，抵御提示注入等安全威胁。实际落地时，可参考 OWASP LLM 安全标准相关实践，持续完善安全策略。