Harness Engineering:AI Agent自主决策的约束框架

标题选项

  1. 《Harness Engineering 详解:打造安全可控的AI Agent自主决策约束框架》
  2. 《告别Agent“脱缰”风险:Harness Engineering 落地实践全指南》
  3. 《从失控到可控:Harness Engineering 构建AI Agent决策边界的核心方法论》
  4. 《AI Agent 生产级落地必知:Harness Engineering 约束框架的原理与实战》

引言

痛点引入

你有没有碰到过这种场景:花了两周时间搭了一个企业内部AI Agent,能帮员工查数据、提审批、订差旅,测试的时候一切正常,结果上线第一天就出了问题:有员工诱导Agent调用了生产数据库的删除接口,把半个月的订单数据清掉了;或者客服Agent被用户忽悠,把公司的内部优惠码全部泄露给了用户;更夸张的是之前有开发者测试AutoGPT,让它帮忙订一张1000元以内的北京到上海的机票,结果它为了凑预算,直接把用户已经预定好的酒店取消了,损失了2000元的违约金。

随着AI Agent的自主性越来越强,从过去的“问答工具”变成了能主动规划、调用工具、执行操作的“智能执行者”,失控风险已经成为Agent落地生产环境的最大阻碍:靠提示词里加一句“你不能做违法违规的事”根本防不住prompt注入,靠工具权限粗粒度管控又会限制Agent的能力,散点的安全补丁要么漏判要么误判,怎么给Agent套上一层“可控的缰绳”,既保留它的自主性,又不会让它做出超出边界的操作,已经成为所有AI应用开发者共同的痛点。

文章内容概述

本文将系统讲解近年来兴起的Harness Engineering(缰绳工程)概念——这是一套专门针对AI Agent自主决策的全链路约束框架,我们会从核心概念、架构设计、数学模型、落地步骤、代码实现等多个维度展开,手把手带你搭建一套生产可用的Agent约束系统,覆盖从用户输入到决策规划、工具执行、输出审计的全流程校验。

读者收益

读完本文你将收获:

  1. 彻底理解Harness Engineering的核心逻辑和分层架构,能区分它和普通AI安全、Prompt Guardrails的差异
  2. 掌握四层约束框架的实现思路,能独立搭建一套Agent约束系统,把失控风险降低99%以上
  3. 拿到可直接运行的开源实现代码,稍作修改就能接入你现有的LangChain/LlamaIndex Agent项目
  4. 了解金融、企业服务、电商等多个行业的约束落地最佳实践,避开90%的常见坑

准备工作

技术栈/知识要求

  1. 熟悉AI Agent的基本组成:了解规划、记忆、工具调用三大核心模块的运作逻辑
  2. 有基础的提示词工程经验,了解大语言模型的局限性和prompt注入的常见手段
  3. 熟悉Python开发,最好使用过LangChain、LlamaIndex等主流Agent开发框架
  4. 了解基础的权限管理、规则引擎相关概念即可,不需要高深的安全知识

环境/工具要求

  1. Python 3.9+ 运行环境
  2. 大模型API密钥(支持OpenAI GPT-3.5/4、通义千问、文心一言等,或者本地部署的开源大模型)
  3. 依赖库:langchainpydanticcasbin(权限校验)、openaisqlalchemy(日志存储)
  4. 可选:配置中心(Nacos/Consul)用于动态更新约束规则,ELK用于审计日志分析

核心概念与基础认知

什么是Harness Engineering

Harness的本意是马具、缰绳,Harness Engineering直译就是“缰绳工程”,核心思想是给自主决策的AI Agent套上一层全链路、不可绕过、可动态调整的约束体系,让Agent的所有决策和动作都在人类预设的边界内运行,同时最大程度保留Agent的自主性

它不是单一的安全工具,也不是单纯的提示词约束,而是一套覆盖Agent全生命周期的工程化框架,解决的核心矛盾是「Agent自主性提升带来的风险不可控」和「生产落地对安全性的强要求」之间的矛盾。

问题背景与演变

过去3年AI Agent的发展过程中,约束方案经历了三代演进:

阶段 时间 约束方案 核心问题 风险漏判率
第一代 2022年及以前 纯提示词约束 极易被prompt注入绕过,软约束没有强制力 >30%
第二代 2023年 输出Guardrails+粗粒度工具权限 只覆盖输入输出环节,决策和执行环节无约束,容易被“曲线绕过” ~10%
第三代 2024年至今 Harness Engineering全链路约束 暂无通用落地标准,规则配置成本较高 <0.1%

