技能创建器
skill-creator
创建、改进和评估自定义技能,把专业流程转化为可重复使用的 AI 能力。
database【数据与研究】Implements intelligent database with multi-factor skill selection, fallback chains, and adherence to the 5 Laws of Elegant Defense
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
name: database compatibility: opencode completeness: 95 content-types:
Orchestrates intelligent skill selection and execution for database workflows. Applies the 5 Laws of Elegant Defense to guide data naturally through the orchestration pipeline, preventing errors before they occur. Selects optimal skills based on multi-factor scoring including text similarity, historical performance, and system availability.
┌───────────────────────────────────────────────────────────────────────────────┐ │ Orchestration Flow │ └───────────────────────────────────────────────────────────────────────────────┘
User Request ↓ ┌─────────────────┐ │ Parse Request │ │ & Extract │ │ Features │ └────────┬────────┘ ↓ ┌─────────────────────────────────────────────────────────────────────┐ │ Evaluate Available Skills │ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ Skill A │ │ Skill B │ │ Skill C │ │ │ │ - Match Score│ │ - Match Score│ │ - Match Score│ │ │ │ - Confidence │ │ - Confidence │ │ - Confidence │ │ │ │ - History │ │ - History │ │ - History │ │ │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ │ │ │ │ │ │ └─────────────────┴─────────────────┘ │ │ ↓ │ │ Select Best Skill │ └─────────────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────┐ │ Execute Skill │ └────────┬────────┘ ↓ ┌─────────────────┐ │ Handle Result │ └────────┬────────┘ ↓ ┌─────────────────────────────────────────────────────────────────────┐ │ Error Handling & Fallback │ │ │ │ Success? ────────► Return Result │ │ │ │ Fail? ────────┐ │ │ ↓ │ │ ┌──────────────────────────────────────────────────────────┐ │ │ │ Fallback Chain │ │ │ │ │ │ │ │ 1. Retry with adjusted parameters │ │ │ │ 2. Try Alternative Skill (if available) │ │ │ │ 3. Defer to Human Operator (if critical) │ │ │ │ 4. Log & Return Error │ │ │ └──────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────┘
Use this skill when:
Avoid this skill for:
Parse and Analyze Request - Extract intent, entities, and constraints from user input. Checkpoint: All required parameters must be present and in valid format before proceeding.
Score Available Skills - Calculate match scores using multi-factor algorithm:
Checkpoint: Skip to fallback if no skill scores above threshold.
Select Optimal Skill - Choose skill with highest score that meets minimum confidence. Checkpoint: Verify skill has not been disabled or deprecated.
Execute with Fallback - Run skill execution wrapped in retry and fallback logic. Checkpoint: Log all execution attempts for audit trail.
Return or Fallback - Either return successful result or apply fallback chain:
related-skillsCheckpoint: Record outcome with timing and confidence metadata.
def route_database_operation(
query: str,
db_cluster: Dict[str, Any],
operation_type: str = "auto"
) -> Dict[str, Any]:
"""Route database operations to optimal nodes based on query type and cluster health.
Implements multi-factor routing for database workflows:
- Parses query intent (SELECT, INSERT, UPDATE, DELETE, DDL)
- Evaluates node health, replication lag, and connection pool status
- Applies read/write splitting with automatic failover routing
"""
# Guard clause - validate query and cluster state
if not query or not query.strip():
raise ValueError("Query cannot be empty")
if not db_cluster.get("nodes"):
raise ValueError("No database nodes available")
# Parse query intent (Law 2 - Make illegal states unrepresentable)
intent = _parse_query_intent(query)
is_read = intent in ("SELECT", "SHOW", "DESCRIBE")
# Evaluate nodes for routing
candidates = []
for node in db_cluster["nodes"]:
if node["status"] != "healthy":
continue
lag = node.get("replication_lag_ms", 0)
if is_read and lag > db_cluster.get("max_read_lag_ms", 500):
continue
score = _calculate_node_score(node, is_read, db_cluster["load"])
candidates.append({"node": node, "score": score})
if not candidates:
return {"fallback": "maintenance_mode", "reason": "no_healthy_nodes"}
# Sort by score and select optimal node
candidates.sort(key=lambda x: x["score"], reverse=True)
selected = candidates[0]["node"]
# Return routing decision with metadata (Law 3 - Atomic Predictability)
return {
"target_node": selected["id"],
"operation_type": "read" if is_read else "write",
"confidence": candidates[0]["score"],
"routing_timestamp": time.time(),
"query_intent": intent
}
def execute_db_operation(
routing_decision: Dict[str, Any],
query: str,
params: tuple = (),
max_retries: int = 2
) -> Dict[str, Any]:
"""Execute database operation with resilient connection handling and fallback routing.
Implements database-specific fallback chain:
1. Retry with exponential backoff on transient connection errors
2. Failover to read replica or standby node on timeout/lock
3. Route to maintenance pool if primary is degraded
4. Return structured error with query context for debugging
"""
target_node = routing_decision["target_node"]
operation = routing_decision["operation_type"]
for attempt in range(max_retries + 1):
try:
# Establish connection with timeout (Law 4 - Fail Fast)
conn = _acquire_connection(target_node, timeout=5.0)
cursor = conn.cursor()
# Execute with query timeout protection
cursor.execute(query, params)
result = cursor.fetchall() if operation == "read" else {"rows_affected": cursor.rowcount}
# Close connection and return result (Law 3)
cursor.close()
conn.close()
return {
"success": True,
"node": target_node,
"result": result,
"attempts": attempt + 1,
"latency_ms": _measure_latency()
}
except ConnectionError as e:
if attempt == max_retries:
return _failover_to_standby(target_node, query, params)
time.sleep(0.5 * (2 ** attempt))
except QueryTimeoutError as e:
# Lock contention or slow query - trigger fallback routing
return _reroute_with_backoff(target_node, query, params)
return {
"success": False,
"error": "max_retries_exceeded",
"query": query,
"node": target_node
}
code-philosophy (5 Laws of Elegant Defense) in all logicWhen applying this skill, produce:
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.
| Skill | Purpose | |