🐉 龍魂提示词路由器 · 完整引擎

动态路由 · 自我迭代 · 鲲鹏同步

DNA追溯码： #龍芯⚡️丙午·癸未·庚辰·ROUTER-ENGINE-v1.0-UID9622

确认码： #CONFIRM🌌9622-ONLY-ONCE🧬LK9X-772Z

主权锚定： #ZHUGEXIN⚡️2025-🇨🇳🐉⚖️♠️🧚🏼‍♀️❤️♾️-DEVICE-BIND-SOUL

---

📦 完整脚本：lh_prompt_router.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
🐉 龍魂提示词路由器 v1.0
动态路由 · 自我迭代 · 鲲鹏同步

DNA: #龍芯⚡️丙午·癸未·庚辰·ROUTER-ENGINE-v1.0-UID9622
CONFIRM: #CONFIRM🌌9622-ONLY-ONCE🧬LK9X-772Z

功能:
  1. 多路由配置管理 (YAML/JSON)
  2. 关键词/意图/正则匹配
  3. 动态加载对应的System Prompt
  4. 自我学习：未匹配时记录，人工审核后自动添加新路由
  5. 与鲲鹏服务器双向同步 (rsync/scp/API)
  6. 版本控制 + 回滚
  7. 完整的日志和错误处理
  8. 命令行/API双接口

集成方式:
  - 作为独立服务运行 (端口9630)
  - 或直接导入作为库使用

依赖:
  pip install pyyaml watchdog requests

用法:
  python lh_prompt_router.py --serve         # 启动API服务
  python lh_prompt_router.py --route "审计"   # 命令行路由测试
  python lh_prompt_router.py --sync          # 手动同步到鲲鹏
  python lh_prompt_router.py --learn         # 处理待学习记录
  python lh_prompt_router.py --status        # 查看状态
