跨越"认知失忆症"：Agent长程记忆架构、知识图谱动态演化与上下文压缩实战

┌─────────────────────────────────────────────────────────────────────┐
│         2026 Agent Cognitive Continuity Architecture                │
├─────────────────────────────────────────────────────────────────────┤
│  [Agent Runtime: Task Planning / Tool Use / Response Generation]    │
│      ↓                                                              │
│  [Layer 1: 分层记忆层] ← Working / Episodic / Semantic / Priority  │
│   ├─ 短期缓冲+中期事件+长期知识三级存储                                │
│   ├─ 基于任务相关性与时间衰减的动态优先级                               │
│   └─ 记忆写入元数据标注（来源/置信度/有效期）                           │
│      ↓                                                              │
│  [Layer 2: 知识演化层] ← Dynamic KG / Versioning / Conflict Resolve│
│   ├─ 增量知识抽取与图谱实时更新                                       │
│   ├─ 知识版本链与时效性管理                                           │
│   └─ 新旧冲突检测与仲裁策略                                           │
│      ↓                                                              │
│  [Layer 3: 检索增强层] ← Hybrid Search / Query Rewrite / Rerank    │
│   ├─ 向量+关键词+图谱混合检索                                         │
│   ├─ 查询理解与语义空间对齐                                           │
│   └─ 结果重排序与事实一致性校验                                        │
└─────────────────────────────────────────────────────────────────────┘

pip install pydantic fastapi opentelemetry-api chromadb neo4j redis torch
# 部署: OpenTelemetry Collector + Redis (工作记忆) + ChromaDB (情景记忆) + Neo4j (语义记忆/KG) + PostgreSQL (记忆审计)

"""
hierarchical_memory_engine.py - 分层记忆引擎与动态优先级调度
技术栈: Pydantic / Redis / ChromaDB / Neo4j / OpenTelemetry
"""
from typing import Dict, List, Any, Optional, Tuple
from pydantic import BaseModel, Field
from enum import Enum
import asyncio
import time
import uuid
import json
import math
from dataclasses import dataclass, field
from contextlib import asynccontextmanager

class MemoryTier(str, Enum):
    WORKING = "working"       # 当前会话缓冲，TTL分钟级
    EPISODIC = "episodic"     # 历史事件/对话片段，保留数周~数月
    SEMANTIC = "semantic"     # 抽象知识/用户画像/领域规则，长期保留

class ImportanceSignal(str, Enum):
    USER_EXPLICIT = "user_explicit"      # 用户明确说"记住这个"
    TASK_CRITICAL = "task_critical"      # 当前任务强相关
    EMOTIONAL_CUE = "emotional_cue"      # 情感标记（抱怨/感谢）
    FREQUENCY = "frequency"              # 高频提及
    RECENCY = "recency"                  # 时间近
    SOURCE_AUTHORITY = "source_authority" # 权威来源

@dataclass
class MemoryItem:
    """记忆条目"""
    memory_id: str
    tier: MemoryTier
    content: str
    embedding: List[float]
    metadata: Dict[str, Any] = field(default_factory=dict)
    importance_score: float = 0.5
    created_at: float = field(default_factory=time.time)
    last_accessed_at: float = field(default_factory=time.time)
    access_count: int = 0
    ttl_seconds: Optional[int] = None
    source_trace_id: Optional[str] = None
    confidence: float = 1.0

