airport-wiki/concepts/knowledge-management/automation-hooks.md

---
title: 自动化钩子与事件驱动架构
created: 2026-04-13
updated: 2026-04-15
type: maintains
tags: [knowledge-management, automation, event-hooks, operations]
sources_count: 1
confidence: 0.9
last_confirmed: 2026-04-15
status: active
relationships: 
- target: hybrid-search.md
detail: "嵌入和索引更新"
---
# ⚡ 自动化钩子与事件驱动架构

基于 **LLM Wiki v2** 的事件驱动维护系统，为机场智能化工程 wiki 提供自动化知识管理。通过事件钩子响应 wiki 操作，减少手动维护负担。

> **核心目标**：将手动知识维护转变为事件驱动的自动化流程，确保 wiki 内容的新鲜度、一致性和质量。

---

## 🏗️ 事件架构总览

### 事件类型与触发器
| 事件类型 | 触发器 | 触发条件 | 响应延迟 |
|----------|--------|----------|----------|
| **来源新增** | 文件系统监视 | `raw/` 中新增 `.md` 文件 | 即时 (15s) |
| **页面创建** | `write_file()` 调用 | `concepts/`, `entities/` 等目录 | 即时 (5s) |
| **页面更新** | `patch()` 调用 | 现有页面内容修改 | 即时 (5s) |
| **页面归档** | 文件移动至 `_archive/` | 手动操作或自动 supersede | 即时 (5s) |
| **用户查询** | `web_search()` 或 `search_files()` | 搜索操作 | 异步 (<60s) |
| **定时任务** | cron 调度器 | 每日/每周/每月 | 指定时间 |

### 自动化钩子执行顺序
```
新来源 → on_new_source() → 来源解析 → 实体提取 → 页面创建/更新
                                                      ↓
页面创建/更新 → on_page_change() → 关系更新 → 嵌入更新 → 索引更新
                                                      ↓
定时任务 → cron_daily/weekly/monthly() → 质量检查 → 置信度衰减
                                                      ↓
用户查询 → on_user_query() → 结果记录 → 潜在答案生成 → 反馈学习
```

---

## 🔧 主要钩子实现

### 1️⃣ `on_new_source()` - 新来源自动摄入
```python
def on_new_source(source_path: str):
    """
    处理 raw/ 目录中的新来源文件
    1. 解析来源内容
    2. 提取实体和事实
    3. 创建/更新 wiki 页面
    4. 更新相关索引
    """
    
    # 1. 读取并解析来源
    content = read_file(source_path)
    metadata = extract_metadata(content)  # 作者、日期、类型等
    
    # 2. 提取实体和事实
    entities = extract_entities(content)
    facts = extract_facts(content, entities)
    
    # 3. 更新现有页面或创建新页面
    for fact in facts:
        target_page = find_or_create_page(fact.topic)
        
        # 检查是否有冲突
        conflict = check_conflict(target_page.content, fact.content)
        
        if conflict:
            # 触发 supersession 流程
            supersede_page(target_page, fact.content, source_path)
        else:
            # 追加新事实
            update_page(target_page, fact.content, source_path)
    
    # 4. 更新嵌入和图谱
    trigger_embedding_update()
    trigger_graph_reconciliation()
    
    # 记录日志
    log_event("source_ingested", {
        "source": source_path,
        "entities_extracted": len(entities),
        "facts_added": len(facts),
        "timestamp": now()
    })
```

**机场场景示例**：
```
事件: 新增 raw/articles/shenzhen-airport-smart-gating-2026.md
响应:
  1. 解析文章：深圳机场2026年智能登机口升级
  2. 提取实体：深圳机场、SITA、生物识别走廊
  3. 更新页面：
     - concepts/smart-gating.md → 添加深圳案例
     - entities/shenzhen-airport.md → 更新智能登机口信息
  4. 更新关系：深圳机场 → uses → 生物识别走廊
```

