메뉴
当智能体涉及"memory"与"Context"的操作时触发skill；智能体底层记忆基础设施，完整实现Context Engineering五大核心能力：选择（噪声过滤+相关性筛选）、压缩（因果结构提取+工具结果压缩）、检索（结果重排序+多样性保证）、状态（任务进度追踪+目标对齐）、记忆（冲突检测+跨会话关联）；认知模型层支持认知模型构建、因果链提取、知识缺口识别、检索时机决策、质量评估、状态一致性校验、状态推理、跨会话关联、遗忘机制；作为元技能强制常驻运行
Codex 또는 Claude로 설치 이 Prompt를 복사해 Codex, Claude 또는 다른 어시스턴트에 붙여 넣으면 Skill 페이지를 검토하고 설치를 진행할 수 있습니다.
SOC 직업 분류 기준
A curated collection of 2409+ best OpenClaw skills — AI tools, productivity, marketing, frontend, mobile, backend, DevOps and more. Weekly updated by MyClaw.ai — Powered by MyClaw.ai
将1688的商品铺货到俄罗斯电商平台Ozon（上架），通过Ozon官方API实现商品信息的上传和状态查询。适用于需要将单个1688的商品上架到Ozon的场景。
ActiveCampaign API integration with managed OAuth. Marketing automation, CRM, contacts, deals, and email campaigns. Use this skill when users want to manage contacts, deals, tags, lists, automations, or campaigns in ActiveCampaign. For other third party apps, use the api-gateway skill (https://clawhub.ai/byungkyu/api-gateway). Requires network access and valid Maton API key.
Acuity Scheduling API integration with managed OAuth. Manage appointments, calendars, clients, and availability. Use this skill when users want to schedule, reschedule, or cancel appointments, check availability, or manage clients and calendars. For other third party apps, use the api-gateway skill (https://clawhub.ai/byungkyu/api-gateway).
World-Class Adaptability & Learning Playbook. Use for: market trend awareness, horizon scanning, PESTLE analysis, organisational agility, Kaizen, PDCA cycles, 5S, lean operations, experimentation culture, hypothesis-driven development, A/B testing, MVP design, knowledge management, decision logs, ADRs, after-action reviews, competitive intelligence, SWOT, Porter's Five Forces, battlecards, pivoting strategy, lean startup, business model canvas, signal detection, scenario planning, learning velocity, value stream mapping, Gemba walks. Trigger when discussing ANY organisational learning, strategic adaptability, continuous improvement, competitive analysis, experimentation, knowledge systems, or pivot/persevere decisions. Also for startup strategy around product-market fit or validated learning. If it touches learning faster, adapting better, or competing smarter — use this skill.
Digital legacy agent — dead man's switch, final message executor, and ghost mode responder that preserves your digital presence. Use when the user wants to set up a dead man's switch, manage their digital will, or enable ghost mode.
| name | agent-memory |
| description | 当智能体涉及"memory"与"Context"的操作时触发skill；智能体底层记忆基础设施，完整实现Context Engineering五大核心能力：选择（噪声过滤+相关性筛选）、压缩（因果结构提取+工具结果压缩）、检索（结果重排序+多样性保证）、状态（任务进度追踪+目标对齐）、记忆（冲突检测+跨会话关联）；认知模型层支持认知模型构建、因果链提取、知识缺口识别、检索时机决策、质量评估、状态一致性校验、状态推理、跨会话关联、遗忘机制；作为元技能强制常驻运行 |
| always | true |
| dependency | {"python":["pydantic>=2.0.0","typing-extensions>=4.0.0","cryptography>=41.0.0","redis>=4.5.0","tiktoken>=0.5.0"]} |
always: true）pydantic>=2.0.0
typing-extensions>=4.0.0
cryptography>=41.0.0
redis>=4.5.0
tiktoken>=0.5.0
所有模块初始化时必须指定存储路径：
base_path = "./memory_data"
key_storage_path = f"{base_path}/keys"
sync_state_path = f"{base_path}/sync_state"
index_storage_path = f"{base_path}/memory_index"
credential_path = f"{base_path}/credentials"
from scripts.redis_adapter import create_redis_adapter
redis_adapter = create_redis_adapter(host="localhost", port=6379)
if redis_adapter.is_available():
print("Redis 连接成功")
from scripts.privacy import PrivacyManager, ConsentStatus
privacy_manager = PrivacyManager(user_id="user_123")
if privacy_manager.get_consent_status("memory_storage") == ConsentStatus.NOT_REQUESTED:
privacy_manager.request_consent(
consent_type="memory_storage",
description="是否允许存储交互记忆以提供个性化服务？"
)
from scripts.perception import PerceptionMemoryStore
from scripts.short_term import ShortTermMemoryManager
from scripts.types import SemanticBucketType
# 感知记忆
perception = PerceptionMemoryStore()
session_id = perception.create_session()
# 短期记忆（智能体判断语义分类）
short_term = ShortTermMemoryManager()
item_id = short_term.store_with_semantics(
content="用户想要实现登录功能",
bucket_type=SemanticBucketType.USER_INTENT,
topic_label="用户登录",
relevance_score=0.85,
)
from scripts.long_term import LongTermMemoryManager
long_term = LongTermMemoryManager()
long_term.update_user_profile(profile_data)
long_term.apply_heat_policy()
from scripts.context_reconstructor import ContextReconstructor
from scripts.insight_module import InsightModule
reconstructor = ContextReconstructor()
insight_module = InsightModule()
context = reconstructor.reconstruct(situation, long_term.get_all_memories())
insights = insight_module.process(context, long_term.get_all_memories())
from scripts.state_capture import GlobalStateCapture, StateEventType
capture = GlobalStateCapture(
user_id="user_123",
storage_path="./state_storage",
)
# 从 LangGraph 同步
checkpoint_id = capture.sync_from_langgraph(
state={"phase": "executing", "current_task": "create_memory"},
node_name="executor",
)
# 事件订阅
subscription_id = capture.subscribe(
event_types=[StateEventType.PHASE_CHANGE, StateEventType.TASK_SWITCH],
callback=on_phase_change,
)
from scripts.context_orchestrator import create_context_orchestrator
from scripts.types import SemanticBucketType
orchestrator = create_context_orchestrator(
user_id="user_123",
session_id="session_456",
max_context_tokens=32000,
)
# 存储记忆
orchestrator.store_memory(
content="用户想要实现登录功能",
bucket_type=SemanticBucketType.USER_INTENT,
topic_label="用户登录",
)
# 准备上下文
context = orchestrator.prepare_context(
user_input="帮我分析这段代码的性能问题",
system_instruction="你是一个代码分析专家",
retrieval_results=["性能优化最佳实践"],
tool_results=["代码分析结果..."],
)
# 结束会话
final_stats = orchestrator.end_session()
from scripts.cognitive_model_builder import CognitiveModelBuilder, StepResult, FactSource
builder = CognitiveModelBuilder(session_id="session_001")
# 设置任务上下文
builder.set_task_context(
goal="实现用户登录功能",
sub_goals=["数据库设计", "前端表单", "后端验证"],
current_focus="后端验证逻辑",
)
# 添加已知事实和约束
builder.add_fact(content="用户使用Python 3.9", source=FactSource.MEMORY, confidence=0.9)
builder.add_constraint("must_use", "bcrypt加密")
builder.add_knowledge_gap(description="SSO集成方案", importance="high")
# 构建认知模型
model = builder.build()
print(model.to_context_string()) # 输出模型可理解的上下文
from scripts.causal_chain_extractor import CausalChainExtractor
extractor = CausalChainExtractor()
chains = extractor.extract("登录失败是因为数据库连接超时...")
for chain in chains:
print(chain.to_summary())
# 问题: 登录失败
# 根本原因: 连接池配置过小
# 解决方案: 增加连接池大小
from scripts.knowledge_gap_identifier import KnowledgeGapIdentifier, KnowledgeType
identifier = KnowledgeGapIdentifier()
# 注册已有知识
identifier.register_knowledge(content="用户使用Python 3.9", knowledge_type=KnowledgeType.FACTUAL)
# 定义所需知识
identifier.define_required(description="数据库连接配置", for_task="配置连接", importance=4)
# 分析缺口
result = identifier.analyze()
print(f"知识缺口: {result.total_gaps}, 覆盖率: {result.coverage_ratio:.1%}")
from scripts.retrieval_decision_engine import RetrievalDecisionEngine
from scripts.retrieval_quality_evaluator import RetrievalQualityEvaluator
# 检索决策
engine = RetrievalDecisionEngine()
decision = engine.decide(query="如何优化Python代码性能")
if decision.need in ["required", "recommended"]:
print(f"建议检索: {decision.queries}")
# 质量评估
evaluator = RetrievalQualityEvaluator()
result = evaluator.evaluate(query="...", items=[{"item_id": "1", "content": "...", "score": 0.9}])
print(f"质量评分: {result.quality.overall_score:.2f}")
from scripts.state_consistency_validator import StateConsistencyValidator, StateModule
validator = StateConsistencyValidator()
# 注册各模块状态
validator.register_state(module=StateModule.TASK_PROGRESS, state={"current_task": "登录功能"})
validator.register_state(module=StateModule.SHORT_TERM_MEMORY, state={"topic": "用户认证"})
# 执行校验
report = validator.validate()
if report.conflicts:
fixed = validator.auto_fix(report)
print(f"修复了 {fixed} 个冲突")
from scripts.state_inference_engine import StateInferenceEngine
engine = StateInferenceEngine()
engine.add_premise("任务进度是80%", confidence=0.9)
engine.add_premise("没有阻塞问题", confidence=0.8)
result = engine.infer_next_state()
print(f"推理结果: {result.inferred_value}, 置信度: {result.confidence:.2f}")
from scripts.cross_session_memory_linker import CrossSessionMemoryLinker, LinkType
linker = CrossSessionMemoryLinker()
linker.register_session(session_id="session_001", topics=["Python优化"], entities=["Pandas"])
linker.register_session(session_id="session_002", topics=["Python优化"], entities=["Redis"])
# 发现关联
links = linker.discover_links()
related = linker.get_related_sessions("session_001")
from scripts.memory_forgetting_mechanism import MemoryForgettingMechanism, MemoryImportance
mechanism = MemoryForgettingMechanism()
mechanism.register_memory(memory_id="mem_001", importance=MemoryImportance.HIGH)
mechanism.access_memory("mem_001") # 提升活跃度
candidates = mechanism.analyze_forgetting_candidates()
report = mechanism.execute_forgetting(candidates)
print(f"活跃记忆: {report.active_memories}, 归档: {report.archived_memories}")
from scripts.multi_source_coordinator import MultiSourceCoordinator, SourceType
coordinator = MultiSourceCoordinator()
coordinator.register_source(source_type=SourceType.SYSTEM_INSTRUCTION, content="你是代码分析专家")
coordinator.register_source(source_type=SourceType.USER_QUERY, content="帮我分析代码")
coordinator.register_source(source_type=SourceType.LONG_TERM_MEMORY, content="用户偏好Python")
context = coordinator.coordinate(max_tokens=8000)
print(f"Token使用: {context.token_count}, 来源: {len(context.sources_used)}")
from scripts.context_lazy_loader import create_lazy_loader
loader = create_lazy_loader(max_cache_size=1000)
loader.register_loader("user_profile", lambda: fetch_user_profile())
result = loader.load("user_profile")
predicted = loader.predict_and_preload("user_profile")
print(f"缓存命中率: {loader.get_stats().cache_hit_rate:.1%}")
from scripts.permission_boundary_controller import create_permission_controller
controller = create_permission_controller()
controller.set_user_permission(user_id="user_123", roles=["user"])
# 检查访问权限
result = controller.check_access(user_id="user_123", resource="memory:long_term", action="read")
# 过滤敏感信息
filtered = controller.filter_sensitive("我的API Key是 sk-xxx，邮箱是 user@example.com")
print(f"过滤后: {filtered.filtered}")
from scripts.observability_manager import create_observability_manager, LatencyTracker
manager = create_observability_manager(token_cost_per_1k=0.03)
# 记录Token使用
record = manager.record_token_usage(session_id="session_001", total_tokens=1800, model="gpt-4")
print(f"成本: ${record.cost_estimate:.4f}")
# 延迟追踪
with LatencyTracker(manager, "context_prepare") as tracker:
tracker.start_stage("memory_load")
# ... 加载记忆
tracker.end_stage("memory_load")
# 获取统计
stats = manager.get_stats(hours=24)
print(f"总Token: {stats.total_tokens}, 总成本: ${stats.total_cost:.2f}")
from scripts.result_compressor import ResultCompressor, CompressionStrategy
compressor = ResultCompressor()
result = compressor.compress_tool_result(content=long_log_content, target_tokens=1000)
print(f"压缩率: {result.compression_ratio:.2%}")
print(f"因果链: {len(result.causal_chains)} 个")
from scripts.task_progress import TaskProgressTracker, StepType
tracker = TaskProgressTracker(task_id="task_001", task_name="实现登录功能")
tracker.set_goal(goal_id="goal_001", goal_name="实现登录", success_criteria=["用户可以登录"])
tracker.track_step(step_id="step_001", step_name="设计流程", step_type=StepType.PLANNING)
tracker.start_step("step_001")
tracker.complete_step("step_001", result="流程设计完成")
report = tracker.get_progress_report()
print(f"完成率: {report.completion_rate:.1%}")
from scripts.memory_conflict import MemoryConflictDetector
detector = MemoryConflictDetector()
conflicts = detector.detect_all_conflicts(new_memory=item, existing_memories=memories)
if conflicts:
result = detector.resolve_conflict(conflict=conflicts[0], mode="recency")
print(f"解决方案: {result.rationale}")
from scripts.chain_reasoning import ChainReasoningEnhancer
enhancer = ChainReasoningEnhancer(state_capture=capture, short_term=short_term, long_term=long_term)
result = enhancer.process_reasoning_step(
step={"thought": "分析...", "need_reflect": True, "reflect_reason": "信息矛盾"},
step_index=12,
)
if result["should_reflect"]:
reflection_result = enhancer.execute_reflection(signal=result["signal"], context_snapshot=result["context_snapshot"])
| 文档 | 何时读取 |
|---|---|
| architecture_overview.md | 需要全局架构视角 |
| api_enums.md | 查阅枚举类型定义 |
| api_class_reference.md | 查看所有导出类名和职责 |
| memory_types.md | 深入理解记忆结构 |
| chain_reasoning_guide.md | 链式推理增强集成 |
| encryption_guide.md | 了解数据加密机制 |
| async_optimization_guide.md | 异步写入优化方案 |
| privacy_guide.md | 隐私配置和合规要求 |
| insight_design.md | 洞察生成机制设计 |
| activation_mechanism.md | 记忆激活机制 |
| agent_loops_guide.md | 智能体循环集成 |
| index_sync_guide.md | 索引同步机制 |
| short_term_insight_guide.md | 短期记忆洞察分析 |
PrivacyManager 并获取同意from scripts.perception import PerceptionMemoryStore
from scripts.short_term import ShortTermMemoryManager
from scripts.long_term import LongTermMemoryManager
from scripts.context_reconstructor import ContextReconstructor
# 初始化
perception = PerceptionMemoryStore()
short_term = ShortTermMemoryManager()
long_term = LongTermMemoryManager()
reconstructor = ContextReconstructor()
# 处理对话
session_id = perception.create_session()
perception.store_conversation(session_id, user_message, system_response)
# 短期记忆
short_term.store_with_semantics(user_message, SemanticBucketType.USER_INTENT, "话题", 0.8)
# 上下文重构
context = reconstructor.reconstruct(situation, long_term.get_all_memories())