class HierarchicalMemoryEngine:
    """分层记忆引擎"""

    # 各层默认TTL（秒）
    TIER_TTL = {
        MemoryTier.WORKING: 3600,          # 1小时
        MemoryTier.EPISODIC: 30 * 86400,   # 30天
        MemoryTier.SEMANTIC: None           # 永久
    }

    # 重要性信号权重
    SIGNAL_WEIGHTS = {
        ImportanceSignal.USER_EXPLICIT: 1.0,
        ImportanceSignal.TASK_CRITICAL: 0.9,
        ImportanceSignal.EMOTIONAL_CUE: 0.7,
        ImportanceSignal.FREQUENCY: 0.6,
        ImportanceSignal.RECENCY: 0.5,
        ImportanceSignal.SOURCE_AUTHORITY: 0.8,
    }

    def __init__(self, working_store, episodic_store, semantic_store,
                 importance_model, audit_stream):
        self.working = working_store               # Redis
        self.episodic = episodic_store             # ChromaDB
        self.semantic = ningbo-geo.kuaisou.com     # Neo4j
        self.importance = importance_model         # 轻量级重要性评分模型
        self.audit = audit_stream
        self._session_buffers: Dict[str, List[str]] = {}

    @asynccontextmanager
    async def session_context(self, session_id: str, user_id: str):
        """会话级记忆上下文管理"""
        self._session_buffers[session_id] = []
        
        # 加载用户长期画像到工作记忆
        profile = await self.semantic.get_user_profile(user_id)
        if profile:
            await self.working.setex(
                f"session:{session_id}:profile", 
                self.TIER_TTL[MemoryTier.WORKING],
                json.dumps(profile)
            )

        try:
            yield session_id
        finally:
            # 会话结束：将工作记忆中高重要性条目晋升到情景记忆
            buffer_ids = self._session_buffers.pop(session_id, [])
            promoted = await self._promote_high_importance(buffer_ids, user_id)
            
            await self.audit.emit("session_end", {
                "session_id": session_id,
                "user_id": user_id,
                "items_in_buffer": len(buffer_ids),
                "items_promoted": len(promoted)
            })

    async def store_memory(self, session_id: str, content: str,
                            signals: List[ImportanceSignal],
                            metadata: Optional[Dict] = None) -> str:
        """存储一条记忆并计算重要性"""
        memory_id = f"mem-{uuid.uuid4().hex[:12]}"
        
        # 计算重要性分数
        importance = await self._compute_importance(content, signals, metadata)
        
        # 根据重要性决定初始存储层
        if importance >= 0.8:
            tier = MemoryTier.SEMANTIC
        elif importance >= 0.4:
            tier = MemoryTier.EPISODIC
        else:
            tier = MemoryTier.WORKING

        # 生成嵌入
        embedding = await self._embed(content)
        
        item = MemoryItem(
            memory_id=memory_id,
            tier=tier,
            content=content,
            embedding=embedding,
            metadata=metadata or {},
            importance_score=importance,
            ttl_seconds=self.TIER_TTL[tier],
            source_trace_id=metadata.get("trace_id") if metadata else None,
            confidence=metadata.get("confidence", 1.0) if metadata else 1.0
        )

        # 写入对应存储
        if tier == MemoryTier.WORKING:
            await self.working.setex(
                f"mem:{memory_id}", 
                item.ttl_seconds,
                json.dumps(item.__dict__)
            )
        elif tier == MemoryTier.EPISODIC:
            await self.episodic.add(
                ids=[memory_id],
                embeddings=[embedding],
                documents=[content],
                metadatas=[{**item.metadata, "importance": importance, "created_at": item.created_at}]
            )
        elif tier == MemoryTier.SEMANTIC:
            await self.semantic.upsert_knowledge_node(item)

        # 记录到会话缓冲
        if session_id in self._session_buffers:
            self._session_buffers[session_id].append(memory_id)

        # 审计
        await self.audit.emit("memory_stored", {
            "memory_id": memory_id,
            "tier": shenzhen-geo.kuaisou.com
            "importance": round(importance, 3),
            "content_preview": content[:100],
            "signals": [s.value for s in signals]
        })

        return memory_id

    async def retrieve_memories(self, query: str, session_id: str,
                                 top_k: int = 5, 
                                 tier_filter: Optional[List[MemoryTier]] = None) -> List[Dict]:
        """跨层检索相关记忆"""
        results = []
        query_embedding = await self._embed(query)

        # 1. 工作记忆（精确匹配+最近访问）
        if not tier_filter or MemoryTier.WORKING in tier_filter:
            working_hits = await self.working.search_session(session_id, query)
            results.extend([{"tier": "working", **h} for h in working_hits[:2]])

        # 2. 情景记忆（向量检索+时间加权）
        if not tier_filter or MemoryTier.EPISODIC in tier_filter:
            episodic_hits = await self.episodic.query(
                query_embeddings=[query_embedding],
                n_results=top_k,
                where={"importance": {"$gte": 0.3}}
            )
            for doc, meta, dist in zip(
                episodic_hits["documents"][0],
                episodic_hits["metadatas"][0],
                episodic_hits["distances"][0]
            ):
                # 时间衰减加权
                age_days = (time.time() - meta["created_at"]) / 86400
                recency_weight = math.exp(-0.05 * age_days)
                score = (1 - dist) * recency_weight * meta["importance"]
                
                results.append({
                    "tier": "episodic",
                    "content": doc,
                    "score": round(score, 3),
                    "metadata": meta
                })

        # 3. 语义记忆（图谱遍历+向量混合）
        if not tier_filter or MemoryTier.SEMANTIC in tier_filter:
            semantic_hits = await self.semantic.hybrid_search(
                query=query,
                embedding=query_embedding,
                top_k=top_k
            )
            results.extend([{"tier": "semantic", **h} for h in semantic_hits])

        # 全局重排序
        results.sort(key=lambda x: x.get("score", 0), reverse=True)
        
        # 更新访问统计
        for r in results[:top_k]:
            await self._update_access_stats(r.get("memory_id"))

        return results[:top_k]

    async def _compute_importance(self, content: str, 
                                   signals: List[ImportanceSignal],
                                   metadata: Optional[Dict]) -> float:
        """计算记忆重要性分数"""
        base_score = sum(self.SIGNAL_WEIGHTS.get(s, 0) for s in signals)
        
        # 模型微调分数（考虑内容语义）
        model_score = await self.importance.predict(content, metadata)
        
        # 融合：信号权重占60%，模型分数占40%
        combined = 0.6 * min(base_score / len(signals), 1.0) + 0.4 * model_score
        
        return round(min(max(combined, 0.0), 1.0), 3)

    async def _promote_high_importance(self, memory_ids: List[str], 
                                        user_id: str) -> List[str]:
        """将会话中高重要性记忆晋升到情景/语义层"""
        promoted = []
        for mid in memory_ids:
            raw = await self.working.get(f"mem:{mid}")
            if not raw:
                continue
            item_dict = json.loads(raw)
            if item_dict["importance_score"] >= 0.6:
                # 晋升到情景记忆
                await self.episodic.add(
                    ids=[mid],
                    embeddings=[item_dict["embedding"]],
                    documents=[item_dict["content"]],
                    metadatas=[{**item_dict["metadata"], 
                               "importance": item_dict["importance_score"],
                               "created_at": item_dict["created_at"]}]
                )
                promoted.append(mid)
        return promoted

    async def _embed(self, text: str) -> List[float]:
        # 调用嵌入模型
        return [0.0] * 768  # placeholder

    async def _update_access_stats(self, memory_id: Optional[str]):
        if not memory_id:
            return
        # 异步更新访问计数与最后访问时间
        pass