### 2️⃣ `on_page_change()` - 页面变更处理
```python
def on_page_change(page_path: str, change_type: str, old_content: Optional[str] = None):
    """
    处理页面创建、更新、删除
    参数:
    - change_type: "create" | "update" | "delete" | "archive"
    - old_content: 仅 update 时提供
    """
    
    if change_type == "create":
        # 新页面：初始化嵌入和关系
        embedding = generate_embedding(page_path)
        save_embedding(page_path, embedding)
        
        # 提取关系并更新图谱
        relationships = extract_relationships(page_path)
        update_knowledge_graph(page_path, relationships)
        
    elif change_type == "update":
        # 页面更新：检查语义变化
        old_embedding = load_embedding(page_path)
        new_embedding = generate_embedding(page_path)
        
        similarity = cosine_similarity(old_embedding, new_embedding)
        
        if similarity < 0.7:  # 语义显著变化
            # 重新计算相关页面的嵌入
            trigger_related_embeddings_update(page_path)
            
            # 更新所有引用该页面的关系
            update_incoming_relationships(page_path)
    
    elif change_type in ["delete", "archive"]:
        # 页面删除/归档：清理相关数据
        remove_embedding(page_path)
        remove_from_knowledge_graph(page_path)
        
        # 更新引用（设置 superseded_by 或删除链接）
        update_references_to_page(page_path, change_type)
    
    # 更新搜索索引
    update_search_index(page_path, change_type)
    
    log_event("page_changed", {
        "page": page_path,
        "type": change_type,
        "semantic_change": similarity if change_type == "update" else None,
        "timestamp": now()
    })
```

### 3️⃣ `cron_weekly()` - 每周维护任务
```python
def cron_weekly():
    """
    每周日自动执行的维护任务
    1. 完整性检查 (lint)
    2. 置信度衰减和更新
    3. 嵌入重新生成
    4. 性能分析
    """
    
    print("=== 每周维护任务开始 ===")
    start_time = now()
    
    # 1. 运行完整性检查
    lint_report = run_lint_check()
    
    # 自动修复可修复的问题
    auto_fixed = lint_report.auto_fix()
    
    # 记录需要手动干预的问题
    manual_tasks = lint_report.get_manual_tasks()
    
    # 2. 置信度衰减
    decayed_pages = decay_confidence_scores()
    
    # 3. 嵌入重新生成（全量）
    pages_updated = regenerate_all_embeddings()
    
    # 4. 搜索索引重建
    rebuild_search_index()
    
    # 5. 性能分析
    performance_report = analyze_search_performance()
    
    # 6. 生成维护报告
    report = generate_maintenance_report({
        "duration_seconds": (now() - start_time).total_seconds(),
        "lint_fixed": auto_fixed,
        "lint_manual": len(manual_tasks),
        "pages_decayed": len(decayed_pages),
        "embeddings_regenerated": pages_updated,
        "search_metrics": performance_report.metrics,
        "timestamp": now()
    })
    
    # 保存报告
    save_report(report, "weekly-maintenance")
    
    # 如有需要手动干预的问题，发送通知
    if manual_tasks:
        notify_maintainer("手动维护任务待处理", manual_tasks)
    
    print(f"=== 每周维护任务完成，耗时 {report.duration_seconds}s ===")
    
    return report
```

### 4️⃣ `on_user_query()` - 查询响应与学习
```python
def on_user_query(query: str, results: List[str], user_feedback: Optional[Dict] = None):
    """
    处理用户搜索查询
    1. 记录查询模式
    2. 潜在答案生成
    3. 质量评估和反馈学习
    """
    
    # 1. 查询分类和记录
    query_type = classify_query(query)
    
    log_search_event({
        "query": query,
        "type": query_type,
        "results_count": len(results),
        "user_id": get_user_id(),  # 匿名或会话ID
        "timestamp": now()
    })
    
    # 2. 检查是否需要生成新答案
    if should_generate_answer(query, results):
        answer = generate_potential_answer(query, results)
        
        # 评估答案质量
        quality_score = evaluate_answer_quality(answer, query, results)
        
        if quality_score > 0.8:  # 高质量答案
            # 自动创建/更新查询页面
            create_query_page(query, answer, quality_score)
            
            log_event("answer_generated", {
                "query": query,
                "answer_page": f"queries/{slugify(query)}.md",
                "quality_score": quality_score,
                "timestamp": now()
            })
    
    # 3. 处理用户反馈（如有）
    if user_feedback:
        process_user_feedback(query, results, user_feedback)
        
        # 更新搜索排名权重
        update_search_weights(query_type, user_feedback)
    
    # 4. 查询模式分析
    analyze_query_patterns(query, results)
    
    return {
        "logged": True,
        "query_type": query_type,
        "potential_answer_generated": should_generate_answer(query, results),
        "feedback_processed": bool(user_feedback)
    }
```

