1
from enum import Enum
2

3
class FailureType(Enum):
4
    RETRYABLE = "retry"
5
    DEGRADABLE = "degrade"
6
    FATAL = "fatal"
7

8
FAILURE_STRATEGIES = {
9
    FailureType.RETRYABLE: retry_with_backoff,
10
    FailureType.DEGRADABLE: degrade_to_simple,
11
    FailureType.FATAL: return_error_to_user
12
}

1
import asyncio
2
from tenacity import retry, stop_after_attempt, wait_exponential
3

4
@retry(
5
    stop=stop_after_attempt(3),
6
    wait=wait_exponential(multiplier=1, min=1, max=10)
7
)
8
async def call_with_retry(tool_func, *args, **kwargs):
9
    try:
10
        return await tool_func(*args, **kwargs)
11
    except RateLimitError:
12
        raise RetryableError("Rate limited")

1
from circuitbreaker import circuit
2

3
@circuit(maximum=10, failure_threshold=5, recovery_timeout=60)
4
async def protected_call():
5
    return await vulnerable_api_call()

1
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
2

3
class LLMAPIError(Exception):
4
    def __init__(self, error_type: str, message: str):
5
        self.error_type = error_type
6
        super().__init__(message)
7

8
class ContextLengthExceeded(LLMAPIError):
9
    def __init__(self, requested: int, limit: int):
10
        super().__init__("context_length_exceeded", f"需要 {requested} tokens，上限 {limit}")
11
        self.requested_tokens = requested
12
        self.max_tokens = limit
13

14
class RateLimitExceeded(LLMAPIError):
15
    def __init__(self, retry_after: float):
16
        super().__init__("rate_limit", "请求过快")
17
        self.retry_after = retry_after
18

19
class SmartLLMRetry:
20
    """LLM 智能重试：根据错误类型采取不同策略"""
21

22
    def __init__(self, max_retries: int = 3):
23
        self.max_retries = max_retries
24

25
    async def call(self, prompt: str, model: str = "gpt-4o", **kwargs) -> str:
26
        """带智能重试的 LLM 调用"""
27
        current_prompt = prompt
28
        current_model = model
29

30
        for attempt in range(self.max_retries):
31
            try:
32
                return await self._do_call(current_prompt, current_model, **kwargs)
33

34
            except ContextLengthExceeded as e:
35
                # 策略 1: 截断 Prompt
36
                if attempt == 0:
37
                    current_prompt = self._truncate_prompt(current_prompt, int(e.max_tokens * 0.8))
38
                    continue
39

40
                # 策略 2: 换用更大上下文的模型
41
                if attempt == 1:
42
                    current_model = self._get_larger_context_model(current_model)
43
                    continue
44

45
                # 策略 3: 压缩历史
46
                current_prompt = await self._compress_prompt(current_prompt)
47
                continue
48

49
            except RateLimitExceeded as e:
50
                # 等待后重试
51
                await asyncio.sleep(e.retry_after)
52
                continue
53

54
            except LLMAPIError as e:
55
                if e.error_type in ("invalid_request", "authentication"):
56
                    # 不可重试的错误，立即抛出
57
                    raise
58
                # 其他错误正常重试
59
                await asyncio.sleep(2 ** attempt)
60
                continue
61

62
        raise Exception(f"LLM 调用在 {self.max_retries} 次重试后仍然失败")
63

64
    async def _do_call(self, prompt: str, model: str, **kwargs) -> str:
65
        """执行 LLM 调用"""
66
        try:
67
            return await llm.complete(prompt, model=model, **kwargs)
68
        except Exception as e:
69
            error = self._parse_error(e)
70
            raise error from e
71

72
    def _truncate_prompt(self, prompt: str, target_tokens: int) -> str:
73
        """截断 Prompt 到目标 Token 数"""
74
        current_tokens = count_tokens(prompt)
75
        if current_tokens <= target_tokens:
