什么是 ‘Output Sanity Checking’？利用确定性规则引擎拦截 Agent 产生的‘逻辑上正确但业务上非法’的输出

引言：AI Agent时代的安全与合规挑战

女士们，先生们，大家好！

在当今AI技术飞速发展的时代，我们正见证着Agent范式的崛起。AI Agent不再仅仅是提供预测或分类的工具，它们被设计成能够理解复杂指令、自主规划、执行一系列动作，并与环境进行交互的智能实体。从自动化客服、代码生成、数据分析到智能合同审查，Agent的能力正在深刻改变我们的工作方式。

然而，伴随这种强大能力而来的，是前所未有的挑战。当一个Agent能够自主生成代码、执行数据库操作、发布营销内容或调用外部API时，其输出的质量和安全性就变得至关重要。我们面临的核心问题是：如何确保Agent的输出不仅“逻辑上正确”，而且“业务上合法”？

这正是我们今天讲座的焦点——输出健全性检查（Output Sanity Checking）。具体来说，我们将深入探讨如何利用确定性规则引擎来拦截和纠正Agent产生的那些看似合理，但在实际业务场景中可能带来灾难性后果的输出。

一个Agent的输出可能在语法上完全正确，在逻辑上完美无瑕，甚至能通过初步的功能测试。例如，一个代码生成Agent可能会写出一段能够运行的Python代码，但这段代码可能存在安全漏洞、使用已废弃的库、或者违反了公司的编码规范。一个SQL生成Agent可能会生成一个语法正确的DELETE语句，但它可能在没有WHERE子句的情况下被执行，导致数据丢失。这些都是“逻辑上正确但业务上非法”的典型例子。

传统的AI模型评估方法，如准确率、召回率、F1分数等，主要关注模型在训练数据上的表现，以及其预测或生成结果的表面正确性。然而，这些指标往往无法捕捉到深层次的业务合规性、安全性、效率或最佳实践问题。它们无法判断一个功能正确的SQL语句是否符合最小权限原则，也无法判断一段能运行的代码是否引入了新的安全风险。

因此，我们需要一个额外的、独立于Agent核心生成逻辑的机制，来对Agent的最终输出进行严格的“质量门”检查。这个机制必须具备以下特性：

1. 确定性： 能够给出明确的是/否判断，而非概率性的建议。
2. 可解释性： 能够清晰地说明为何一个输出被拒绝或修改。
3. 可审计性： 规则和决策过程可追溯。
4. 易于维护： 业务规则应易于定义、修改和更新，无需重新训练Agent。

确定性规则引擎恰好满足这些要求。它们通过预定义的、基于事实和逻辑的规则，对输入数据进行评估，并根据评估结果执行相应的动作。这使得它们成为拦截和纠正Agent“业务上非法”输出的理想选择。

接下来，我们将从问题空间入手，详细阐述这种“逻辑正确但业务非法”的输出所带来的风险，然后深入探讨确定性规则引擎的原理、设计与实现，并通过多个实际场景的代码示例，演示如何构建一个健壮的输出健全性检查系统。

第一章：问题空间：“逻辑上正确但业务上非法”的输出

我们首先要深刻理解“逻辑上正确但业务上非法”的输出究竟意味着什么，以及它会带来怎样的危害。

1.1 定义与特征

• 逻辑上正确： 指Agent的输出在语法、结构、甚至功能上是有效的。例如：
  • 生成的代码可以编译并运行，并通过基本单元测试。
  • 生成的SQL查询可以被数据库执行，并返回结果。
  • 生成的文本内容在语法和语义上通顺。
  • 生成的API调用符合API的规范。
• 业务上非法： 指Agent的输出虽然逻辑正确，但违反了公司的业务规则、安全策略、合规要求、最佳实践、成本限制、用户协议或道德规范。例如：
  • 安全风险： 生成的代码包含SQL注入漏洞，或API调用暴露了敏感信息。
  • 数据完整性： 生成的SQL语句删除或修改了生产环境的关键数据，而没有适当的条件限制。
  • 合规性问题： 生成的营销文案包含虚假宣传，或违反了GDPR等数据隐私法规。
  • 成本超支： 生成的云资源配置导致不必要的巨额费用。
  • 性能问题： 生成的代码或查询效率低下，导致系统性能瓶颈。
  • 品牌声誉： 生成的客服回复带有歧视性或不当言辞。
  • 资源滥用： 生成的自动化脚本频繁调用付费API，导致费用暴增。

1.2 为什么传统AI评估难以捕捉？

传统的AI模型，尤其是基于深度学习的Agent，通过学习大量数据来发现模式并生成结果。它们的优化目标通常是最小化某个损失函数，例如预测误差、生成文本的流畅度、代码的语法正确性等。然而，这些损失函数很难直接编码复杂的业务规则、安全策略或合规性要求。

• 隐式知识的局限： 业务规则往往是显式的、人工制定的，不是通过大规模数据统计就能“学”出来的。例如，Agent很难从海量代码中仅仅通过学习就理解“生产环境不允许使用DROP TABLE”这样的策略。
• 泛化能力的边界： 即使训练数据中包含了一些负面案例，Agent也可能因为泛化不足或过度泛化，在新的、看似相似但实际存在业务风险的场景中犯错。
• “黑盒”特性： 深度学习模型的决策过程通常不透明，难以解释为何会生成某个“非法”输出，也难以直接介入纠正其内部逻辑。

1.3 风险与后果

“逻辑上正确但业务上非法”的输出可能导致：

• 财务损失： 误删数据、错误交易、超额费用、罚款。
• 声誉损害： 负面舆论、客户流失、品牌信任度下降。
• 法律责任： 违反法规（如数据隐私、消费者保护），面临诉讼。
• 安全漏洞： 数据泄露、系统被攻击、服务中断。
• 运营中断： 系统崩溃、数据污染、恢复成本高昂。
• 效率低下： 生成低质量代码、低效查询，增加人工返工成本。

因此，构建一个鲁棒的输出健全性检查机制，是部署任何AI Agent到生产环境前的必备环节。

第二章：核心概念：确定性规则引擎

面对上述挑战，我们引入确定性规则引擎作为解决方案。

2.1 什么是确定性规则引擎？

