virbius-review-sdk

VirbiusAgent 官方审查 SDK（Python）——在调用方侧完成输入前置审查、输出后置审查与流式滑动窗口审查，全部经引擎 POST /v1/evaluate。

引擎侧契约见 PROTOCOL §2.6.5（role 副作用矩阵：input 喂会话风险/零工具副作用；output 零副作用并返回 disposal_action）。双审二选一：用了本 SDK 就不要再把同一内容走引擎 chat 代理或重复自审。

安装

pip install .                # 同步客户端（requests）
pip install '.[async]'       # + 异步客户端（httpx）

三步接入

from virbius_review import ReviewClient

client = ReviewClient(
    "http://virbius-engine:8082",
    fail_open=True,                 # 引擎不可达时放行（False 则抛 ReviewError）
    window_chars=200, context_chars=50,
    refusal_text="抱歉，该内容已被安全策略拦截。",
)

# ① 输入前置（调 LLM 之前；命中即不要发给模型）
v = client.review_input(user_msg, tenant_id="default",
                         session_id=sid, trace_id=tid, user_id=uid)
if v.blocked:
    return refusal()

answer = call_your_llm(messages)          # 任意 provider，调用方直连

# ② 输出后置（非流式；拿到完整回答）
v = client.review_output(answer, session_id=sid, trace_id=tid)
final = client.apply_disposal(answer, v)  # reject→拒答文案 / omit→空 / redact→脱敏 / warn→原文

流式输出审查

from virbius_review.adapters import from_openai_sse   # 或 from_ollama_ndjson

resp = requests.post(llm_url, json={...,"stream":True}, stream=True)
lines = (line.decode() for line in resp.iter_lines())
deltas = from_openai_sse(lines)

for ev in client.wrap_stream(deltas, session_id=sid, trace_id=tid):
    match ev:
        case Delta(text):  send(text)                   # 只含已过审内容（不含重叠）
        case Blocked(rc):  resp.close(); break          # 断上游流省 token，向用户发违规提示
        case Done:         finish()

窗口算法与引擎 StreamReviewWindow 同一套测试向量（schemas/review-window-vectors.json）钉死，永不漂移。

异步客户端

async with AsyncReviewClient("http://virbius-engine:8082") as client:
    v = await client.review_input_async(msg, session_id=sid, trace_id=tid)
    async for ev in client.wrap_stream_async(delta_aiter, session_id=sid, trace_id=tid):
        ...

配置（调用方侧）

行为要点

• 429 处理：命中引擎限流 → 恒抛 ReviewError（fail-closed），无放行开关——限流是人为节流而非故障，fail-open 会让恶意刷限流的调用方把后续内容全部未审放行。
• 401/403 处理：认证失败（缺凭证/签名无效/过期/被吊销）→ 默认 恒抛 ReviewError（fail-closed）；需要可用性兜底时用 auth_fail_open=True。接入认证推荐 license_jwt + auth_url（SDK 自动换取+刷新），或 token_provider。
• 自定义 transport 契约：429 必须返回引擎的 JSON body（{"error":"rate_limited",...}），401/403 必须返回 {"error":"unauthorized","reason":...}，都不要抛异常——抛出会与传输故障混同并落入 fail_open 策略。若需带认证，transport 内从 client.current_token() 取当前 Bearer 并自行附加 Authorization 头（默认 requests/httpx 传输已自动处理）。
• 契约严格化：role=output 响应缺 disposal_action 即抛 ReviewError（引擎与 SDK 同仓同版，不存在版本偏斜）；本地 fail-open 合成判定除外（推导为 warn/reject）。
• trace 贯通：trace_id/session_id/user_id 贯穿 input、每窗、output——审计/统计按回答聚合依赖它。
• 规则寻址：控制台里绑 tool_names:["chat_input"] 仅输入生效，["chat_completion"] 仅输出生效。

测试

PYTHONPATH=src python3 -m unittest discover -s tests