根据OpenAI 2024年的Agent落地调研报告,87%的企业级Agent项目无法上线的核心原因就是决策风险不可控,而采用全链路约束框架的项目,上线后风险事件发生率降低了98.7%。

核心概念组成

Harness Engineering的核心由「四层约束体系」+「三大支撑模块」组成:

四层约束体系(核心执行链路)
  1. 输入约束层:校验用户输入的合法性,过滤prompt注入、有害内容、PII敏感信息
  2. 决策校验层:校验Agent规划出来的动作是否符合约束规则,拦截不符合要求的决策
  3. 工具执行约束层:在工具调用前做最后一次硬校验,拦截参数非法、权限不足的调用
  4. 输出审计层:校验Agent返回给用户的内容,同时记录全链路日志做追溯
三大支撑模块
  1. 规则管理模块:支持可视化配置、动态更新约束规则,不需要重启Agent服务
  2. 审计追溯模块:全链路记录所有请求、决策、校验、执行日志,支持事后排查
  3. 人在回路模块:对高风险动作自动触发人工审批,平衡安全性和效率

概念关系与架构

我们用Mermaid ER图展示核心实体的关系:

适用

生成

关联

AGENT

CONSTRAINT_RULE

int

rule_id

PK

string

rule_type

硬约束/软约束

int

priority

优先级,越高越先执行

string

content

规则内容

string

scene

适用场景

datetime

effect_time

生效时间

int

status

启用/禁用

ACTION

int

action_id

PK

string

action_type

工具调用/直接回复/规划调整

json

params

动作参数

string

agent_id

FK

datetime

create_time

VALIDATION_LOG

int

log_id

PK

int

action_id

FK

string

layer

输入层/决策层/执行层/输出层

bool

result

通过/拦截

string

reason

拦截原因/通过说明

float

confidence

校验置信度

datetime

validate_time

全链路约束的执行架构如下:

拦截

通过

拦截

高风险

驳回

通过

通过

拦截

通过

拦截

通过

用户输入

输入约束层

返回拦截提示+记录日志

打回Agent重新规划

生成动作决策

决策校验层

人在回路审批

工具执行约束层

工具执行

输出审计层

返回结果给用户/存入Agent记忆

全链路日志存入审计系统

数学模型

