agent-memory

当智能体涉及"memory"与"Context"的操作时触发skill；智能体底层记忆基础设施，完整实现Context Engineering五大核心能力：选择（噪声过滤+相关性筛选）、压缩（因果结构提取+工具结果压缩）、检索（结果重排序+多样性保证）、状态（任务进度追踪+目标对齐）、记忆（冲突检测+跨会话关联）；认知模型层支持认知模型构建、因果链提取、知识缺口识别、检索时机决策、质量评估、状态一致性校验、状态推理、跨会话关联、遗忘机制；作为元技能强制常驻运行

# Agent Memory System

## 任务目标

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

### 存储路径（必需）

所有模块初始化时**必须指定存储路径**：

```python
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 连接（推荐）

```python
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: 隐私配置（必需）

```python
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: 感知与短期记忆

```python
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: 长期记忆

```python
from scripts.long_term import LongTermMemoryManager

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

### Step 4: 上下文重构与洞察

```python
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集成）

```python
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（总控层）

```python
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: 认知模型构建

```python
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: 因果链提取

```python
from scripts.causal_chain_extractor import CausalChainExtractor

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

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

### Step 9: 知识缺口识别

```python
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: 检索决策与评估

```python
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: 状态一致性校验

```python
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: 状态推理

```python
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: 跨会话关联

```python
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: 遗忘机制

```python
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: 多源协调

```python
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: 上下文懒加载

```python
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_sta