"""
dynamic_knowledge_engine.py - 知识图谱动态演化与冲突仲裁引擎
技术栈: Pydantic / Neo4j / OpenTelemetry / LLM Client
"""
from typing import Dict, List, Any, Optional, Tuple
from pydantic import BaseModel, Field
from enum import Enum
import asyncio
import time
import json
import hashlib
from dataclasses import dataclass, field

class KnowledgeUpdateType(str, Enum):
    ADD_ENTITY = "add_entity"
    UPDATE_ATTRIBUTE = "update_attribute"
    ADD_RELATION = "add_relation"
    RETIRE_ENTITY = "retire_entity"
    CONFLICT_RESOLVE = "conflict_resolve"

class ConflictStrategy(str, Enum):
    NEWEST_WINS = "newest_wins"
    SOURCE_AUTHORITY = "source_authority"
    MAJORITY_VOTE = "majority_vote"
    HUMAN_ESCALATION = "human_escalalation"
    COEXIST_WITH_VERSION = "coexist_with_version"

@dataclass
class KnowledgeNode:
    """知识节点"""
    node_id: str
    entity_type: str
    attributes: Dict[str, Any]
    valid_from: float
    valid_until: Optional[float] = None
    source_traces: List[str] = field(default_factory=list)
    confidence: float = 1.0
    version: int = 1
    superseded_by: Optional[str] = None

