Agent Memory System

任务目标

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

前置准备

依赖

pydantic>=2.0.0
typing-extensions>=4.0.0
cryptography>=41.0.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.credential_manager import CredentialManager

cred_manager = CredentialManager(storage_path="./memory_data/credentials")
cred_manager.store("github_token", "ghp_xxxxx")  # 自动加密
token = cred_manager.get("github_token")         # 自动解密

操作步骤

Step 0: 隐私配置（必需）

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 1: 感知记忆

from scripts.perception import PerceptionMemoryStore

store = PerceptionMemoryStore()
session_id = store.create_session()

should_store, reason = privacy_manager.should_store_memory("episodic", user_message)
if should_store:
    store.store_conversation(session_id, user_message, system_response)

situation = store.detect_situation()

Step 2: 短期记忆

推荐方式：智能体判断语义分类后存储。

from scripts.short_term import ShortTermMemoryManager
from scripts.types import SemanticBucketType

short_term = ShortTermMemoryManager()

item_id = short_term.store_with_semantics(
    content="用户想要实现登录功能",
    bucket_type=SemanticBucketType.USER_INTENT,  # 智能体判断
    topic_label="用户登录",                       # 智能体指定
    relevance_score=0.85,
)

语义桶分类：

话题聚合与提炼：

summary = short_term.get_topic_summary()        # 话题聚合摘要
topics = short_term.get_active_topics()         # 活跃话题

if short_term.should_extract():
    from scripts.short_term import AsynchronousExtractor
    AsynchronousExtractor(short_term, long_term).extract()

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

reconstructor = ContextReconstructor()
reconstructor.bind_state_capture(capture)  # P0: 状态感知优化

context = reconstructor.reconstruct(situation, long_term.get_all_memories())
state_ctx = reconstructor.get_state_aware_context()

reconstructor.unbind_state_capture()

Step 5: 洞察生成

from scripts.insight_module import InsightModule

insight_module = InsightModule()
insight_module.bind_state_capture(capture)  # P1: 状态模式分析

insights = insight_module.process(context, long_term.get_all_memories())
high_priority = insight_module.get_high_priority_insights()
state_insights = insight_module.get_state_pattern_insights()  # P1新增

insight_module.unbind_state_capture()

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

from scripts.state_capture import GlobalStateCapture, StateEventType

capture = GlobalStateCapture(
    user_id="user_123",
    storage_path="./state_storage",
    default_ttl_hours=168,
)

# 从 LangGraph 同步
checkpoint_id = capture.sync_from_langgraph(
    state={"phase": "executing", "current_task": "create_memory"},
    node_name="executor",
)

# 检查点管理
checkpoints = capture.list_checkpoints(thread_id="thread_xxx")
state = capture.restore_checkpoint("cp_xxx")

# 事件订阅
subscription_id = capture.subscribe(
    event_types=[StateEventType.PHASE_CHANGE, StateEventType.TASK_SWITCH],
    callback=on_phase_change,
)

Step 7: 工具管理

# 记录工具使用效果
long_term.update_tool_usage(
    tool_name="code_interpreter",
    task_type="code_debugging",
    outcome="success",
    effectiveness_score=0.85,
    checkpoint_id=checkpoint_id,  # P0: 状态关联
    phase="executing",
    scenario="coding",
)

# 获取工具推荐
rec = long_term.get_tool_recommendation(
    task_type="code_debugging",
    phase="executing",    # P0: 状态过滤
    scenario="coding",
)

Step 8: 链式推理增强（P0 新增）

用途：模型自检测反思、反思结果持久化、元学习数据提取。

from scripts.chain_reasoning import ChainReasoningEnhancer
from scripts.types import ReflectionOutcome

# 初始化
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": "检测到信息矛盾",
        "reflect_confidence": 0.85,
    },
    step_index=12,
    task_type="analysis",
)

if result["should_reflect"]:
    # 执行反思
    reflection_result = enhancer.execute_reflection(
        signal=result["signal"],
        context_snapshot=result["context_snapshot"],
    )

    if not reflection_result.passed:
        # 验证失败处理
        if reflection_result.need_verification:
            verification = enhancer.execute_verification(
                reflection_result=reflection_result,
                context_snapshot=result["context_snapshot"],
            )

            if verification.result == "failed":
                # 回退处理
                rollback_info = enhancer.handle_rollback(
                    step_index=12,
                    reason="验证失败",
                )

    # 持久化反思结果
    memory_id = enhancer.persist_reflection_result(
        trigger_record=result["trigger_record"],
        reflection_result=reflection_result,
        verification_result=verification if reflection_result.need_verification else None,
        final_outcome=ReflectionOutcome.CORRECTED,
    )

LangGraph 集成：

# 作为节点嵌入工作流
workflow.add_node("reflection", enhancer.as_reflection_node())
workflow.add_node("verification", enhancer.as_verification_node())

# 条件边
workflow.add_conditional_edges(
    "reasoning",
    ChainReasoningEnhancer.should_reflect,
    {"reflection": "reflection", "continue": "next_node"},
)

元学习数据提取：

# 提取训练数据（用于优化"何时反思"能力）
training_data = enhancer.extract_meta_learning_data(
    min_learning_value=LearningValue.MEDIUM,
    limit=100,
)

资源索引

核心脚本

参考文档

注意事项

1. 路径必传：所有存储路径无默认值，必须显式传入
2. 隐私优先：处理用户数据前必须初始化 PrivacyManager 并获取同意
3. 敏感数据：系统自动识别密码、账号等敏感信息，默认不存储
4. 加密存储：敏感数据推荐 AES-256-GCM 加密
5. 类型安全：所有函数必须有类型注解，禁止使用裸 dict
6. 异步优先：提炼、热度计算等后台异步执行
7. 超然洞察：洞察模块独立运行，提供非强制性建议
8. 降级策略：模块故障时自动降级，保证核心流程可用
9. 数据权利：用户可随时导出、删除所有数据

快速开始

from scripts.perception import PerceptionMemoryStore
from scripts.short_term import ShortTermMemoryManager, AsynchronousExtractor
from scripts.long_term import LongTermMemoryManager
from scripts.context_reconstructor import ContextReconstructor
from scripts.insight_module import InsightModule

# 初始化
perception = PerceptionMemoryStore()
short_term = ShortTermMemoryManager()
long_term = LongTermMemoryManager()
reconstructor = ContextReconstructor()
insight_module = InsightModule()

# 处理对话
session_id = perception.create_session()
perception.store_conversation(session_id, user_message, system_response)
situation = perception.detect_situation()

# 短期记忆 + 提炼
short_term.store_with_semantics(user_message, SemanticBucketType.USER_INTENT, "话题", 0.8)
if short_term.should_extract():
    AsynchronousExtractor(short_term, long_term).extract()

# 上下文重构 + 洞察
context = reconstructor.reconstruct(situation, long_term.get_all_memories())
insights = insight_module.process(context, long_term.get_all_memories())
print(insight_module.format_insights_for_context())