从 pip install 到生产部署：AI 自愈 Agent 10 分钟上线指南

本文是一份实操指南。目标：从零开始，将一个普通的 OpenAI 调用改造成具有多 Provider 容灾、级联自愈、实时可观测性的生产级 AI Agent。

第一步：安装 SDK

pip install neuralbridge-sdk

预期结果：SDK 约 375 KB，唯一依赖是 httpx。安装时间 < 10 秒。

验证安装：

import neuralbridge
print(neuralbridge.__version__)
# 期望输出：5.2.11

第二步：一行代码替换现有 OpenAI 调用

改造前——裸调用（无自愈能力）：

import openai
client = openai.OpenAI(api_key="sk-...")
resp = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}]
)
print(resp.choices[0].message.content)

自愈率：0%。Provider 挂了 → 直接崩溃。

改造后——带自愈能力：

import neuralbridge as nb
client = nb.NeuralBridge(api_key="nb-...")
resp = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}]
)
print(resp.choices[0].message.content)

自愈引擎就绪。Provider 挂了 → 自动切换。

关键说明：只需要改 import 和 client 实例化。completions.create() 的接口保持和 OpenAI SDK 一致。

第三步：配置多 Provider

创建 neuralbridge.yml：

providers:
  - name: openai
    api_key: ${OPENAI_API_KEY}
    priority: 1
    models:
      - gpt-4o
      - gpt-4o-mini
  - name: anthropic
    api_key: ${ANTHROPIC_API_KEY}
    priority: 2
    models:
      - claude-sonnet-4-20250514
  - name: deepseek
    api_key: ${DEEPSEEK_API_KEY}
    priority: 3
    models:
      - deepseek-chat

self_healing:
  enabled: true
  quick_retry:
    max_attempts: 2
    backoff_base: 1s
  circuit_breaker:
    failure_threshold: 3
    recovery_timeout: 30s
  output_validation:
    schema_check: true

配置加载：

client = nb.NeuralBridge(
    api_key="nb-...",
    config_path="./neuralbridge.yml"
)

第四步：启动控制台（可选）

from neuralbridge.gateway import start_console
start_console(port=8765)

访问 http://localhost:8765 即可看到实时监控面板：

所有 API 调用的延迟分布（P50/P95/P99）
自愈事件的时间线和详情
Provider 的在线状态和健康评分
级联恢复的触发层级分布

第五步：生产配置调优

超时配置

不同场景推荐不同的超时参数：

timeout:
  connect: 5s     # 连接超时（5s 足够判断网络可用性）
  read: 30s       # 读取超时（长文本需要更长时间）
  total: 35s      # 总超时（connect + read 之和）

Provider 权重

如果你更倾向于用某些 Provider：

providers:
  - name: openai
    priority: 1
    weight: 50    # 50% 流量
  - name: deepseek
    priority: 2
    weight: 30    # 30% 流量
  - name: anthropic
    priority: 3
    weight: 20    # 20% 流量

告警阈值

alerting:
  p95_latency_ms: 5000    # P95 超过 5s 告警
  error_rate: 0.05        # 错误率超过 5% 告警
  drift_detected: true    # 检测到漂移时告警

生产环境清单

上线前检查：

[ ] 至少配置了 3 个不同 Provider
[ ] Provider 的 API Key 都通过环境变量注入，不硬编码
[ ] 超时参数已按场景调优（非默认值）
[ ] 输出完整性验证已启用（至少 Schema 校验）
[ ] 检查点持久化已配置存储后端
[ ] 控制台地址不对外暴露（仅内网访问）
[ ] 降级后的输出有明确标注（告知用户当前是非主模型响应）

实时验证

用以下脚本验证自愈功能按预期工作：

# test_healing.py
import neuralbridge as nb

client = nb.NeuralBridge(api_key="nb-...")

# 测试正常调用
resp = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Return 'OK'"}]
)
assert resp.choices[0].message.content == "OK"
print("✅ 正常调用通过")

# 测试故障注入（模拟 Provider 不可用）
resp = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Return 'OK'"}],
    extra_headers={"X-NB-Inject-Fault": "500"}
)
assert resp.choices[0].message.content == "OK"  # 自愈后仍应返回
print("✅ 故障注入自愈通过")

NeuralBridge SDK v5.2.11 兼容 Python 3.10–3.12，支持 OpenAI SDK 接口，提供进程内 MAPE-K 级联自愈 + 输出完整性验证 + 实时可观测性。pip install neuralbridge-sdk 即可开始。

推荐订阅源

DEV Community