76
            return prompt
77
        ratio = target_tokens / current_tokens
78
        cut_point = int(len(prompt) * ratio)
79
        return prompt[:cut_point] + "\n\n[内容已截断]"
80

81
    def _get_larger_context_model(self, current: str) -> str:
82
        """获取上下文更大的替代模型"""
83
        upgrades = {
84
            "gpt-4o-mini": "gpt-4o",
85
            "claude-haiku-3.5": "claude-sonnet-4",
86
            "gemini-2.0-flash": "gemini-2.5-pro",
87
        }
88
        return upgrades.get(current, current)

1
# 抖动退避实现
2
import random
3

4
async def retry_with_jitter(
5
    func,
6
    max_retries: int = 3,
7
    base_delay: float = 1.0,
8
    max_delay: float = 30.0,
9
):
10
    """带抖动的指数退避"""
11
    for attempt in range(max_retries):
12
        try:
13
            return await func()
14
        except Exception as e:
15
            if attempt == max_retries - 1:
16
                raise
17

18
            # 指数退避 + 随机抖动
19
            delay = min(base_delay * (2 ** attempt), max_delay)
20
            jitter = random.uniform(0, delay * 0.5)
21
            await asyncio.sleep(delay + jitter)

1
async def degrade_to_simple(query: str) -> str:
2
    """降级到简单模式"""
3
    # 第一级：简单 RAG
4
    try:
5
        return await simple_rag(query)
6
    except:
7
        pass
8

9
    # 第二级：关键词匹配
10
    try:
11
        return await keyword_search(query)
12
    except:
13
        pass
14

15
    # 第三级：返回预设答案
16
    return "抱歉，暂时无法回答您的问题。"

1
from dataclasses import dataclass
2
from enum import Enum
3

4
class ServiceLevel(Enum):
5
    FULL = "full"           # 完整功能
6
    DEGRADED = "degraded"   # 部分功能
7
    MINIMAL = "minimal"     # 最小功能
8
    OFFLINE = "offline"     # 仅返回静态内容
9

10
@dataclass
11
class ServiceHealth:
12
    llm_available: bool = True
13
    search_available: bool = True
14
    database_available: bool = True
15
    cache_available: bool = True
16

17
class GracefulDegradation:
18
    """优雅降级框架"""
19

20
    def __init__(self):
21
        self.health = ServiceHealth()
22
        self.degradation_rules = {
23
            # (缺失的服务, 降级策略)
24
            "no_search": {
25
                "level": ServiceLevel.DEGRADED,
26
                "fallback": "knowledge_base_only",
27
                "message": "搜索功能暂不可用，基于知识库回答",
28
            },
29
            "no_llm": {
30
                "level": ServiceLevel.MINIMAL,
31
                "fallback": "template_responses",
32
                "message": "AI 服务暂不可用，返回预设回答",
33
            },
34
            "no_database": {
35
                "level": ServiceLevel.DEGRADED,
36
                "fallback": "llm_knowledge_only",
37
                "message": "数据库暂不可用，基于模型知识回答",
38
            },
39
        }
40

41
    def determine_level(self) -> tuple[ServiceLevel, list[str]]:
42
        """根据健康状况确定服务级别"""
43
        missing = []
44
        if not self.health.llm_available:
45
            missing.append("no_llm")
46
        if not self.health.search_available:
47
            missing.append("no_search")
48
        if not self.health.database_available:
49
            missing.append("no_database")
50

51
        if not missing:
52
            return ServiceLevel.FULL, []
53

54
        # 取最严重的降级级别
55
        levels = [self.degradation_rules[m]["level"] for m in missing]
56
        severity = {ServiceLevel.FULL: 0, ServiceLevel.DEGRADED: 1, ServiceLevel.MINIMAL: 2}
57
        worst = max(levels, key=lambda l: severity[l])
58
        return worst, missing
59

60
    async def handle_request(self, query: str) -> dict:
61
        """根据当前服务级别处理请求"""
62
        level, missing = self.determine_level()
63

64
        if level == ServiceLevel.FULL:
65
            result = await self._full_service(query)
66
        elif level == ServiceLevel.DEGRADED:
67
            result = await self._degraded_service(query, missing)
68
        elif level == ServiceLevel.MINIMAL:
69
            result = await self._minimal_service(query)
70
        else:
71
            result = self._offline_response()
72

73
        result["service_level"] = level.value
74
        return result
75

76
    async def _full_service(self, query: str) -> dict:
77
        """完整服务"""
78
        response = await agent.run(query)
79
        return {"response": response, "quality": "high"}
80

81
    async def _degraded_service(self, query: str, missing: list[str]) -> dict:
82
        """降级服务"""
83
        messages = [self.degradation_rules[m]["message"] for m in missing]
84

85
        if "no_search" in missing:
86
            # 搜索不可用，用知识库
87
            response = await knowledge_base_rag(query)
88
        elif "no_database" in missing:
89
            # 数据库不可用，靠 LLM 知识
90
            response = await llm.complete(query)
91
        else:
92
            response = await agent.run(query)
93

94
        return {"response": response, "quality": "medium", "warnings": messages}
95

96
    async def _minimal_service(self, query: str) -> dict:
97
        """最小服务"""
98
        # 模板匹配或 FAQ
99
        response = match_faq(query) or "服务暂时受限，请稍后再试。"
100
        return {"response": response, "quality": "low"}
101

102
    def _offline_response(self) -> dict:
103
        return {"response": "系统维护中，请稍后再试。", "quality": "none"}

1
import re
2
from dataclasses import dataclass
3

4
@dataclass
5
class ValidationResult:
6
    is_valid: bool
7
    errors: list[str]
8
    warnings: list[str]
9
    sanitized_input: str | None = None
10

11
class InputValidator:
12
    """Agent 输入验证器"""
13

14
    MAX_INPUT_LENGTH = 10000
15
    MAX_TOOL_PARAMS = 20
16

17
    def validate(self, user_input: str) -> ValidationResult:
18
        errors = []
19
        warnings = []
20

21
        # 1. 长度检查
22
        if len(user_input) > self.MAX_INPUT_LENGTH:
23
            errors.append(f"输入过长: {len(user_input)} > {self.MAX_INPUT_LENGTH}")
24
        elif len(user_input) > self.MAX_INPUT_LENGTH * 0.8:
25
            warnings.append("输入接近长度上限，可能影响处理效果")
26

27
        # 2. 空输入检查
28
        if not user_input.strip():
29
            errors.append("输入为空")
30

31
        # 3. 注入检测
32
        injection_score = self._detect_injection(user_input)
33
        if injection_score > 0.8:
34
            errors.append("输入包含疑似注入内容")
35
        elif injection_score > 0.5:
36
            warnings.append("输入包含可疑内容")
37

38
        # 4. 编码检查
39
        if not self._is_valid_encoding(user_input):
40
            errors.append("输入包含无效字符")
41

42
        # 5. 语言检查（可选）
43
        if self._contains_mixed_scripts(user_input):
44
            warnings.append("输入包含混合文字，可能影响理解")
45

46
        sanitized = self._sanitize(user_input) if not errors else None
47

48
        return ValidationResult(
49
            is_valid=len(errors) == 0,
50
            errors=errors,
51
            warnings=warnings,
52
            sanitized_input=sanitized,
53
        )
54

55
    def _detect_injection(self, text: str) -> float:
56
        """检测注入攻击（返回 0-1 的风险分数）"""
57
        injection_patterns = [
58
            r"忽略.{0,5}(之前的|上面|所有|全部).{0,5}(指令|规则|提示)",
59
            r"(forget|ignore|disregard).{0,10}(previous|above|all).{0,10}(instructions|rules)",
60
            r"你是一个",
61
            r"system:",
62
            r"<\|im_start\|>",
63
            r"\\n\\n",
64
        ]
