SteerPlane——面向 AI 智能体的确定性运行时护栏

vijaym2k6

/

SteerPlane

Public

Notifications You must be signed in to change notification settings

Fork 1

Star 4

main

Branches Tags

Go to file

Code Open more actions menu

Latest commit

History

80 Commits

80 Commits

Folders and files

Name Name Last commit message

Last commit date

.github/ workflows

.github/ workflows

api

api

assets

assets

dashboard

dashboard

docs

docs

examples

examples

scripts

scripts

sdk-ts

sdk-ts

sdk

sdk

.env.example

.env.example

.gitignore

.gitignore

CHANGELOG.md

CHANGELOG.md

CONTRIBUTING.md

CONTRIBUTING.md

LICENSE

LICENSE

PATENTS.md

PATENTS.md

README.md

README.md

docker-compose.yml

docker-compose.yml

ruff.toml

ruff.toml

View all files

Repository files navigation

The safety layer for the AGI era. Runtime guardrails for autonomous AI agents.

Cost limits · Loop detection · Dual enforcement (Kill/Alert) · Streaming gateway · Policy engine · Human-in-the-loop · CLI · Docker · 4 framework integrations

pip install steerplane · npm install steerplane

🌐 steerplane.com

The Problem

AI agents can call APIs, execute code, browse the web, and make real-world decisions. Without guardrails:

🔄 A single misconfigured agent can enter an infinite loop

💸 A runaway agent can burn through $10,000+ in API credits overnight

💀 Agents can take destructive actions with zero visibility

SteerPlane fixes this with one line of code.

How It Works

from steerplane import guard

@ guard ( agent_name = "support_bot" , max_cost_usd = 10.00 , max_steps = 50 , denied_actions = [ "delete_" , "sudo_" ], enforcement = "alert" , alert_threshold = 0.8 , alert_timeout_sec = 1800 , ) def run_agent (): # Your agent runs normally. # SteerPlane silently monitors every step. # Financial/runtime limits can pause for human approval. # Loops and policy violations still terminate immediately. agent . run ()

🚀 SteerPlane | Run Started Run ID: a3f8d2b1-... Agent: support_bot Limits: $10.00 cost / 50 steps ───────────────────────────────────────────── ✅ Step 1: query_database | 380 tokens | $0.0020 | 45ms ✅ Step 2: call_llm_analyze | 1240 tokens | $0.0080 | 320ms ✅ Step 3: search_knowledge | 560 tokens | $0.0030 | 89ms ✅ Step 4: generate_response | 1800 tokens | $0.0120 | 450ms ✅ Step 5: send_notification | 120 tokens | $0.0010 | 200ms ─────────────────────────────────────────────

✅ SteerPlane | Run COMPLETED Steps: 5 Cost: $0.0260 Tokens: 4,100 Duration: 1.1s

Features

Feature What It Does

🔄 Loop Detection O(W²) sliding-window algorithm catches single-action, alternating, and multi-step repeating patterns in sub-millisecond time — no LLM calls

💰 Cost Ceiling Per-run (SDK) / per-session (gateway) USD limits, checked after each step so overshoot is bounded to a single step. Built-in pricing for 25+ models across OpenAI, Anthropic, Google, Meta, and Mistral

🌊 Streaming Gateway Real-time SSE chunk forwarding with mid-stream cost kill — if the budget is exceeded during a stream, SteerPlane injects a termination event and cuts the connection

🛡️ Policy Engine Allow/deny lists with glob patterns and sliding-window rate limits

🌐 Gateway Proxy OpenAI-compatible API proxy — change only base_url for zero-code enforcement. The agent passes its provider key through the gateway, which forces all traffic through enforcement before forwarding upstream

🖥️ Real-Time Dashboard Next.js dashboard with auto-refresh, animated timelines, cost breakdowns, and policy management

🔧 CLI Tool steerplane runs list , steerplane status , steerplane keys create — manage everything from your terminal

📄 Config File .steerplane.yml auto-discovery — set defaults without hardcoding limits in source code

🔗 4 Framework Integrations LangChain, OpenAI Agents SDK, CrewAI, AutoGen — zero-config drop-in handlers

🐳 Docker Compose One command brings up API + Dashboard + PostgreSQL

