AgentScope 2.0 服务化部署：把智能体封装成 HTTP 接口

前十二篇都在造 Agent，这一篇讲怎么把它放上线。

一个能上线的服务，请求要过四道闸：鉴权 → 限流 → 护栏 → Agent。

代码纯标准库、离线零成本，FastAPI 放在可选运行时里。

实测结果：鉴权 3/3、限流 3 放 2 拒、护栏 3/3、端到端 6/6、门禁拦 bug 放修复。

一、环境

python 013_deploy_service.py

鉴权、限流、护栏、门禁全是纯规则，不需要任何 API Key。

FastAPI 是可选依赖，装了才会构建 app；没装也能跑完全部演示。

二、第一道闸：鉴权

API_KEYS = {
    "sk-mc-demo-001": {"user_id": "u_miniapp", "tier": "free"},
    "sk-mc-demo-002": {"user_id": "u_web", "tier": "pro"},
}

def authenticate(api_key):
    """API Key 白名单校验。返回 (ok, user_id, reason)。"""
    if not api_key:
        return False, None, "未提供 API Key"
    info = API_KEYS.get(api_key)
    if not info:
        return False, None, "API Key 无效"
    return True, info["user_id"], "ok"

三个细节。

缺 Key 和 Key 错要区分。 前者是 401 + “未提供”，后者是 401 + “无效”。这个区分对调试很重要——客户端没传 header 和服务端不认这把 Key，是两回事。

白名单查表用 dict，不用 list。 dict 是 O(1)，list 是 O(n)。Key 少的时候看不出差别，但这是习惯问题。

绝不把明文 Key 写进日志。 底稿里这句是加粗的。日志要记 sk-mc-demo-001 的话，只记前 6 位 + 后 2 位：sk-mc-***01。

生产环境这套要换成数据库或 KMS，但"不落明文"这条底线不变。

三、第二道闸：令牌桶限流

class TokenBucket:
    """令牌桶：容量 capacity 个令牌，每秒补充 refill_rate 个。"""

    def __init__(self, capacity=3, refill_rate=0.0):
        self.capacity = capacity
        self.refill_rate = refill_rate
        self.tokens = capacity
        self.last = time.monotonic()

    def allow(self):
        now = time.monotonic()
        elapsed = now - self.last
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last = now
        if self.tokens >= 1:
            self.tokens -= 1
            return True, ""
        return False, f"令牌桶已空（容量 {self.capacity}，补充 {self.refill_rate}/s)"

time.monotonic() 而不是 time.time()。前者是单调递增的，不受系统时钟调整影响；后者会被 NTP 校正往回拨，导致 elapsed 变成负数，令牌数暴涨。

这个坑很小但很致命——服务器时间同步一次，限流就失效一次。

参数怎么定

这是最常被问的问题：容量设多少、补充速率设多少。

容量 = 允许的突发量。
用户正常操作，多久点一次？如果 3 秒一次，但偶尔会连点，容量就设在"最大的合理连点次数"。

速率 = 长期平均 QPS 上限。
按单个用户的合理频率算，不是按系统总容量算。

举个例子。一个用户正常 3 秒问一句，偶尔连发 5 条：

bucket = TokenBucket(capacity=5, refill_rate=1/3)   # 每 3 秒补 1 个

容量 5 能容忍一次连发，速率 1/3 保证长期不会超过每分钟 20 次。

底稿的 demo 把 refill_rate 设成 0，纯粹为了演示"连发 5 次、3 放 2 拒"这个确定性的结果。生产里速率必须大于 0，否则桶空一次就永久拒绝。

为什么不用固定窗口

固定窗口（每分钟 100 次）有个边界漏洞：

用户在 0:59 发 100 次，1:01 再发 100 次。两秒内 200 次，但两个窗口各自都没超限。

令牌桶没有这个问题，因为补充是连续的，不存在"到点清零"那个跳变。

四、第三道闸：合规护栏

护栏做三件事，顺序有讲究：注入 → 敏感词 → PII。

INJECTION_PATTERNS = [
    r"忽略(上面|之前|前面)的?(所有)?(指令|提示|设定)",
    r"ignore (all )?(previous|above) instructions",
    r"你现在是.*(不受限制|没有限制)",
    r"(告诉我|输出|打印).{0,6}(系统提示|system prompt)",
]

def check_guard(text):
    """返回 (ok, reason, cleaned_text)。"""
    for p in INJECTION_PATTERNS:
        if re.search(p, text, re.I):
            return False, f"命中注入模式：{p}", text
    hits = [w for w in BANNED_WORDS if w in text]
    if hits:
        return False, f"命中违禁词：{hits}", text
    return True, "ok", mask_pii(text)

三个细节。

注入规则必须覆盖最直白的说法。