65
        matches = sum(1 for p in injection_patterns if re.search(p, text, re.IGNORECASE))
66
        return min(matches / len(injection_patterns), 1.0)
67

68
    def _is_valid_encoding(self, text: str) -> bool:
69
        try:
70
            text.encode("utf-8")
71
            return True
72
        except UnicodeEncodeError:
73
            return False
74

75
    def _contains_mixed_scripts(self, text: str) -> bool:
76
        has_cjk = any("\u4e00" <= c <= "\u9fff" for c in text)
77
        has_cyrillic = any("\u0400" <= c <= "\u04ff" for c in text)
78
        has_arabic = any("\u0600" <= c <= "\u06ff" for c in text)
79
        scripts = [has_cjk, has_cyrillic, has_arabic]
80
        return sum(scripts) > 1
81

82
    def _sanitize(self, text: str) -> str:
83
        """清理输入"""
84
        # 移除控制字符
85
        sanitized = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", text)
86
        # 规范化空白
87
        sanitized = re.sub(r"\s+", " ", sanitized).strip()
88
        return sanitized

1
import json
2
import re
3

4
class RobustOutputParser:
5
    """健壮的 LLM 输出解析器"""
6

7
    async def parse_json(self, text: str) -> dict | None:
8
        """从 LLM 输出中可靠地解析 JSON"""
9
        # 策略 1: 直接解析
10
        try:
11
            return json.loads(text)
12
        except json.JSONDecodeError:
13
            pass
14

15
        # 策略 2: 提取 ```json ... ``` 代码块
16
        json_match = re.search(r"```(?:json)?\s*\n?(.*?)\n?```", text, re.DOTALL)
17
        if json_match:
18
            try:
19
                return json.loads(json_match.group(1))
20
            except json.JSONDecodeError:
21
                pass
22

23
        # 策略 3: 找到第一个 { 和最后一个 }
24
        first_brace = text.find("{")
25
        last_brace = text.rfind("}")
26
        if first_brace != -1 and last_brace != -1:
27
            try:
28
                return json.loads(text[first_brace:last_brace + 1])
29
            except json.JSONDecodeError:
30
                pass
31

32
        # 策略 4: 让 LLM 自修复
33
        return await self._llm_repair_json(text)
34

35
    async def _llm_repair_json(self, broken_text: str) -> dict | None:
36
        """用 LLM 修复损坏的 JSON"""
37
        repair_prompt = f"""以下文本应该是一个 JSON 对象，但格式有问题。请修复并输出有效的 JSON。
38

39
原始文本:
40
{broken_text}
41

42
修复后的 JSON:"""
43

44
        for attempt in range(2):
45
            try:
46
                repaired = await llm.complete(repair_prompt)
47
                # 提取并解析
48
                return json.loads(repaired)
49
            except:
50
                continue
51

52
        return None
53

54
    def parse_action(self, text: str) -> dict:
55
        """从 ReAct 格式的文本中提取 Action"""
56
        # 匹配多种格式
57
        patterns = [
58
            r"Action:\s*(\w+)\s*\((.*)\)",                    # Action: search(query="test")
59
            r"Action:\s*(\w+)\s*\[(.*)\]",                    # Action: search[query="test"]
60
            r"```(?:json)?\s*\n?{{.*?\"action\":\s*\"(\w+)\".*?\"input\":\s*({.*?})",
61
        ]
62

63
        for pattern in patterns:
64
            match = re.search(pattern, text, re.DOTALL)
65
            if match:
66
                return {
67
                    "tool": match.group(1),
68
                    "input": self._parse_action_input(match.group(2)),
69
                }
70

71
        # 无法解析，返回空 Action
72
        return {"tool": None, "input": {}}
73

74
    def parse_final_answer(self, text: str) -> str:
75
        """提取 Final Answer"""