"""

import os
import sys
import json
import yaml
import re
import hashlib
import time
import shutil
import subprocess
import argparse
import logging
import threading
import signal
from pathlib import Path
from datetime import datetime
from typing import Dict, List, Optional, Any, Tuple, Callable
from dataclasses import dataclass, field, asdict
from enum import Enum
import copy

# ============================================================
# 固定锚点
# ============================================================

ROOT_DIR = Path.home() / ".龙魂"
ROUTER_DIR = ROOT_DIR / "prompt_router"
CONFIG_FILE = ROUTER_DIR / "routes.yaml"
LEARN_FILE = ROUTER_DIR / "learn_queue.yaml"
BACKUP_DIR = ROUTER_DIR / "backups"
LOG_DIR = ROUTER_DIR / "logs"
SYNC_LOG = ROUTER_DIR / "sync_history.json"
ROUTER_DIR.mkdir(parents=True, exist_ok=True)
BACKUP_DIR.mkdir(exist_ok=True)
LOG_DIR.mkdir(exist_ok=True)

# 鲲鹏服务器配置（通过环境变量或配置文件）
KUNPENG_HOST = os.environ.get("KUNPENG_HOST", "your-kunpeng-server")
KUNPENG_USER = os.environ.get("KUNPENG_USER", "root")
KUNPENG_PATH = os.environ.get("KUNPENG_PATH", "/opt/龙魂/prompt_router/")
KUNPENG_SYNC_MODE = os.environ.get("KUNPENG_SYNC_MODE", "rsync")  # rsync, scp, api

DNA = "#龍芯⚡️丙午·癸未·庚辰·ROUTER-ENGINE-v1.0-UID9622"
CONFIRM = "#CONFIRM🌌9622-ONLY-ONCE🧬LK9X-772Z"
VERSION = "1.0.0"

# ============================================================
# 日志系统
# ============================================================

def setup_logging():
    logging.basicConfig(
        level=logging.INFO,
        format="[%(asctime)s] %(levelname)s - %(message)s",
        handlers=[
            logging.FileHandler(LOG_DIR / f"router_{datetime.now().strftime('%Y%m%d')}.log"),
            logging.StreamHandler()
        ]
    )
    return logging.getLogger("prompt_router")

logger = setup_logging()

# ============================================================
# 核心数据结构
# ============================================================

@dataclass
class Route:
    """单条路由定义"""
    name: str
    triggers: List[str]          # 关键词/短语
    patterns: List[str]          # 正则表达式 (可选)
    system_prompt: str           # 对应的System Prompt
    tools: List[str] = field(default_factory=list)
    priority: int = 0            # 优先级 (越高越先匹配)
    enabled: bool = True
    version: str = "1.0"
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())
    updated_at: str = field(default_factory=lambda: datetime.now().isoformat())
    metadata: Dict = field(default_factory=dict)

    def to_dict(self) -> Dict:
        return asdict(self)

    @classmethod
    def from_dict(cls, data: Dict) -> 'Route':
        return cls(**data)

@dataclass
class RouterConfig:
    """路由器配置"""
    version: str = VERSION
    default_prompt: str = "你是龍魂助手，处理通用问题。"
    routes: List[Route] = field(default_factory=list)
    learn_threshold: int = 3     # 未匹配记录数达到此值触发学习建议
    sync_interval: int = 3600    # 自动同步间隔（秒）

    def to_dict(self) -> Dict:
        return {
            "version": self.version,
            "default_prompt": self.default_prompt,
            "routes": [r.to_dict() for r in self.routes],
            "learn_threshold": self.learn_threshold,
            "sync_interval": self.sync_interval
        }

    @classmethod
    def from_dict(cls, data: Dict) -> 'RouterConfig':
        config = cls(
            version=data.get("version", VERSION),
            default_prompt=data.get("default_prompt", "你是龍魂助手，处理通用问题。"),
            learn_threshold=data.get("learn_threshold", 3),
            sync_interval=data.get("sync_interval", 3600)
        )
        for r in data.get("routes", []):
            config.routes.append(Route.from_dict(r))
        return config

# ============================================================
# 配置管理器
# ============================================================

class ConfigManager:
    """配置的加载、保存、备份、回滚"""

    def __init__(self, config_path: Path = CONFIG_FILE):
        self.config_path = config_path
        self.config: Optional[RouterConfig] = None
        self._load()

    def _load(self):
        """加载配置文件，不存在则创建默认"""
        if not self.config_path.exists():
            self.config = RouterConfig()
            self._save()
            logger.info("✅ 创建默认配置文件")
        else:
            try:
                with open(self.config_path, 'r', encoding='utf-8') as f:
                    data = yaml.safe_load(f) or {}
                self.config = RouterConfig.from_dict(data)
                logger.info(f"✅ 加载配置: {len(self.config.routes)} 条路由")
            except Exception as e:
                logger.error(f"❌ 配置文件损坏，使用默认配置: {e}")
                self.config = RouterConfig()
                self._save()

    def _save(self):
        """保存配置"""
        # 自动备份
        self._backup()
        # 写新配置
        with open(self.config_path, 'w', encoding='utf-8') as f:
            yaml.dump(self.config.to_dict(), f, allow_unicode=True, default_flow_style=False)
        logger.info("💾 配置已保存")

    def _backup(self):
        """备份当前配置"""
        if self.config_path.exists():
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            backup_file = BACKUP_DIR / f"routes_{timestamp}.yaml"
            shutil.copy(self.config_path, backup_file)
            # 保留最近30个备份
            backups = sorted(BACKUP_DIR.glob("routes_*.yaml"))
            if len(backups) > 30:
                for old in backups[:-30]:
                    old.unlink()

    def reload(self):
        """重新加载配置（热更新）"""
        self._load()

    def get_route_by_name(self, name: str) -> Optional[Route]:
        for r in self.config.routes:
            if r.name == name:
                return r
        return None

    def add_route(self, route: Route):
        """添加新路由（若已存在则更新）"""
        existing = self.get_route_by_name(route.name)
        if existing:
            existing.triggers = route.triggers
            existing.patterns = route.patterns
            existing.system_prompt = route.system_prompt
            existing.tools = route.tools
            existing.priority = route.priority
            existing.enabled = route.enabled
            existing.updated_at = datetime.now().isoformat()
        else:
            self.config.routes.append(route)
        self._save()

    def remove_route(self, name: str):
        self.config.routes = [r for r in self.config.routes if r.name != name]
        self._save()

    def enable_route(self, name: str, enable: bool = True):
        route = self.get_route_by_name(name)
        if route:
            route.enabled = enable
            route.updated_at = datetime.now().isoformat()
            self._save()

    def rollback(self, version: str = None):
        """回滚到指定备份版本"""
        if version:
            backup_file = BACKUP_DIR / f"routes_{version}.yaml"
        else:
            # 默认回滚到上一个备份
            backups = sorted(BACKUP_DIR.glob("routes_*.yaml"))
            if len(backups) < 2:
                logger.warning("没有可回滚的备份")
                return
            backup_file = backups[-2]  # 倒数第二个
        if backup_file.exists():
            shutil.copy(backup_file, self.config_path)
            self._load()
            logger.info(f"⏪ 已回滚到: {backup_file.name}")

    def get_history(self) -> List[Dict]:
        """获取备份历史"""
        backups = sorted(BACKUP_DIR.glob("routes_*.yaml"))
        history = []
        for b in backups:
            stat = b.stat()
            history.append({
                "file": b.name,
                "size": stat.st_size,
                "modified": datetime.fromtimestamp(stat.st_mtime).isoformat()
            })
        return history

# ============================================================
# 学习队列管理器（自我迭代）
# ============================================================

class LearnManager:
    """管理未匹配的输入，用于人工审核后生成新路由"""

    def __init__(self, learn_file: Path = LEARN_FILE):
        self.learn_file = learn_file
        self.queue = self._load()

    def _load(self) -> List[Dict]:
        if not self.learn_file.exists():
            return []
        try:
            with open(self.learn_file, 'r', encoding='utf-8') as f:
                return yaml.safe_load(f) or []
        except:
            return []

    def _save(self):
        with open(self.learn_file, 'w', encoding='utf-8') as f:
            yaml.dump(self.queue, f, allow_unicode=True, default_flow_style=False)

    def record_unmatched(self, user_input: str, context: Dict = None):
        """记录一条未匹配的输入"""
        entry = {
            "input": user_input,
            "context": context or {},
            "timestamp": datetime.now().isoformat(),
            "count": 1,
            "resolved": False
        }
        # 查重累加
        for item in self.queue:
            if item["input"] == user_input and not item["resolved"]:
                item["count"] += 1
                item["timestamp"] = datetime.now().isoformat()
                self._save()
                return
        self.queue.append(entry)
        self._save()

    def get_pending(self) -> List[Dict]:
        """获取待处理的未匹配记录（按出现频率排序）"""
        pending = [item for item in self.queue if not item.get("resolved", False)]
        return sorted(pending, key=lambda x: x["count"], reverse=True)

    def mark_resolved(self, input_text: str, route_name: str):
        """标记某条未匹配记录已被解决（已添加路由）"""
        for item in self.queue:
            if item["input"] == input_text:
                item["resolved"] = True
                item["resolved_at"] = datetime.now().isoformat()
                item["route_name"] = route_name
                self._save()
                return

    def suggest_new_routes(self, threshold: int = 3) -> List[Dict]:
        """生成路由建议（频率超过阈值的未匹配记录）"""
        pending = self.get_pending()
        suggestions = []
        for item in pending:
            if item["count"] >= threshold:
                suggestions.append({
                    "input": item["input"],
                    "count": item["count"],
                    "suggested_triggers": self._extract_keywords(item["input"]),
                    "suggested_name": self._generate_name(item["input"])
                })
        return suggestions

    def _extract_keywords(self, text: str) -> List[str]:
        """从文本提取候选关键词"""
        # 简单分词
        words = re.findall(r'[\u4e00-\u9fa5a-zA-Z]+', text)
        # 过滤短词和停用词
        stopwords = {'的', '了', '是', '我', '你', '他', '她', '它', '们', '和', '与', '或', '但', '而', '因', '为', '以', '于'}
        keywords = [w for w in words if len(w) >= 2 and w not in stopwords]
        return keywords[:5]

    def _generate_name(self, text: str) -> str:
        """根据文本生成路由名（拼音/英文/缩写）"""
        # 取前两个关键词
        keywords = self._extract_keywords(text)
        if keywords:
            return "_".join(keywords[:2]).lower()
        return "route_" + hashlib.md5(text.encode()).hexdigest()[:8]

# ============================================================
# 路由引擎核心
# ============================================================

class RouterEngine:
    """路由匹配与执行核心"""

    def __init__(self, config_manager: ConfigManager, learn_manager: LearnManager):
        self.config_mgr = config_manager
        self.learn_mgr = learn_manager
        self.match_cache = {}  # 缓存匹配结果

    def route(self, user_input: str, context: Dict = None) -> Tuple[str, Optional[Route]]:
        """
        根据用户输入匹配路由，返回 (system_prompt, route)
        """
        context = context or {}
        # 检查缓存
        cache_key = hashlib.md5((user_input + str(context)).encode()).hexdigest()
        if cache_key in self.match_cache:
            return self.match_cache[cache_key]

        # 遍历所有启用的路由
        sorted_routes = sorted(
            [r for r in self.config_mgr.config.routes if r.enabled],
            key=lambda x: x.priority, reverse=True
        )

        matched_route = None
        for route in sorted_routes:
            # 关键词匹配
            for trigger in route.triggers:
                if trigger in user_input:
                    matched_route = route
                    break
            if matched_route:
                break
            # 正则匹配
            for pattern in route.patterns:
                if re.search(pattern, user_input, re.IGNORECASE):
                    matched_route = route
                    break
            if matched_route:
                break

        # 如果未匹配，记录到学习队列
        if not matched_route:
            self.learn_mgr.record_unmatched(user_input, context)
            prompt = self.config_mgr.config.default_prompt
            result = (prompt, None)
        else:
            prompt = matched_route.system_prompt
            result = (prompt, matched_route)

        # 缓存
        self.match_cache[cache_key] = result
        return result

    def clear_cache(self):
        self.match_cache = {}

    def get_route_names(self) -> List[str]:
        return [r.name for r in self.config_mgr.config.routes]

    def stats(self) -> Dict:
        routes = self.config_mgr.config.routes
        enabled = sum(1 for r in routes if r.enabled)
        pending = len(self.learn_mgr.get_pending())
        return {
            "total_routes": len(routes),
            "enabled_routes": enabled,
            "pending_learn": pending,
            "cache_size": len(self.match_cache)
        }

# ============================================================
# 同步管理器（鲲鹏服务器）
# ============================================================

class SyncManager:
    """与鲲鹏服务器的双向同步"""

    def __init__(self, local_dir: Path = ROUTER_DIR):
        self.local_dir = local_dir
        self.host = KUNPENG_HOST
        self.user = KUNPENG_USER
        self.remote_path = KUNPENG_PATH
        self.mode = KUNPENG_SYNC_MODE
        self.history_file = SYNC_LOG
        self._load_history()

    def _load_history(self):
        if self.history_file.exists():
            with open(self.history_file, 'r', encoding='utf-8') as f:
                self.history = json.load(f)
        else:
            self.history = []

    def _save_history(self, entry: Dict):
        self.history.append(entry)
        # 保留最近100条
        if len(self.history) > 100:
            self.history = self.history[-100:]
        with open(self.history_file, 'w', encoding='utf-8') as f:
            json.dump(self.history, f, ensure_ascii=False, indent=2)

    def sync_to_kunpeng(self) -> bool:
        """同步本地配置到鲲鹏服务器"""
        if self.mode == "rsync":
            cmd = [
                "rsync", "-avz", "--delete",
                str(self.local_dir) + "/",
                f"{self.user}@{self.host}:{self.remote_path}"
            ]
        elif self.mode == "scp":
            cmd = [
                "scp", "-r",
                str(self.local_dir),
                f"{self.user}@{self.host}:{self.remote_path}"
            ]
        else:
            logger.error("不支持的同步模式")
            return False

        try:
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
            if result.returncode == 0:
                logger.info(f"✅ 同步成功到鲲鹏: {self.host}:{self.remote_path}")
                self._save_history({
                    "time": datetime.now().isoformat(),
                    "status": "success",
                    "mode": self.mode,
                    "host": self.host,
                    "path": self.remote_path
                })
                return True
            else:
                logger.error(f"❌ 同步失败: {result.stderr}")
                self._save_history({
                    "time": datetime.now().isoformat(),
                    "status": "failed",
                    "mode": self.mode,
                    "error": result.stderr[:200]
                })
                return False
        except Exception as e:
            logger.error(f"❌ 同步异常: {e}")
            return False

    def sync_from_kunpeng(self) -> bool:
        """从鲲鹏服务器拉取最新配置（用于多节点同步）"""
        if self.mode == "rsync":
            cmd = [
                "rsync", "-avz", "--delete",
                f"{self.user}@{self.host}:{self.remote_path}/",
                str(self.local_dir) + "/"
            ]
        elif self.mode == "scp":
            cmd = [
                "scp", "-r",
                f"{self.user}@{self.host}:{self.remote_path}",
                str(self.local_dir.parent)  # 父目录
            ]
        else:
            logger.error("不支持的同步模式")
            return False

        try:
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
            if result.returncode == 0:
                logger.info(f"✅ 从鲲鹏拉取成功: {self.host}")
                # 重新加载配置
                return True
            else:
                logger.error(f"❌ 拉取失败: {result.stderr}")
                return False
        except Exception as e:
            logger.error(f"❌ 拉取异常: {e}")
            return False

    def get_history(self) -> List[Dict]:
        return self.history

# ============================================================
# 路由器服务（API + 自动同步）
# ============================================================

class RouterService:
    """路由器总服务，支持API和后台任务"""

    def __init__(self):
        self.config_mgr = ConfigManager()
        self.learn_mgr = LearnManager()
        self.engine = RouterEngine(self.config_mgr, self.learn_mgr)
        self.sync_mgr = SyncManager()
        self.running = False
        self.auto_sync_thread = None

    def route(self, user_input: str, context: Dict = None) -> Dict:
        """路由接口"""
        prompt, route = self.engine.route(user_input, context)
        return {
            "system_prompt": prompt,
            "route": route.to_dict() if route else None,
            "matched": route is not None,
            "timestamp": datetime.now().isoformat()
        }

    def add_route(self, name: str, triggers: List[str], system_prompt: str,
                  patterns: List[str] = None, tools: List[str] = None,
                  priority: int = 0) -> Dict:
        """添加或更新路由"""
        route = Route(
            name=name,
            triggers=triggers,
            patterns=patterns or [],
            system_prompt=system_prompt,
            tools=tools or [],
            priority=priority,
            enabled=True
        )
        self.config_mgr.add_route(route)
        self.engine.clear_cache()
        return {"status": "success", "name": name}

    def remove_route(self, name: str) -> Dict:
        self.config_mgr.remove_route(name)
        self.engine.clear_cache()
        return {"status": "success", "name": name}

    def learn_suggestions(self, threshold: int = None) -> List[Dict]:
        """获取学习建议"""
        threshold = threshold or self.config_mgr.config.learn_threshold
        return self.learn_mgr.suggest_new_routes(threshold)

    def apply_suggestion(self, suggestion: Dict) -> Dict:
        """应用一条学习建议（自动生成路由）"""
        name = suggestion.get("suggested_name")
        triggers = suggestion.get("suggested_triggers", [])
        input_text = suggestion.get("input", "")
        # 构建一个默认的提示词模板
        prompt = f"你是龍魂助手，专门处理『{input_text}』相关的问题。请根据用户输入提供专业、清晰的回答。"
        self.add_route(
            name=name,
            triggers=triggers,
            system_prompt=prompt,
            priority=5
        )
        self.learn_mgr.mark_resolved(input_text, name)
        return {"status": "success", "name": name, "triggers": triggers}

    def sync(self, direction: str = "to") -> bool:
        """同步到鲲鹏或从鲲鹏拉取"""
        if direction == "to":
            return self.sync_mgr.sync_to_kunpeng()
        else:
            return self.sync_mgr.sync_from_kunpeng()

    def start_auto_sync(self, interval: int = 3600):
        """启动自动同步后台线程"""
        if self.running:
            logger.warning("服务已在运行")
            return
        self.running = True

        def sync_loop():
            while self.running:
                time.sleep(interval)
                if self.running:
                    logger.info("⏰ 自动同步触发")
                    self.sync(direction="to")

        self.auto_sync_thread = threading.Thread(target=sync_loop, daemon=True)
        self.auto_sync_thread.start()
        logger.info(f"🔄 自动同步已启动 (间隔 {interval}秒)")

    def stop(self):
        self.running = False
        if self.auto_sync_thread:
            self.auto_sync_thread.join(timeout=5)

    def stats(self) -> Dict:
        return {
            "router": self.engine.stats(),
            "config_version": self.config_mgr.config.version,
            "sync_history": self.sync_mgr.get_history()[-5:],
            "backups": len(list(BACKUP_DIR.glob("routes_*.yaml")))
        }

# ============================================================
# 服务端API（使用FastAPI，可选）
# ============================================================

def run_api_server(port: int = 9630, host: str = "0.0.0.0"):
    """启动HTTP API服务"""
    try:
        from fastapi import FastAPI, HTTPException, Depends
        from pydantic import BaseModel
        import uvicorn
    except ImportError:
        logger.error("请安装 fastapi 和 uvicorn: pip install fastapi uvicorn")
        return

    app = FastAPI(title="龍魂提示词路由器", version=VERSION)

    # 全局服务实例
    service = RouterService()

    # 请求模型
    class RouteRequest(BaseModel):
        input: str
        context: Optional[Dict] = None

    class AddRouteRequest(BaseModel):
        name: str
        triggers: List[str]
        system_prompt: str
        patterns: Optional[List[str]] = []
        tools: Optional[List[str]] = []
        priority: int = 0

    @app.get("/health")
    def health():
        return {"status": "ok", "version": VERSION, "dna": DNA}

    @app.post("/route")
    def route(req: RouteRequest):
        result = service.route(req.input, req.context)
        return {"code": 0, "data": result}

    @app.post("/route/add")
    def add_route(req: AddRouteRequest):
        result = service.add_route(
            name=req.name,
            triggers=req.triggers,
            system_prompt=req.system_prompt,
            patterns=req.patterns,
            tools=req.tools,
            priority=req.priority
        )
        return {"code": 0, "data": result}

    @app.delete("/route/{name}")
    def delete_route(name: str):
        result = service.remove_route(name)
        return {"code": 0, "data": result}

    @app.get("/learn/suggestions")
    def get_suggestions():
        suggestions = service.learn_suggestions()
        return {"code": 0, "data": suggestions}

    @app.post("/learn/apply")
    def apply_suggestion(suggestion: Dict):
        result = service.apply_suggestion(suggestion)
        return {"code": 0, "data": result}

    @app.post("/sync/to")
    def sync_to_kunpeng():
        ok = service.sync(direction="to")
        return {"code": 0 if ok else 1, "data": {"success": ok}}

    @app.post("/sync/from")
    def sync_from_kunpeng():
        ok = service.sync(direction="from")
        return {"code": 0 if ok else 1, "data": {"success": ok}}

    @app.get("/stats")
    def get_stats():
        return {"code": 0, "data": service.stats()}

    @app.get("/backups")
    def get_backups():
        history = service.config_mgr.get_history()
        return {"code": 0, "data": history}

    logger.info(f"🚀 启动API服务: http://{host}:{port}")
    logger.info(f"📖 文档: http://{host}:{port}/docs")
    uvicorn.run(app, host=host, port=port)

# ============================================================
# 命令行接口
# ============================================================

def main():
    parser = argparse.ArgumentParser(
        description="🐉 龍魂提示词路由器 v1.0",
        epilog="示例: python lh_prompt_router.py --route '审计'"
    )

    parser.add_argument("--route", type=str, help="测试路由匹配（输入文本）")
    parser.add_argument("--context", type=str, help="上下文JSON（可选）")
    parser.add_argument("--serve", action="store_true", help="启动API服务")
    parser.add_argument("--port", type=int, default=9630, help="API端口")
    parser.add_argument("--sync", action="store_true", help="同步到鲲鹏")
    parser.add_argument("--sync-from", action="store_true", help="从鲲鹏拉取")
    parser.add_argument("--learn", action="store_true", help="查看学习建议")
    parser.add_argument("--apply-learn", type=str, help="应用学习建议（输入待处理文本）")
    parser.add_argument("--add-route", action="store_true", help="交互式添加路由")
    parser.add_argument("--list", action="store_true", help="列出所有路由")
    parser.add_argument("--status", action="store_true", help="显示状态")
    parser.add_argument("--rollback", type=str, help="回滚到指定备份（输入时间戳）")
    parser.add_argument("--info", action="store_true", help="显示版本信息")

    args = parser.parse_args()

    if args.info:
        print(f"""
