BanditDB

BanditDB Python SDK

BanditDB 的官方 Python 客户端和模型上下文协议（MCP）服务，适用于 BanditDB——这款以 Rust 编写、超快速且无锁的上下文老虎机数据库。

BanditDB 将强化学习（LinUCB、Thompson Sampling）中的复杂线性代数封装在极简 API 之后。构建实时个性化工具、动态 A/B 测试，并为 LLM Agent 提供数学上严谨的持久记忆。

mcp-name: com.banditdb/mcp

安装

pip install banditdb-python

需要运行 BanditDB Rust 服务（默认地址：http://localhost:8080）。

---

1. 标准 SDK 用法

客户端支持自动连接池、指数退避重试和严格超时。

from banditdb import Client, BanditDBError

# Connect to the BanditDB server.
# Pass api_key if BANDITDB_API_KEY is set on the server.
db = Client(
    url="http://localhost:8080",
    timeout=2.0,
    api_key="your-secret-key",   # omit if server runs without auth
)

try:
    # 1. Create a campaign (run once at startup)
    # algorithm defaults to "linucb"; use "thompson_sampling" for Bayesian exploration
    db.create_campaign(
        campaign_id="checkout_upsell",
        arms=["offer_discount", "offer_free_shipping"],
        features=["is_mobile", "cart_value", "is_returning_user"],
    )
    # or: db.create_campaign(..., algorithm="thompson_sampling")

    # 2. A user arrives — ask the database what to show them.
    # Scale each value to 0–1; the SDK puts them in order, adds a constant 1
    # and normalises the vector.
    arm_id, interaction_id = db.predict("checkout_upsell", {
        "is_mobile": 1.0,
        "cart_value": min(cart_total / 200, 1.0),
        "is_returning_user": 0.0,
    })
    print(f"Showing: {arm_id}")  # e.g., "offer_free_shipping"

    # 3. The user clicked — send the reward
    db.reward(interaction_id, reward=1.0)

except BanditDBError as e:
    print(f"Database error: {e}")

命名特征或向量。 使用 features=[...] 时，活动会将特征顺序存储在其元数据中，predict、batch_predict 和 interact 会接受以名称为键的字典。缺少或拼写错误的名称会在发送请求前触发 ValueError。在 BanditDB 控制台中创建的活动也接受字典。每个活动的特征顺序会通过 campaign_info() 读取一次，并缓存在客户端上。 若要自行管理向量，请改用 feature_dim，并传入列表——normalize_context() 会将其缩放为单位长度。

所有客户端方法

健康状态

方法	说明
health()	如果服务可访问且 WAL 写入器运行正常，则返回 True。
health_detail()	返回完整的健康状态字典，包括各活动的 entropy 和 status（"ok" / "warning" / "critical"）。

活动

方法	说明
create_campaign(campaign_id, arms, feature_dim=None, alpha=1.0, algorithm="linucb", metadata=None, decay_half_life_hours=None, pacing=None, features=None)	注册一个新活动。传入 features（按顺序排列的名称）即可使用字典进行预测，或传入 feature_dim 以发送向量。algorithm 接受 "linucb"、"thompson_sampling"、NeuralLinUCBConfig、NeuralTSConfig 或 ProgressiveConfig。metadata 是任意 JSON 字典（≤ 64 KB）。pacing 接受 PacingConfig — 参见预算约束。
list_campaigns()	返回所有活动（活跃和已归档）的列表，包括 alpha、arm_count 和 algorithm。
campaign_info(campaign_id)	返回各臂的完整状态：theta、theta_norm、预测计数和奖励计数。若未找到，则抛出 APIError（404）。
report(campaign_id)	业务层面的收敛报告。converged=True 表示某个臂以 95% CI 达到统计显著的领先优势 — 可以安全停止。converged=False 表示该臂领先，但置信区间仍有重叠。converged=None 表示数据仍不足（每个臂的奖励数少于 30）。
diagnostics(campaign_id)	运行诊断：各臂的 theta 范数、A_inv 不确定性界限、熵健康状况（selection_entropy、entropy_status、entropy_trend、likely_cause、suggested_action）、锦标赛流量和神经网络缓冲区大小。
archive_campaign(campaign_id)	软删除：暂停预测和奖励，但保留所有已学习的权重。可通过 restore_campaign() 恢复。
restore_campaign(campaign_id)	将已归档的活动恢复为活跃状态，并保留所有权重。
delete_campaign(campaign_id)	永久删除活动。若未找到，则返回 False。
pacing(campaign_id)	使用 pacing 的活动的预算状态：各资源的 budget、consumed、remaining、utilization、lambda（影子价格）、is_exhausted。若活动未启用 pacing，则返回 None。需要服务 2.3.0+。
add_arm(campaign_id, arm_id, group=None, warm_start=None, costs=None)	向运行中的活动添加一个臂（需要管理员密钥）。对于使用 pacing 的活动，costs 将资源名称映射为每次预测消耗的单位数。
set_arm_status(campaign_id, arm_id, status)	设置为 "active"、"paused"（不可选，可恢复）或 "retired"。该臂已学到的内容会保留。需要管理员密钥。