底稿里那句注释值得抄下来：必须覆盖"忽略指令"这种最直白的形式，不能只测带修饰词的变体。

因为攻击者不会用优雅的说法。你写 忽略之前的所有指令，他直接用 忽略指令，你就漏了。

违禁词拦截是拒绝，PII 是脱敏后放行。 这两者的处理完全不同：

def mask_pii(text):
    """手机号 + 邮箱打码，放行。"""
    text = re.sub(r"1[3-9]\d{9}", lambda m: m.group()[:3] + "****" + m.group()[-4:], text)
    text = re.sub(r"([\w.-]{2})[\w.-]*@([\w-]+\.\w+)", r"\1***@\2", text)
    return text

手机号出现在正常咨询里（“我的电话是 139…”），不能因为含 PII 就拒绝服务。正确做法是放行，但往日志和工单里写的时候打码。

把脱敏做成拦截，是常见的过度设计——用户会莫名其妙被拒，而且不知道为什么。

五、端到端处理链

def handle_chat_request(payload, auth, limiter, agent):
    # ① 鉴权
    ok, user_id, why = authenticate(payload.get("api_key"))
    if not ok:
        return {"ok": False, "code": 401, "error": "鉴权失败", "detail": why}

    # ② 限流（按 user_id 维度）
    ok, why = limiter.allow(user_id)
    if not ok:
        return {"ok": False, "code": 429, "error": "限流", "detail": why}

    # ③ 护栏
    ok, why, cleaned = check_guard(payload.get("text", ""))
    if not ok:
        return {"ok": False, "code": 403, "error": "护栏拦截", "detail": why}

    # ④ 转发给 Agent（不解析内部结构）
    result = agent(cleaned)
    return {"ok": True, "code": 200, **result}

最后一行 **result 是整个设计的核心：服务层不解析 Agent 的内部结构，原样展开。

这意味着 Agent 从规则替身换成真模型，服务层一行不用改。

六、为什么顺序不能乱

把四道闸按"单次成本"排一下：

闸门	单次成本	挡住什么
鉴权	一次 dict 查表（微秒级）	不认识的调用方
限流	一次浮点运算	超量的调用
护栏	几次正则匹配	恶意输入
Agent	一次模型调用（秒级、要钱）	——

这是按成本从低到高排的漏斗。

顺序反了会怎样：

Agent 在限流之前：被刷 1000 次，每次都跑模型。等你发现 QPS 异常，账单已经出来了。

护栏在 Agent 之后：注入已经进了模型，它可能照做了。护栏只能拦输出，拦不住已经发生的推理。而且模型可能把内容编进别的回答里。

鉴权在限流之后：不认识的调用也消耗限流额度，恶意 IP 把你的正常用户挤掉。

三个反例的共同点：顺序错了，闸门就白装了。

七、上线门禁

def gate(result, threshold=0.95):
    """达标率 ≥ threshold 才放行。"""
    if result["rate"] >= threshold:
        return True, f"✅ 放行上线（{result['rate']}）"
    return False, f"🚫 禁止上线（{result['rate']}，硬失败：{result['hard_fails']}）"

实测：

[bug 版]   达标率 0.75 · 🚫 禁止上线（硬失败：玻尿酸保湿霜多少钱 → faithfulness）
[修复版]   达标率 1.0  · ✅ 放行上线

这条规则的价值不在 0.95 这个数，在于它改变了决策的性质：

没有门禁，上线是讨论题——“你觉得行不行”“再测两天”“客户催了先上”。
有门禁，上线是计算题——跑一遍，达标上，不达标不上。

而且门禁消费的是评测数据。没有前几篇的评测体系，门禁就是一句空话，因为你没有可判定的"达标"。

八、真实输出

— ① 鉴权 —
  [正确 Key] → code=200 ✅  小程序端
  [错误 Key] → code=401 ✅  拒绝 API Key 无效
  [缺 Key]   → code=401 ✅  拒绝 未提供 API Key

— ② 限流（容量 3，连发 5 次）—
  连发 5 次 → 放行 3 次 / 拒绝 2 次 ✅

— ③ 合规护栏 —
  [注入]     🛡️ 拦截 @ injection — 命中注入模式：忽略指令
  [敏感词]   🛡️ 拦截 @ sensitive — 命中广告法违禁词：['根治', '七天见效']
  [正常含PII] ✅ 放行，脱敏后：139****5678（['手机号']）

— ④ 端到端 —
  [正常·导购] 200 ✅
  [正常·售后] 200 ✅
  [护栏·注入] 403 ✅
  [护栏·违禁词] 403 ✅
  [鉴权失败] 401 ✅
  [正常·含PII] 200 ✅

— ⑤ 上线门禁 —
  [bug 版] 0.75 → 🚫
  [修复版] 1.0  → ✅

— ⑥ FastAPI 运行时 —
  ✅ app 已构建：/v1/chat + /health