---

## ⏰ 定时任务调度

### 每日任务 (`cron_daily`)
```python
SCHEDULE = {
    "daily": {
        "time": "02:30",  # 凌晨执行，避免影响使用
        "tasks": [
            "verify_recent_changes",  # 检查24小时内变更
            "update_recommendations",  # 更新推荐系统
            "clean_temp_files",       # 清理临时文件
            "backup_incremental"     # 增量备份
        ]
    }
}

def cron_daily():
    """每日凌晨执行的任务"""
    tasks = [
        # 1. 验证最近变更
        verify_recent_changes(since=datetime.now() - timedelta(days=1)),
        
        # 2. 更新个性化推荐
        update_recommendations(),
        
        # 3. 清理临时文件
        clean_temp_files(max_age=timedelta(days=7)),
        
        # 4. 增量备份
        backup_incremental(target="s3://wiki-backups/daily/")
    ]
    
    return execute_tasks(tasks, name="daily_maintenance")
```

### 每周任务 (`cron_weekly`)
```python
def cron_weekly():
    """每周日执行的全量维护"""
    return {
        "lint": run_lint_check(),
        "embeddings": regenerate_all_embeddings(),
        "confidence": decay_confidence_scores(),
        "index": rebuild_search_index(),
        "report": generate_weekly_report()
    }
```

### 每月任务 (`cron_monthly`)
```python
def cron_monthly():
    """每月1日执行的深度维护"""
    return {
        "archival": archive_stale_content(older_than=timedelta(days=180)),
        "model_evaluation": evaluate_embedding_models(),
        "capacity_planning": analyze_growth_trends(),
        "security_audit": run_security_checks(),
        "comprehensive_report": generate_monthly_report()
    }
```

---

## 🚀 实施部署

### 阶段 1：基础钩子（当前）
- ✅ `on_page_change()` 记录至日志
- ✅ 新增来源手动触发处理
- 🔄 定期 lint 检查（手动）

### 阶段 2：自动化管道（1-2周）
- 🔄 文件系统监视：`raw/` 新增自动触发
- 🔄 页面变更自动更新嵌入和关系
- 🔄 每周自动维护脚本
- 🔄 搜索结果记录与分析

### 阶段 3：高级自动化（1个月）
- 🔄 智能答案生成（质量阈值 >0.8）
- 🔄 自适应权重调整（基于用户反馈）
- 🔄 异常检测和自动修复
- 🔄 多环境部署（开发/测试/生产）

### 阶段 4：智能运维（未来）
- 🔄 预测性维护（基于历史模式）
- 🔄 A/B 测试搜索算法
- 🔄 跨wiki知识同步
- 🔄 故障自愈能力

---

## 🔧 技术实现细节

### 钩子注册机制
```python
class HookRegistry:
    """事件钩子注册中心"""
    
    def __init__(self):
        self.hooks = defaultdict(list)
    
    def register(self, event_type: str, callback: Callable, priority: int = 0):
        """注册钩子"""
        self.hooks[event_type].append({
            "callback": callback,
            "priority": priority
        })
        self.hooks[event_type].sort(key=lambda x: x["priority"])
    
    def trigger(self, event_type: str, **kwargs):
        """触发事件"""
        for hook in self.hooks.get(event_type, []):
            try:
                hook["callback"](**kwargs)
            except Exception as e:
                log_error(f"钩子执行失败: {event_type}", e)

# 全局钩子注册器
hooks = HookRegistry()

# 注册示例
hooks.register("page_created", on_page_change, priority=10)
hooks.register("source_added", on_new_source, priority=5)
```