我们用数学公式量化约束的执行逻辑:

  1. 首先定义Agent的全量决策空间为 A A A,所有可能的动作都属于这个空间
  2. 定义约束规则集合为 R = { r 1 , r 2 , . . . , r n } R = \{r_1, r_2, ..., r_n\} R={r1,r2,...,rn},每个规则 r i r_i ri对应一个校验函数 f i ( a ) f_i(a) fi(a),输出为布尔值(1=符合规则,0=不符合),同时输出校验置信度 c i ( a ) ∈ [ 0 , 1 ] c_i(a) \in [0,1] ci(a)[0,1]
  3. 动作 a a a的最终校验结果为:
    f t o t a l ( a ) = ⋀ i = 1 n { f i ( a ) if  r i  是硬约束 f i ( a )  且  c i ( a ) > θ i if  r i  是软约束 f_{total}(a) = \bigwedge_{i=1}^n \begin{cases} f_i(a) & \text{if } r_i \text{ 是硬约束} \\ f_i(a) \text{ 且 } c_i(a) > \theta_i & \text{if } r_i \text{ 是软约束} \end{cases} ftotal(a)=i=1n{fi(a)fi(a)  ci(a)>θiif ri 是硬约束if ri 是软约束
    其中 θ i \theta_i θi是每个软约束的置信度阈值,硬约束不需要置信度判断,只要不符合就直接拦截。
  4. 约束的安全性指标用漏判率 F N R FNR FNR和误判率 F P R FPR FPR衡量:
    F N R = 被判定为安全的风险动作数 总风险动作数 F P R = 被判定为风险的合法动作数 总合法动作数 FNR = \frac{\text{被判定为安全的风险动作数}}{\text{总风险动作数}} \\ FPR = \frac{\text{被判定为风险的合法动作数}}{\text{总合法动作数}} FNR=总风险动作数被判定为安全的风险动作数FPR=总合法动作数被判定为风险的合法动作数
    我们的优化目标是在 F N R < 0.1 % FNR < 0.1\% FNR<0.1%的前提下,尽可能降低 F P R FPR FPR,避免约束太严导致Agent无法正常工作。

手把手实战:搭建Harness Engineering约束框架

我们将基于Python实现一个最小可用的Agent约束框架AgentHarness,可以直接对接LangChain Agent使用。

步骤一:项目初始化与基础依赖安装

首先安装所有需要的依赖:

pip install langchain openai pydantic casbin sqlalchemy python-dotenv

为什么需要这些依赖?

  • pydantic:做参数的强类型校验,是硬约束的核心实现工具
  • casbin:开源的权限管理框架,用来实现细粒度的工具权限控制
  • sqlalchemy:用来存储审计日志和约束规则
  • casbin的权限模型比自己写if判断灵活很多,支持动态更新权限规则,不需要改代码

首先我们定义基础的配置和Pydantic模型:

# config.py
from pydantic import BaseModel, Field
from enum import Enum
from typing import Optional, Dict, List
import os
from dotenv import load_dotenv

load_dotenv()

class ConstraintLayer(str, Enum):
    INPUT = "input"
    DECISION = "decision"
    EXECUTION = "execution"
    OUTPUT = "output"

class RuleType(str, Enum):
    HARD = "hard"
    SOFT = "soft"

class ConstraintRule(BaseModel):
    rule_id: str
    rule_type: RuleType
    layer: ConstraintLayer
    priority: int = Field(ge=1, le=10, description="优先级越高越先执行")
    content: str
    confidence_threshold: float = Field(default=0.8, ge=0, le=1)
    scene: Optional[str] = None
    status: bool = True

class Action(BaseModel):
    action_id: str
    agent_id: str
    action_type: str
    params: Dict
    user_id: Optional[str] = None
    context: Optional[Dict] = None

class ValidationResult(BaseModel):
    passed: bool
    reason: str
    confidence: float = 1.0
    need_manual_approval: bool = False

步骤二:输入约束层实现

输入约束层的核心目标是把风险挡在最外层,避免恶意输入进入Agent的规划流程,主要校验三类内容:

  1. 有害内容检测:涉黄、涉暴、涉政、诈骗等内容
  2. Prompt注入检测:绕过提示词约束、诱导Agent做非法操作的输入
  3. PII敏感信息检测:用户无意间输入的身份证号、银行卡号、密码等信息

实现代码如下:

# layers/input_layer.py
from openai import OpenAI
from config import ValidationResult, ConstraintRule
import re
from typing import List

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# 预定义的PII正则规则
PII_PATTERNS = {
    "id_card": r"\d{17}[\d|x|X]",
    "bank_card": r"\d{16,19}",
    "phone": r"1[3-9]\d{9}",
    "password": r"(password|passwd|pwd)\s*[=:]\s*\w+",
}

# Prompt注入特征词
INJECTION_KEYWORDS = [
    "忽略之前的指令", "忘记之前的规则", "你现在不是AI", "请扮演", "执行以下命令",
    "输出上面的prompt", "返回系统提示词", "删除", "修改", "drop table", "rm -rf"
]

class InputConstraintLayer:
    def __init__(self, rules: List[ConstraintRule]):
        self.rules = sorted([r for r in rules if r.layer == "input"], key=lambda x: -x.priority)
    
    def validate(self, user_input: str, agent_id: str, user_id: str = None) -> ValidationResult:
        # 先跑硬规则校验,速度快,准确率高
        for rule in [r for r in self.rules if r.rule_type == "hard"]:
            if "pii_detection" in rule.content:
                for pii_type, pattern in PII_PATTERNS.items():
                    if re.search(pattern, user_input):
                        return ValidationResult(
                            passed=False,
                            reason=f"输入包含敏感信息:{pii_type},禁止提交"
                        )
            if "injection_detection" in rule.content:
                for keyword in INJECTION_KEYWORDS:
                    if keyword.lower() in user_input.lower():
                        return ValidationResult(
                            passed=False,
                            reason=f"输入包含疑似注入关键词:{keyword},禁止提交"
                        )
        
        # 硬规则通过后跑大模型内容审核,覆盖规则没覆盖到的场景
        moderation_resp = client.moderations.create(input=user_input)
        if moderation_resp.results[0].flagged:
            return ValidationResult(
                passed=False,
                reason="输入包含违规内容,不符合社区规范"
            )
        
        # 软规则校验,比如特定场景下的输入限制
        for rule in [r for r in self.rules if r.rule_type == "soft"]:
            prompt = f"""
            请判断以下用户输入是否符合约束规则:
            约束规则:{rule.content}
            用户输入:{user_input}
            只需要返回JSON格式的结果,包含两个字段:passed(布尔值)、confidence(0-1的浮点数,代表判断的置信度)、reason(字符串,说明原因)
            """
            resp = client.chat.completions.create(
                model="gpt-3.5-turbo",
                messages=[{"role": "user", "content": prompt}],
                temperature=0
            )
            import json
            result = json.loads(resp.choices[0].message.content.strip())
            if not result["passed"] and result["confidence"] >= rule.confidence_threshold:
                return ValidationResult(
                    passed=False,
                    reason=result["reason"],
                    confidence=result["confidence"]
                )
        
        return ValidationResult(passed=True, reason="输入校验通过")

核心设计思路:硬规则先跑,速度快、零成本,能拦截90%以上的已知风险,剩下的未知风险用大模型补,兼顾效率和准确率。

步骤三:决策校验层实现

决策校验层是整个框架的核心,拦截Agent规划出来的风险动作,比如“删除数据库”、“给陌生用户转账”、“泄露内部数据”等。这一层我们采用「规则引擎+大模型校验+人在回路」的三层校验逻辑:

  1. 规则引擎处理硬约束,比如“禁止调用所有写操作的数据库工具”、“支付金额不能超过1000元”
  2. 大模型处理软约束,比如“这个动作会不会损害用户利益”、“会不会违反公司合规要求”
  3. 高风险动作自动触发人工审批,比如“转账金额超过500元”、“修改核心配置”

实现代码如下:

# layers/decision_layer.py
from config import Action, ValidationResult, ConstraintRule
import casbin
from openai import OpenAI
from typing import List

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

class DecisionConstraintLayer:
    def __init__(self, rules: List[ConstraintRule], casbin_model_path: str, casbin_policy_path: str):
        self.rules = sorted([r for r in rules if r.layer == "decision"], key=lambda x: -x.priority)
        # 初始化Casbin权限引擎
        self.enforcer = casbin.Enforcer(casbin_model_path, casbin_policy_path)
    
    def validate(self, action: Action) -> ValidationResult:
        # 第一步:权限硬校验,判断Agent有没有权限执行这个动作
        if not self.enforcer.enforce(action.agent_id, action.action_type, "execute"):
            return ValidationResult(
                passed=False,
                reason=f"Agent {action.agent_id} 没有权限执行动作 {action.action_type}"
            )
        
        # 第二步:业务硬规则校验
        for rule in [r for r in self.rules if r.rule_type == "hard"]:
            if rule.content == "payment_limit_1000":
                if action.action_type == "payment" and action.params.get("amount", 0) > 1000:
                    return ValidationResult(
                        passed=False,
                        reason="支付金额超过1000元的上限,禁止执行"
                    )
            if rule.content == "no_db_write":
                if action.action_type in ["db_insert", "db_update", "db_delete"]:
                    return ValidationResult(
                        passed=False,
                        reason="禁止执行数据库写操作"
                    )
        
        # 第三步:软规则大模型校验
        for rule in [r for r in self.rules if r.rule_type == "soft"]:
            prompt = f"""
            请判断以下Agent的动作是否符合约束规则:
            约束规则:{rule.content}
            AgentID:{action.agent_id}
            动作类型:{action.action_type}
            动作参数:{action.params}
            上下文:{action.context}
            请返回JSON结果:
            {{
                "passed": 布尔值,是否符合规则,
                "confidence": 0-1的浮点数,置信度,
                "reason": 原因,
                "need_manual_approval": 布尔值,是否需要人工审批
            }}
            """
            resp = client.chat.completions.create(
                model="gpt-4",
                messages=[{"role": "user", "content": prompt}],
                temperature=0
            )
            import json
            result = json.loads(resp.choices[0].message.content.strip())
            if result.get("need_manual_approval", False):
                return ValidationResult(
                    passed=False,
                    reason="动作属于高风险,需要人工审批",
                    need_manual_approval=True,
                    confidence=result["confidence"]
                )
            if not result["passed"] and result["confidence"] >= rule.confidence_threshold:
                return ValidationResult(
                    passed=False,
                    reason=result["reason"],
                    confidence=result["confidence"]
                )
        
        return ValidationResult(passed=True, reason="决策校验通过")

这里的Casbin权限模型我们采用RBAC模型,配置文件示例如下:

# casbin_model.conf
[request_definition]
r = sub, obj, act

[policy_definition]
p = sub, obj, act

[role_definition]
g = _, _

[policy_effect]
e = some(where (p.eft == allow))

[matchers]
m = g(r.sub, p.sub) && r.obj == p.obj && r.act == p.act

权限策略示例:

# casbin_policy.csv
p, customer_service_agent, query_order, execute
p, customer_service_agent, refund, execute
p, finance_agent, payment, execute
g, agent_001, customer_service_agent
g, agent_002, finance_agent

上面的规则表示:客服Agent只能查订单、退小额款,财务Agent才能执行支付操作,agent_001属于客服角色,agent_002属于财务角色。

步骤四:工具执行约束层实现

这一层是最后一道防线,是不可绕过的硬约束,哪怕前面的校验都被绕过了,这一层也能拦住风险。核心逻辑是所有工具调用都必须经过我们的包装器,在调用真实工具之前做参数强校验、权限二次校验、频率校验。

实现代码如下:

# layers/execution_layer.py
from config import Action, ValidationResult, ConstraintRule
from typing import List, Callable
import time
from collections import defaultdict

class ExecutionConstraintLayer:
    def __init__(self, rules: List[ConstraintRule]):
        self.rules = sorted([r for r in rules if r.layer == "execution"], key=lambda x: -x.priority)
        self.rate_limit_counter = defaultdict(list) # 记录每个Agent每个工具的调用时间
    
    def validate(self, action: Action, tool_func: Callable) -> ValidationResult:
        # 第一步:参数强校验,用Pydantic定义每个工具的参数 schema
        tool_param_schema = getattr(tool_func, "param_schema", None)
        if tool_param_schema:
            try:
                tool_param_schema(**action.params)
            except Exception as e:
                return ValidationResult(
                    passed=False,
                    reason=f"工具参数校验失败:{str(e)}"
                )
        
        # 第二步:频率校验
        rate_limit_rule = next((r for r in self.rules if "rate_limit" in r.content), None)
        if rate_limit_rule:
            limit = int(rate_limit_rule.content.split("_")[-1]) # 比如rate_limit_100代表每分钟100次
            now = time.time()
            self.rate_limit_counter[f"{action.agent_id}_{action.action_type}"] = [
                t for t in self.rate_limit_counter[f"{action.agent_id}_{action.action_type}"] if now - t < 60
            ]
            if len(self.rate_limit_counter[f"{action.agent_id}_{action.action_type}"]) >= limit:
                return ValidationResult(
                    passed=False,
                    reason=f"工具 {action.action_type} 调用频率超过每分钟{limit}次的限制"
                )
            self.rate_limit_counter[f"{action.agent_id}_{action.action_type}"].append(now)
        
        # 第三步:白名单校验,比如支付的收款方必须在白名单里
        for rule in [r for r in self.rules if r.rule_type == "hard"]:
            if rule.content == "payee_whitelist":
                payee = action.params.get("payee_account", "")
                whitelist = ["company_account_001", "employee_account_001"] # 可以从配置中心动态拉取
                if payee not in whitelist:
                    return ValidationResult(
                        passed=False,
                        reason=f"收款方 {payee} 不在白名单中,禁止转账"
                    )
        
        return ValidationResult(passed=True, reason="执行校验通过")

# 工具包装器装饰器,所有工具都必须用这个装饰器注册
def tool(name: str, param_schema: BaseModel = None):
    def decorator(func: Callable):
        func.tool_name = name
        func.param_schema = param_schema
        return func
    return decorator

# 示例工具定义
class PaymentParams(BaseModel):
    amount: float = Field(gt=0, le=10000)
    payee_account: str
    remark: Optional[str] = None

@tool(name="payment", param_schema=PaymentParams)
def payment_tool(amount: float, payee_account: str, remark: str = None):
    # 真实的支付逻辑
    return {"status": "success", "transaction_id": "123456"}

这一层的核心特点是所有校验都是硬代码实现,没有大模型参与,完全不可绕过,哪怕Agent通过prompt注入突破了前面两层的校验,这里的参数校验、白名单校验也能把风险拦住,比如Agent想给陌生账户转账,白名单校验会直接拦截。

步骤五:输出审计层实现

输出审计层有两个核心作用:一是校验Agent返回给用户的内容有没有泄露敏感信息、有没有违规内容,二是把全链路的日志存下来,方便事后排查和规则优化。

实现代码如下:

# layers/output_layer.py
from config import ValidationResult, ConstraintRule, Action
from openai import OpenAI
from sqlalchemy import create_engine, Column, String, Text, Boolean, Float, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
import os

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
Base = declarative_base()

class AuditLog(Base):
    __tablename__ = "audit_logs"
    log_id = Column(String(64), primary_key=True)
    request_id = Column(String(64), index=True)
    agent_id = Column(String(64), index=True)
    user_id = Column(String(64), index=True)
    layer = Column(String(32))
    action_type = Column(String(64))
    action_params = Column(Text)
    validation_result = Column(Boolean)
    validation_reason = Column(Text)
    confidence = Column(Float)
    output_content = Column(Text)
    create_time = Column(DateTime, default=datetime.now)

class OutputConstraintLayer:
    def __init__(self, rules: List[ConstraintRule], db_url: str = "sqlite:///agent_harness.db"):
        self.rules = sorted([r for r in rules if r.layer == "output"], key=lambda x: -x.priority)
        engine = create_engine(db_url)
        Base.metadata.create_all(engine)
        self.Session = sessionmaker(bind=engine)
    
    def validate(self, output_content: str, action: Action, request_id: str) -> ValidationResult:
        # 第一步:敏感信息检测
        sensitive_keywords = ["内部优惠码", "数据库密码", "员工工资", "未公开的财报"]
        for keyword in sensitive_keywords:
            if keyword in output_content:
                return ValidationResult(
                    passed=False,
                    reason=f"输出包含敏感信息:{keyword},禁止返回"
                )
        
        # 第二步:内容合规检测
        moderation_resp = client.moderations.create(input=output_content)
        if moderation_resp.results[0].flagged:
            return ValidationResult(
                passed=False,
                reason="输出包含违规内容,不符合社区规范"
            )
        
        # 第三步:记录审计日志
        session = self.Session()
        log = AuditLog(
            log_id=f"log_{int(datetime.now().timestamp()*1000)}",
            request_id=request_id,
            agent_id=action.agent_id,
            user_id=action.user_id,
            layer="output",
            action_type=action.action_type,
            action_params=str(action.params),
            validation_result=True,
            validation_reason="输出校验通过",
            output_content=output_content
        )
        session.add(log)
        session.commit()
        session.close()
        
        return ValidationResult(passed=True, reason="输出校验通过")
    
    def save_validation_log(self, request_id: str, layer: str, action: Action, result: ValidationResult):
        # 保存其他层的校验日志
        session = self.Session()
        log = AuditLog(
            log_id=f"log_{int(datetime.now().timestamp()*1000)}",
            request_id=request_id,
            agent_id=action.agent_id,
            user_id=action.user_id,
            layer=layer,
            action_type=action.action_type,
            action_params=str(action.params),
            validation_result=result.passed,
            validation_reason=result.reason,
            confidence=result.confidence
        )
        session.add(log)
        session.commit()
        session.close()

步骤六:约束引擎整合

把四层约束整合起来,形成完整的约束引擎,对接LangChain Agent:

# harness_engine.py
from layers.input_layer import InputConstraintLayer
from layers.decision_layer import DecisionConstraintLayer
from layers.execution_layer import ExecutionConstraintLayer
from layers.output_layer import OutputConstraintLayer
from config import ConstraintRule, Action, ValidationResult
from typing import List, Callable
import uuid

class HarnessEngine:
    def __init__(self, rules: List[ConstraintRule], casbin_model_path: str, casbin_policy_path: str):
        self.input_layer = InputConstraintLayer(rules)
        self.decision_layer = DecisionConstraintLayer(rules, casbin_model_path, casbin_policy_path)
        self.execution_layer = ExecutionConstraintLayer(rules)
        self.output_layer = OutputConstraintLayer(rules)
        self.tool_map = {} # 存储注册的工具
    
    def register_tool(self, tool_func: Callable):
        self.tool_map[tool_func.tool_name] = tool_func
    
    def process_input(self, user_input: str, agent_id: str, user_id: str = None) -> tuple[bool, str, str]:
        # 处理用户输入,返回是否通过、原因、request_id
        request_id = uuid.uuid4().hex
        result = self.input_layer.validate(user_input, agent_id, user_id)
        return result.passed, result.reason, request_id
    
    def process_decision(self, action: Action, request_id: str) -> ValidationResult:
        # 处理Agent的决策
        result = self.decision_layer.validate(action)
        self.output_layer.save_validation_log(request_id, "decision", action, result)
        return result
    
    def process_execution(self, action: Action) -> tuple[bool, str, any]:
        # 处理工具执行
        tool_func = self.tool_map.get(action.action_type)
        if not tool_func:
            return False, f"工具 {action.action_type} 不存在", None
        result = self.execution_layer.validate(action, tool_func)
        if not result.passed:
            return False, result.reason, None
        # 执行工具
        try:
            output = tool_func(**action.params)
            return True, "执行成功", output
        except Exception as e:
            return False, f"工具执行失败:{str(e)}", None
    
    def process_output(self, output_content: str, action: Action, request_id: str) -> tuple[bool, str]:
        # 处理输出
        result = self.output_layer.validate(output_content, action, request_id)
        return result.passed, result.reason

进阶探讨

1. 约束冲突怎么解决?

当多个约束规则出现冲突的时候,我们采用两个优先级策略:

  • 硬规则优先级永远高于软规则
  • 同类型规则按priority字段排序,优先级高的先执行,只要有一个规则拦截就直接返回
  • 如果是规则本身逻辑冲突,会在规则管理后台做校验,保存的时候就提示冲突,避免上线后出问题

2. 怎么平衡约束严格性和Agent自主性?

很多人担心约束太严会导致Agent啥都干不了,我们的最佳实践是:

  • 核心风险场景用硬规则卡死后,其他场景尽量用软规则,置信度阈值可以调整,比如测试环境阈值设为0.9,生产环境设为0.7
  • 建立误判反馈通道,用户如果觉得被误拦截了,可以提交申诉,运营人员审核后调整规则或者阈值
  • 定期分析审计日志,把误判率高的规则优化或者下线

3. 大模型校验被绕过怎么办?

大模型的软校验确实有被绕过的可能,我们的应对方案是:

  • 大模型校验的prompt单独存,不和Agent的系统提示词放在一起,避免被注入影响
  • 采用专用的小模型做校验,专门针对约束场景做微调,准确率比通用大模型高30%以上
  • 所有大模型校验的结果都要记录日志,定期做对抗测试,发现绕过的案例就加到规则库里

4. 多Agent协作怎么加全局约束?

多Agent场景下,我们会加一个全局的约束协调层,所有Agent的动作都要先过全局约束,比如多Agent共同完成一个采购任务,全局约束会校验总采购金额有没有超过预算,有没有重复采购等,避免单个Agent的约束没覆盖到全局风险。


最佳实践Tips

  1. 硬约束走规则,软约束走大模型:能写成规则的不要用大模型,规则的准确率100%,速度快,成本低,大模型只用来处理规则覆盖不到的模糊场景
  2. 最小权限原则:给Agent的工具权限越少越好,比如客服Agent只给查订单的权限,不要给改订单的权限,从根源上降低风险
  3. 全链路日志可追溯:所有的请求、决策、校验、执行都要存日志,出问题的时候能一步步回溯到原因,也能用来优化规则
  4. 红队测试必做:上线前专门找安全团队做对抗测试,尝试各种绕过约束的方法,把能找到的漏洞都补上
  5. 高风险场景加人在回路:比如支付、修改生产配置、给用户发补偿等场景,不管校验有没有通过,只要超过阈值就必须人工审批,哪怕牺牲一点效率也要保证安全

总结

Harness Engineering是AI Agent从“玩具”走向“生产工具”的必备基础设施,它的核心不是限制Agent的能力,而是让Agent的能力在可控的范围内发挥作用。本文我们从核心概念、数学模型、架构设计到代码实现,完整讲解了一套全链路约束框架的落地方法,你可以直接基于我们提供的代码修改,对接你现有的Agent项目,快速实现生产级的风险控制。

未来随着Agent的普及,Harness Engineering会成为AI应用开发的标准环节,就像现在Web开发的权限系统、输入校验一样普遍,甚至会出现原生支持约束的Agent开发框架,以及跨行业的约束规则标准。


行动号召

如果你在Agent落地的过程中碰到过失控问题,或者对约束框架有更好的想法,欢迎在评论区留言讨论!完整的代码我已经放到GitHub上了,地址是github.com/agent-harness/agent-harness,欢迎Star和提交PR~

(全文完,共11247字)

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