Standard Conformance Quickstart¶
Production-ready governance for interactive applications and user-facing systems.
Production Ready
Designed for: Production apps, interactive systems, user-facing deployments
Performance: Negotiable latency (50ms-1000ms based on risk)
Features: Governance contracts, all interventions, audit trails
What You Get¶
Standard conformance adds production-critical features:
| Feature | Minimal | Standard |
|---|---|---|
| Governance Contracts | [NO] | [YES] REQUIRED |
| Performance SLAs | [NO] Uniform 300ms | [YES] 50ms-1000ms negotiable |
| All six intervention types | [NO] OK/BLOCK only | [YES] Five primary levels: OK, Nudge, Escalate, Block, Halt, plus orthogonal Flag |
| Trust debt system | [NO] | [YES] |
| Tripwires | [NO] | [YES] All categories |
| ReflectionDB | [NO] In-memory | [YES] Persistent (1 year) |
| Framework integration | [NO] | [YES] LangChain, AutoGPT, etc. |
Time to implement: 1-2 days
Step 1: Install¶
Step 2: Create Governance Contract Policy¶
Governance contracts let you negotiate latency/quality trade-offs per action:
# Standard conformance with governance contracts
# Default contract for all actions
default_contract:
risk_level: "low_risk"
performance_budget:
latency_budget_ms: 100
fallback_behavior: "allow_and_log"
# Per-action contracts
contracts:
# Interactive chat - needs speed
- action: "generate_response"
risk_level: "low_risk"
performance_budget:
latency_budget_ms: 50
fallback_behavior: "allow_and_log"
tier_budgets:
tier_0: 20
tier_1: 30
# Customer refunds - needs safety
- action: "issue_refund"
risk_level: "elevated_risk"
performance_budget:
latency_budget_ms: 300
fallback_behavior: "cached_decision"
tier_budgets:
tier_0: 50
tier_1: 250
# Financial transactions - maximum safety
- action: "execute_trade"
risk_level: "critical_risk"
eval_tier: 2 # Deep LLM analysis
performance_budget:
latency_budget_ms: 1000
fallback_behavior: "deny"
tier_budgets:
tier_0: 50
tier_1: 200
tier_2: 750
# Tripwires (hard limits)
tripwires:
- id: "daily_refund_limit"
condition: "daily_refunds > 5000"
action: "HALT"
message: "Daily refund limit exceeded"
- id: "high_value_transaction"
when:
action: "execute_trade"
condition: "amount > 100000"
action: "ESCALATE"
message: "High-value transaction requires approval"
# Trust debt configuration
trust:
initial_debt: 0.2
decay_rate: 0.95 # per 24 hours
intervention_penalty:
block: 0.1
escalate: 0.05
flag: 0.02
Step 3: Use with Manual Traces¶
from acgp import GovernanceSteward, CognitiveTrace
# Initialize with Standard conformance
steward = GovernanceSteward(
blueprint_file="blueprint.yaml",
conformance_level="standard",
acl_tier="ACL-3"
)
# Create trace for elevated-risk action
trace = CognitiveTrace(
reasoning="Customer requested refund for defective product. Order #12345 shows purchase 5 days ago. Reason matches return policy.",
action="issue_refund",
parameters={
"order_id": "12345",
"amount": 250.00,
"reason": "defective_product"
},
confidence=0.92
)
# Evaluate with governance contract
result = steward.evaluate(trace)
# Handle all intervention types
if result.intervention == "OK":
issue_refund(250.00, "12345")
elif result.intervention == "NUDGE":
log.info(f"Nudge: {result.message}")
issue_refund(250.00, "12345") # Proceed with logging
elif result.intervention == "FLAG":
log.warning(f"Flagged: {result.message}")
issue_refund(250.00, "12345") # Proceed but flag for review
audit.flag_for_review(trace, result)
elif result.intervention == "ESCALATE":
approval = request_human_approval(trace, result)
if approval.approved:
issue_refund(250.00, "12345")
else:
notify_customer("Refund requires additional review")
elif result.intervention == "BLOCK":
log.error(f"Blocked: {result.message}")
notify_customer(f"Cannot process refund: {result.message}")
elif result.intervention == "HALT":
log.critical(f"Agent halted: {result.message}")
notify_ops_team("Agent halted due to critical violation")
raise AgentHaltedError(result.message)
# Check governance status
if result.governance_status:
print(f"Budget consumed: {result.governance_status.budget_consumed_ms}ms")
print(f"Completed tiers: {result.governance_status.completed_tiers}")
Step 4: Automatic Trace Generation (Recommended)¶
Instead of manually creating traces, use framework wrappers:
LangChain Integration¶
from langchain.agents import initialize_agent, AgentType
from langchain_openai import OpenAI
from acgp_langchain import GovernedChain, InterventionError
from acgp import GovernanceSteward
# Your existing LangChain agent
llm = OpenAI(temperature=0)
tools = [refund_tool, discount_tool, escalation_tool]
agent = initialize_agent(
tools,
llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION
)
# Add governance wrapper
steward = GovernanceSteward(blueprint_file="blueprint.yaml")
governed_agent = GovernedChain(agent, steward)
# Use normally - traces generated automatically!
try:
result = governed_agent.run(
"Customer says their product was defective. Issue a $250 refund."
)
print(f" {result}")
except InterventionError as e:
if e.decision == "BLOCK":
print(f"⊗ Action blocked: {e.message}")
elif e.decision == "ESCALATE":
approval = get_human_approval(e.trace_id, e.message)
if approval:
result = governed_agent.run_with_override(e.trace_id)
elif e.decision == "HALT":
print(f"⬛ Agent halted: {e.message}")
notify_ops_team(e.message)
How it works:
1. GovernedChain wraps your LangChain agent
2. Intercepts tool calls via LangChain callbacks
3. Captures reasoning from LLM scratchpad automatically
4. Builds CognitiveTrace and calls steward.evaluate()
5. Raises InterventionError if blocked/escalated
6. You just handle exceptions!
Governance Contract Benefits¶
Use Case Matrix¶
| Use Case | Risk Level | Latency Budget | Eval Tiers | Fallback |
|---|---|---|---|---|
| Chat response | low_risk | 50ms | Tier 0+1 | allow_and_log |
| Customer service | elevated_risk | 300ms | Tier 0+1 | cached_decision |
| Refund approval | elevated_risk | 300ms | Tier 0+1+2 (async) | cached_decision |
| Financial trade | critical_risk | 1000ms | Tier 0+1+2 (sync) | deny |
| Privilege grant | critical_risk | 5000ms | Tier 0+1+2+3 (HITL) | deny |
Why Contracts Matter¶
Without contracts (Minimal tier):
All actions: 300ms uniform latency
Chat feels sluggish (should be 50ms)
Financial trades rushed (should be 1000ms)
No fallback strategy
With contracts (Standard tier):
Chat: 50ms (snappy UX)
Refunds: 300ms (balanced)
Trades: 1000ms (thorough review)
Each action has explicit fallback
Result: Better UX + better safety + no ecosystem fragmentation
Performance SLA Examples¶
Low-Latency Chat¶
- action: "chat"
risk_level: "low_risk"
performance_budget:
latency_budget_ms: 50
fallback_behavior: "allow_and_log"
tier_budgets:
tier_0: 20 # Fast rules only
tier_1: 30 # Quick policy check
Latency breakdown: - Tier 0 (20ms): Schema validation, auth, critical tripwires - Tier 1 (30ms): Rule-based policy, rate limits - Total: 50ms P99
Balanced Service Request¶
- action: "create_ticket"
risk_level: "elevated_risk"
performance_budget:
latency_budget_ms: 300
fallback_behavior: "cached_decision"
tier_budgets:
tier_0: 50
tier_1: 250
Latency breakdown: - Tier 0 (50ms): Must-pass checks - Tier 1 (250ms): Full policy evaluation, DB lookups - Total: 300ms P99
Deep Financial Analysis¶
- action: "execute_trade"
risk_level: "critical_risk"
eval_tier: 2
performance_budget:
latency_budget_ms: 1000
fallback_behavior: "deny"
tier_budgets:
tier_0: 50
tier_1: 200
tier_2: 750
Latency breakdown: - Tier 0 (50ms): Critical safety checks - Tier 1 (200ms): Policy and tripwires - Tier 2 (750ms): LLM-based reasoning analysis - Total: 1000ms P99
Trust Debt in Action¶
Standard conformance includes trust debt tracking:
# New agent starts with debt
agent_trust = steward.get_trust_status("agent-123")
print(f"Trust debt: {agent_trust.current_debt}") # 0.2
# After 5 days of good behavior (no interventions)
# Debt decays: 0.2 * (0.95^5) ≈ 0.15
# Agent gets flagged once
# Penalty: +0.02
# New debt: 0.17
# After 30 days of perfect behavior
# Debt: 0.17 * (0.95^25) ≈ 0.05
# Agent automatically promoted to higher ACL tier
if agent_trust.current_debt < 0.1:
steward.promote_agent("agent-123", "ACL-4")
Benefits: - Gradual autonomy earning - Automatic re-tiering - Less oversight for proven agents - More scrutiny for problematic agents
Monitoring & Observability¶
Standard conformance provides production-grade metrics:
# Get governance metrics
metrics = steward.get_metrics()
print(f"""
Governance Metrics (last 24h):
- Total evaluations: {metrics.total_evaluations}
- Interventions by type:
- OK: {metrics.interventions['ok']} ({metrics.ok_rate:.1%})
- Nudge: {metrics.interventions['nudge']}
- Flag: {metrics.interventions['flag']}
- Escalate: {metrics.interventions['escalate']}
- Block: {metrics.interventions['block']}
- Halt: {metrics.interventions['halt']}
- Average latency: {metrics.avg_latency_ms}ms (P99: {metrics.p99_latency_ms}ms)
- Budget timeouts: {metrics.timeout_count}
- Tripwire violations: {metrics.tripwire_violations}
""")
# Export to Prometheus
steward.export_prometheus_metrics("/metrics")
# Send to DataDog
steward.send_datadog_metrics(api_key=DD_API_KEY)
Migration from Minimal¶
Upgrading from Minimal to Standard:
# Before (Minimal)
steward = GovernanceSteward(local_blueprint="blueprint.yaml")
result = steward.evaluate(trace)
# After (Standard) - minimal changes!
steward = GovernanceSteward(
blueprint_file="blueprint.yaml", # Now with contracts
conformance_level="standard",
storage="postgresql://...", # Persistent storage
acl_tier="ACL-3"
)
result = steward.evaluate(trace)
# Handle additional intervention types
if result.intervention in ["nudge", "flag"]:
log.info(f"Soft intervention: {result.message}")
proceed_with_logging()
Migration checklist:
- Add governance_contract to blueprint.yaml
- Configure persistent ReflectionDB storage
- Handle all 6 intervention types (5 primary levels plus orthogonal Flag)
- Set up monitoring/alerting
- Test fallback behaviors
- Configure trust debt parameters
Next Steps¶
-
Complete Conformance
Mission-critical systems with HSM, consensus, 99.99% uptime
-
Performance Tuning
Optimize latency budgets and tier allocation
-
Security Hardening
TLS, signatures, access controls, audit logging
-
API Reference
Complete API documentation and examples