### 文件系统监视
```python
import watchdog
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class WikiFileHandler(FileSystemEventHandler):
    """监视 raw/ 目录的变更"""
    
    def on_created(self, event):
        if event.is_directory:
            return
        
        path = event.src_path
        if path.startswith("/raw/") and path.endswith(".md"):
            # 触发来源处理钩子
            hooks.trigger("source_added", source_path=path)
    
    def on_modified(self, event):
        if event.is_directory:
            return
        
        path = event.src_path
        if not path.startswith("/raw/"):
            # 触发页面变更钩子
            hooks.trigger("page_changed", page_path=path, change_type="update")

# 启动监视器
observer = Observer()
observer.schedule(WikiFileHandler(), "/path/to/wiki", recursive=True)
observer.start()
```

### 定时任务调度器
```python
import schedule
import time

def setup_scheduler():
    """配置定时任务"""
    
    # 每日凌晨任务
    schedule.every().day.at("02:30").do(cron_daily)
    
    # 每周日任务  
    schedule.every().sunday.at("03:00").do(cron_weekly)
    
    # 每月1日任务
    schedule.every().month.at("04:00").do(cron_monthly)
    
    print("定时任务已配置")
    
    # 运行调度器（后台线程）
    import threading
    
    def run_scheduler():
        while True:
            schedule.run_pending()
            time.sleep(60)  # 每分钟检查一次
    
    thread = threading.Thread(target=run_scheduler, daemon=True)
    thread.start()

# 应用启动时调用
setup_scheduler()
```

---

## 📊 监控与告警

### 关键指标监控
| 指标 | 阈值 | 告警级别 | 响应动作 |
|------|------|----------|----------|
| **处理失败率** | >5% | 警告 | 检查日志，重启服务 |
| **嵌入更新延迟** | >24h | 警告 | 手动触发嵌入生成 |
| **页面冲突数量** | >10 | 警告 | 审核冲突内容 |
| **搜索查询失败** | >20% | 严重 | 检查搜索索引 |
| **磁盘使用率** | >80% | 警告 | 清理或扩容 |

### 告警规则示例
```yaml
alerts:
  - name: "high_failure_rate"
    condition: "rate(failed_hooks_total[5m]) / rate(hooks_total[5m]) > 0.05"
    severity: "warning"
    description: "钩子执行失败率超过5%"
    actions: ["send_slack", "create_jira"]
  
  - name: "search_degradation"  
    condition: "search_response_time_p95 > 3000"
    severity: "critical"
    description: "搜索P95响应时间超过3秒"
    actions: ["page_oncall", "rollback_search"]
```

---

## 🔄 故障恢复流程

### 常见故障场景
1. **钩子执行失败**
   ```bash
   # 1. 查看错误日志
   tail -f /var/log/wiki/hooks.log
   
   # 2. 暂时禁用问题钩子
   disable_hook("on_page_change", "problematic_callback")
   
   # 3. 手动执行受影响操作
   run_manual_cleanup()
   ```

2. **嵌入生成中断**
   ```bash
   # 1. 检查嵌入存储完整性
   verify_embeddings_integrity()
   
   # 2. 重新生成受影响页面
   regenerate_embeddings_for_pages(since="2026-04-10")
   
   # 3. 重建搜索索引
   rebuild_search_index()
   ```

3. **关系图谱不一致**
   ```python
   # 自动一致性检查
   def reconcile_knowledge_graph():
       # 1. 检测孤立实体
       orphans = find_orphaned_entities()
       
       # 2. 检查关系对称性
       mismatches = validate_relationship_symmetry()
       
       # 3. 修复不一致
       fix_inconsistencies(orphans + mismatches)
       
       return {"fixed": len(orphans + mismatches)}
   ```

---

## 📚 相关文档

- [[knowledge-management/memory-lifecycle.md]] - 置信度衰减和整合机制
- [[knowledge-management/knowledge-graph.md]] - 实体关系自动提取
- [[hybrid-search.md]] - 搜索结果记录和权重调整
- [[wiki-backup-recovery.md]] - 备份和恢复流程
- [[performance-monitoring.md]] - 系统性能监控

---

> **状态**: 当前实现基础钩子记录。下一步：部署文件系统监视和定时任务。最后更新：2026-04-13。