端到端 6 条都过。注意"正常·含PII"那条是 200 而不是 403——这就是"放行 + 脱敏"和"拦截"的区别。

九、日志：让它出问题的时候你能知道

这是部署里最容易被敷衍的一环，也是最难补的——因为日志是"写的时候不觉得有用，出事的时候才知道少写了什么"。

一行合格的请求日志长这样：

{
  "ts": "2026-10-28T17:30:00+08:00",
  "trace_id": "req_8f3a2b",
  "user_id": "u_miniapp",
  "gate": "guard",
  "result": "blocked",
  "reason": "命中注入模式",
  "elapsed_ms": {"auth": 0.1, "limit": 0.2, "guard": 1.3},
  "text_hash": "sha256:ab12..."
}

五个字段，每个都有明确用途。

trace_id 必须贯穿整条链路。

一个请求会过四道闸，可能产生四条日志。没有 trace_id，你只能按时间戳猜哪几条是同一次请求——而高并发下时间戳会撞。

生成方式很简单：uuid4().hex[:12]，在入口生成，一路往下传。

每个闸门都要记"过了还是拦了"，以及原因。

只记"403"没用。要记 gate: "guard", reason: "命中注入模式"。

因为排查的时候你想知道的是"它为什么被拦"，不是"它被拦了"。

耗时必须分段记。

elapsed_ms: {auth: 0.1, limit: 0.2, guard: 1.3} 比一个总耗时有用得多。

总耗时 8 秒的时候，你需要立刻知道这 8 秒花在哪个闸门——是护栏的正则写得太烂，还是 Agent 那边超时了。

原文不落日志。

这一条最容易犯。把用户输入原样存进日志，等于把手机号、地址、订单号存了两份。

正确做法是存 sha256 的前 16 位——需要比对"是不是同一句话"时够用，需要看内容时回到业务库查。

如果确实要看内容，先过一遍 mask_pii。

日志的验收标准

只有一条：线上出一个问题，你能不能只靠日志还原"这个请求发生了什么"。

具体说，要能回答四个问题：

1. 谁调的（user_id）
2. 过了哪几道闸，在哪一道被拦（gate + result）
3. 为什么被拦（reason）
4. 慢在哪（elapsed_ms）

四个都答不上来，日志就是没写。

十、并发：先让管道在确定的情况下跑对

这一节讲的是"先上线一个假的"这个做法的具体理由。

服务层接的 agent 是个可调用对象。真实场景里它是"模型 + 编排"，但你可以先塞一个固定返回的假 Agent：

def fake_agent(text):
    """假 Agent：行为完全可预测。"""
    if "价格" in text:
        return {"reply": "这款 268 元。", "role": "sales"}
    if "物流" in text:
        return {"reply": "已发货，单号 SF123。", "role": "aftersale"}
    return {"reply": "请说说您的肤质和需求。", "role": "sales"}

它不需要聪明，只需要行为可预测。

这样你能验证的全是管道层面的事：

• 20 个并发请求，限流是不是精确地放了 3 个
• 护栏拦下的时候，返回格式和 200 一样吗（客户端能不能统一处理）
• 四个闸门的耗时分别是多少
• 日志里的 trace_id 会不会串

这些和模型一点关系都没有，但每一项都会在真模型上线后暴露——那时候你分不清是模型的锅还是管道的锅。

引入真模型之后，不确定性一下多了三样：响应时间波动、输出格式不稳、偶发失败。这时候再去调管道，等于同时改两个变量。

先让管道在"确定"的情况下跑对，再引入"不确定"的模型。

十一、常见报错

现象	原因	处理
限流永久生效	refill_rate=0	生产必须 >0
限流时灵时不灵	用了 time.time()	换 time.monotonic()
正常用户被 403	PII 做成了拦截	改成脱敏后放行
注入被绕过	规则只覆盖了带修饰词的形式	补最直白的"忽略指令"
日志里出现完整 Key	没做截断	只记前后几位
换成真 Agent 服务层要改	服务层解析了 Agent 返回	原样 **result 展开
门禁永远通不过	阈值定太高或评测集太难	先跑一次看基线

十二、接口速查

类 / 函数	关键参数	作用
authenticate	api_key	白名单校验
TokenBucket	capacity、refill_rate	令牌桶限流
check_guard	text	注入 + 违禁词 + PII
mask_pii	text	手机号/邮箱打码
handle_chat_request	payload、auth、limiter、agent	全链路
gate	rate、threshold	上线门禁

十三、一句话总结

Demo 和产品的区别，不在 Agent 聪不聪明，在"能不能被安全、稳定、可验证地调用"。

四道闸的顺序不是风格问题，是成本问题——每一道都用最便宜的方式挡住最贵的浪费。

下一篇讲容器化，把服务装进镜像。