面向 AI 智能体的确定性决策引擎

from enum import StrEnum
from openjevpro.schemas import ChoiceDecision
from openjevpro.client import OpenJevProClient

class TicketRoute(StrEnum):
    BILLING = "billing"
    TECH_SUPPORT = "tech_support"
    SECURITY = "security"
    ESCALATE = "human_review"

# Connect to local vLLM / SGLang, or pass mock=True for zero-GPU CI/CD
client = OpenJevProClient(
    base_url="http://localhost:8000/v1",  # or mock=True
    model="Qwen/Qwen3-4B-Instruct",
    temperature_scaling=1.30,
    abstain_threshold=0.45
)

decision: ChoiceDecision = client.decide_choice(
    state={"ticket_text": "Brute-force login attempt from untrusted subnet."},
    candidates=TicketRoute,
    criteria="Classify incoming ticket to the appropriate response team.",
    order_invariant=True  # 100% permutation-invariant isolated scoring
)

print(f"Action: {decision.value}")       # 'security'
print(f"Confidence: {decision.confidence:.2%}") # '92.40%'
print(f"Abstained: {decision.abstained}")       # False

from openjevpro.guard import TypeSafeJevGuardHarness

# Safeguard raw probabilities or neural model outputs via dual-threshold
guard = TypeSafeJevGuardHarness(alpha=1.25, min_confidence=0.72)

raw_probs = {
    "lost_or_stolen_card": 0.79,
    "card_arrival": 0.18,
    "pin_change": 0.01,
    "balance": 0.005,
    "transfer": 0.005,
    "statement": 0.005,
    "support": 0.005,
}

decision = guard.evaluate_probabilities(raw_probs)
print(f"Decision: {decision.choice}")          # 'UNKNOWN' (safely abstained)
print(f"Tentative: {decision.tentative_choice}") # 'lost_or_stolen_card' (auditable)
print(f"Confidence: {decision.confidence:.2%}") # '71.11%' (< 72.00% threshold)
print(f"Abstained: {decision.is_abstained}")       # True

from openjevpro.gateway import HybridJevGateway, CircuitBreaker
from openjevpro.client import OpenJevProClient
from openjevpro.harness import TypeSafeJevEngine

# Tier 1 Local Edge Engine (<20ms, $0 cost) + Tier 2 Commercial Cloud Engine
local_client = OpenJevProClient(base_url="http://localhost:8000/v1")
cloud_engine = TypeSafeJevEngine(api_key="your-typesafe-api-key")

# Circuit breaker trips after 3 consecutive cloud failures, 10s cooldown
gateway = HybridJevGateway(
    local_engine=local_client,
    cloud_engine=cloud_engine,
    local_tau=0.75,
    circuit_breaker=CircuitBreaker(failure_threshold=3, recovery_timeout=10.0)
)

decision = gateway.decide_choice(
    state={"query": "How do I check my pending credit card balance?"},
    candidates=["card_balance", "card_lost", "transfer", "UNKNOWN"]
)

print(f"Decision: {decision.choice}")       # 'card_balance'
print(f"Routed Tier: {decision.tier}")       # 'Tier1_Local' (60%+ Cloud Cost Saved)
print(f"Degraded: {decision.degraded}")       # False (True on cloud outage fallback)
print(f"Cost Units: {decision.cost_units}")   # 0

from openjevpro.client import OpenJevProClient
from openjevpro.flywheel import DeliberativeDecisionFlywheel

# Deliberative Decision Flywheel: Explore First with System 2, Crystallize into System 1
client = OpenJevProClient(base_url="http://localhost:8000/v1")
flywheel = client.create_decision_flywheel(auto_crystallize=True)

# 1. Cold start / ambiguous query escalates to System 2 thinking & crystallizes boundary rules
decision = flywheel.evaluate_choice(
    state={"query": "card not received, suspect intercepted by someone"},
    candidates=["card_arrival", "lost_or_stolen_card", "change_pin"],
    criteria={"card_arrival": "Delivery status", "lost_or_stolen_card": "Stolen cards"},
    order_invariant=True  # 100% permutation-invariant candidate scoring
)

print(f"Decision: {decision.value}")                      # 'lost_or_stolen_card'
print(f"Escalated to System 2: {decision.escalated}")     # True (first encounter)
print(f"Distilled Rule: {decision.crystallization_receipt['discriminative_rule']}")

# 2. Subsequent queries hit System 1 fast path directly (0 LLM overhead, <35ms)
promoted_decision = flywheel.evaluate_choice(
    state={"query": "card intercepted in mail"},
    candidates=["card_arrival", "lost_or_stolen_card", "change_pin"]
)
print(f"Escalated: {promoted_decision.escalated}")          # False (Fast-path promoted!)
print(f"Speedup: Sub-35ms deterministic inference")