35：Multimodal Memory系统设计：短期记忆、长期记忆、结构化+Graph记忆架构

from collections import deque

class ShortTermMemory:
    """短期记忆"""
    def __init__(self, capacity=100):
        """初始化短期记忆"""
        self.capacity = capacity
        self.memory = deque(maxlen=capacity)
    
    def add(self, item):
        """添加项目到短期记忆"""
        self.memory.append(item)
    
    def get(self, index=-1):
        """获取短期记忆中的项目"""
        if not self.memory:
            return None
        return self.memory[index]
    
    def get_all(self):
        """获取所有短期记忆"""
        return list(self.memory)
    
    def clear(self):
        """清空短期记忆"""
        self.memory.clear()
    
    def size(self):
        """获取短期记忆大小"""
        return len(self.memory)

# 示例使用
stm = ShortTermMemory(capacity=10)
stm.add({"type": "text", "content": "Hello, how are you?"})
stm.add({"type": "image", "content": "image_path.jpg"})
stm.add({"type": "video", "content": "video_path.mp4"})

print("短期记忆内容:")
for item in stm.get_all():
    print(item)

from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

class LongTermMemory:
    """长期记忆"""
    def __init__(self, persist_directory="./long_term_memory"):
        """初始化长期记忆"""
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
        self.vectorstore = Chroma(
            persist_directory=persist_directory,
            embedding_function=self.embeddings
        )
    
    def add(self, content, metadata=None):
        """添加项目到长期记忆"""
        if metadata is None:
            metadata = {}
        self.vectorstore.add_texts([content], metadatas=[metadata])
    
    def search(self, query, k=5):
        """从长期记忆中搜索信息"""
        results = self.vectorstore.similarity_search(query, k=k)
        return results
    
    def delete(self, query):
        """删除长期记忆中的信息"""
        # 实现删除逻辑
        pass

# 示例使用
ltm = LongTermMemory()
ltm.add("Agentic RAG是一种先进的检索增强生成技术", {"type": "text", "category": "AI"})
ltm.add("Multimodal Memory系统支持多种模态", {"type": "text", "category": "AI"})

# 搜索
results = ltm.search("RAG技术")
print("长期记忆搜索结果:")
for result in results:
    print(result.page_content)

class StructuredMemory:
    """结构化记忆"""
    def __init__(self):
        """初始化结构化记忆"""
        self.structure = {}
    
    def add(self, category, key, value):
        """添加结构化信息"""
        if category not in self.structure:
            self.structure[category] = {}
        self.structure[category][key] = value
    
    def get(self, category, key=None):
        """获取结构化信息"""
        if category not in self.structure:
            return None
        if key is None:
            return self.structure[category]
        return self.structure[category].get(key, None)
    
    def delete(self, category, key=None):
        """删除结构化信息"""
        if category not in self.structure:
            return
        if key is None:
            del self.structure[category]
        else:
            if key in self.structure[category]:
                del self.structure[category][key]
    
    def get_all(self):
        """获取所有结构化信息"""
        return self.structure

# 示例使用
sm = StructuredMemory()
sm.add("AI", "RAG", "检索增强生成")
sm.add("AI", "LLM", "大型语言模型")
sm.add("ML", "CNN", "卷积神经网络")

print("结构化记忆内容:")
print(sm.get_all())
print("AI类别:")
print(sm.get("AI"))

import networkx as nx

class GraphMemory:
    """Graph记忆"""
    def __init__(self):
        """初始化Graph记忆"""
        self.graph = nx.DiGraph()
    
    def add_node(self, node_id, attributes=None):
        """添加节点"""
        if attributes is None:
            attributes = {}
        self.graph.add_node(node_id, **attributes)
    
    def add_edge(self, source, target, relationship=None):
        """添加边"""
        if relationship is None:
            relationship = {}
        self.graph.add_edge(source, target, **relationship)
    
    def get_node(self, node_id):
        """获取节点"""
        if node_id in self.graph.nodes:
            return self.graph.nodes[node_id]
        return None
    
    def get_edges(self, node_id):
        """获取节点的边"""
        if node_id in self.graph.nodes:
            return list(self.graph.edges(node_id, data=True))
        return []
    
    def search_path(self, source, target):
        """搜索路径"""
        try:
            path = nx.shortest_path(self.graph, source, target)
            return path
        except nx.NetworkXNoPath:
            return None

