zhiqiang feng
99d66a9601
- 新增知识管理系统核心文档 - automation-hooks.md: 自动化钩子和事件驱动架构 - knowledge-lifecycle.md: 知识生命周期和遗忘曲线 - quality-control.md: 质量控制和自我纠正机制 - llm-wiki-v2-reference.md: v2 参考文档 - 更新索引和架构 - index.md: 重构为目录导航，移除详细列表 - SCHEMA.md: 更新知识管理领域定义 - 新增辅助文档 - hybrid-search.md: 混合搜索系统 - systems-design/aodb-system-architecture.md: AODB系统架构 版本: LLM Wiki v2.1.0 特性: 置信度衰减、智能整合、质量验证、事件驱动自动化
17 KiB
17 KiB
title, created, updated, type, tags, confidence, sources_count, last_confirmed, status, relationships
| title | created | updated | type | tags | confidence | sources_count | last_confirmed | status | relationships | ||||||||||||||||||||||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 自动化钩子与事件驱动架构 | 2026-04-13 | 2026-04-13 | concept |
|
0.9 | 5 | 2026-04-13 | active |
|
⚡ 自动化钩子与事件驱动架构
基于 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() - 新来源自动摄入
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() - 页面变更处理
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() - 每周维护任务
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() - 查询响应与学习
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)
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)
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)
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知识同步
- 🔄 故障自愈能力
🔧 技术实现细节
钩子注册机制
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)
文件系统监视
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()
定时任务调度器
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% | 警告 | 清理或扩容 |
告警规则示例
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. 查看错误日志 tail -f /var/log/wiki/hooks.log # 2. 暂时禁用问题钩子 disable_hook("on_page_change", "problematic_callback") # 3. 手动执行受影响操作 run_manual_cleanup() -
嵌入生成中断
# 1. 检查嵌入存储完整性 verify_embeddings_integrity() # 2. 重新生成受影响页面 regenerate_embeddings_for_pages(since="2026-04-10") # 3. 重建搜索索引 rebuild_search_index() -
关系图谱不一致
# 自动一致性检查 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。