106 lines
3.0 KiB
Python
106 lines
3.0 KiB
Python
|
"""Hook 配置持久化 - Phase 7.3"""
|
|||
|
|
|
|||
|
|
import json
|
|||
|
|
import os
|
|||
|
|
from dataclasses import asdict, dataclass
|
|||
|
|
from typing import Any
|
|||
|
|
|
|||
|
|
from app.agents.tools.hooks.manager import get_hook_manager
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class HookConfigEntry:
|
|||
|
|
"""Hook 配置条目"""
|
|||
|
|
|
|||
|
|
name: str
|
|||
|
|
hook_type: str
|
|||
|
|
enabled: bool
|
|||
|
|
tool_names: list[str] | None = None
|
|||
|
|
categories: list[str] | None = None
|
|||
|
|
priority: int = 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
class HookConfigPersistence:
|
|||
|
|
"""Hook 配置持久化"""
|
|||
|
|
|
|||
|
|
def __init__(self, config_path: str | None = None):
|
|||
|
|
"""
|
|||
|
|
Args:
|
|||
|
|
config_path: 配置文件路径，None 则使用默认路径
|
|||
|
|
"""
|
|||
|
|
if config_path is None:
|
|||
|
|
config_path = os.path.join(
|
|||
|
|
os.path.dirname(__file__), "..", "..", "..", "..", "config", "hooks.json"
|
|||
|
|
)
|
|||
|
|
self.config_path = config_path
|
|||
|
|
|
|||
|
|
def load_config(self) -> list[HookConfigEntry]:
|
|||
|
|
"""从文件加载 Hook 配置"""
|
|||
|
|
if not os.path.exists(self.config_path):
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
with open(self.config_path, "r", encoding="utf-8") as f:
|
|||
|
|
data = json.load(f)
|
|||
|
|
return [HookConfigEntry(**entry) for entry in data]
|
|||
|
|
except Exception:
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
def save_config(self, entries: list[HookConfigEntry]) -> bool:
|
|||
|
|
"""保存 Hook 配置到文件"""
|
|||
|
|
try:
|
|||
|
|
os.makedirs(os.path.dirname(self.config_path), exist_ok=True)
|
|||
|
|
with open(self.config_path, "w", encoding="utf-8") as f:
|
|||
|
|
json.dump([asdict(e) for e in entries], f, indent=2, ensure_ascii=False)
|
|||
|
|
return True
|
|||
|
|
except Exception:
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
def apply_config(self) -> int:
|
|||
|
|
"""应用配置到 HookManager
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
应用的 Hook 数量
|
|||
|
|
"""
|
|||
|
|
from app.agents.tools.hooks.types import HookType
|
|||
|
|
|
|||
|
|
manager = get_hook_manager()
|
|||
|
|
entries = self.load_config()
|
|||
|
|
count = 0
|
|||
|
|
|
|||
|
|
for entry in entries:
|
|||
|
|
if entry.enabled:
|
|||
|
|
from app.agents.tools.hooks.types import HookDefinition, HookTrigger
|
|||
|
|
|
|||
|
|
trigger = HookTrigger(
|
|||
|
|
tool_names=entry.tool_names,
|
|||
|
|
categories=entry.categories,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# 创建空的 handler，只是注册配置
|
|||
|
|
hook_def = HookDefinition(
|
|||
|
|
name=entry.name,
|
|||
|
|
hook_type=HookType(entry.hook_type),
|
|||
|
|
trigger=trigger,
|
|||
|
|
handler=lambda ctx, *args: ctx,
|
|||
|
|
priority=entry.priority,
|
|||
|
|
enabled=True,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
manager.register(hook_def)
|
|||
|
|
count += 1
|
|||
|
|
|
|||
|
|
return count
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 全局单例
|
|||
|
|
_persistence: HookConfigPersistence | None = None
|
|||
|
|
|
|||
|
|
|
|||
|
|
def get_hook_config_persistence() -> HookConfigPersistence:
|
|||
|
|
"""获取全局 Hook 配置持久化实例"""
|
|||
|
|
global _persistence
|
|||
|
|
if _persistence is None:
|
|||
|
|
_persistence = HookConfigPersistence()
|
|||
|
|
return _persistence
|