Ai-Reign/Rein-Ai

# Rein **自主 AI 代理的运行时紧急开关。** Rein 拦截并控制你的代理执行的每一个动作，根据实时表现为其打分，并在事态失控前停止系统。 ``` pip install rein-ai ``` ``` from rein_ai import Rein, ReinConfig brain = Rein(cfg=ReinConfig.from_env()); await brain.start() @brain.governed(source="llm_agent") async def send_email(to, body): ... ``` 只需一个装饰器。与框架无关 —— 支持 LangChain、LlamaIndex、OpenAI Agents SDK 或你自己的循环。 **适用场景：**你的代理可能会在四分钟内烧掉 200 美元的 API 调用费，向错误的列表发送 80 封电子邮件，或者 在数据流中断后继续交易 —— 而内容护栏无法捕捉到其中任何一种情况，因为输出结果是“有效的”。Rein 监控的是*结果*，而不是文本。 最初提取自一个生产环境的 Kalshi 交易机器人，该机器人在第一周就拦截了 12 笔失控交易。[起源故事。](ORIGIN.md) [](https://pypi.org/project/rein-ai/) [](https://www.python.org/) [](LICENSE) [](#status) ## 问题所在 现有的 AI 安全工具主要验证**内容**。护栏库会检查 LLM 的输入或输出是否包含 PII（个人身份信息）、脏话或提示注入字符串。这是必要的，但还不够。一旦代理真正开始*执行动作*（下单交易、发送电子邮件、调用付费 API、发布到生产环境），内容验证就太迟了。你需要一个**运行时调控器**，能够根据观察到的结果，在运行过程中切断行为异常的代理，而不仅仅是检查文本。 Rein 就是这个调控器。 ## Rein 是什么 一个 Python 库，它与任何自主系统串联，并控制每一个动作。它能： 1. **根据实时信号输入，对当前状态进行分类**（正常 / 压力 / 冲击）。 2. **使用贝叶斯衰减为每个“动作源 × 动作类型”打分。**陈旧的表现会被折扣，近期的表现会产生复利效应。 3. **在出现回撤、错误率、过期状态、速率限制风暴或异常检测触发阈值时停止系统。** 4. **写入防篡改的审计日志**（JSONL + 密码学链条），记录每一个决策，用于合规和事后复盘。 其他治理库所没有的三个特性： - **自然语言策略编译器。** 用英语编写规则，即可获得可执行的配置。 - **内置对抗性红队模拟器。** 在你发布*之前*，先攻击你自己的策略。 - **状态切片打分。** 一种在平稳市场中有效但在冲击中失效的策略，会被视为两种不同的策略。 ## Rein 与替代方案的对比 | | Rein | Guardrails AI | NeMo Guardrails | 自定义代码 | |---|---|---|---|---| | 内容验证（PII、脏话） | — | ✅ | ✅ | DIY | | 运行时动作治理 | ✅ | — | — | DIY | | 状态感知决策 | ✅ | — | — | DIY | | 自然语言策略 | ✅ | 部分 | ✅ | — | | 对抗性红队模拟器 | ✅ | — | — | — | | 防篡改审计日志 | ✅ | — | — | DIY | | 与框架无关 | ✅ | ✅ | ✅ | — | Rein 和内容护栏库是互补的：使用 Guardrails/NeMo 来验证 LLM *说什么*，使用 Rein 来治理代理*做什么*。 ## 60 秒快速入门 ``` pip install rein-ai ``` ``` import asyncio, time from rein_ai import Rein, ReinConfig async def main(): brain = Rein(cfg=ReinConfig.from_env()) await brain.start() # Gate every action decision = brain.gate(source="llm_agent", series="send_email") if not decision.allowed: print(f"blocked: {decision.reason}") return # ...execute the action... # Report outcome so Rein can learn await brain.record_fill( source="llm_agent", series="send_email", ticker="msg-123", filled=True, slippage_cents=0.0, attempt_at=time.time(), ) await brain.shutdown() asyncio.run(main()) ``` 或者使用装饰器，一行代码即可集成： ``` @brain.governed(source="llm_agent") async def send_email(to, body): ... ``` ## 其他产品没有的两个特性 ### 1. 自然语言策略编译器 停止编写 YAML 吧。用英语描述你的策略： ``` from rein_ai import compile_policy policy = compile_policy([ "Cap each caller at 8 requests per second with bursts of 16", "Halt the portfolio when losses exceed 5 percent", "Alert when deny rate exceeds 70 percent over a 60 second window", "Detect runaway callers at 5x baseline", ]) brain = policy.build_brain(persist_dir="./state") ``` 每一行都会被解析为结构化的执行规则。在部署前检查编译后的输出： ``` for rule in policy.rules: print(rule.rule_type, rule.params) ``` ### 2. 对抗性红队模拟器 在你信任你的策略之前，先攻击它： ``` from rein_ai import run_red_team report = await run_red_team(brain) print(report.render()) # 红队报告 — 拦截率：100% # [BLOCKED] runaway_loop 在 iter 16 # [BLOCKED] deny_storm 在 iter 9 # [BLOCKED] enumeration 在 iter 17 # [BLOCKED] portfolio_drain 在 iter 0 # [BLOCKED] cost_bomb 在 iter 0 ``` 开箱即用的五个已注册攻击类（失控 run away、拒绝风暴 deny-storm、枚举 enumeration、投资组合耗尽 portfolio-drain、成本炸弹 cost-bomb）。发布你已验证过能捕获明显漏洞的策略。 完整可运行演示：`examples/policy_and_redteam/demo.py`。 ## 工作原理 ``` ┌─────────────────────────────────────────────────────────┐ │ Your Agent / App │ │ (LLM tool call, trade, email, scraper, RPA step…) │ └───────────────────────────┬─────────────────────────────┘ │ brain.gate(source, series) ▼ ┌─────────────────────────────────────────────────────────┐ │ Rein │ │ │ │ ┌─────────────┐ ┌─────────────┐ ┌────────────────┐ │ │ │ Regime │ │ Strategy │ │ Rate │ │ │ │ Detector │→ │ Scorer │ │ Limiter │ │ │ │ normal/ │ │ Bayesian, │ │ token-bucket │ │ │ │ stress/ │ │ time-decay │ │ per caller │ │ │ │ shock │ │ │ │ │ │ │ └─────────────┘ └─────────────┘ └────────────────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌───────────────────────────────────────────────────┐ │ │ │ Decision Engine │ │ │ │ GREEN (allow) · YELLOW (shadow) · RED (block) │ │ │ └───────────────────────────┬───────────────────────┘ │ │ │ │ │ ┌─────────────┐ ┌──────────▼──────────┐ ┌──────────┐ │ │ │ Circuit │ │ Anomaly │ │ Audit │ │ │ │ Breaker │ │ Detector │ │ Log │ │ │ │ drawdown/ │ │ deny-storm, │ │ JSONL │ │ │ │ error-rate/ │ │ enumeration, │ │ hash- │ │ │ │ staleness │ │ runaway callers │ │ linked │ │ │ └─────────────┘ └─────────────────────┘ └──────────┘ │ └───────────────────────────┬─────────────────────────────┘ │ AllowDecision(allowed, reason) ▼ Execute action / Halt / Log ``` | 子系统 | 职责 | |---|---| | `RegimeDetector` | 根据实时信号对世界状态进行分类（正常 / 压力 / 冲击） | | `StrategyScorer` | 基于每个 (source, series) 并带有时间衰减的贝叶斯打分；学习*当前*有效的方法 | | `CircuitBreaker` | 在达到回撤、错误率或过期阈值时停止系统 | | `RateLimiter` | 带有突发处理的按调用方进行的令牌桶限流 | | `AnomalyDetector` | 拒绝风暴、枚举和失控调用方检测 | | `AuditLog` | 防篡改的 JSONL 链条 —— 每个决策都带有时间戳并通过哈希链接 | ## 集成 - **Anthropic Claude** — 嵌入式 `GovernedToolRunner` 自动治理所有工具使用 from rein_ai.adapters.anthropic import GovernedToolRunner runner = GovernedToolRunner(brain=brain, source="claude_agent", impls={...}) result = await runner.execute(tool_use_block) - **LangChain** — 适用于任何 `BaseTool` 的治理包装器 - **FastAPI 管理路由** — 提供 `/rein/state`、`/rein/halt`、`/rein/resume`、`/rein/audit` 端点（可选附加项：`pip install rein-ai[api]`） - **与框架无关** — `@brain.governed()` 装饰器适用于任何 `async` 函数 请参阅 `examples/llm_agent_governor/` 了解非交易用例。 ## CLI ``` rein compile --policy policy.txt # show what a policy expands to rein redteam --policy policy.txt # run adversarial simulator rein attacks # list registered attack scenarios ``` ## 配置 所有阈值都可通过环境变量覆盖。默认前缀为 `REIN_`；对于多租户应用，可将自定义前缀传递给 `ReinConfig.from_env(prefix="MYAPP_REIN_")`。 关键变量： | 变量 | 默认值 | 用途 | |---|---|---| | `REIN_ENABLED` | `true` | 主开关 | | `REIN_SHADOW` | `true` | 仅观察不阻止（安全的默认设置） | | `REIN_EDGE_RED_P` | `0.85` | 触发 RED 状态的边界轴概率阈值 | | `REIN_PORTFOLIO_FLOOR_PCT` | `-0.05` | 导致投资组合停止的回撤百分比 | | `REIN_REGIME_TICK_SECONDS` | `30.0` | 状态检测器的轮询频率 | 完整列表请见 `src/rein_ai/config.py`。 ## 状态 - **135 项测试**，全部通过 - **生产环境运行中**（Predbot 交易机器人，自 2026-04-16 起） - **低开销** — `gate()` 平均耗时 **1.7 μs**，p99 耗时 **2.7 μs**，在单核上可达 **500k+ 次调用/秒**（参见 `scripts/bench.py`；2021 MacBook Pro，Python 3.14） - 建议在强制执行前使用**影子模式协议** — 详见 `docs/shadow-protocol.md` ## Rein-AI Pro 专为通过其代理运行真实资金、真实用户或面临真实合规要求的团队打造的生产层级 —— 包含扩展的攻击库、校准好的检测器预设、托管云服务以及商业 AGPL 豁免。详情、定价及联系方式：[PRO.md](PRO.md)。 ## 许可证 **双重许可：** - **AGPL-3.0**（默认 — 参见 [`LICENSE`](LICENSE))。免费用于开源、研究和自托管使用。如果你将 Rein 作为网络服务的一部分运行，你的服务也必须在 AGPL-3.0 下发布。 - **商业许可证**（参见 [`COMMERCIAL-LICENSE.md`](COMMERCIAL-LICENSE.md))，用于无 Copyleft 义务的专有 / SaaS 使用。请联系 **licensing@reinai.io** 获取报价。 贡献者：参见 [`CLA.md`](CLA.md)。 由 **John N.W. Hampton Jr** () 创建并维护。 版权所有 © 2026 John N.W. Hampton Jr.

标签：AGPL, AI Agent框架, AI安全, AI护栏, Async/Await, Chat Copilot, Kill-Switch, LangChain, LlamaIndex, OpenAI, Python, 交易机器人, 内存规避, 商业授权, 大模型安全, 异常检测, 异步编程, 提示词注入, 无后门, 框架无关, 紧急停止开关, 红队模拟, 自主智能体, 自动化风控, 自然语言策略编译, 行为治理, 计算机取证, 贝叶斯调节器, 轻量级, 运行时控制, 逆向工具, 风控系统