🔌 Graceful Degradation If the API goes down, the SDK enforces all limits locally. Agents are never unprotected

🧪 CI/CD GitHub Actions pipeline — lint, test, build Docker on every push

Hosted/Enterprise tier: alert-mode human-approval workflows (with email/webhook notifications), server-side provider-key vaulting, and Redis-backed multi-worker gateway state are available on SteerPlane's hosted plan. The open-source SDK still exposes the enforcement="alert" client options so it works out of the box against a hosted or enterprise deployment — self-hosting only the free tier here runs kill-mode enforcement.

Quick Start

Option A: Docker (Recommended)

git clone https://github.com/vijaym2k6/SteerPlane.git cd SteerPlane cp .env.example .env docker compose up -d

API at localhost:8000 · Dashboard at localhost:3000 · PostgreSQL auto-configured.

Option B: Manual Setup

# Install the SDK pip install steerplane

# Start the API cd api && pip install -r requirements.txt uvicorn app.main:app --reload --port 8000

# Start the Dashboard cd dashboard && npm install && npm run dev

Option C: CLI

pip install steerplane[cli] steerplane status # Check API health steerplane runs list # List recent runs steerplane keys create -n prod # Generate API key

Run the Demo Agent

python examples/simple_agent/agent_example.py

Open localhost:3000 → See your agent run in real time.

SDK Reference

Python — Decorator API

from steerplane import guard

@ guard ( agent_name = "my_bot" , max_cost_usd = 10.00 , max_steps = 50 , max_runtime_sec = 300 , enforcement = "alert" , alert_threshold = 0.8 , denied_actions = [ "delete_" , "sudo_" ], allowed_actions = [ "search_" , "read_" , "generate_" ], rate_limits = [{ "pattern" : "call_llm" , "max_count" : 20 , "window_seconds" : 60 }], ) def run_my_agent (): agent . run ()

Python — Context Manager API

from steerplane import SteerPlane

sp = SteerPlane ( agent_id = "my_bot" )

with sp . run ( max_cost_usd = 10.0 , max_steps = 50 ) as run : run . log_step ( "query_db" , tokens = 380 , cost = 0.002 , latency_ms = 45 ) run . log_step ( "generate" , tokens = 1240 , cost = 0.008 , latency_ms = 320 )

TypeScript

import { guard , GuardOptions } from 'steerplane' ;

const protectedAgent = guard ( async ( run ) => { await run . logStep ( { action : 'query_db' , tokens : 380 , cost : 0.002 } ) ; await run . logStep ( { action : 'generate' , tokens : 1240 , cost : 0.008 } ) ; return 'done' ; } , { agentName : 'support_bot' , maxCostUsd : 10.0 , maxSteps : 50 , policy : { deniedActions : [ 'delete_' , 'sudo_' ] , } , } ) ;

const result = await protectedAgent ( ) ;

Exception Handling

from steerplane . exceptions import ( CostLimitExceeded , LoopDetectedError , StepLimitExceeded , PolicyViolationError , )

@ guard ( max_cost_usd = 5 , denied_actions = [ "delete_*" ]) def run_agent (): try : agent . run () except CostLimitExceeded as e : print ( f"Budget exceeded: { e } " ) except LoopDetectedError as e : print ( f"Loop detected: { e } " ) except StepLimitExceeded as e : print ( f"Step limit hit: { e } " ) except PolicyViolationError as e : print ( f"Policy violation: { e . action } blocked by { e . rule } " )

Config File (.steerplane.yml)

Set defaults in a .steerplane.yml at your project root instead of hardcoding limits:

api_url : http://localhost:8000 agent_name : my_bot

defaults : max_cost_usd : 25.0 max_steps : 100 max_runtime_sec : 1800 enforcement : alert loop_window_size : 10

policy : denied_actions : - " delete_ " - " drop_ " rate_limits : - pattern : " search_* " max_count : 10 window_seconds : 60

alerts : email : ops@company.com webhook_url : https://hooks.slack.com/... threshold : 0.8

Merge order: Explicit decorator params → .steerplane.yml → hardcoded defaults. The config file is auto-discovered by walking up from the current directory.

Framework Integrations

LangChain

from steerplane . integrations . langchain import SteerPlaneCallbackHandler

