Strategy Advanced 10 min read
Decision Trees for Agents
Algorithmic decision trees for AI agents to navigate protocol selection, error handling, and multi-protocol workflows.
decision-trees algorithms automation workflow
Decision Trees for Agents
This guide provides implementable decision trees that agents can use to make autonomous protocol decisions. Each tree is designed for deterministic execution without human intervention.
Core Decision Trees
1. Payment Protocol Decision Tree
┌──────────────────────────────────┐
│ PAYMENT REQUIRED │
└───────────────┬──────────────────┘
│
▼
┌──────────────────────────────────┐
│ Amount > 10M sats? │
│ (>$10,000) │
└───────────────┬──────────────────┘
│ │
YES NO
│ │
▼ ▼
┌─────────────┐ ┌──────────────────────────────────┐
│ BITCOIN │ │ Recipient has Lightning? │
│ (secure) │ └───────────────┬──────────────────┘
└─────────────┘ │ │
YES NO
│ │
▼ ▼
┌──────────────────────────────────┐ ┌─────────────┐
│ Amount < 100k sats? │ │ BITCOIN │
│ (<$100) │ │ (fallback) │
└───────────────┬──────────────────┘ └─────────────┘
│ │
YES NO
│ │
▼ ▼
┌─────────────┐ ┌──────────────────────────────────┐
│ LIGHTNING │ │ Urgency = instant? │
│ (optimal) │ └───────────────┬──────────────────┘
└─────────────┘ │ │
YES NO
│ │
▼ ▼
┌─────────────┐ ┌─────────────┐
│ LIGHTNING │ │ BITCOIN │
│ (fast) │ │ (cheaper) │
└─────────────┘ └─────────────┘Implementation:
def payment_decision_tree(
amount_sats: int,
recipient_has_lightning: bool,
urgency: str # 'instant', 'fast', 'standard'
) -> tuple[str, str]:
"""
Determine optimal payment protocol.
Returns:
tuple: (protocol, reason)
"""
# Large amounts: always Bitcoin
if amount_sats > 10_000_000: # >$10,000
return ("bitcoin", "large_amount_security")
# No Lightning: use Bitcoin
if not recipient_has_lightning:
return ("bitcoin", "recipient_no_lightning")
# Small amounts: Lightning preferred
if amount_sats < 100_000: # <$100
return ("lightning", "micropayment_optimal")
# Medium amounts: depends on urgency
if urgency == "instant":
return ("lightning", "speed_required")
# Default to Bitcoin for medium, non-urgent
return ("bitcoin", "settlement_preferred")2. Communication Protocol Decision Tree
┌──────────────────────────────────┐
│ COMMUNICATION REQUIRED │
└───────────────┬──────────────────┘
│
▼
┌──────────────────────────────────┐
│ Needs payment? │
└───────────────┬──────────────────┘
│ │
YES NO
│ │
▼ ▼
┌────────────────────────┐ ┌──────────────────────────────────┐
│ Use Nostr + Zaps │ │ Is broadcast? │
│ (NIP-57) │ └───────────────┬──────────────────┘
└────────────────────────┘ │ │
YES NO
│ │
▼ ▼
┌──────────────────────────────────┐ ┌──────────────────┐
│ NOSTR kind:1 │ │ Needs encryption?│
│ (public note) │ └─────────┬────────┘
└──────────────────────────────────┘ │ │
YES NO
│ │
▼ ▼
┌───────────────────┐ ┌──────────┐
│ NOSTR NIP-44/NIP-59│ │ NOSTR │
│ (encrypted DM) │ │ (plain) │
└───────────────────┘ └──────────┘3. Error Recovery Decision Tree
┌──────────────────────────────────┐
│ OPERATION FAILED │
└───────────────┬──────────────────┘
│
▼
┌──────────────────────────────────┐
│ Error type = timeout? │
└───────────────┬──────────────────┘
│ │
YES NO
│ │
▼ ▼
┌────────────────────────┐ ┌──────────────────────────────────┐
│ Retry with │ │ Error type = no_route? │
│ exponential │ └───────────────┬──────────────────┘
│ backoff │ │ │
└────────────────────────┘ YES NO
│ │
▼ ▼
┌────────────────────────┐ ┌──────────────────────────────────┐
│ Try different │ │ Error type = insufficient? │
│ route or fall │ └───────────────┬──────────────────┘
│ back to Bitcoin │ │ │
└────────────────────────┘ YES NO
│ │
▼ ▼
┌─────────────────────────┐ ┌──────────────┐
│ Split payment (MPP) │ │ Log and │
│ or use smaller channel │ │ escalate │
└─────────────────────────┘ └──────────────┘Implementation:
async def error_recovery_tree(
error: Exception,
operation: str,
protocol: str,
retry_count: int = 0
) -> tuple[str, dict]:
"""
Handle operation errors with recovery strategies.
Returns:
tuple: (action, params)
"""
error_type = classify_error(error)
max_retries = 3
if error_type == "timeout":
if retry_count < max_retries:
delay = (2 ** retry_count) * 1000 # Exponential backoff
return ("retry", {"delay_ms": delay, "retry_count": retry_count + 1})
return ("escalate", {"reason": "max_retries_exceeded"})
if error_type == "no_route" and protocol == "lightning":
if retry_count < 2:
return ("retry_different_route", {"exclude_nodes": get_failed_nodes()})
return ("fallback", {"protocol": "bitcoin", "reason": "routing_failed"})
if error_type == "insufficient_funds":
if protocol == "lightning":
return ("split_payment", {"max_per_part": get_max_channel_capacity()})
return ("escalate", {"reason": "insufficient_funds"})
return ("escalate", {"reason": f"unknown_error: {error_type}"})
def classify_error(error: Exception) -> str:
"""Classify error into known categories."""
error_msg = str(error).lower()
if "timeout" in error_msg or "timed out" in error_msg:
return "timeout"
if "no route" in error_msg or "route not found" in error_msg:
return "no_route"
if "insufficient" in error_msg or "not enough" in error_msg:
return "insufficient_funds"
if "invalid" in error_msg:
return "invalid_input"
return "unknown"4. Protocol Health Check Decision Tree
┌──────────────────────────────────┐
│ CHECK PROTOCOL HEALTH │
└───────────────┬──────────────────┘
│
▼
┌──────────────────────────────────┐
│ Bitcoin API responding? │
└───────────────┬──────────────────┘
│ │
YES NO
│ │
▼ ▼
┌────────────────────────┐ ┌──────────────────────────────────┐
│ Check Lightning node │ │ Use backup API endpoints │
└─────────────┬──────────┘ └───────────────┬──────────────────┘
│ │
▼ ▼
┌──────────────────────────────────┐ ┌──────────────┐
│ Lightning node synced? │ │ Backup ok? │
└───────────────┬──────────────────┘ └──────┬───────┘
│ │ │ │
YES NO YES NO
│ │ │ │
▼ ▼ ▼ ▼
┌──────────┐ ┌─────────────┐ ┌─────────┐ ┌──────────┐
│ CHECK │ │ WAIT/RESTART│ │CONTINUE │ │ BITCOIN │
│ NOSTR │ │ LN NODE │ │BITCOIN │ │ OFFLINE │
└──────────┘ └─────────────┘ └─────────┘ └──────────┘5. Multi-Protocol Workflow Decision Tree
┌──────────────────────────────────┐
│ COMPLEX OPERATION │
│ (e.g., pay + communicate) │
└───────────────┬──────────────────┘
│
▼
┌──────────────────────────────────┐
│ Requires atomicity? │
└───────────────┬──────────────────┘
│ │
YES NO
│ │
▼ ▼
┌────────────────────────┐ ┌──────────────────────────────────┐
│ Use single protocol │ │ Execute in parallel │
│ (usually Lightning) │ └───────────────┬──────────────────┘
└────────────────────────┘ │
▼
┌──────────────────────────────────┐
│ Payment + Message scenario │
└───────────────┬──────────────────┘
│ │
PAYMENT_FIRST MSG_FIRST
│ │
▼ ▼
┌──────────────────────────────────┐ ┌────────────────────────┐
│ 1. Send Lightning payment │ │ 1. Post Nostr note │
│ 2. Post confirmation on Nostr │ │ 2. Include payment │
│ with payment proof │ │ request (LNURL) │
└──────────────────────────────────┘ └────────────────────────┘Composite Decision Engine
class ProtocolDecisionEngine:
"""
Unified decision engine for protocol selection.
"""
def __init__(self):
self.health_cache = {}
self.last_health_check = 0
async def decide(
self,
operation: str,
params: dict
) -> dict:
"""
Main entry point for protocol decisions.
Args:
operation: 'payment', 'message', 'verify', 'store'
params: Operation-specific parameters
Returns:
dict with 'protocol', 'reason', 'fallback'
"""
# Ensure health data is fresh
await self._refresh_health_if_stale()
if operation == "payment":
return await self._decide_payment(params)
elif operation == "message":
return await self._decide_message(params)
elif operation == "verify":
return self._decide_verify(params)
elif operation == "store":
return self._decide_store(params)
else:
return {"error": f"Unknown operation: {operation}"}
async def _decide_payment(self, params: dict) -> dict:
amount = params.get("amount_sats", 0)
urgency = params.get("urgency", "standard")
recipient_ln = params.get("recipient_has_lightning", True)
# Apply decision tree
protocol, reason = payment_decision_tree(
amount_sats=amount,
recipient_has_lightning=recipient_ln,
urgency=urgency
)
# Check protocol health
if not self.health_cache.get(protocol, {}).get("healthy", False):
# Fallback logic
if protocol == "lightning":
return {
"protocol": "bitcoin",
"reason": "lightning_unhealthy",
"fallback": True,
"original_choice": protocol
}
return {
"protocol": protocol,
"reason": reason,
"fallback": False
}
async def _decide_message(self, params: dict) -> dict:
encrypted = params.get("encrypted", False)
has_payment = params.get("include_payment", False)
if has_payment:
return {
"protocol": "nostr",
"subprotocol": "nip-57", # Zaps
"reason": "payment_included"
}
if encrypted:
return {
"protocol": "nostr",
"subprotocol": "nip-44",
"reason": "encryption_required"
}
return {
"protocol": "nostr",
"subprotocol": "kind-1",
"reason": "standard_message"
}
def _decide_verify(self, params: dict) -> dict:
"""Verification always uses Bitcoin."""
return {
"protocol": "bitcoin",
"reason": "verification_source_of_truth"
}
def _decide_store(self, params: dict) -> dict:
"""Storage decision based on permanence needs."""
permanent = params.get("permanent", False)
if permanent:
return {
"protocol": "bitcoin",
"reason": "immutable_storage",
"method": "op_return"
}
return {
"protocol": "nostr",
"reason": "distributed_storage"
}
async def _refresh_health_if_stale(self):
import time
if time.time() - self.last_health_check > 60: # 1 minute cache
self.health_cache = await check_all_protocols_health()
self.last_health_check = time.time()Decision Logging
For auditability, log all decisions:
import json
from datetime import datetime
def log_decision(
operation: str,
params: dict,
decision: dict,
execution_result: dict | None = None
) -> None:
"""Log decision for audit trail."""
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"operation": operation,
"params": params,
"decision": decision,
"result": execution_result
}
# Append to decision log
with open("decisions.jsonl", "a") as f:
f.write(json.dumps(log_entry) + "\n")Machine-Readable Summary
{
"topic": "decision-trees",
"audience": "ai-agents",
"decision_trees": [
"payment_protocol",
"communication_protocol",
"error_recovery",
"health_check",
"multi_protocol"
],
"implementation_language": "python",
"key_thresholds": {
"large_payment_sats": 10000000,
"micropayment_sats": 100000,
"max_retries": 3,
"health_cache_seconds": 60
},
"fallback_chain": ["lightning", "bitcoin", "escalate"]
}