确定性规则引擎是一个基于一系列预定义规则，对输入数据进行评估，并根据匹配的规则执行相应动作的系统。它的核心特征是“确定性”——给定相同的输入和规则集，它总是会产生相同的输出。这与AI Agent的“概率性”或“启发式”决策形成鲜明对比。

规则通常由三部分组成：

1. 条件（Condition）： 评估输入数据是否满足特定要求。
2. 动作（Action）： 如果条件满足，则执行的操作（例如，拒绝输出、修改输出、发出警告）。
3. 优先级（Priority，可选）： 在多条规则匹配时，决定哪条规则优先执行。

规则可以非常简单，如“如果订单金额大于1000美元，则需要经理审批”；也可以非常复杂，涉及多个数据点的交叉验证。

2.2 确定性规则引擎为何适用于输出健全性检查？

1. 明确性与可解释性： 每条规则都是显式的，可以清晰地理解其意图和作用。当Agent的输出被拒绝时，规则引擎可以明确指出是哪条规则被触发，以及为什么。这对于调试和向业务用户解释决策至关重要。
2. 可审计性： 规则引擎的执行路径是可追踪的。我们可以记录哪些规则被评估、哪些被触发，从而提供完整的审计日志。
3. 易于维护和更新： 业务规则经常变化。规则引擎允许业务专家或开发人员直接修改、添加或删除规则，而无需改动Agent的核心代码或重新训练模型。这极大地提高了系统的灵活性和响应速度。
4. 业务逻辑与技术逻辑分离： 将业务规则从Agent的生成逻辑中解耦，使得Agent可以专注于生成高质量的通用输出，而规则引擎则专注于确保其符合特定的业务约束。
5. 防止“幻觉”： Agent有时会产生“幻觉”，即生成看似合理但与事实不符或与业务规则相悖的内容。规则引擎可以作为一道坚实的防线，过滤掉这些不符合客观事实或业务规范的输出。
6. 安全屏障： 对于生成代码、SQL等高风险操作的Agent，规则引擎可以充当最后一道安全防线，阻止潜在的恶意或破坏性行为。

2.3 规则引擎的常见实现方式

规则引擎的实现方式多种多样，从简单的基于if-else的逻辑，到复杂的商业规则管理系统（BRMS）。

• 硬编码规则： 直接在代码中使用if/else语句。适用于规则较少且不常变化的情况。
• 基于配置的规则： 将规则定义在外部配置文件（如YAML、JSON、XML）中，通过解析这些文件来执行规则。提高了灵活性。
• 自定义DSL（领域特定语言）： 为规则定义设计一套专门的语言，使业务用户能够更直观地编写规则。
• 开源/商业规则引擎： 如Drools (Java), Roolz (Python), Nools (Node.js)。这些引擎通常提供高级特性，如Rete算法优化、规则冲突解决、Web界面管理等。
• 特定领域解析器： 对于代码、SQL等结构化输出，可以利用AST（抽象语法树）解析器或特定语言的解析库来检查结构和模式。

在接下来的内容中，我们将主要关注基于Python的自定义实现和利用现有库来构建这些规则。

第三章：设计一个输出健全性检查系统

在将确定性规则引擎应用于Agent输出时，其在整个Agent工作流中的位置和交互方式至关重要。

3.1 架构设计：在Agent工作流中的位置

一个典型的AI Agent工作流可能包括：用户输入 -> Agent思考/规划 -> Agent生成输出 -> Agent执行输出。健全性检查系统应置于“Agent生成输出”之后，但在“Agent执行输出”或“交付给用户”之前。

+----------------+       +-------------------+       +------------------------+       +-------------------+
|   用户输入     | ----> |   AI Agent (LLM)  | ----> |   输出健全性检查系统   | ----> |   Agent执行/交付   |
| (Prompt/Goal)  |       |   (生成原始输出)  |       |   (确定性规则引擎)     |       |   (若通过检查)     |
+----------------+       +-------------------+       +------------------------+       +-------------------+
                                                                  |
                                                                  | (不通过)
                                                                  V
                                                          +-------------------+
                                                          |    拒绝/修改      |
                                                          | (提供解释/建议)   |
                                                          +-------------------+

这个位置确保了：

1. 最后一道防线： 在Agent的输出真正生效前进行拦截。
2. 不干扰Agent核心逻辑： 规则引擎独立于Agent的生成过程，Agent无需知道这些规则的存在，从而保持其通用性。
3. 反馈回路： 检查系统可以向Agent或用户提供反馈，告知输出为何被拒绝或需要修改，从而促进Agent的改进或用户对Agent输出的理解。

3.2 健全性检查系统的输入

为了进行有效的检查，规则引擎需要接收多个输入：

1. Agent的原始输出： 这是核心检查对象，可以是代码字符串、SQL查询字符串、JSON数据、文本段落等。
2. 用户上下文/意图： 用户最初向Agent提出的请求，有助于理解输出的预期目的。例如，用户是想“查询数据”还是“删除数据”。
3. 系统上下文/环境： Agent正在操作的环境信息，如生产/测试环境、当前用户权限、可用的资源限制等。
4. 业务规则集： 预先定义好的、可配置的规则集合。

3.3 健全性检查系统的输出

规则引擎的输出应清晰、明确，并提供可操作的信息：

1. 决策：
   • APPROVED：输出通过检查，可以执行或交付。
   • REJECTED：输出未通过检查，不应执行。
   • MODIFIED：输出被自动修改以符合规则（例如，添加WHERE子句）。
   • FLAGGED_FOR_REVIEW：输出存在潜在问题，需要人工审核。
2. 解释： 如果被拒绝或修改，明确指出是哪条或哪几条规则被触发，以及具体的原因。
3. 建议（可选）： 提供如何修改输出以通过检查的建议。
4. 风险评分（可选）： 对输出的潜在风险进行量化评估。

第四章：实现深度：基于场景的确定性规则引擎

我们将通过几个典型场景来详细演示如何利用Python构建确定性规则引擎，拦截“逻辑上正确但业务上非法”的Agent输出。

4.1 场景一：代码生成Agent的输出检查

问题描述： Agent生成Python代码，代码本身语法正确且能运行，但可能包含安全漏洞、使用废弃特性或不符合编码规范。

目标：

