agent-memory

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

Agent Memory System

任务目标

• 本 Skill 用于：为智能体构建完整的记忆能力基础设施，实现 Context Engineering 核心能力
• 触发条件：元技能，强制常驻运行（always: true）
• 架构总览：详见 references/architecture_overview.md
• 枚举参考：详见 references/api_enums.md

前置准备

依赖

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"

Redis 连接（推荐）

from scripts.redis_adapter import create_redis_adapter

redis_adapter = create_redis_adapter(host="localhost", port=6379)
if redis_adapter.is_available():
    print("Redis 连接成功")

操作步骤

Step 1: 隐私配置（必需）

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="是否允许存储交互记忆以提供个性化服务？"
    )

Step 2: 感知与短期记忆

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,
)

Step 3: 长期记忆

from scripts.long_term import LongTermMemoryManager

long_term = LongTermMemoryManager()
long_term.update_user_profile(profile_data)
long_term.apply_heat_policy()

Step 4: 上下文重构与洞察

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())

Step 5: 全局状态捕捉（LangGraph集成）

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,
)

Step 6: Context Orchestrator（总控层）

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()

Step 7: 认知模型构建

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())  # 输出模型可理解的上下文

Step 8: 因果链提取

from scripts.causal_chain_extractor import CausalChainExtractor

extractor = CausalChainExtractor()
chains = extractor.extract("登录失败是因为数据库连接超时...")

for chain in chains:
    print(chain.to_summary())
    # 问题: 登录失败
    # 根本原因: 连接池配置过小
    # 解决方案: 增加连接池大小

Step 9: 知识缺口识别

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%}")

Step 10: 检索决策与评估

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}")

Step 11: 状态一致性校验

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} 个冲突")

Step 12: 状态推理

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}")

Step 13: 跨会话关联

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")

Step 14: 遗忘机制

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}")

Step 15: 多源协调

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)}")

Step 16: 上下文懒加载

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%}")

Step 17: 权限边界控制

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}")

Step 18: 可观测性管理

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}")

Step 19: 结果压缩

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)} 个")

Step 20: 任务进度追踪

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%}")

Step 21: 记忆冲突检测

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}")

Step 22: 链式推理增强

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"])

资源索引

核心脚本（40个）

参考文档

注意事项

1. 路径必传：所有存储路径无默认值，必须显式传入
2. 隐私优先：处理用户数据前必须初始化 PrivacyManager 并获取同意
3. 敏感数据：系统自动识别密码、账号等敏感信息，默认不存储
4. 类型安全：所有函数必须有类型注解，禁止使用裸 dict
5. 异步优先：提炼、热度计算等后台异步执行
6. 降级策略：模块故障时自动降级，保证核心流程可用

快速开始

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())