--- name: knowledge-federation-4b754c7f3be2 description: 知识共享框架。多Agent跨项目学习机制，规则版本管理，社群排行榜，冲突协调（本地优先/社群优先/合并/版本管理），实现OpenClaw系统的分布式演化。 model: minimax-portal/MiniMax-M2.7 effort: high --- # Knowledge Federation — 知识共享框架 ## 功能概述 将单个Agent的学习经验扩展到整个OpenClaw社群，实现跨项目、跨Agent的规则共享、版本管理、冲突协调，形成一个自主进化的分布式学习网络。 ## 核心功能 ### 1. 本地规则库 (LocalRuleRegistry) 每个Agent维护独立的本地规则库，持久化到 `.local-rules/` 目录： ```python registry = LocalRuleRegistry(workspace_dir) # 注册新规则 version = registry.register_rule( rule_id="check_funds_transfer", content={"condition": "amount > 1000", "require_approval": True}, effectiveness=85.0, # 初始效能评分 description="初始版本" ) # 检索规则 rule = registry.get_rule("check_funds_transfer") # 列出所有本地规则 all_rules = registry.list_rules() ``` **存储结构**： - 文件位置：`.local-rules/{rule_id}_{version_id}.json` - 内容：RuleVersion对象 (version_id, rule_id, parent_version, author_agent, timestamp, content, effectiveness_score, status, tags, breaking_changes) ### 2. 规则版本管理 (RuleVersion) 每条规则都有完整的版本链，支持演化追踪： ```python @dataclass class RuleVersion: version_id: str # 版本ID（8字符UUID） rule_id: str # 规则ID parent_version: Optional[str] # 父版本 (用于追踪演化) author_agent: str # 作者Agent ID timestamp: str # 创建时间 content: Dict # 规则实际内容 effectiveness_score: float # 效能评分 (0-100) status: str # draft / published / deprecated tags: List[str] # 标签 (security, finance, common, etc.) description: str # 版本说明 breaking_changes: List[str] # 破坏性改动说明 ``` **版本生命周期**： ``` draft (草稿) └→ 编辑、测试 └→ published (已发布) ├→ 保持活跃（高效能） ├→ deprecated (已废弃) └→ replaced (被取代) ``` ### 3. 冲突协调 (ConflictResolver) 多Agent更新同一规则时的智能冲突处理： ```python # 检测冲突 conflict = ConflictResolver.detect_conflicts(local_rule, community_rule) # 解决冲突（4种策略） resolver = ConflictResolver() resolved = resolver.resolve_conflict(conflict) ``` **4种冲突解决策略**： | 策略 | 说明 | 使用场景 | |------|------|---------| | `LOCAL_PRIORITY` | 保留本地版本 | 项目特定规则，本地自定义 | | `COMMUNITY_PRIORITY` | 使用社群版本 | 社群规则更优且更新 | | `MERGE` | 合并两个版本 | 互补的改进可以合并 | | `VERSION` | 比较效能选择 | 让数据决策（高效能优先） | **冲突解决示例**： ``` 本地规则 (v1): - 内容: {"timeout": 10s, "retry": 3} - 效能: 75分 社群规则 (v2): - 内容: {"timeout": 5s, "retry": 5, "cache": true} - 效能: 88分 策略: VERSION 或 COMMUNITY_PRIORITY ↓ 选择: v2 (效能88 > 75) ↓ 结果: 新主规则已更新，v1保存为历史版本 ``` ### 4. 社群排行榜 (CommunityLeaderboard) 实时追踪所有规则的效能排名、采纳人数、演化历史： ```python leaderboard = CommunityLeaderboard() # 添加规则 leaderboard.add_rule(community_rule) # 更新效能得分 leaderboard.update_effectiveness("rule_123", 82.5) # 记录被采纳 leaderboard.record_adoption("rule_123") # 获取排行前10 top_10 = leaderboard.get_top_rules(limit=10) # 输出排行榜 board = leaderboard.get_leaderboard() # [ # {"position": 1, "rule_id": "rule_123", "score": 88.5, "adoption_count": 15}, # {"position": 2, "rule_id": "rule_456", "score": 85.2, "adoption_count": 12}, # ... # ] ``` **排行榜特性**： - ✅ 实时计分（基于历史效能均值） - ✅ 采纳计数（有多少Agent使用） - ✅ 项目分类（按标签聚类） - ✅ 趋势追踪（效能历史曲线） ### 5. 知识共享系统 (KnowledgeFederation) 核心编排层，整合上述所有功能： ```python fed = KnowledgeFederation( workspace_dir="~/.openclaw/workspace", central_api="http://localhost:8000" # 可选：中央知识库API ) # 1. 发布规则到社群 version_id = fed.publish_rule( rule_id="security_check", content={"verify_signature": True}, effectiveness=88.5, tags=["security", "critical"] ) # 2. 订阅社群规则（可选过滤） community_rules = fed.subscribe_community_rules( filters={ "min_score": 75, "tags": ["security"], "project": "finance" } ) # 3. 集成社群规则（自动处理冲突） result = fed.integrate_community_rule(community_rule) # 4. 获取规则演化历史 genealogy = fed.get_rule_genealogy("rule_id") # 返回按时间排序的版本链 # 5. 查看系统统计 stats = fed.get_statistics() # { # "agent_id": "agent_xyz", # "project_id": "openclaw", # "local_rules_count": 42, # "community_rules_count": 156, # "conflicts_detected": 3, # "leaderboard_top_10": [...] # } ``` --- ## 使用方式 ### Python API ```python from skills.knowledge_federation.scripts.knowledge_federation import KnowledgeFederation # 初始化 fed = KnowledgeFederation() # 场景1：发布高效规则到社群 new_rule = fed.publish_rule( rule_id="avoid_typo_in_model_field", content={ "checker": "regex_pattern", "pattern": r"^(claude|gpt|mistral)", "action": "warn_user" }, effectiveness=92.5, tags=["content-safety", "model-names"] ) print(f"规则已发布: {new_rule}") # 场景2：订阅并集成社群规则 community_rules = fed.subscribe_community_rules( filters={"min_score": 80, "tags": ["security"]} ) for rule in community_rules[:5]: try: integrated = fed.integrate_community_rule(rule) print(f"✅ 已集成: {rule.rule_id} (效能: {integrated.effectiveness_score})") except Exception as e: print(f"⚠️ 集成失败: {e}") # 场景3：分析规则演化 genealogy = fed.get_rule_genealogy("my_critical_rule") print(f"规则演化链: {len(genealogy)} 个版本") for i, version in enumerate(genealogy): print(f" v{i}: {version.author_agent} @ {version.timestamp}") ``` ### CLI 使用 ```bash # 发布规则 python3 knowledge_federation.py publish my_rule \ --content '{"action":"check"}' \ --effectiveness 85.0 \ --tags security critical # 订阅规则 python3 knowledge_federation.py subscribe \ --min-score 75 \ --tags security finance \ --workspace ~/.openclaw/workspace # 查看统计 python3 knowledge_federation.py stats --workspace ~/.openclaw/workspace ``` ### 与 Rule Optimizer 集成 ```python from skills.rule_optimizer.scripts.rule_optimizer import RuleOptimizer from skills.knowledge_federation.scripts.knowledge_federation import KnowledgeFederation # 自动循环：优化 → 社群共享 def evolve_rules(): optimizer = RuleOptimizer() fed = KnowledgeFederation() # 1. 找出高效规则 for rule_id in [r.rule_id for r in fed.local_registry.list_rules()]: metrics = optimizer.evaluate_rule_effectiveness(rule_id) # 2. 效能优秀？发布到社群 if metrics.effectiveness_score > 85: fed.publish_rule( rule_id=rule_id, content=metrics.rule_content, effectiveness=metrics.effectiveness_score, tags=metrics.tags ) print(f"✅ 发布高效规则: {rule_id}") # 3. 效能低？寻求社群改进 elif metrics.effectiveness_score < 50: community_versions = fed.subscribe_community_rules({ "min_score": 75, "tags": metrics.tags }) if community_versions: best = sorted(community_versions, key=lambda r: r.leaderboard_score)[-1] fed.integrate_community_rule(best) print(f"⬆️ 更新规则 {rule_id} 为社群版本") ``` ### 与 Behavior Analyzer 集成 ```python from skills.behavior_analyzer.scripts.behavior_analyzer import SessionBehaviorAnalyzer from skills.knowledge_federation.scripts.knowledge_federation import KnowledgeFederation # 异常行为 → 禁用社群规则，回到本地基线 analyzer = SessionBehaviorAnalyzer() fed = KnowledgeFederation() session_health = analyzer.analyze_session("session_123") if session_health["health_score"] < 40: # 会话异常，撤销社群规则，使用本地稳定版本 for rule in fed.local_registry.list_rules(): if rule.status == "published": print(f"回退规则: {rule.rule_id} (会话异常)") ``` --- ## 数据模型 ### RuleSource（规则来源） ```python class RuleSource(Enum): LOCAL = "local" # 本地生成 COMMUNITY = "community" # 社群共享 VERIFIED = "verified" # 已验证通过 ``` ### ConflictResolution（冲突策略） ```python class ConflictResolution(Enum): LOCAL_PRIORITY = "local_priority" COMMUNITY_PRIORITY = "community_priority" MERGE = "merge" VERSION = "version" ``` ### RuleConflict（冲突记录） ```python @dataclass class RuleConflict: conflict_id: str # 冲突ID local_rule: RuleVersion # 本地版本 community_rule: RuleVersion # 社群版本 detected_at: str # 检测时间 resolution_strategy: ConflictResolution # 解决策略 resolution_result: Optional[RuleVersion] # 解决后的版本 user_decision: Optional[str] # 用户决策（可选） ``` ### CommunityRule（社群规则） ```python @dataclass class CommunityRule: rule_id: str # 规则ID versions: List[RuleVersion] # 版本链 effectiveness_history: List[Tuple[str, float]] # 效能历史 adoption_count: int # 采纳数 project_tags: Set[str] # 项目标签 leaderboard_position: Optional[int] # 排行位置 leaderboard_score: float # 排行分数 ``` --- ## 文件格式 ### 本地规则存储 (`.local-rules/`) ```json // .local-rules/check_funds_v1a2b3c4d.json { "version_id": "v1a2b3c4d", "rule_id": "check_funds", "parent_version": null, "author_agent": "agent_production_01", "timestamp": "2026-04-11T10:30:00", "content": {"threshold": 1000, "require_approval": true}, "effectiveness_score": 87.5, "status": "published", "tags": ["finance", "critical"], "description": "Initial version, requires approval for large transfers", "breaking_changes": [] } ``` ### 联邦日志 (`.federation-log.jsonl`) ```json {"timestamp": "2026-04-11T10:30:00", "agent_id": "agent_1", "project_id": "default", "rule_id": "check_funds", "version_id": "v1a2b3c4d", "effectiveness": 87.5, "tags": ["finance", "critical"]} ``` ### 冲突日志 (`.conflict-log.jsonl`) ```json {"timestamp": "2026-04-11T10:35:00", "conflict_id": "c_xyz", "local_rule_id": "v123", "community_rule_id": "v456", "resolution": "community_priority"} ``` --- ## 工作流示例 ### 例子1：规则优化 → 社群推广 ``` 1. 规则执行反馈记录 (via rule_optimizer) optimizer.record_rule_application("check_funds", fixed=True, satisfaction=4.8) 2. 评估效能 metrics = optimizer.evaluate_rule_effectiveness("check_funds") → effectiveness_score = 91.2 (优秀) 3. 发布到社群 fed.publish_rule("check_funds", {...}, 91.2, ["finance", "verified"]) 4. 社群排行榜 leaderboard_position = 2 adoption_count = 24 (24个Agent采纳) 5. 继续监控 新的反馈 → 效能更新 → 排行调整 ``` ### 例子2：社群规则集成 + 冲突解决 ``` 1. Agent A 本地规则 "timeout": 10s, "retry": 3 效能: 65 分 2. 社群规则（更新） "timeout": 5s, "retry": 5, "cache": true 效能: 88 分 3. 检测冲突 ConflictResolver.detect_conflicts(local, community) → True 4. 选择解决策略 ConflictResolution.VERSION (让数据说话) 5. 自动选择 88 > 65 → 使用社群版本 6. 更新与记录 fed.integrate_community_rule(community_rule) → 新版本成为主规则，旧版本保存为历史 ``` ### 例子3：异常检测 + 规则回退 ``` 1. 会话行为分析 behavior_analyzer.analyze_session("s_123") → health_score = 35 (异常) 2. 检测异常 anomaly_patterns = ["repeated_errors", "role_drift"] 3. 紧急措施 禁用最近集成的社群规则 恢复到 local_registry 的稳定版本 4. 通知用户 "检测到异常行为，已降级为保守模式" 5. 恢复后升级 继续监控 health_score health_score > 60 → 逐步恢复高效能规则 ``` --- ## 集成点 ### 与 Week 1-4 系统 - ✅ **rule-optimizer**: 发布/接收经过A/B验证的高分规则 - ✅ **behavior-analyzer**: 异常时回退社群规则 - ✅ **fusion-engine**: 社群规则可信度 (membership in community → +信任) - ✅ **evolve**: 演化出的新规则自动加入知识库 ### 与 OpenClaw 2.0 钩子 - `session:end` — 上报本会话生成的新规则 - `rule:published` — 中央库接收规则发布 - `rule:adopted` — 其他Agent采纳此规则 - `federation:conflict_detected` — 触发冲突解决 --- ## 性能指标 - **发布耗时**: < 50ms - **集成耗时**: < 200ms (包括冲突检测) - **排行刷新**: < 100ms (1000+ 规则) - **内存开销**: < 50MB - **数据保留**: 无限制（支持日期分片） - **并发**: 支持 ≥ 50 Agent 并发发布 --- ## 测试覆盖 ```bash pytest skills/knowledge-federation/tests/ -v # 29/29 通过 ``` **测试覆盖**： - ✅ 版本管理 (4个测试) - ✅ 冲突检测与解决 (6个测试) - ✅ 排行榜管理 (7个测试) - ✅ 规则库操作 (5个测试) - ✅ 端到端集成 (2个测试) --- ## 故障排查 ### 问题：冲突不断发生 **原因**：多Agent频繁编辑同一规则，或冲突解决策略选择错误 **解决**： 1. 检查 `.conflict-log.jsonl` 的冲突历史 2. 调整 `resolution_strategy`（建议使用 VERSION） 3. 为高冲突规则添加 `project_tags` 以隔离作用域 ### 问题：社群规则效能显示为低分 **原因**：初始采纳少、反馈数据不足 **解决**： 1. 增加 `adoption_count` 阈值，减少噪声 2. 监控一周后重新评估 3. 检查 `effectiveness_history` 是否在增长 ### 问题：规则族谱链中断 **原因**：parent_version 指向的文件被误删或迁移 **解决**： 1. 检查 `.local-rules/` 目录完整性 2. 使用 `get_rule_genealogy()` 诊断断链位置 3. 从备份恢复或创建新版本 --- ## 许可 遵循项目主许可证。 --- Source attribution: ybbms777 Original project: https://github.com/ybbms777/openclaw-claude-code-integration/blob/5252e859006d3779cb9d9bbbda35a5cc3d1eb3a5/skills/knowledge-federation/SKILL.md License: MIT License information: https://opensource.org/license/mit Original upstream instructions; downloaded from Mega AI Hub. ## Upstream license notice MIT License Copyright (c) 2026 OpenClaw × Claude Code Integration Contributors Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. # Mega AI Hub resource metadata Creator / publisher: ybbms777 Original project: https://github.com/ybbms777/openclaw-claude-code-integration/blob/5252e859006d3779cb9d9bbbda35a5cc3d1eb3a5/skills/knowledge-federation/SKILL.md Attribution basis: Source repository record Fields of work: Data & Analytics, Development & Engineering, Product Management Roles: Data Scientist, ML / AI Engineer, Product Manager Specialties: API Tools: OpenAI API, Openclaw Country applicability: Country not specified Classification basis: Source catalog labels Instruction SHA-256: 8b5ebcb69d420ba6db6facaa0748c9d16270d031fca6b8168ad7a74909a41791