# 示例使用
gm = GraphMemory()
gm.add_node("RAG", {"type": "AI技术"})
gm.add_node("LLM", {"type": "AI模型"})
gm.add_node("VectorStore", {"type": "存储"})
gm.add_edge("RAG", "LLM", {"relationship": "使用"})
gm.add_edge("RAG", "VectorStore", {"relationship": "依赖"})

print("Graph记忆节点:")
print(list(gm.graph.nodes(data=True)))
print("Graph记忆边:")
print(list(gm.graph.edges(data=True)))
print("RAG到VectorStore的路径:")
print(gm.search_path("RAG", "VectorStore"))

from transformers import CLIPProcessor, CLIPModel
import torch

class MultimodalFusion:
    """多模态融合"""
    def __init__(self):
        """初始化多模态融合"""
        self.model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
        self.processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
    
    def fuse(self, text=None, image=None):
        """融合文本和图像"""
        inputs = self.processor(
            text=text,
            images=image,
            return_tensors="pt",
            padding=True
        )
        
        with torch.no_grad():
            outputs = self.model(**inputs)
        
        return outputs

# 示例使用
fusion = MultimodalFusion()

# 融合文本和图像
text = "A cat"
image = "cat.jpg"  # 实际使用时需要加载图像

outputs = fusion.fuse(text=text)
print("多模态融合结果:")
print(outputs)

from collections import deque
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
import networkx as nx
from transformers import CLIPProcessor, CLIPModel

class ShortTermMemory:
    """短期记忆"""
    def __init__(self, capacity=100):
        """初始化短期记忆"""
        self.capacity = capacity
        self.memory = deque(maxlen=capacity)
    
    def add(self, item):
        """添加项目到短期记忆"""
        self.memory.append(item)
    
    def get(self, index=-1):
        """获取短期记忆中的项目"""
        if not self.memory:
            return None
        return self.memory[index]
    
    def get_all(self):
        """获取所有短期记忆"""
        return list(self.memory)
    
    def clear(self):
        """清空短期记忆"""
        self.memory.clear()
    
    def size(self):
        """获取短期记忆大小"""
        return len(self.memory)

class LongTermMemory:
    """长期记忆"""
    def __init__(self, persist_directory="./long_term_memory"):
        """初始化长期记忆"""
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
        self.vectorstore = Chroma(
            persist_directory=persist_directory,
            embedding_function=self.embeddings
        )
    
    def add(self, content, metadata=None):
        """添加项目到长期记忆"""
        if metadata is None:
            metadata = {}
        self.vectorstore.add_texts([content], metadatas=[metadata])
    
    def search(self, query, k=5):
        """从长期记忆中搜索信息"""
        results = self.vectorstore.similarity_search(query, k=k)
        return results

class StructuredMemory:
    """结构化记忆"""
    def __init__(self):
        """初始化结构化记忆"""
        self.structure = {}
    
    def add(self, category, key, value):
        """添加结构化信息"""
        if category not in self.structure:
            self.structure[category] = {}
        self.structure[category][key] = value
    
    def get(self, category, key=None):
        """获取结构化信息"""
        if category not in self.structure:
            return None
        if key is None:
            return self.structure[category]
        return self.structure[category].get(key, None)
    
    def get_all(self):
        """获取所有结构化信息"""
        return self.structure

class GraphMemory:
    """Graph记忆"""
    def __init__(self):
        """初始化Graph记忆"""
        self.graph = nx.DiGraph()
    
    def add_node(self, node_id, attributes=None):
        """添加节点"""
        if attributes is None:
            attributes = {}
        self.graph.add_node(node_id, **attributes)
    
    def add_edge(self, source, target, relationship=None):
        """添加边"""
        if relationship is None:
            relationship = {}
        self.graph.add_edge(source, target, **relationship)
    
    def get_node(self, node_id):
        """获取节点"""
        if node_id in self.graph.nodes:
            return self.graph.nodes[node_id]
        return None
    
    def get_edges(self, node_id):
        """获取节点的边"""
        if node_id in self.graph.nodes:
            return list(self.graph.edges(node_id, data=True))
        return []