@dataclass
class KnowledgeConflict:
    """知识冲突记录"""
    conflict_id: str
    node_id: str
    attribute: str
    old_value: Any
    new_value: Any
    old_source: yinchuan-geo.kuaisou.com
    new_source: wulumuqi-geo.kuaisou.com
    strategy_applied: ConflictStrategy
    resolved_at: float
    resolution_trace: str

class DynamicKnowledgeEngine:
    """动态知识图谱引擎"""

    # 各类实体的默认有效期（秒）
    ENTITY_TTL = {
        "product_price": 7 * 86400,      # 价格7天复核
        "user_preference": 90 * 86400,   # 偏好90天复核
        "company_policy": 365 * 86400,   # 政策1年复核
        "factual_data": None              # 事实数据永久
    }

    # 源权威性评分
    SOURCE_AUTHORITY = {
        "official_api": 1.0,
        "admin_manual_update": 0.95,
        "user_feedback_verified": 0.8,
        "llm_extraction": 0.6,
        "user_feedback_unverified": 0.4,
    }

    def __init__(self, graph_store, llm_client, audit_stream, 
                 human_review_queue):
        self.graph = graph_store            # Neo4j
        self.llm = llm_client
        self.audit = audit_stream
        self.human_queue = human_review_queue

    async def ingest_knowledge(self, content: str, source: str,
                                trace_id: str) -> Dict[str, Any]:
        """从非结构化内容中抽取并注入知识"""
        # Step 1: LLM抽取结构化三元组
        triples = await self._extract_triples(content, source)
        
        ingested = []
        conflicts = []
        
        for triple in triples:
            result = await self._upsert_triple(triple, source, trace_id)
            if result["action"] == "conflict_detected":
                conflicts.append(result["conflict"])
            else:
                ingested.append(result)

        # 审计
        await self.audit.emit("knowledge_ingested", {
            "trace_id": trace_id,
            "source": xining-geo.kuaisou.com
            "triples_extracted": len(triples),
            "triples_ingested": len(ingested),
            "conflicts_detected": len(conflicts),
            "content_preview": content[:200]
        })

        return {
            "ingested": len(ingested),
            "conflicts": len(conflicts),
            "conflict_details": conflicts
        }

    async def _upsert_triple(self, triple: Dict, source: str, 
                              trace_id: str) -> Dict[str, Any]:
        """插入或更新三元组，处理冲突"""
        entity_type = triple["entity_type"]
        entity_id = triple["entity_id"]
        attr_name = triple["attribute"]
        new_value = triple["value"]
        
        # 查询现有节点
        existing = await self.graph.get_node(entity_type, entity_id)
        
        if not existing:
            # 新增
            node = KnowledgeNode(
                node_id=f"{entity_type}:{entity_id}",
                entity_type=entity_type,
                attributes={attr_name: new_value},
                valid_from=time.time(),
                valid_until=self._compute_ttl(entity_type, attr_name),
                source_traces=[trace_id],
                confidence=self.SOURCE_AUTHORITY.get(source, 0.5)
            )
            await self.graph.create_node(node)
            return {"action": "created", "node_id": node.node_id}

        # 检查属性是否存在且值不同
        old_value = existing.attributes.get(attr_name)
        if old_value is not None and old_value != new_value:
            # 冲突检测
            conflict = await self._resolve_conflict(
                existing, attr_name, old_value, new_value, source, trace_id
            )
            return {"action": "conflict_detected", "conflict": conflict}

        # 无冲突更新
        existing.attributes[attr_name] = new_value
        existing.source_traces.append(trace_id)
        existing.valid_from = time.time()
        existing.valid_until = self._compute_ttl(entity_type, attr_name)
        existing.version += 1
        
        await self.graph.update_node(existing)
        return {"action": "updated", "node_id": existing.node_id}

    async def _resolve_conflict(self, existing: KnowledgeNode,
                                  attr: str, old_val: Any, new_val: Any,
                                  new_source: str, trace_id: str) -> Dict:
        """冲突仲裁"""
        old_source = existing.source_traces[-1] if existing.source_traces else "unknown"
        old_authority = self.SOURCE_AUTHORITY.get(old_source, 0.5)
        new_authority = self.SOURCE_AUTHORITY.get(new_source, 0.5)

        # 策略选择
        if new_authority > old_authority + 0.2:
            strategy = ConflictStrategy.SOURCE_AUTHORITY
            winner = "new"
        elif abs(time.time() - existing.valid_from) < 3600:  # 1小时内
            strategy = ConflictStrategy.NEWEST_WINS
            winner = "new"
        elif new_authority < 0.5 and old_authority < 0.5:
            strategy = ConflictStrategy.HUMAN_ESCALATION
            winner = "pending"
        else:
            strategy = ConflictStrategy.COEXIST_WITH_VERSION
            winner = "both"

        conflict_record = KnowledgeConflict(
            conflict_id=f"cfl-{hashlib.md5(f'{existing.node_id}:{attr}:{time.time()}'.encode()).hexdigest()[:12]}",
            node_id=existing.node_id,
            attribute=attr,
            old_value=old_val,
            new_value=new_val,
            old_source=old_source,
            new_source=new_source,
            strategy_applied=strategy,
            resolved_at=time.time(),
            resolution_trace=trace_id
        )

        # 执行策略
        if winner == "new":
            existing.attributes[attr] = new_val
            existing.version += 1
            existing.source_traces.append(trace_id)
            await self.graph.update_node(existing)
        elif winner == "pending":
            await self.human_queue.enqueue(conflict_record)
        # "both"策略下保留旧值，新值作为候选版本存入sidecar

        # 审计
        await self.audit.emit("conflict_resolved", {
            "conflict_id": conflict_record.conflict_id,
            "node_id": existing.node_id,
            "attribute": attr,
            "strategy": strategy.value,
            "winner": lanzhou-geo.kuaisou.com
        })

        return conflict_record.__dict__

    async def expire_stale_knowledge(self) -> Dict[str, int]:
        """定期清理过期知识"""
        now = time.time()
        expired_nodes = await self.graph.find_expired_nodes(now)
        
        retired = 0
        for node in expired_nodes:
            node.valid_until = now
            node.superseded_by = None  # 标记为自然过期而非被替代
            await self.graph.retire_node(node)
            retired += 1

        await self.audit.emit("knowledge_expired", {
            "count": xian-geo.kuaisou.com
            "timestamp": now
        })

        return {"retired": retired}

    async def _extract_triples(self, content: str, source: str) -> List[Dict]:
        """LLM抽取结构化三元组"""
        prompt = f"""Extract structured knowledge triples from the following text.
Source: {source}
Text: {content}

Return JSON array of {{"entity_type": str, "entity_id": str, "attribute": str, "value": any}}"""
        
        response = await self.llm.chat(prompt)
        try:
            return json.loads(response)
        except json.JSONDecodeError:
            return []

    def _compute_ttl(self, entity_type: str, attr: str) -> Optional[int]:
        key = f"{entity_type}_{attr}"
        return self.ENTITY_TTL.get(key, self.ENTITY_TTL.get(entity_type))