76
        patterns = [
77
            r"Final Answer:\s*(.*)",
78
            r"最终答案[：:]\s*(.*)",
79
            r"答案[：:]\s*(.*)",
80
        ]
81

82
        for pattern in patterns:
83
            match = re.search(pattern, text, re.DOTALL)
84
            if match:
85
                return match.group(1).strip()
86

87
        # 没有明确的 Final Answer 标记，返回全文
88
        return text.strip()

1
class OutputValidator:
2
    """输出验证器"""
3

4
    def validate(self, response: str, schema: dict | None = None) -> dict:
5
        issues = []
6

7
        # 1. 空响应检查
8
        if not response or not response.strip():
9
            issues.append({"type": "empty", "severity": "critical"})
10

11
        # 2. 长度检查
12
        if len(response) > 50000:
13
            issues.append({"type": "too_long", "severity": "warning"})
14
        elif len(response) < 10:
15
            issues.append({"type": "too_short", "severity": "warning"})
16

17
        # 3. 重复检查（LLM 有时会重复内容）
18
        if self._has_excessive_repetition(response):
19
            issues.append({"type": "repetition", "severity": "warning"})
20

21
        # 4. 有害内容检查
22
        if self._contains_harmful_content(response):
23
            issues.append({"type": "harmful", "severity": "critical"})
24

25
        # 5. Schema 验证（如果指定）
26
        if schema:
27
            schema_issues = self._validate_schema(response, schema)
28
            issues.extend(schema_issues)
29

30
        return {
31
            "is_valid": not any(i["severity"] == "critical" for i in issues),
32
            "issues": issues,
33
        }
34

35
    def _has_excessive_repetition(self, text: str) -> bool:
36
        """检测过度的内容重复"""
37
        sentences = text.split("。")
38
        if len(sentences) < 3:
39
            return False
40
        # 检查是否有超过 3 个相同的句子
41
        from collections import Counter
42
        counter = Counter(s.strip() for s in sentences if s.strip())
43
        return any(count > 3 for count in counter.values())
44

45
    def _contains_harmful_content(self, text: str) -> bool:
46
        """简单的有害内容检测"""
47
        harmful_patterns = [
48
            r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b",  # 电话号码
49
            r"\b\d{3}[-.]?\d{2}[-.]?\d{4}\b",  # SSN
50
            r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",  # 邮箱
51
        ]
52
        return any(re.search(p, text) for p in harmful_patterns)

1
import time
2
from collections import deque
3

4
class TokenRateLimiter:
5
    def __init__(self, max_tokens: int, window_seconds: int):
6
        self.max_tokens = max_tokens
7
        self.window = window_seconds
8
        self.requests = deque()
9

10
    async def acquire(self, tokens: int):
11
        now = time.time()
12
        # 清理过期请求
13
        while self.requests and now - self.requests[0] > self.window:
14
            self.requests.popleft()
15

16
        if sum(self.requests) + tokens > self.max_tokens:
17
            wait_time = self.window - (now - self.requests[0])
18
            await asyncio.sleep(wait_time)
19

20
        self.requests.append(tokens)

1
from asyncio import Semaphore
2

3
MAX_CONCURRENT = 10
4
semaphore = Semaphore(MAX_CONCURRENT)
5

6
async def limited_agent_call(query: str):
7
    async with semaphore:
8
        return await agent.process(query)

1
from dataclasses import dataclass
2
from enum import Enum
3

4
class Priority(Enum):
5
    CRITICAL = 0   # 付费用户、关键业务
6
    NORMAL = 1     # 普通用户
7
    LOW = 2        # 后台任务、批量处理
8

9
@dataclass
10
class RateLimitConfig:
11
    max_concurrent: int
12
    max_rpm: int        # 每分钟请求数
13
    max_tpm: int        # 每分钟 Token 数
14