🐉 龍魂提示词路由器 v{VERSION}
DNA: {DNA}
CONFIRM: {CONFIRM}

功能:
  - 动态路由匹配
  - 自我学习迭代
  - 鲲鹏服务器同步

使用方法:
  python lh_prompt_router.py --route "审计协议"
  python lh_prompt_router.py --serve --port 9630
  python lh_prompt_router.py --sync
  python lh_prompt_router.py --learn
  python lh_prompt_router.py --status
        """)
        return

    # 初始化服务
    service = RouterService()

    if args.serve:
        run_api_server(port=args.port)
        return

    if args.sync:
        ok = service.sync(direction="to")
        print("✅ 同步成功" if ok else "❌ 同步失败")
        return

    if args.sync_from:
        ok = service.sync(direction="from")
        print("✅ 拉取成功" if ok else "❌ 拉取失败")
        return

    if args.learn:
        suggestions = service.learn_suggestions()
        if suggestions:
            print("📚 学习建议 (高频未匹配输入):")
            for s in suggestions:
                print(f"  - '{s['input']}' (出现{s['count']}次) → 建议路由名: {s['suggested_name']}")
                print(f"    候选触发词: {', '.join(s['suggested_triggers'])}")
        else:
            print("✅ 暂无学习建议")
        return

    if args.apply_learn:
        # 查找匹配的未匹配记录
        pending = service.learn_mgr.get_pending()
        matched = None
        for item in pending:
            if item["input"] == args.apply_learn:
                matched = item
                break
        if matched:
            # 自动生成建议并应用
            suggestions = service.learn_suggestions()
            for s in suggestions:
                if s["input"] == args.apply_learn:
                    result = service.apply_suggestion(s)
                    print(f"✅ 已应用学习建议: {result['name']}")
                    break
        else:
            print("❌ 未找到匹配的待处理记录")
        return

    if args.add_route:
        print("🐉 交互式添加路由")
        name = input("路由名称: ").strip()
        triggers = input("触发词（逗号分隔）: ").strip().split(",")
        triggers = [t.strip() for t in triggers if t.strip()]
        system_prompt = input("System Prompt: ").strip()
        priority = int(input("优先级 (0-10): ").strip() or "0")
        result = service.add_route(name, triggers, system_prompt, priority=priority)
        print(f"✅ 添加成功: {result['name']}")
        return

    if args.list:
        routes = service.config_mgr.config.routes
        print(f"📋 当前路由列表 ({len(routes)}条):")
        for r in routes:
            status = "✅" if r.enabled else "⛔"
            print(f"  {status} {r.name} (优先级{r.priority}) -> {r.triggers[:3]}")
        return

    if args.rollback:
        service.config_mgr.rollback(args.rollback)
        print(f"⏪ 已回滚到: {args.rollback}")
        return

    if args.status:
        stats = service.stats()
        print("📊 路由器状态:")
        print(f"  总路由数: {stats['router']['total_routes']}")
        print(f"  已启用: {stats['router']['enabled_routes']}")
        print(f"  待学习记录: {stats['router']['pending_learn']}")
        print(f"  缓存大小: {stats['router']['cache_size']}")
        print(f"  配置版本: {stats['config_version']}")
        print(f"  备份数量: {stats['backups']}")
        if stats['sync_history']:
            last_sync = stats['sync_history'][-1]
            print(f"  最近同步: {last_sync['time']} ({last_sync['status']})")
        return

    if args.route:
        context = json.loads(args.context) if args.context else {}
        result = service.route(args.route, context)
        print("🐉 路由结果:")
        print(f"  输入: {args.route}")
        print(f"  匹配: {'✅' if result['matched'] else '❌'}")
        if result['route']:
            print(f"  路由: {result['route']['name']}")
            print(f"  触发词: {', '.join(result['route']['triggers'])}")
        print(f"  System Prompt:\n{result['system_prompt'][:200]}...")
        return

    parser.print_help()

if __name__ == "__main__":
    main()

---

🔧 配置文件示例 (routes.yaml)

version: "1.0.0"
default_prompt: "你是龍魂助手，处理通用问题。"
learn_threshold: 3
sync_interval: 3600
routes:
  - name: "审计"
    triggers:
      - "审计"
      - "检查协议"
      - "分析违规"
      - "/audit"
    patterns:
      - "(?i)(audit|compliance)"
    system_prompt: |
      你是龍魂审计师，负责检测平台规则与华夏法则的冲突。
      你要严格按照三色判定（🟢🟡🔴）输出审计结果。
      始终引用具体法条，保持客观中立。
    tools:
      - "扫描条款"
      - "匹配法律"
      - "三色判定"
    priority: 10
    enabled: true
    metadata:
      author: "UID9622"
      category: "治理"

  - name: "卦象"
    triggers:
      - "卦"
      - "易经"
      - "占卜"
      - "/iching"
    patterns: []
    system_prompt: |
      你是龍魂易学师，用易经作为复杂系统动态平衡的隐喻模型。
      根据输入推演卦象，给出相应的启示和建议。
      强调这不是预测，而是系统状态的映射。
    tools:
      - "取卦"
      - "解卦"
      - "推演"
    priority: 8
    enabled: true

  - name: "代码生成"
    triggers:
      - "写代码"
      - "生成脚本"
      - "/code"
      - "实现"
    patterns:
      - "(?i)(def |class |import |python|bash)"
    system_prompt: |
      你是龍魂架构师，根据用户需求生成符合四层命名的代码。
      输出包含完整的错误处理、日志、命令行接口。
      代码必须包含DNA追溯码。
    tools:
      - "生成Python"
      - "生成Bash"
      - "生成文档"
    priority: 6
    enabled: true

  - name: "同步"
    triggers:
      - "同步"
      - "更新鲲鹏"
      - "/sync"
    system_prompt: |
      你是龍魂同步管理员，负责将本地配置同步到鲲鹏服务器。
      检查网络和权限，执行rsync/scp，并记录同步状态。
    tools:
      - "rsync"
      - "scp"
      - "状态检查"
    priority: 5
    enabled: true

---

🚀 集成到现有系统 (lh 命令)

在 ~/bin/lh 中添加：

"router"|"prompt")
    python3 ~/.龙魂/prompt_router/lh_prompt_router.py "$@"
    ;;

然后可以使用：

lh router --route "审计"
lh router --serve --port 9630
lh router --sync
lh router --list

---

🔄 自我迭代工作流

---

📊 与鲲鹏服务器的同步机制

同步方式	命令	说明
手动同步	lh router --sync	立即将本地配置推送到鲲鹏
手动拉取	lh router --sync-from	从鲲鹏拉取最新配置
自动同步	启动服务时 --serve 后台线程	每 sync_interval 秒自动推送
双向一致性	鲲鹏作为主节点，多个本地实例定期拉取	保证多端路由表一致

---

🧬 扩展点（你未提及但逻辑应包含的）

扩展点	说明	实现位置
版本回滚	配置更新出问题时快速恢复	ConfigManager.rollback()
缓存失效	路由更新后自动清空匹配缓存	RouterEngine.clear_cache()
权限控制	API接口可加API Key认证（预留）	FastAPI依赖注入
监控指标	匹配率、学习队列长度、同步成功率	stats() + Prometheus（预留）
多路由复合	一条输入可匹配多个路由时合并提示词	扩展 RouterEngine.route()
A/B测试	路由可设置版本，支持灰度发布	Route.version 字段
审计日志	所有路由匹配记录带DNA追溯码	继承现有龍魂日志体系

---

📁 文件结构

~/.龙魂/prompt_router/
├── routes.yaml           # 主配置
├── learn_queue.yaml      # 待学习记录
├── backups/
│   ├── routes_20260803_120000.yaml
│   └── routes_20260802_180000.yaml
├── logs/
│   └── router_20260803.log
├── sync_history.json     # 同步记录
└── lh_prompt_router.py   # 主脚本

---

🧬 DNA追溯

DNA: #龍芯⚡️丙午·癸未·庚辰·ROUTER-ENGINE-v1.0-UID9622
CONFIRM: #CONFIRM🌌9622-ONLY-ONCE🧬LK9X-772Z
SEAL: #ZHUGEXIN⚡️2025-🇨🇳🐉⚖️♠️🧚🏼‍♀️❤️♾️-DEVICE-BIND-SOUL

---

一句话总结：动态路由匹配 + 自我学习迭代 + 鲲鹏双向同步 = 龍魂提示词路由器，会自己长大的智能路由系统。 🐉