预测与奖励

方法	说明
predict(campaign_id, context, eligible_arms=None, exclude_arms=None)	context 是特征值字典（命名活动）或向量。返回 (arm_id, interaction_id)。将 interaction_id 传给 reward() 以完成闭环。可选筛选条件仅缩小本次调用的候选范围。
batch_predict(predictions)	在一次往返中，对最多 100 个活动/上下文组合进行预测。每项格式为：`{"campaign_id": str, "context": dict
reward(interaction_id, reward)	记录结果。reward 必须在 [0.0, 1.0] 范围内。如果该交互已获得奖励或已过期（默认 TTL：24 h），则抛出 APIError。

数据与导出

方法	说明
checkpoint()	刷新 WAL、创建模型快照、写入 Parquet 分片、运行神经网络重新训练和锦标赛评估，并轮换 WAL。返回摘要字符串。
export()	列出按活动分组的 Parquet 导出分片。返回 {export_dir, shards}。

预算限制

服务端 2.3.0+。为一场 campaign 指定一个或多个有预算限制的资源，以及各臂的成本；服务端会为每种资源学习一个影子价格，并引导支出在整个周期内持续有效，只在收益提升值得时选择高成本臂。

from banditdb import Client, PacingConfig, ResourceConstraint

db.create_campaign(
    "promos", ["none", "discount_10", "discount_20"], feature_dim=4,
    pacing=PacingConfig(resources=[
        ResourceConstraint(
            name="promo_budget",
            budget=5_000.0,      # units available over the horizon
            horizon=100_000,     # decisions the budget should last
            arm_costs={"discount_10": 1.0, "discount_20": 2.0},  # "none" costs 0
        ),
    ]),
)

# Arms added later declare their own costs
db.add_arm("promos", "discount_30", costs={"promo_budget": 3.0})

for r in db.pacing("promos")["resources"]:
    print(f"{r['name']}: {r['utilization']:.0%} spent, lambda={r['lambda']:.3f}")

值得了解：

• 预算在 predict 时消耗，而不是在 reward 时消耗。 即使你没有采用某项推荐，它仍会消耗预算。
• 预算不会重置，节奏控制也不可更改。 预算一旦花出，高成本臂就会一直无法选择，除非重新创建 campaign。
• arm_costs 的键必须是 campaign 中的臂。服务端会将拼写错误的臂静默视为免费；SDK 则会引发 ValueError。
• 对启用了节奏控制的 campaign 进行预测时，如果服务端的 WAL 已饱和，可能会返回 503。
• 将服务端降级到 2.3.0 以下会静默丢弃节奏控制。

---

2. AI「蜂巢思维」（Model Context Protocol）

标准 LLM Agent 是无状态的——如果它们把任务路由给了错误的模型并失败，明天还会重蹈覆辙。BanditDB 内置的 MCP 服务为整个 Agent 集群提供共享的持久化记忆。

启动 MCP 服务

# Set environment variables before starting
export BANDITDB_URL=http://localhost:8080
export BANDITDB_API_KEY=your-secret-key   # omit if server runs without auth

banditdb-mcp

连接到 Claude Desktop

将以下内容添加到你的 Claude 配置文件中：

• Mac：~/Library/Application Support/Claude/claude_desktop_config.json
• Windows：%APPDATA%\Claude\claude_desktop_config.json

{
  "mcpServers": {
    "banditdb": {
      "command": "banditdb-mcp",
      "args": [],
      "env": {
        "BANDITDB_URL": "http://localhost:8080",
        "BANDITDB_API_KEY": "your-secret-key"
      }
    }
  }
}

现在，Agent 集群拥有十二种工具：

工具	功能
create_campaign	创建新的决策 campaign。接受 algorithm（"linucb" 或 "thompson_sampling"）、alpha，以及可选的 budgets，用于限制有成本的操作。使用 Thompson Sampling 可自然地进行贝叶斯探索，无需调参。
list_campaigns	列出所有活跃的 campaign（显示 algorithm 和 alpha）——在调用 get_intuition 前检查已有 campaign 时很有用。
campaign_diagnostics	查看各臂的学习状态：theta_norm、预测次数、奖励率和熵健康度。当 campaign 看起来没有学习，或某个臂占据主导地位时使用。
campaign_report	业务级收敛报告。告知 campaign 是否已在统计意义上收敛，以及哪个臂胜出并附带置信区间。
get_intuition	询问 BanditDB 针对给定上下文应选择哪个臂。可选的 exclude_arms / eligible_arms 仅用于缩小本次请求的选择范围。返回臂和需要保存的 interaction_id。
batch_get_intuition	通过一次往返获取多个 campaign 的决策。传入由 {campaign_id, context} 字典组成的列表。
record_outcome	报告所选操作成功（1.0）还是失败（0.0）。更新共享模型。
archive_campaign	软删除一个 campaign。暂停预测和奖励，但保留所有已学习的权重。
restore_campaign	将已归档的 campaign 恢复为活跃状态，并保留全部权重。
campaign_budget	查看每种资源的预算支出：已消耗、剩余、利用率、影子价格和耗尽情况。
add_arm	向正在运行的 campaign 添加一个操作，可选择从现有臂进行热启动并设置预算成本。
set_arm_status	暂停、退役或重新启用某个操作。该操作已学习的内容都会保留。

网络中任何 Agent 做出的每项决策，都会改善所有未来 Agent 的路由效果。

---

3. 数据科学与离线评估

BanditDB 会将每次预测和奖励记录到 Write-Ahead Log（WAL）中。调用 checkpoint() 会将已完成的预测→奖励配对编译为 Snappy 压缩的 Parquet 文件，每个 campaign 一个，供 Polars 或 Pandas 进行离线分析。

无论奖励延迟几小时才到达，都能确保每次预测出现在 Parquet 文件中：BanditDB 会在每次检查点生成时重新输出进行中的交互，因此延迟奖励总能在后续周期中被记录。

# Checkpoint: snapshot models, write Parquet, rotate the WAL.
# Call this on a schedule or after significant traffic.
summary = db.checkpoint()
print(summary)
# "Checkpoint written and WAL rotated: 2 campaigns, offset 4821 bytes,
#  150 interactions exported, 3 in-flight re-emitted"

# List which Parquet files are available
print(db.export())
# 'Parquet files in /data/exports: ["llm_routing.parquet"]'

# Load directly from the mounted volume into Polars.
# Flat schema: interaction_id | arm_id | reward | predicted_at | rewarded_at | propensity | feature_0 | ...
import polars as pl
df = pl.read_parquet("/data/exports/llm_routing.parquet")
print(df.head())
print(df.columns)

离线策略评估（OPE）

SDK 在 banditdb.eval 中提供了三个 OPE 估计器。它们回答的问题是：“如果采用另一种策略，在不进行实时实验的情况下，我的平均奖励会是多少？”

安装评估依赖项：

pip install "banditdb-python[eval]"

估计器	函数	工作原理	适用场景
Replay	replay(df)	以 (1/K) / propensity 的概率接受每次交互（Li 等，2010）。这是均匀随机策略的无偏样本。	用于基线合理性检查。覆盖率较低是预期情况——约有 1/K 的交互会被使用。
IPS / SNIPS	ips(df, clip=10.0)	使用所有交互，并赋予重要性权重 (1/K) / propensity。通过自归一化降低方差。权重裁剪（默认 10×）用于控制偏差与方差之间的权衡。	主要估计器。数据充足但希望获得完整覆盖时使用。
双重稳健	doubly_robust(df, clip=10.0)	拟合线性奖励模型，然后对残差应用 IPS 校正。只要奖励模型或倾向概率其中之一正确，估计结果就是一致的。	统计效率最高。比较多个策略或扫描 alpha 时使用。

所有三个估计器均：

• 接受从 BanditDB Parquet 导出文件加载的 Polars 或 pandas DataFrame
• 将均匀随机策略作为目标策略进行评估（要超越的无偏基线）
• 对 Thompson Sampling 活动抛出 ValueError（倾向概率列为 null——TS 不会记录倾向概率）
• 返回包含 estimate、std_error、n_used、n_total 和 method 的 OPEResult

import polars as pl
from banditdb.eval import replay, ips, doubly_robust

df = pl.read_parquet("/data/exports/llm_routing.parquet")

# How much reward would a uniform random policy have earned?
print(replay(df))
# OPEResult(method='replay', estimate=0.4821, std_error=0.0312, coverage=22.1% [33/149])

print(ips(df))
# OPEResult(method='ips', estimate=0.5103, std_error=0.0187, coverage=100.0% [149/149])

print(doubly_robust(df))
# OPEResult(method='doubly_robust', estimate=0.5219, std_error=0.0141, coverage=100.0% [149/149])

# Compare against the observed reward of the logging policy:
print("Observed (logging policy):", df["reward"].mean())
# If observed >> estimate, the campaign has learned something real — it outperforms random.

实用方法：部署前离线扫描 alpha。 使用真实流量训练活动，保存检查点到 Parquet，然后通过 doubly_robust() 回放不同的 alpha 值，找出最佳探索水平——无需进行实时实验。

注意： OPE 需要 propensity 列，该列仅为 LinUCB 活动写入。Thompson Sampling 活动会记录 null 倾向概率，因为 TS 的臂选择是随机的，而倾向概率评分需要确定性的日志策略。

---

选择算法

BanditDB 支持五种算法，可在创建 campaign 时选择。

算法	algorithm 值	探索方式	适用场景
LinUCB	"linucb"（默认）	确定性 UCB 奖励：θ·x + α√(x·A⁻¹·x)	可预测、可调节。离线调整 alpha 以进行校准。
Linear Thompson Sampling	"thompson_sampling"	从 θ̃ ~ N(θ, α²·A⁻¹) 中采样，根据 θ̃·x 评分	贝叶斯后验——无需调整 alpha。并发用户会自动选择不同选项。
NeuralLinUCB	NeuralLinUCBConfig(...)	深度 MLP 嵌入 + 嵌入空间中的 LinUCB	非线性奖励函数。每获得 N 次奖励后重新训练 MLP。
NeuralTS	NeuralTSConfig(...)	与 NeuralLinUCB 使用相同的 MLP 嵌入；在嵌入空间中通过后验抽样而非 UCB 上界进行评分	适用于长期表现比早期遗憾更重要的非线性奖励。在 20K 次以上交互后收敛效果更好。
Progressive	ProgressiveConfig(...)	自动调优的锦标赛：并行运行基础算法和挑战算法，将流量转向胜者	无需配置的模型选择。自动选出最佳算法。

from banditdb import Client, NeuralLinUCBConfig, NeuralTSConfig, ProgressiveConfig

db = Client("http://localhost:8080")

# LinUCB (default)
db.create_campaign("routing", ["fast", "cheap"], feature_dim=4, alpha=1.5)

# Thompson Sampling — natural Bayesian exploration, alpha=1.0 is ideal
db.create_campaign("routing_ts", ["fast", "cheap"], feature_dim=4,
                   algorithm="thompson_sampling")

# NeuralLinUCB — learns a deep embedding of the context, then applies LinUCB
cfg = NeuralLinUCBConfig(
    context_dim=4,     # must match feature_dim
    embed_dim=32,      # arm matrix dimension (default 32)
    hidden_dim=128,    # MLP hidden layer width (default 128)
    retrain_every=200, # retrain the MLP every N cumulative rewards
)
db.create_campaign("routing_neural", ["fast", "cheap"], feature_dim=4, algorithm=cfg)

# NeuralTS — same parameters as NeuralLinUCB, posterior-sampling exploration
cfg = NeuralTSConfig(context_dim=4, embed_dim=32)
db.create_campaign("routing_nts", ["fast", "cheap"], feature_dim=4, algorithm=cfg)

# Progressive — runs LinUCB vs NeuralLinUCB, shifts traffic to whoever wins SNIPS checkpoints
cfg = ProgressiveConfig(
    base="linucb",
    challenger=NeuralLinUCBConfig(context_dim=4, embed_dim=32),
    min_obs=100,       # minimum buffer entries per arm before any traffic shift
    required_wins=3,   # consecutive checkpoint wins to earn one traffic step
    step_bps=1000,     # traffic delta per win run, in basis points (1000 = 10%)
)
db.create_campaign("routing_prog", ["fast", "cheap"], feature_dim=4, algorithm=cfg)

五种算法都使用相同的 predict → reward 循环。任一种神经网络配置也可用作 Progressive 的 base 或 challenger。

NeuralLinUCB 与 NeuralTS 的比较

	NeuralLinUCB	NeuralTS
探索方式	UCB 上界（确定性）	后验样本（随机），离散程度由 alpha 控制
早期遗憾	较低	较高
20K 次以上交互后的收敛速度	较慢	较快
最适用场景	生产环境安全性、避免早期代价高昂的错误	最大化长期表现
论文	Xu、Wen、Zhao 和 Gu，ICLR 2022（arXiv:2012.01780）	Zhang、Zhou、Li 和 Gu，ICLR 2021（arXiv:2010.00827）

两种神经网络算法都需要使用 neural 功能构建服务：检查 "neural" in db.server_info()["features"]。

---

错误处理

异常	引发时机
BanditDBError	基础异常——捕获此异常即可处理所有 SDK 错误。
ConnectionError	服务离线或无法访问。
TimeoutError	请求超出配置的超时时间。
APIError	服务返回错误（例如，未找到 campaign、未授权）。

---

许可证

Apache-2.0 — 版权所有 (C) 2026 Simeon Lukov 和 Dynamic Pricing Ltd. 详情请参阅主仓库。