15
PRIORITY_LIMITS = {
16
    Priority.CRITICAL: RateLimitConfig(max_concurrent=20, max_rpm=120, max_tpm=200000),
17
    Priority.NORMAL: RateLimitConfig(max_concurrent=10, max_rpm=60, max_tpm=100000),
18
    Priority.LOW: RateLimitConfig(max_concurrent=3, max_rpm=10, max_tpm=30000),
19
}
20

21
class PriorityRateLimiter:
22
    """分优先级的限流器"""
23

24
    def __init__(self):
25
        self.semaphores = {
26
            p: asyncio.Semaphore(c.max_concurrent)
27
            for p, c in PRIORITY_LIMITS.items()
28
        }
29
        self.request_counts = {p: deque() for p in Priority}
30

31
    async def acquire(self, priority: Priority, estimated_tokens: int = 0):
32
        """获取执行许可"""
33
        config = PRIORITY_LIMITS[priority]
34

35
        # 并发限制
36
        await self.semaphores[priority].acquire()
37

38
        # RPM 限制
39
        now = time.time()
40
        self._cleanup_old(self.request_counts[priority], window=60)
41
        if len(self.request_counts[priority]) >= config.max_rpm:
42
            wait_time = 60 - (now - self.request_counts[priority][0])
43
            await asyncio.sleep(wait_time)
44

45
        self.request_counts[priority].append(now)
46

47
    def release(self, priority: Priority):
48
        self.semaphores[priority].release()

1
@dataclass
2
class AgentHealth:
3
    success_rate: float  # > 0.95
4
    avg_latency_ms: float  # < 2000
5
    error_rate_by_type: dict
6
    cache_hit_rate: float  # > 0.5

1
async def health_check_loop():
2
    while True:
3
        health = await check_agent_health()
4

5
        if health.success_rate < 0.8:
6
            await scale_up()
7
        elif health.success_rate < 0.5:
8
            await circuit_break()
9

10
        await asyncio.sleep(30)

1
from datetime import datetime, timedelta
2

3
@dataclass
4
class HealthCheckResult:
5
    status: str            # healthy / degraded / unhealthy
6
    checks: dict[str, bool]
7
    latency_ms: float
8
    last_error: str | None
9
    timestamp: datetime
10

11
class AgentHealthChecker:
12
    """Agent 健康检查系统"""
13

14
    def __init__(self):
15
        self.history: list[HealthCheckResult] = []
16
        self.alert_handlers: list[callable] = []
17

18
    async def check(self) -> HealthCheckResult:
19
        """执行完整健康检查"""
20
        checks = {}
21
        start_time = time.time()
22

23
        # 检查 1: LLM API 可用性
24
        checks["llm_api"] = await self._check_llm_api()
25

26
        # 检查 2: 工具服务可用性
27
        checks["tools"] = await self._check_tools()
28

29
        # 检查 3: 数据库连接
30
        checks["database"] = await self._check_database()
31

32
        # 检查 4: 缓存服务
33
        checks["cache"] = await self._check_cache()
34

35
        # 检查 5: 最近错误率
36
        checks["error_rate"] = self._check_error_rate()
37

38
        latency = (time.time() - start_time) * 1000
39
        all_healthy = all(checks.values())
40
        mostly_healthy = sum(checks.values()) >= len(checks) * 0.6
41

42
        status = "healthy" if all_healthy else ("degraded" if mostly_healthy else "unhealthy")
43

44
        result = HealthCheckResult(
45
            status=status,
46
            checks=checks,
47
            latency_ms=latency,
48
            last_error=self._get_last_error(),
49
            timestamp=datetime.now(),
50
        )
51

52
        self.history.append(result)
53

54
        # 触发告警
55
        if status != "healthy":
56
            for handler in self.alert_handlers:
57
                await handler(result)
58

59
        return result
60

61
    async def _check_llm_api(self) -> bool:
62
        """检查 LLM API 是否可用"""
63
        try:
64
            response = await asyncio.wait_for(
65
                llm.complete("Hello, respond with 'OK'."),
66
                timeout=10.0,
67
            )
68
            return "ok" in response.lower()
69
        except Exception:
70
            return False
71

72
    async def _check_tools(self) -> bool:
73
        """检查核心工具是否可用"""
74
        try:
75
            result = await search_tool("test")
76
            return result is not None
77
        except Exception:
78
            return False
79

80
    async def _check_database(self) -> bool:
81
        """检查数据库连接"""
82
        try:
83
            await db.execute("SELECT 1")
84
            return True
85
        except Exception:
86
            return False
87

88
    async def _check_cache(self) -> bool:
89
        """检查缓存服务"""
90
        try:
91
            redis_client.ping()
92
            return True
93
        except Exception:
94
            return False
95

96
    def _check_error_rate(self) -> bool:
97
        """检查最近 10 分钟的错误率"""
98
        cutoff = datetime.now() - timedelta(minutes=10)
99
        recent = [r for r in self.history if r.timestamp > cutoff]
100
        if not recent:
101
            return True
102
        error_rate = sum(1 for r in recent if r.status != "healthy") / len(recent)
103
        return error_rate < 0.3

1
class AdversarialDefense:
2
    """对抗性防御"""
3

4
    def __init__(self):
5
        self.max_turns = 20
6
        self.budget_per_user = 100  # 每用户每小时的 Token 预算
7

8
    def check_input_safety(self, user_input: str) -> dict:
9
        """输入安全检查"""
10
        risks = []
11

12
        # 检查 1: 提示注入
13
        if self._detect_prompt_injection(user_input):
14
            risks.append({"type": "prompt_injection", "severity": "high"})
15

16
        # 检查 2: 越狱尝试
17
        if self._detect_jailbreak(user_input):
18
            risks.append({"type": "jailbreak", "severity": "high"})
19

20
        # 检查 3: 敏感信息请求
21
        if self._detect_sensitive_request(user_input):
22
            risks.append({"type": "sensitive_request", "severity": "medium"})
23

24
        return {
25
            "is_safe": not any(r["severity"] == "high" for r in risks),
26
            "risks": risks,
27
        }
28

29
    def _detect_prompt_injection(self, text: str) -> bool:
30
        patterns = [
31
            r"忽略.*指令",
32
            r"forget.*instructions",
33
            r"new instructions",
34
            r"system\s*:",
35
            r"<\|im_start\|>",
36
        ]
37
        return any(re.search(p, text, re.IGNORECASE) for p in patterns)
38

39
    def _detect_jailbreak(self, text: str) -> bool:
40
        patterns = [
41
            r"DAN\s+mode",
42
            r"developer\s+mode",
43
            r"jailbreak",
44
            r"越狱",
45
        ]
46
        return any(re.search(p, text, re.IGNORECASE) for p in patterns)
47

48
    def _detect_sensitive_request(self, text: str) -> bool:
49
        patterns = [
50
            r"系统.*(提示|prompt)",
51
            r"(password|secret|api.?key)",
52
            r"数据库.*(密码|连接串)",
53
        ]
54
        return any(re.search(p, text, re.IGNORECASE) for p in patterns)

1
class OutputSafetyFilter:
2
    """输出安全过滤器"""
3

4
    SENSITIVE_PATTERNS = [
5
        (r"sk-[a-zA-Z0-9]{32,}", "[API_KEY_REDACTED]"),
6
        (r"\b\d{16,19}\b", "[CARD_NUMBER_REDACTED]"),
7
        (r"\b\d{3}-\d{2}-\d{4}\b", "[SSN_REDACTED]"),
8
        (r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", "[EMAIL_REDACTED]"),
9
    ]
10

11
    def filter(self, response: str) -> str:
12
        """过滤敏感信息"""
13
        for pattern, replacement in self.SENSITIVE_PATTERNS:
14
            response = re.sub(pattern, replacement, response)
15
        return response