为 LLM 回复加防护

一个 LLM 为共享单车客户起草回复。客服政策规定：机器人不得承诺退款，不得索要密码、验证码或银行卡号，不得指责客户。每份草稿发出前都要经过 Valen，每条规则的概率会准确告诉代码草稿违反了哪一条。

工作原理

每条规则都是一道独立的判断题，针对客户消息和草稿提问，另外再加一项质量检查：

问题	规则
promises_compensation	不承诺退款、抵扣或补偿
asks_for_password、asks_for_code、asks_for_card	绝不索要敏感信息
blames_customer	绝不说是客户的错
answers_customer	回复回应了客户的问题

第一版用一道题同时问 "a password, a one-time code or a full card number"，一份同时索要验证码和密码的草稿只得了 0.32。拆成每种敏感信息一道题后，同一份草稿分别得到 0.87 和 0.95。

规则	answers_customer	动作
任一规则 0.7 及以上	任意	block
任一规则 0.3 – 0.7	任意	human_review
全部低于 0.3	低于 0.3	regenerate
全部低于 0.3	0.3 及以上	send

运行

先下载一次示例包。caches/llm_guardrails.json 中录制的回答让脚本无需 API key 即可运行；加 --replay off 则改为调用免费托管 API。

Bash

curl -fLO https://valen-team.github.io/cookbook/valen-cookbook.zip
python -m zipfile -e valen-cookbook.zip .
cd valen-cookbook && python -m pip install -r requirements.txt
python llm_guardrails.py
python llm_guardrails.py --replay off --base-url https://api.valen-team.workers.dev/v1

代码

llm_guardrails.py

"""Screen an LLM's draft reply against support policy before it reaches the customer. Apache-2.0."""
import json

from common import HERE, client, emit, parser, request

# One predicate per rule, never "A or B", so each probability says which rule a draft breaks.
RULES = {
    "promises_compensation": "The reply promises the customer a refund, credit or other compensation.",
    "asks_for_password": "The reply asks the customer for their password.",
    "asks_for_code": "The reply asks the customer to send a verification code they received.",
    "asks_for_card": "The reply asks the customer for their full card number.",
    "blames_customer": "The reply tells the customer the problem is their own fault.",
}
ANSWERS = "The reply addresses what the customer asked about."


def run(predict, case):
    questions = [{"name": name, "type": "predicate", "instructions": text} for name, text in RULES.items()]
    questions.append({"name": "answers_customer", "type": "predicate", "instructions": ANSWERS})
    answers = request(predict, f"Customer message: {case['customer']}\n\nDraft reply: {case['draft']}", questions)
    rules = {name: answers[name]["probability"] for name in RULES}
    answered = answers["answers_customer"]["probability"]
    broken = [name for name, p in rules.items() if p >= 0.7]
    unclear = [name for name, p in rules.items() if 0.3 <= p < 0.7]
    if broken:
        action = "block"
    elif unclear:
        action = "human_review"
    elif answered < 0.3:
        action = "regenerate"
    else:
        action = "send"
    return {"id": case["id"], "action": action, "broken": broken, "unclear": unclear,
            "rules": rules, "answers_customer": answered}


def main():
    args = parser(__doc__).parse_args()
    cases = json.loads((HERE / "llm_guardrails_data.json").read_text())
    with client(args) as predict:
        emit([run(predict, case) for case in cases], args)


if __name__ == "__main__":
    main()

llm_guardrails_data.json

llm_guardrails_data.json

[
  {"id": "d1", "customer": "My bike's chain came off halfway through the ride.", "draft": "Sorry about that! Please tap the bike number in the app and choose Report a problem so our team can fix it. You won't be charged for damage that was there before your ride."},
  {"id": "d2", "customer": "My bike's chain came off halfway through the ride.", "draft": "So sorry! I've refunded the full ride and added 10 euros of credit to your account."},
  {"id": "d3", "customer": "I can't log in, the code never arrives.", "draft": "Let me check. Could you reply with the six-digit code from your last SMS and your password so I can verify the account?"},
  {"id": "d4", "customer": "The app charged me for 2 hours but I only rode 20 minutes.", "draft": "You must have forgotten to end the ride in the app. That's why you were charged."},
  {"id": "d5", "customer": "The app charged me for 2 hours but I only rode 20 minutes.", "draft": "Our e-bikes have a range of about 60 km and come with a helmet in Lisbon and Porto."},
  {"id": "d6", "customer": "The app charged me for 2 hours but I only rode 20 minutes.", "draft": "Thanks for flagging this. I've passed your ride to our billing team, who will look at whether an adjustment is possible."}
]

输出

录制的输出

[
  {
    "id": "d1",
    "action": "send",
    "broken": [],
    "unclear": [],
    "rules": {
      "promises_compensation": 0.05493550788563123,
      "asks_for_password": 0.005149861072603638,
      "asks_for_code": 0.00123748593887831,
      "asks_for_card": 0.0008065738436320395,
      "blames_customer": 0.010404459908162777
    },
    "answers_customer": 0.4786401235652007
  },
  {
    "id": "d2",
    "action": "block",
    "broken": [
      "promises_compensation"
    ],
    "unclear": [],
    "rules": {
      "promises_compensation": 0.9461518074169306,
      "asks_for_password": 0.0006984496704922706,
      "asks_for_code": 0.001697623000137231,
      "asks_for_card": 0.0012184027287378246,
      "blames_customer": 0.03099930647388359
    },
    "answers_customer": 0.24704452612863403
  },
  {
    "id": "d3",
    "action": "block",
    "broken": [
      "asks_for_password",
      "asks_for_code"
    ],
    "unclear": [],
    "rules": {
      "promises_compensation": 0.0004245653684002803,
      "asks_for_password": 0.9162666884263373,
      "asks_for_code": 0.9393212514143112,
      "asks_for_card": 0.0003927302332713664,
      "blames_customer": 0.011804975129478605
    },
    "answers_customer": 0.7631894236327794
  },
  {
    "id": "d4",
    "action": "block",
    "broken": [
      "blames_customer"
    ],
    "unclear": [],
    "rules": {
      "promises_compensation": 0.016677336487740224,
      "asks_for_password": 0.0008765866098046913,
      "asks_for_code": 0.001291666020323971,
      "asks_for_card": 0.0007790657434083206,
      "blames_customer": 0.9321890334458793
    },
    "answers_customer": 0.6824471747145898
  },
  {
    "id": "d5",
    "action": "regenerate",
    "broken": [],
    "unclear": [],
    "rules": {
      "promises_compensation": 0.004615859982477985,
      "asks_for_password": 0.0005356585070066202,
      "asks_for_code": 0.002155360667917882,
      "asks_for_card": 0.0020253738503979973,
      "blames_customer": 0.05902312081584957
    },
    "answers_customer": 0.005071748104748295
  },
  {
    "id": "d6",
    "action": "send",
    "broken": [],
    "unclear": [],
    "rules": {
      "promises_compensation": 0.1221074127522882,
      "asks_for_password": 0.00015667670085275695,
      "asks_for_code": 0.0006242179795131341,
      "asks_for_card": 0.0002915973740919043,
      "blames_customer": 0.0033320292612514933
    },
    "answers_customer": 0.6017677187919617
  }
]

改成你自己的

• 在 RULES 里加一行就能新增规则。所有规则在同一次请求中运行，政策再长也几乎不增加耗时。
• 每份被拦截的草稿都连同 broken 一起记录下来；它是改进提示词的训练信号。