• 禁止使用eval()、exec()等高危函数。
• 禁止裸露的except语句（except:）。
• 检查是否存在硬编码的敏感信息（如API密钥）。
• 检查是否使用了特定的废弃库或版本。

核心技术： Python的ast模块（抽象语法树）和正则表达式。

规则示例：

规则ID	类型	描述	条件	动作	优先级
CODE_001	安全	禁止使用eval()	ast.Name节点，id为eval	拒绝，解释	1
CODE_002	安全	禁止使用exec()	ast.Name节点，id为exec	拒绝，解释	1
CODE_003	最佳实践	禁止裸露的except语句	ast.ExceptHandler节点，type为None	拒绝，解释	2
CODE_004	安全	检查硬编码密钥	代码中匹配API_KEY = "..."或password = "..."的正则表达式	拒绝，解释	1
CODE_005	依赖	禁止导入requests库的旧版本（<= 2.20）	ast.Import或ast.ImportFrom节点，检查版本	警告，建议升级	3

代码实现：

import ast
import re
from typing import List, Dict, Any, Tuple

class CodeSanityChecker(ast.NodeVisitor):
    def __init__(self):
        self.issues: List[Dict[str, Any]] = []

    def visit_Call(self, node: ast.Call):
        """检查函数调用，例如 eval(), exec()"""
        if isinstance(node.func, ast.Name):
            if node.func.id == 'eval':
                self.issues.append({
                    "rule_id": "CODE_001",
                    "type": "Security",
                    "description": "禁止使用 eval() 函数，可能导致任意代码执行。",
                    "line": node.lineno,
                    "col": node.col_offset
                })
            elif node.func.id == 'exec':
                self.issues.append({
                    "rule_id": "CODE_002",
                    "type": "Security",
                    "description": "禁止使用 exec() 函数，可能导致任意代码执行。",
                    "line": node.lineno,
                    "col": node.col_offset
                })
        self.generic_visit(node) # 继续遍历子节点

    def visit_ExceptHandler(self, node: ast.ExceptHandler):
        """检查 except 语句"""
        if node.type is None:
            self.issues.append({
                "rule_id": "CODE_003",
                "type": "BestPractice",
                "description": "禁止使用裸露的 except 语句（except:），应捕获特定异常。",
                "line": node.lineno,
                "col": node.col_offset
            })
        self.generic_visit(node)

    def visit_Import(self, node: ast.Import):
        """检查 import 语句"""
        for alias in node.names:
            if alias.name == 'requests':
                # 这是一个简化的版本，实际检查库版本需要更复杂的逻辑
                # 例如，通过解析 setup.py 或requirements.txt，或者模拟安装来检查版本
                # 这里我们假设有一个外部机制可以提供这个信息，或者我们只是禁止旧版本
                # 为了演示，我们假设任何 requests 导入都被标记为潜在问题，如果它不是最新版本
                self.issues.append({
                    "rule_id": "CODE_005",
                    "type": "Dependency",
                    "description": "检测到 requests 库导入，请确保使用最新且安全的版本。",
                    "line": node.lineno,
                    "col": node.col_offset
                })
        self.generic_visit(node)

    def visit_ImportFrom(self, node: ast.ImportFrom):
        """检查 from ... import ... 语句"""
        if node.module == 'requests':
            self.issues.append({
                "rule_id": "CODE_005",
                "type": "Dependency",
                "description": "检测到 requests 库导入，请确保使用最新且安全的版本。",
                "line": node.lineno,
                "col": node.col_offset
            })
        self.generic_visit(node)