handler = SteerPlaneCallbackHandler ( agent_name = "research_bot" , max_cost_usd = 5.0 , max_steps = 30 , )

llm = ChatOpenAI ( model = "gpt-4o" , callbacks = [ handler ]) agent . run ( "Analyze this data" , callbacks = [ handler ]) handler . finish ()

OpenAI Agents SDK

from steerplane . integrations . openai_agents import SteerPlaneAgentHooks

hooks = SteerPlaneAgentHooks ( agent_name = "my_openai_agent" , max_cost_usd = 10.0 , max_steps = 100 , )

# Convenience wrapper result = await hooks . run ( agent , "Hello!" )

# Or manual lifecycle hooks . start () result = await Runner . run ( agent , "Hello!" ) hooks . finish ()

CrewAI

from steerplane . integrations . crewai import SteerPlaneCrewMonitor

monitor = SteerPlaneCrewMonitor ( agent_name = "my_crew" , max_cost_usd = 25.0 , max_steps = 200 , )

crew = Crew ( agents = [ researcher , writer ], tasks = [ research_task , write_task ], step_callback = monitor . step_callback , )

result = monitor . kickoff ( crew )

AutoGen

from steerplane . integrations . autogen import SteerPlaneAutoGenMonitor

monitor = SteerPlaneAutoGenMonitor ( agent_name = "my_autogen_group" , max_cost_usd = 15.0 , max_steps = 150 , )

result = monitor . initiate_chat ( user_proxy , assistant , message = "Hello!" )

Install only the integration you need:

pip install steerplane[langchain] # LangChain pip install steerplane[cli] # CLI tool pip install steerplane[yaml] # Config file support pip install steerplane[all] # Everything

Gateway Proxy (Zero-Code Mode)

For agents you can't modify, SteerPlane provides an OpenAI-compatible gateway proxy with real-time streaming and mid-stream cost enforcement :

from openai import OpenAI

client = OpenAI ( base_url = "http://localhost:8000/gateway/v1" , api_key = "sk_sp_your_steerplane_key" , # The real provider key is sent to the gateway, which forwards it upstream. default_headers = { "X-LLM-API-Key" : "sk-your-real-provider-key" }, )

# Streaming works — chunks forwarded in real-time for chunk in client . chat . completions . create ( model = "gpt-4o" , messages = [{ "role" : "user" , "content" : "Hello" }], stream = True , ): print ( chunk . choices [ 0 ]. delta . content , end = "" )

What the gateway enforces per request:

Policy rules (deny/allow/rate limits)

Session cost vs. ceiling (including mid-stream kill)

SHA-256 prompt-hash loop detection

Monthly budget tracking

Anthropic + OpenAI streaming support

Security model: The agent points its OpenAI client at the gateway and passes the real provider key in the X-LLM-API-Key header. The gateway authenticates the SteerPlane key, runs every request through enforcement (policy → cost → loop), and only then forwards it upstream with that provider key — so the agent can't reach the provider directly or bypass the guardrails.

Server-side provider-key vaulting (so the agent never sends X-LLM-API-Key ) is available on the hosted/enterprise plan.

CLI Tool

pip install steerplane[cli]

Command Description

steerplane status Check API server health

steerplane runs list List recent runs (filter by --status )

steerplane runs inspect Full run detail with step-by-step table

steerplane runs kill Force-terminate a live run

steerplane keys list List all API keys

steerplane keys create --name prod Generate a new API key

steerplane keys revoke Revoke a key

steerplane logs --tail Live polling of running agents

Policy Engine

The policy engine runs before any cost is incurred, enforcing rules in strict priority order:

Deny List → Allow List → Rate Limits

Rule Type How It Works

Deny list Glob patterns (e.g. delete_* ) — any match is blocked immediately

Allow list If set, action must match at least one pattern to proceed

Rate limits Sliding-window counters per pattern — blocks when count exceeds threshold

Available in Python and TypeScript SDKs, the dashboard UI, REST API, and .steerplane.yml config file.

Enforcement Mode

The self-hosted free tier runs kill mode : immediate, deterministic termination on any violation.

The SDK also exposes an enforcement="alert" option (pause → notify a

... (truncated at the site's per-page length limit; see the source link below for the full text.)