class MultimodalFusion:
    """多模态融合"""
    def __init__(self):
        """初始化多模态融合"""
        self.model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
        self.processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
    
    def fuse(self, text=None, image=None):
        """融合文本和图像"""
        inputs = self.processor(
            text=text,
            images=image,
            return_tensors="pt",
            padding=True
        )
        
        with torch.no_grad():
            outputs = self.model(**inputs)
        
        return outputs

class MultimodalMemorySystem:
    """Multimodal Memory系统"""
    def __init__(self):
        """初始化Multimodal Memory系统"""
        self.stm = ShortTermMemory()
        self.ltm = LongTermMemory()
        self.sm = StructuredMemory()
        self.gm = GraphMemory()
        self.fusion = MultimodalFusion()
    
    def add(self, content, metadata=None, category=None, key=None, node_id=None, relationships=None):
        """添加信息到系统"""
        # 添加到短期记忆
        self.stm.add({"content": content, "metadata": metadata})
        
        # 添加到长期记忆
        if metadata is None:
            metadata = {}
        self.ltm.add(content, metadata)
        
        # 添加到结构化记忆
        if category and key:
            self.sm.add(category, key, content)
        
        # 添加到Graph记忆
        if node_id:
            self.gm.add_node(node_id, metadata)
            if relationships:
                for target, rel in relationships.items():
                    self.gm.add_edge(node_id, target, rel)
    
    def retrieve(self, query, k=5):
        """检索信息"""
        # 从长期记忆中检索
        ltm_results = self.ltm.search(query, k=k)
        
        # 从结构化记忆中检索
        sm_results = []
        for category, items in self.sm.get_all().items():
            for key, value in items.items():
                if query in key or query in value:
                    sm_results.append({"category": category, "key": key, "value": value})
        
        # 从Graph记忆中检索
        gm_results = []
        for node in self.gm.graph.nodes:
            node_data = self.gm.get_node(node)
            if query in str(node) or query in str(node_data):
                gm_results.append({"node": node, "data": node_data})
        
        return {
            "ltm_results": ltm_results,
            "sm_results": sm_results,
            "gm_results": gm_results
        }
    
    def fuse_multimodal(self, text=None, image=None):
        """融合多模态信息"""
        return self.fusion.fuse(text=text, image=image)

# 示例使用
memory_system = MultimodalMemorySystem()

# 添加信息
memory_system.add(
    "Agentic RAG是一种先进的检索增强生成技术",
    metadata={"type": "text", "category": "AI"},
    category="AI",
    key="RAG",
    node_id="RAG",
    relationships={"LLM": {"relationship": "使用"}, "VectorStore": {"relationship": "依赖"}}
)

memory_system.add(
    "Multimodal Memory系统支持多种模态",
    metadata={"type": "text", "category": "AI"},
    category="AI",
    key="Multimodal Memory",
    node_id="Multimodal Memory",
    relationships={"RAG": {"relationship": "增强"}}
)

# 检索信息
results = memory_system.retrieve("RAG")
print("检索结果:")
print("长期记忆结果:")
for result in results["ltm_results"]:
    print(result.page_content)

print("结构化记忆结果:")
for result in results["sm_results"]:
    print(result)

print("Graph记忆结果:")
for result in results["gm_results"]:
    print(result)

# app.py
from fastapi import FastAPI, Query
from multimodal_memory import MultimodalMemorySystem

app = FastAPI()

# 初始化Multimodal Memory系统
memory_system = MultimodalMemorySystem()

@app.post("/add")
async def add(
    content: str = Query(..., description="内容"),
    metadata: dict = Query({}, description="元数据"),
    category: str = Query(None, description="类别"),
    key: str = Query(None, description="键"),
    node_id: str = Query(None, description="节点ID"),
    relationships: dict = Query({}, description="关系")
):
    """添加信息到记忆系统"""
    memory_system.add(
        content=content,
        metadata=metadata,
        category=category,
        key=key,
        node_id=node_id,
        relationships=relationships
    )
    return {"message": "信息添加成功"}