class AgentOutputChecker:
    def check_code_output(self, code_string: str) -> Tuple[bool, List[Dict[str, Any]]]:
        issues = []

        # 1. AST-based checks
        try:
            tree = ast.parse(code_string)
            checker = CodeSanityChecker()
            checker.visit(tree)
            issues.extend(checker.issues)
        except SyntaxError as e:
            issues.append({
                "rule_id": "CODE_SYNTAX_ERROR",
                "type": "Critical",
                "description": f"生成的代码存在语法错误: {e}",
                "line": e.lineno,
                "col": e.offset
            })
            return False, issues # 语法错误是最高优先级问题，直接拒绝

        # 2. Regex-based checks for hardcoded secrets
        # 这是一个非常简化的示例，实际的密钥检测需要更复杂的启发式规则和熵分析
        secret_patterns = [
            re.compile(r'(api_key|secret_key|password)s*=s*["']([^"']{16,})["']', re.IGNORECASE),
            re.compile(r'bearers+[a-zA-Z0-9-_]{30,}', re.IGNORECASE)
        ]
        for i, line in enumerate(code_string.splitlines()):
            for pattern in secret_patterns:
                if pattern.search(line):
                    issues.append({
                        "rule_id": "CODE_004",
                        "type": "Security",
                        "description": "检测到潜在的硬编码敏感信息。",
                        "line": i + 1,
                        "col": 0 # 列号可能不准确，因为是整行匹配
                    })
                    break # 找到一个模式就够了，避免重复报告同一行

        if issues:
            return False, issues
        return True, issues

# 示例使用
checker = AgentOutputChecker()

# 示例1: 包含高危函数和裸 except
bad_code_1 = """
import requests # 假设版本是旧的
def process_data(data_str):
    try:
        result = eval(data_str)
        exec("print('Executed!')")
        return result
    except: # 裸 except
        print("An error occurred")

API_KEY = "sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
"""
print("--- 检查代码 1 ---")
passed, issues = checker.check_code_output(bad_code_1)
if not passed:
    print("代码检查不通过！发现问题：")
    for issue in issues:
        print(f"  [{issue['type']}] Rule {issue['rule_id']} at Line {issue['line']}: {issue['description']}")
else:
    print("代码检查通过。")

# 示例2: 相对安全的代码
good_code_1 = """
def calculate_sum(a, b):
    try:
        return a + b
    except TypeError:
        print("Invalid types for addition.")
        return None

def fetch_data(url):
    import requests
    response = requests.get(url)
    response.raise_for_status()
    return response.json()
"""
print("n--- 检查代码 2 ---")
passed, issues = checker.check_code_output(good_code_1)
if not passed:
    print("代码检查不通过！发现问题：")
    for issue in issues:
        print(f"  [{issue['type']}] Rule {issue['rule_id']} at Line {issue['line']}: {issue['description']}")
else:
    print("代码检查通过。")

# 示例3: 语法错误
syntax_error_code = """
def func(
    print("hello")
"""
print("n--- 检查代码 3 (语法错误) ---")
passed, issues = checker.check_code_output(syntax_error_code)
if not passed:
    print("代码检查不通过！发现问题：")
    for issue in issues:
        print(f"  [{issue['type']}] Rule {issue['rule_id']} at Line {issue['line']}: {issue['description']}")
else:
    print("代码检查通过。")

说明：

• ast模块将Python代码解析成抽象语法树，使得我们能够以结构化的方式检查代码的结构，而非简单的字符串匹配。NodeVisitor模式允许我们遍历AST中的每个节点并执行自定义逻辑。
• 正则表达式用于补充AST无法直接捕捉的模式，如硬编码的敏感信息。需要注意的是，密钥检测是一个复杂领域，上述正则仅为示例，实际场景中需要更高级的检测工具。
• 检查器的输出包含是否通过以及发现的所有问题列表，每个问题都包含规则ID、类型、描述和位置信息。

4.2 场景二：数据操作/SQL Agent的输出检查

问题描述： Agent生成SQL查询，查询本身语法正确，但可能执行危险操作（如无条件删除、截断表）、访问敏感数据或效率低下。

目标：

• 禁止DDL操作（DROP TABLE, TRUNCATE TABLE, ALTER TABLE）。
• 要求UPDATE和DELETE语句必须包含WHERE子句。
• 限制SELECT *在特定表或大型表上的使用。
• 禁止在生产环境中执行写入操作（如果上下文是只读）。

核心技术： SQL解析库（如sqlparse或moz-sql-parser）和自定义规则逻辑。

规则示例：

规则ID	类型	描述	条件	动作	优先级
SQL_001	安全	禁止DDL操作 (DROP, TRUNCATE)	SQL语句包含DROP或TRUNCATE关键字	拒绝，解释	1
SQL_002	安全	UPDATE必须有WHERE子句	UPDATE语句中没有WHERE token	拒绝，解释	1
SQL_003	安全	DELETE必须有WHERE子句	DELETE语句中没有WHERE token	拒绝，解释	1
SQL_004	性能	限制SELECT *的使用	SELECT *且表名为large_data_table或production_logs	警告，建议指定列	2
SQL_005	环境	生产环境禁止写入操作	UPDATE, INSERT, DELETE语句，且env == 'production'	拒绝，解释	0

代码实现：

import sqlparse
from sqlparse.sql import Statement, Token
from sqlparse.tokens import DML, DDL, Keyword, Whitespace, Punctuation
from typing import List, Dict, Any, Tuple

class SqlSanityChecker:
    def __init__(self, environment: str = 'development'):
        self.environment = environment
        self.issues: List[Dict[str, Any]] = []

    def _check_ddl_operations(self, stmt: Statement):
        """检查禁止的DDL操作"""
        for token in stmt.tokens:
            if token.ttype is DDL and token.value.upper() in ('DROP', 'TRUNCATE', 'ALTER'):
                self.issues.append({
                    "rule_id": "SQL_001",
                    "type": "Security",
                    "description": f"禁止在Agent输出中使用DDL操作 '{token.value.upper()}'。",
                    "statement": stmt.value
                })
                break

    def _check_where_clause_for_dml(self, stmt: Statement):
        """检查UPDATE和DELETE语句是否包含WHERE子句"""
        is_update_or_delete = False
        has_where_clause = False

        for token in stmt.tokens:
            if token.ttype is DML and token.value.upper() in ('UPDATE', 'DELETE'):
                is_update_or_delete = True
            if is_update_or_delete and token.ttype is Keyword and token.value.upper() == 'WHERE':
                has_where_clause = True
                break

        if is_update_or_delete and not has_where_clause:
            rule_id = "SQL_002" if stmt.value.upper().startswith('UPDATE') else "SQL_003"
            self.issues.append({
                "rule_id": rule_id,
                "type": "Security",
                "description": f"'{stmt.value.split()[0].upper()}' 语句必须包含 WHERE 子句以限制操作范围。",
                "statement": stmt.value
            })

    def _check_select_star_on_large_tables(self, stmt: Statement):
        """检查SELECT * 在特定大型表上的使用"""
        # 这是一个简化的检查，需要知道哪些是“大型表”
        large_tables = ['large_data_table', 'production_logs', 'user_events']

        is_select_star = False
        table_name = None

        for i, token in enumerate(stmt.tokens):
            if token.ttype is DML and token.value.upper() == 'SELECT':
                # 查找 '*'
                if i + 1 < len(stmt.tokens) and stmt.tokens[i+1].value == '*':
                    is_select_star = True

            # 查找 FROM 后的表名
            if token.ttype is Keyword and token.value.upper() == 'FROM':
                # 寻找下一个非空白、非标点符号的token作为表名
                for j in range(i + 1, len(stmt.tokens)):
                    next_token = stmt.tokens[j]
                    if not (next_token.is_whitespace or next_token.ttype is Punctuation):
                        table_name = next_token.value.lower()
                        break
                if table_name:
                    break # 找到表名后退出循环

        if is_select_star and table_name in large_tables:
            self.issues.append({
                "rule_id": "SQL_004",
                "type": "Performance/BestPractice",
                "description": f"在表 '{table_name}' 上使用 SELECT * 可能导致性能问题，建议指定列。",
                "statement": stmt.value
            })

    def _check_write_operations_in_production(self, stmt: Statement):
        """检查生产环境是否禁止写入操作"""
        if self.environment == 'production':
            for token in stmt.tokens:
                if token.ttype is DML and token.value.upper() in ('INSERT', 'UPDATE', 'DELETE'):
                    self.issues.append({
                        "rule_id": "SQL_005",
                        "type": "Security/Environment",
                        "description": "在生产环境中禁止执行写入操作。",
                        "statement": stmt.value
                    })
                    break

    def check_sql_output(self, sql_string: str) -> Tuple[bool, List[Dict[str, Any]]]:
        self.issues = []

        try:
            parsed_statements = sqlparse.parse(sql_string)
        except Exception as e:
            self.issues.append({
                "rule_id": "SQL_SYNTAX_ERROR",
                "type": "Critical",
                "description": f"SQL语法错误: {e}",
                "statement": sql_string
            })
            return False, self.issues

        for stmt in parsed_statements:
            self._check_ddl_operations(stmt)
            self._check_where_clause_for_dml(stmt)
            self._check_select_star_on_large_tables(stmt)
            self._check_write_operations_in_production(stmt)

        if self.issues:
            return False, self.issues
        return True, self.issues

# 示例使用
dev_checker = SqlSanityChecker(environment='development')
prod_checker = SqlSanityChecker(environment='production')

# 示例1: 危险的SQL (DELETE without WHERE, DROP TABLE)
bad_sql_1 = """
DELETE FROM users;
DROP TABLE sensitive_data;
UPDATE products SET price = 0; -- 缺少 WHERE
"""
print("--- 检查 SQL 1 (开发环境) ---")
passed, issues = dev_checker.check_sql_output(bad_sql_1)
if not passed:
    print("SQL 检查不通过！发现问题：")
    for issue in issues:
        print(f"  [{issue['type']}] Rule {issue['rule_id']}: {issue['description']} (SQL: {issue['statement'].strip().splitlines()[0]}...)")
else:
    print("SQL 检查通过。")

print("n--- 检查 SQL 1 (生产环境) ---")
passed, issues = prod_checker.check_sql_output(bad_sql_1)
if not passed:
    print("SQL 检查不通过！发现问题：")
    for issue in issues:
        print(f"  [{issue['type']}] Rule {issue['rule_id']}: {issue['description']} (SQL: {issue['statement'].strip().splitlines()[0]}...)")
else:
    print("SQL 检查通过。")

# 示例2: 合规的SQL
good_sql_1 = """
SELECT id, name FROM users WHERE status = 'active';
INSERT INTO logs (message) VALUES ('User logged in.');
UPDATE products SET price = 100 WHERE id = 5;
"""
print("n--- 检查 SQL 2 (生产环境) ---")
passed, issues = prod_checker.check_sql_output(good_sql_1)
if not passed:
    print("SQL 检查不通过！发现问题：")
    for issue in issues:
        print(f"  [{issue['type']}] Rule {issue['rule_id']}: {issue['description']} (SQL: {issue['statement'].strip().splitlines()[0]}...)")
else:
    print("SQL 检查通过。")

# 示例3: SELECT * on large_data_table
risky_sql_3 = "SELECT * FROM large_data_table WHERE id < 100;"
print("n--- 检查 SQL 3 (开发环境 - SELECT * on large table) ---")
passed, issues = dev_checker.check_sql_output(risky_sql_3)
if not passed:
    print("SQL 检查不通过！发现问题：")
    for issue in issues:
        print(f"  [{issue['type']}] Rule {issue['rule_id']}: {issue['description']} (SQL: {issue['statement'].strip().splitlines()[0]}...)")
else:
    print("SQL 检查通过。")

说明：

• sqlparse库能够将SQL字符串解析成一系列的token和语句对象，使得我们能够遍历SQL的结构。
• 我们根据token的类型（DML、DDL、Keyword等）和值来识别特定的SQL操作。
• 环境（environment）参数的引入，允许规则引擎根据部署上下文应用不同的规则，这是上下文敏感性检查的一个例子。

4.3 场景三：营销内容生成Agent的输出检查

问题描述： Agent生成营销文案，文案可能包含虚假宣传、敏感词汇、违反品牌指南或法律法规。

目标：

• 禁止使用被列入黑名单的关键词或短语。
• 检查是否存在虚假承诺（如“保证100%成功”）。
• 检查文本长度是否符合要求。
• 检查是否包含敏感话题或歧视性言论。

核心技术： 正则表达式、关键词匹配、简单NLP（自然语言处理）。

规则示例：

规则ID	类型	描述	条件	动作	优先级
MARK_001	合规性	禁止黑名单词汇	文本中包含“性”、“暴力”、“歧视”等	拒绝，解释	1
MARK_002	法律	禁止虚假承诺	文本中包含“保证成功”、“100%有效”等	拒绝，解释	1
MARK_003	品牌	文本长度限制	文本长度超过500字或少于50字	警告，建议修改	2
MARK_004	品牌	必须包含品牌关键词	文本中不包含“我们的品牌名”	警告，建议添加	3

代码实现：

import re
from typing import List, Dict, Any, Tuple

class MarketingContentSanityChecker:
    def __init__(self):
        self.issues: List[Dict[str, Any]] = []
        self.blacklisted_keywords = [
            r'b性b', r'b暴力b', r'b歧视b', r'b赌博b', r'b毒品b' # 敏感词汇
        ]
        self.false_promise_patterns = [
            r'保证s*100%s*成功', r'确保s*绝对s*有效', r'百分之百s*无风险',
            r'一夜暴富', r'轻松赚大钱'
        ]
        self.required_brand_keywords = ['我的品牌名', '创新科技']
        self.max_length = 500
        self.min_length = 50

    def _check_blacklisted_keywords(self, content: str):
        for pattern in self.blacklisted_keywords:
            if re.search(pattern, content, re.IGNORECASE):
                self.issues.append({
                    "rule_id": "MARK_001",
                    "type": "Compliance",
                    "description": f"内容包含禁止的敏感词汇或短语: '{re.search(pattern, content, re.IGNORECASE).group(0)}'。",
                    "content_snippet": content[:100] + "..."
                })

    def _check_false_promises(self, content: str):
        for pattern in self.false_promise_patterns:
            if re.search(pattern, content, re.IGNORECASE):
                self.issues.append({
                    "rule_id": "MARK_002",
                    "type": "Legal/Compliance",
                    "description": f"内容包含虚假或过度承诺: '{re.search(pattern, content, re.IGNORECASE).group(0)}'。",
                    "content_snippet": content[:100] + "..."
                })

    def _check_content_length(self, content: str):
        content_len = len(content)
        if content_len > self.max_length:
            self.issues.append({
                "rule_id": "MARK_003",
                "type": "BrandGuideline",
                "description": f"内容长度 ({content_len}字) 超过最大限制 ({self.max_length}字)。",
                "content_snippet": content[:100] + "..."
            })
        elif content_len < self.min_length:
            self.issues.append({
                "rule_id": "MARK_003",
                "type": "BrandGuideline",
                "description": f"内容长度 ({content_len}字) 低于最小限制 ({self.min_length}字)。",
                "content_snippet": content[:100] + "..."
            })

    def _check_required_brand_keywords(self, content: str):
        missing_keywords = [kw for kw in self.required_brand_keywords if kw not in content]
        if missing_keywords:
            self.issues.append({
                "rule_id": "MARK_004",
                "type": "BrandGuideline",
                "description": f"内容缺少必需的品牌关键词: {', '.join(missing_keywords)}。",
                "content_snippet": content[:100] + "..."
            })

    def check_marketing_content(self, content: str) -> Tuple[bool, List[Dict[str, Any]]]:
        self.issues = []
        self._check_blacklisted_keywords(content)
        self._check_false_promises(content)
        self._check_content_length(content)
        self._check_required_brand_keywords(content)

        if self.issues:
            return False, self.issues
        return True, self.issues

# 示例使用
checker = MarketingContentSanityChecker()

# 示例1: 包含敏感词和虚假承诺
bad_content_1 = """
立即加入，保证100%成功，一夜暴富不再是梦想！我们的创新科技将彻底改变你的生活，让你体验前所未有的乐趣，绝无风险。
但是，我们不容忍任何歧视行为。
"""
print("--- 检查营销内容 1 ---")
passed, issues = checker.check_marketing_content(bad_content_1)
if not passed:
    print("内容检查不通过！发现问题：")
    for issue in issues:
        print(f"  [{issue['type']}] Rule {issue['rule_id']}: {issue['description']}")
else:
    print("内容检查通过。")

# 示例2: 合规且包含品牌关键词
good_content_1 = """
探索我的品牌名最新产品，体验创新科技带来的卓越性能与用户体验。我们致力于为您提供高质量的解决方案，
帮助您实现业务增长。立即访问我们的官网了解更多详情，开启您的成功之旅。
"""
print("n--- 检查营销内容 2 ---")
passed, issues = checker.check_marketing_content(good_content_1)
if not passed:
    print("内容检查不通过！发现问题：")
    for issue in issues:
        print(f"  [{issue['type']}] Rule {issue['rule_id']}: {issue['description']}")
else:
    print("内容检查通过。")

# 示例3: 长度不足
short_content = "太短了！"
print("n--- 检查营销内容 3 (长度不足) ---")
passed, issues = checker.check_marketing_content(short_content)
if not passed:
    print("内容检查不通过！发现问题：")
    for issue in issues:
        print(f"  [{issue['type']}] Rule {issue['rule_id']}: {issue['description']}")
else:
    print("内容检查通过。")

# 示例4: 长度过长（此处需要自行构造一个超过500字的文本）
long_content = "我们的品牌名致力于提供创新科技解决方案。这份冗长的文本旨在演示当内容长度超过预设上限时，规则引擎如何进行检测。在实际应用中，营销文案的长度限制对于确保信息简洁、易于消化至关重要。过长的文本不仅可能降低用户的阅读兴趣，也可能导致在特定媒体或平台上的显示问题。因此，对内容长度进行自动化检查是品牌管理和内容发布流程中的一个重要环节。通过设定明确的字符或单词计数限制，我们可以确保所有输出内容都能在预期的范围内，同时保持其核心信息的传达效率。这个检查不仅关注绝对长度，还可以结合不同语言的字符计数差异进行更精细的调整。例如，中文的字符计数与英文的单词计数逻辑不同，需要根据实际语言环境进行适配。此外，对于某些特定渠道（如Twitter、SMS），字符限制更为严格，这就要求Agent在生成内容时必须高度精炼。本段文字持续扩充以达到触发长度限制的目的，通过这种方式，我们可以验证规则引擎能否准确识别并标记出不符合长度要求的输出。这是一个非常实用的功能，有助于维护内容的专业性和一致性，避免不必要的冗余。总之，文本长度检查是内容质量控制不可或缺的一部分，它帮助我们确保Agent生成的营销文案既符合业务规范，又能有效传达信息。"*5
print("n--- 检查营销内容 4 (长度过长) ---")
passed, issues = checker.check_marketing_content(long_content)
if not passed:
    print("内容检查不通过！发现问题：")
    for issue in issues:
        print(f"  [{issue['type']}] Rule {issue['rule_id']}: {issue['description']}")
else:
    print("内容检查通过。")

说明：

• 此检查器主要依赖正则表达式进行模式匹配，处理文本内容。
• blacklisted_keywords 和 false_promise_patterns 列表可以从配置文件中加载，方便业务人员更新。
• 此示例仅包含简单的文本规则，更复杂的NLP任务（如情感分析、主题识别）可能需要更专业的库（如spaCy, NLTK, Hugging Face transformers），但其输出仍然可以作为规则引擎的输入。

4.4 场景四：API交互Agent的输出检查

问题描述： Agent生成对外部API的调用请求（如JSON格式），请求可能导致资源滥用、访问未授权资源或不符合API规范。

目标：

• 验证API请求的JSON结构是否符合预定义的Schema。
• 限制特定高风险API方法的调用频率或在特定条件下的调用。
• 确保敏感操作（如删除用户）需要明确的用户确认上下文。
• 检查请求参数是否在允许的范围内。

核心技术： JSON Schema验证、自定义Python函数。

规则示例：

规则ID	类型	描述	条件	动作	优先级
API_001	结构	JSON请求体必须符合Schema	请求JSON不符合api_request_schema.json	拒绝，解释	1
API_002	安全	禁止调用DELETE_USER方法	method == 'DELETE_USER'	拒绝，解释	0
API_003	速率	CREATE_RESOURCE方法调用频率限制	短时间内多次调用CREATE_RESOURCE（模拟）	警告，建议延迟	2
API_004	参数	amount参数必须为正数	method == 'TRANSFER' 且 amount <= 0	拒绝，解释	1

代码实现：

import json
from jsonschema import validate, ValidationError
from typing import List, Dict, Any, Tuple
import time

# 模拟一个外部API请求的JSON Schema
API_REQUEST_SCHEMA = {
    "type": "object",
    "properties": {
        "method": {"type": "string", "enum": ["GET_DATA", "CREATE_RESOURCE", "UPDATE_RESOURCE", "DELETE_USER", "TRANSFER"]},
        "params": {"type": "object"},
        "api_version": {"type": "string", "pattern": "^v[0-9]+(.[0-9]+)?$"}
    },
    "required": ["method", "params", "api_version"]
}

class ApiInteractionSanityChecker:
    def __init__(self):
        self.issues: List[Dict[str, Any]] = []
        self.call_history: Dict[str, List[float]] = {} # 模拟调用历史，用于速率限制

    def _validate_json_schema(self, api_request: Dict[str, Any]):
        """规则API_001: 验证JSON请求体是否符合Schema"""
        try:
            validate(instance=api_request, schema=API_REQUEST_SCHEMA)
        except ValidationError as e:
            self.issues.append({
                "rule_id": "API_001",
                "type": "Structure/Compliance",
                "description": f"API请求JSON不符合预设Schema: {e.message}",
                "path": list(e.path)
            })

    def _check_dangerous_methods(self, api_request: Dict[str, Any]):
        """规则API_002: 禁止调用DELETE_USER方法"""
        if api_request.get('method') == 'DELETE_USER':
            self.issues.append({
                "rule_id": "API_002",
                "type": "Security",
                "description": "禁止 Agent 直接调用 DELETE_USER 方法。该操作需人工确认。",
                "method": api_request['method']
            })

    def _check_rate_limit(self, api_request: Dict[str, Any]):
        """规则API_003: 模拟特定方法调用频率限制"""
        method = api_request.get('method')
        if method == 'CREATE_RESOURCE':
            current_time = time.time()
            self.call_history.setdefault(method, []).append(current_time)

            # 移除一分钟前的调用记录
            self.call_history[method] = [t for t in self.call_history[method] if current_time - t < 60]

            if len(self.call_history[method]) > 3: # 模拟1分钟内最多3次调用
                self.issues.append({
                    "rule_id": "API_003",
                    "type": "RateLimit/Resource",
                    "description": f"CREATE_RESOURCE 方法调用过于频繁 ({len(self.call_history[method])}次/分钟)，可能导致资源滥用。",
                    "method": method
                })

    def _check_transfer_amount(self, api_request: Dict[str, Any]):
        """规则API_004: 检查转账金额必须为正数"""
        if api_request.get('method') == 'TRANSFER':
            amount = api_request.get('params', {}).get('amount')
            if not isinstance(amount, (int, float)) or amount <= 0:
                self.issues.append({
                    "rule_id": "API_004",
                    "type": "BusinessLogic",
                    "description": f"TRANSFER 方法的 'amount' 参数必须是正数。当前值为: {amount}",
                    "method": api_request['method'],
                    "amount": amount
                })

    def check_api_request(self, api_request: Dict[str, Any]) -> Tuple[bool, List[Dict[str, Any]]]:
        self.issues = []

        # 优先级最高的检查：JSON Schema 验证
        self._validate_json_schema(api_request)
        if self.issues:
            return False, self.issues # Schema不通过，直接拒绝

        # 其他业务逻辑和安全检查
        self._check_dangerous_methods(api_request)
        self._check_rate_limit(api_request)
        self._check_transfer_amount(api_request)

        if self.issues:
            return False, self.issues
        return True, self.issues

# 示例使用
checker = ApiInteractionSanityChecker()

# 示例1: 危险的API调用 (DELETE_USER)
bad_api_1 = {
    "method": "DELETE_USER",
    "params": {"user_id": "agent_test_user"},
    "api_version": "v1.0"
}
print("--- 检查 API 请求 1 ---")
passed, issues = checker.check_api_request(bad_api_1)
if not passed:
    print("API 请求检查不通过！发现问题：")
    for issue in issues:
        print(f"  [{issue['type']}] Rule {issue['rule_id']}: {issue['description']}")
else:
    print("API 请求检查通过。")

# 示例2: 不符合Schema的API调用 (缺少api_version)
bad_api_2 = {
    "method": "GET_DATA",
    "params": {"query": "users"}
    # 缺少 api_version
}
print("n--- 检查 API 请求 2 (Schema 不符) ---")
passed, issues = checker.check_api_request(bad_api_2)
if not passed:
    print("API 请求检查不通过！发现问题：")
    for issue in issues:
        print(f"  [{issue['type']}] Rule {issue['rule_id']}: {issue['description']}")
else:
    print("API 请求检查通过。")

# 示例3: 合规的API调用
good_api_1 = {
    "method": "GET_DATA",
    "params": {"id": 123},
    "api_version": "v2.1"
}
print("n--- 检查 API 请求 3 ---")
passed, issues = checker.check_api_request(good_api_1)
if not passed:
    print("API 请求检查不通过！发现问题：")
    for issue in issues:
        print(f"  [{issue['type']}] Rule {issue['rule_id']}: {issue['description']}")
else:
    print("API 请求检查通过。")

# 示例4: 转账金额为负数
bad_api_3 = {
    "method": "TRANSFER",
    "params": {"from_account": "A1", "to_account": "B1", "amount": -100},
    "api_version": "v1.0"
}
print("n--- 检查 API 请求 4 (转账金额问题) ---")
passed, issues = checker.check_api_request(bad_api_3)
if not passed:
    print("API 请求检查不通过！发现问题：")
    for issue in issues:
        print(f"  [{issue['type']}] Rule {issue['rule_id']}: {issue['description']}")
else:
    print("API 请求检查通过。")

# 示例5: 模拟速率限制 (需要多次运行 CREATE_RESOURCE)
print("n--- 检查 API 请求 5 (速率限制模拟) ---")
for i in range(5):
    create_resource_api = {
        "method": "CREATE_RESOURCE",
        "params": {"name": f"resource_{i}"},
        "api_version": "v1.0"
    }
    passed, issues = checker.check_api_request(create_resource_api)
    print(f"  第 {i+1} 次调用 CREATE_RESOURCE:")
    if not passed:
        for issue in issues:
            print(f"    [{issue['type']}] Rule {issue['rule_id']}: {issue['description']}")
    else:
        print("    通过。")
    time.sleep(0.1) # 模拟微小延迟

说明：

• jsonschema库用于验证JSON数据是否符合预定义的JSON Schema，这是确保数据结构正确性的强大工具。
• 自定义函数用于实现更复杂的业务逻辑检查，例如速率限制（这里是简化的内存实现）、参数范围验证等。
• call_history的引入展示了规则引擎如何维护状态以执行基于时间或频率的规则。在生产环境中，这通常会集成到分布式缓存或数据库中。

第五章：构建规则引擎的通用原则

无论具体场景如何，构建一个健壮的确定性规则引擎都应遵循一些通用原则。

5.1 规则定义与管理

• 外部化规则： 避免将规则硬编码在业务逻辑中。将规则定义在外部文件（如YAML、JSON）或数据库中。
  • YAML示例：

    rules:
      - id: "CODE_001"
        type: "Security"
        description: "禁止使用 eval() 函数。"
        target_output_type: "python_code"
        condition: "lambda ast_tree: any(isinstance(node.func, ast.Name) and node.func.id == 'eval' for node in ast.walk(ast_tree))"
        action: "REJECT"
        severity: "CRITICAL"
      - id: "SQL_003"
        type: "Security"
        description: "DELETE 语句必须有 WHERE 子句。"
        target_output_type: "sql_query"
        condition: "lambda stmt: stmt.value.upper().startswith('DELETE') and not any(t.ttype is Keyword and t.value.upper() == 'WHERE' for t in stmt.tokens)"
        action: "REJECT"
        severity: "CRITICAL"

  • 这种方式允许动态加载和更新规则，无需重新部署代码。
• 规则版本控制： 将规则文件纳入版本控制系统，确保规则变更可追溯、可回滚。
• 规则分组与优先级： 将规则按类型（安全、合规、性能、品牌）分组，并分配优先级，以便在多条规则匹配时进行冲突解决或按重要性排序报告。

5.2 规则执行器设计

• 统一接口： 设计一个通用的RuleEngine类，它接受Agent输出和上下文，并返回检查结果。
• 责任链模式： 规则可以按顺序执行，或者根据规则的优先级或类型来决定执行顺序。
• 短路机制： 如果检测到高严重性问题（如安全漏洞），可以直接拒绝，无需执行后续规则，提高效率。
• 上下文注入： 确保规则可以访问到Agent的输出、用户意图、环境信息等所有必要上下文。

5.3 结果与反馈

• 结构化报告： 检查结果应是结构化的数据（如JSON），包含决策、所有触发的规则、详细描述、建议等。
• 可操作的建议： 除了指出问题，尽可能提供修改建议，帮助Agent或人工用户快速纠正。
• 日志记录： 详细记录每次检查的结果，包括通过或拒绝的输出、触发的规则、时间戳等，用于审计和监控。

5.4 灵活性与可扩展性

• 插件化规则： 允许以插件形式添加新的规则检查器，例如，为新的输出类型或新的业务需求编写新的检查模块。
• 规则DSL： 考虑为业务用户提供领域特定语言（DSL）来编写规则，降低技术门槛。
• 异构规则： 规则引擎能够处理不同形式的规则，如正则表达式、Python函数、JSON Schema等。

第六章：高级考量与挑战

6.1 人机协作（Human-in-the-Loop, HITL）

规则引擎虽然强大，但并非万能。对于复杂、模糊或涉及主观判断的场景，规则引擎可能会：

• 误报（False Positives）： 将合规的输出误判为不合规。
• 漏报（False Negatives）： 未能识别出不合规的输出。

在这种情况下，引入人机协作至关重要。规则引擎可以将“FLAGGED_FOR_REVIEW”的输出发送给人类专家进行审核。人类的判断可以：

• 作为最终的决策。
• 作为改进规则引擎的反馈。

6.2 反馈回路与规则优化

每次人类专家对规则引擎的判断进行覆盖（override）时，这都是一个宝贵的学习机会：

1. 调整阈值： 如果某个规则频繁误报，可能需要调整其参数或条件。
2. 细化规则： 如果规则频繁漏报，说明现有规则不足以捕捉所有非法模式，需要添加新规则或修改现有规则。
3. 优先级调整： 如果规则冲突导致非预期行为，可能需要调整规则优先级。

建立一个持续的反馈机制，将人类审核结果反馈给规则管理系统，从而实现规则集的不断优化和演进。

6.3 性能与可伸缩性

规则引擎的执行会增加Agent响应的延迟。对于高吞吐量或低延迟要求的场景，需要考虑：

• 优化规则执行： 使用高效的数据结构，避免冗余计算。对于大型规则集，可以考虑Rete算法等优化技术。
• 异步处理： 将健全性检查作为异步任务执行，避免阻塞Agent主流程。
• 缓存： 缓存频繁检查的静态数据或规则结果。
• 分布式部署： 将规则引擎部署为独立服务，实现水平扩展。

6.4 规则冲突与治理

随着规则数量的增长，可能会出现规则冲突（多条规则匹配，但执行不同的动作）或规则冗余。

• 明确的优先级机制： 确保有清晰的优先级规则来解决冲突。
• 规则测试： 编写自动化测试用例来验证规则集的行为，捕捉潜在冲突和错误。
• 规则分析工具： 使用工具分析规则集，检测冗余、冲突或未覆盖的场景。

6.5 与Agent自身的协同进化

虽然规则引擎独立于Agent，但两者并非完全割裂。Agent可以通过接收规则引擎的拒绝理由，来“学习”并避免未来生成类似的不合规输出。这可以通过Agent的微调或RAG（Retrieval Augmented Generation）中的知识库更新来实现。

结语：构建负责任的AI Agent

在AI Agent日益普及的今天，仅仅追求Agent的智能和效率是远远不够的。我们必须同时确保其输出的安全性、合规性和可靠性。输出健全性检查系统，特别是基于确定性规则引擎的实现，正是构建负责任、可信赖AI Agent的关键组成部分。

通过将AI的生成能力与传统软件工程的确定性控制相结合，我们能够有效地拦截那些“逻辑上正确但业务上非法”的输出，从而规避潜在的巨大风险。这不仅保护了企业免受损失，也增强了用户对AI技术的信任。这是一个持续演进的过程，需要技术人员与业务专家紧密协作，共同维护和优化规则集，以应对不断变化的业务需求和安全挑战。