🤖 知微 v10.48 · 主动式AI 全面落地

# diagnose() — 异常诊断 diagnose() { local svc=$1 port=$2 log_pattern=$3 echo "=== Diagnose: $svc ===" # 最近日志错误提取 journalctl -u $svc --no-pager -n 10 2>/dev/null | grep -i "error\|fail\|timeout" || true # 系统状态采集 echo " CPU: $(top -bn1 | grep 'Cpu' | head -1)" echo " MEM: $(free -m | awk '/Mem/{print $3"/"$2"MB"}')" echo " DISK: $(df -h / | awk 'NR==2{print $5}')" } # verify() — 修复验证 verify() { local svc=$1 port=$2 url=$3 local start_ts=$(date +%s) sleep 3 local code=$(curl -s -o /dev/null -w "%{http_code}" http://${BASE_URL:-https://x.kddauto.com}:$port${url:-/} 2>/dev/null) local elapsed=$(($(date +%s) - start_ts)) [ "$code" = "200" ] && log "INFO" "$svc 恢复验证成功 (HTTP $code, ${elapsed}s)" && return 0 log "WARN" "$svc 验证失败 (HTTP $code)" return 1 } # adapt() — 写入自愈历史 adapt() { local svc=$1 symptom=$2 success=$3 python3 " import json, sys from datetime import datetime records = [] try: with open('$BASE_DIR/data/self_heal_history.json') as f: data = json.load(f) records = data.get('history', []) except: pass records.append({ 'service': '$svc', 'symptom': '$symptom', 'timestamp': datetime.now().isoformat(), 'success': '$success' == 'true' }) records = records[-500:] with open('$BASE_DIR/data/self_heal_history.json', 'w') as f: json.dump({'history': records}, f, indent=2, ensure_ascii=False) " } # R-11: 3次失败升级人工 escalate_if_needed() { local svc=$1 count=$2 if [ "$count" -ge 3 ]; then log "ESCALATE" "$svc 连续失败 $count 次，升级人工处理" python3 -c " import smtplib, ssl, os from email.mime.text import MIMEText env={} for l in open('$BASE_DIR/.env'): if '=' in l and not l.startswith('#'): k,v=l.strip().split('=',1) env[k.strip()]=v.strip() msg=MIMEText(f'知微P0-2告警: $svc 自愈连续失败 $count 次，已停止自动修复，请人工介入', 'plain') msg['Subject']=f'🚨 自愈升级人工: $svc' msg['From']=env.get('FEISHU_EMAIL_USER','') msg['To']=env.get('FEISHU_NOTIFY_TO','') try: with smtplib.SMTP_SSL(env.get('FEISHU_SMTP_HOST',''),int(env.get('FEISHU_SMTP_PORT','465')), context=ssl.create_default_context()) as s: s.login(msg['From'], env.get('FEISHU_EMAIL_PASS','')) s.send_message(msg) print('ESCALATE EMAIL SENT') except Exception as e: print(f'ESCALATE EMAIL FAIL: {e}') " || log "WARN" "升级邮件发送失败" return 0 fi return 1 }

from queue import PriorityQueue from threading import Lock, Thread import json, time, logging class DynamicScheduler: """P0-3: 动态任务调度器 - 对标LangGraph状态机 + Airflow DAG 特性: PriorityQueue + DAG依赖图 + 并发控制(max=3) + 事件驱动 + API动态注册""" def __init__(self, engine): self.engine = engine # WorkflowEngine实例 self.queue = PriorityQueue() self.dag = {} # {task_id: [downstream_ids]} self.locks = {} # {resource: Lock} self.active_tasks = set() self.max_concurrent = 3 self.retry_limit = 3 self.running = False def register(self, task_id, priority=1, deps=None, triggers=None): """注册任务: priority:0=P0,1=P1,2=P2 | deps:依赖列表 | triggers:事件类型列表""" self.dag[task_id] = deps or [] if triggers: self._subscribe_events(task_id, triggers) return task_id def enqueue(self, task_id, priority=1): if not self._deps_satisfied(task_id): return False # 依赖未满足 self.queue.put((priority, task_id)) return True def run_loop(self): """主循环: 取队→执行→触发下游 (独立线程)""" self.running = True while self.running: if len(self.active_tasks) >= self.max_concurrent: time.sleep(1); continue try: priority, task_id = self.queue.get(timeout=1) self.active_tasks.add(task_id) Thread(target=self._execute, args=(task_id,), daemon=True).start() except: pass def _execute(self, task_id): for attempt in range(self.retry_limit): try: self.engine.run_workflow(task_id) break except Exception as e: logging.error(f"Task {task_id} attempt {attempt+1}: {e}") time.sleep(2 ** attempt) # 指数退避 self.active_tasks.discard(task_id) # 触发下游任务 for downstream in self.dag.get(task_id, []): self.enqueue(downstream)