@app.get("/retrieve")
async def retrieve(
    query: str = Query(..., description="查询"),
    k: int = Query(5, description="返回结果数量")
):
    """从记忆系统中检索信息"""
    results = memory_system.retrieve(query, k=k)
    return results

@app.post("/fuse")
async def fuse(
    text: str = Query(None, description="文本"),
    image: str = Query(None, description="图像路径")
):
    """融合多模态信息"""
    results = memory_system.fuse_multimodal(text=text, image=image)
    return {"message": "多模态融合成功"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

from multimodal_memory import MultimodalMemorySystem

# 初始化Multimodal Memory系统
memory_system = MultimodalMemorySystem()

# 添加客服知识
memory_system.add(
    "如何重置密码？",
    metadata={"type": "text", "category": "客服"},
    category="客服",
    key="重置密码",
    node_id="重置密码",
    relationships={"账户": {"relationship": "相关"}}
)

memory_system.add(
    "如何查询订单状态？",
    metadata={"type": "text", "category": "客服"},
    category="客服",
    key="查询订单",
    node_id="查询订单",
    relationships={"订单": {"relationship": "相关"}}
)

# 处理用户查询
def handle_query(query):
    """处理用户查询"""
    # 检索相关信息
    results = memory_system.retrieve(query)
    
    # 基于检索结果生成回复
    reply = generate_reply(results, query)
    
    # 添加到短期记忆
    memory_system.add(query, metadata={"type": "text", "category": "用户查询"})
    memory_system.add(reply, metadata={"type": "text", "category": "系统回复"})
    
    return reply

# 示例查询
query = "如何重置我的密码？"
reply = handle_query(query)
print(f"用户: {query}")
print(f"客服: {reply}")

from multimodal_memory import MultimodalMemorySystem

# 初始化Multimodal Memory系统
memory_system = MultimodalMemorySystem()

# 添加医疗知识
memory_system.add(
    "糖尿病的症状包括多饮、多尿、多食和体重减轻",
    metadata={"type": "text", "category": "医疗"},
    category="疾病",
    key="糖尿病",
    node_id="糖尿病",
    relationships={"症状": {"relationship": "相关"}}
)

memory_system.add(
    "高血压的治疗包括药物治疗和生活方式干预",
    metadata={"type": "text", "category": "医疗"},
    category="疾病",
    key="高血压",
    node_id="高血压",
    relationships={"治疗": {"relationship": "相关"}}
)

# 处理医疗查询
def handle_medical_query(query):
    """处理医疗查询"""
    # 检索相关信息
    results = memory_system.retrieve(query)
    
    # 基于检索结果生成回复
    reply = generate_medical_reply(results, query)
    
    return reply

# 示例查询
query = "糖尿病的症状有哪些？"
reply = handle_medical_query(query)
print(f"医生: {query}")
print(f"系统: {reply}")

from multimodal_memory import MultimodalMemorySystem

# 初始化Multimodal Memory系统
memory_system = MultimodalMemorySystem()

# 添加教育知识
memory_system.add(
    "Python中的列表是一种可变序列类型",
    metadata={"type": "text", "category": "编程"},
    category="Python",
    key="列表",
    node_id="Python列表",
    relationships={"序列类型": {"relationship": "属于"}}
)

memory_system.add(
    "Python中的字典是一种键值对存储结构",
    metadata={"type": "text", "category": "编程"},
    category="Python",
    key="字典",
    node_id="Python字典",
    relationships={"存储结构": {"relationship": "属于"}}
)

# 处理学习查询
def handle_learning_query(query):
    """处理学习查询"""
    # 检索相关信息
    results = memory_system.retrieve(query)
    
    # 基于检索结果生成回复
    reply = generate_learning_reply(results, query)
    
    # 添加到短期记忆
    memory_system.add(query, metadata={"type": "text", "category": "学习查询"})
    memory_system.add(reply, metadata={"type": "text", "category": "学习回复"})
    
    return reply

# 示例查询
query = "Python中的列表是什么？"
reply = handle_learning_query(query)
print(f"学生: {query}")
print(f"系统: {reply}")

# 安装依赖
pip install langchain openai chromadb networkx transformers torch fastapi uvicorn

# 配置环境变量
export OPENAI_API_KEY=your-api-key

# 运行示例
python app.py