阶段 4：记忆、知识库系统与长期状态

📖 本阶段目标

• 理解 Agent 的记忆系统架构
• 掌握短期记忆和长期记忆的实现
• 学习 RAG（检索增强生成）技术
• 实践构建知识库问答 Agent

🧠 记忆系统架构

Agent 的记忆层次

┌─────────────────────────────────────────┐
│         工作记忆 (Working Memory)        │
│      当前对话上下文，即时可用            │
│         Token 限制：4K-128K              │
└─────────────────┬───────────────────────┘
                  │
┌─────────────────┴───────────────────────┐
│        短期记忆 (Short-term Memory)      │
│      最近的对话历史，会话级别            │
│         存储：内存/Redis                 │
└─────────────────┬───────────────────────┘
                  │
┌─────────────────┴───────────────────────┐
│        长期记忆 (Long-term Memory)       │
│    持久化知识，跨会话可用                │
│    存储：向量数据库 + 结构化数据库       │
└─────────────────────────────────────────┘

💾 短期记忆实现

1. 对话历史管理

# examples/stage4/conversation_memory.py
from typing import List, Dict
from collections import deque

class ConversationMemory:
    """对话记忆管理器"""

    def __init__(self, max_messages=20, max_tokens=4000):
        self.messages = deque(maxlen=max_messages)
        self.max_tokens = max_tokens
        self.system_message = None

    def set_system_message(self, content: str):
        """设置系统消息"""
        self.system_message = {"role": "system", "content": content}

    def add_user_message(self, content: str):
        """添加用户消息"""
        self.messages.append({"role": "user", "content": content})

    def add_assistant_message(self, content: str):
        """添加助手消息"""
        self.messages.append({"role": "assistant", "content": content})

    def get_messages(self) -> List[Dict]:
        """获取所有消息（包括系统消息）"""
        messages = []
        if self.system_message:
            messages.append(self.system_message)
        messages.extend(list(self.messages))
        return messages

    def get_recent_messages(self, n=10) -> List[Dict]:
        """获取最近 n 条消息"""
        messages = []
        if self.system_message:
            messages.append(self.system_message)

        recent = list(self.messages)[-n:]
        messages.extend(recent)
        return messages

    def clear(self):
        """清空对话历史"""
        self.messages.clear()

    def summarize_old_messages(self, llm_client):
        """总结旧消息以节省 token"""
        if len(self.messages) < 10:
            return

        # 获取较旧的消息
        old_messages = list(self.messages)[:5]

        # 使用 LLM 生成摘要
        summary_prompt = "请简要总结以下对话内容：\n\n"
        for msg in old_messages:
            summary_prompt += f"{msg['role']}: {msg['content']}\n"

        # ... 调用 LLM 获取摘要
        # summary = llm_client.chat(summary_prompt)

        # 移除旧消息，保留摘要
        for _ in range(5):
            self.messages.popleft()

        # 添加摘要作为系统消息
        self.messages.appendleft({
            "role": "system",
            "content": f"之前的对话摘要：{summary}"
        })

# 使用示例
if __name__ == "__main__":
    memory = ConversationMemory(max_messages=10)
    memory.set_system_message("你是一个有帮助的助手。")

    memory.add_user_message("你好")
    memory.add_assistant_message("你好！有什么可以帮助你的？")
    memory.add_user_message("我想了解 Python")
    memory.add_assistant_message("Python 是一种高级编程语言...")

    print("所有消息:")
    for msg in memory.get_messages():
        print(f"{msg['role']}: {msg['content']}")

2. 会话状态管理

# examples/stage4/session_manager.py
import json
from datetime import datetime
from typing import Dict, Any

class SessionManager:
    """会话状态管理器"""

    def __init__(self, session_id: str):
        self.session_id = session_id
        self.state = {
            "created_at": datetime.now().isoformat(),
            "last_active": datetime.now().isoformat(),
            "context": {},
            "metadata": {}
        }

    def set_context(self, key: str, value: Any):
        """设置上下文信息"""
        self.state["context"][key] = value
        self._update_activity()

    def get_context(self, key: str, default=None) -> Any:
        """获取上下文信息"""
        return self.state["context"].get(key, default)

    def set_metadata(self, key: str, value: Any):
        """设置元数据"""
        self.state["metadata"][key] = value
        self._update_activity()

    def _update_activity(self):
        """更新活动时间"""
        self.state["last_active"] = datetime.now().isoformat()

    def save(self, filepath: str):
        """保存会话状态"""
        with open(filepath, 'w', encoding='utf-8') as f:
            json.dump(self.state, f, ensure_ascii=False, indent=2)

    def load(self, filepath: str):
        """加载会话状态"""
        with open(filepath, 'r', encoding='utf-8') as f:
            self.state = json.load(f)

# 使用示例
if __name__ == "__main__":
    session = SessionManager("user_123_session_1")

    # 设置上下文
    session.set_context("user_name", "张三")
    session.set_context("topic", "Python 学习")
    session.set_context("skill_level", "初学者")

    # 设置元数据
    session.set_metadata("language", "zh-CN")
    session.set_metadata("device", "mobile")

    # 保存
    session.save("/tmp/session_user_123.json")

    print(f"用户名: {session.get_context('user_name')}")
    print(f"主题: {session.get_context('topic')}")

🗄️ 长期记忆与 RAG

1. 向量数据库基础

# examples/stage4/vector_store.py
from typing import List, Dict
import numpy as np
from openai import OpenAI

client = OpenAI()

class SimpleVectorStore:
    """简单的向量存储（实际应使用专业向量数据库）"""

    def __init__(self):
        self.documents = []
        self.embeddings = []
        self.metadata = []

    def get_embedding(self, text: str) -> List[float]:
        """获取文本的向量表示"""
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return response.data[0].embedding

    def add_document(self, text: str, metadata: Dict = None):
        """添加文档"""
        embedding = self.get_embedding(text)

        self.documents.append(text)
        self.embeddings.append(embedding)
        self.metadata.append(metadata or {})

    def cosine_similarity(self, vec1: List[float], vec2: List[float]) -> float:
        """计算余弦相似度"""
        vec1 = np.array(vec1)
        vec2 = np.array(vec2)

        return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))

    def search(self, query: str, top_k: int = 3) -> List[Dict]:
        """搜索相关文档"""
        query_embedding = self.get_embedding(query)

        # 计算所有文档的相似度
        similarities = []
        for i, doc_embedding in enumerate(self.embeddings):
            similarity = self.cosine_similarity(query_embedding, doc_embedding)
            similarities.append({
                "index": i,
                "similarity": similarity,
                "document": self.documents[i],
                "metadata": self.metadata[i]
            })

        # 排序并返回 top_k
        similarities.sort(key=lambda x: x["similarity"], reverse=True)
        return similarities[:top_k]

# 使用示例
if __name__ == "__main__":
    store = SimpleVectorStore()

    # 添加文档
    documents = [
        "Python 是一种高级编程语言，由 Guido van Rossum 创建于 1991 年。",
        "JavaScript 是一种用于网页开发的脚本语言，主要运行在浏览器中。",
        "Java 是一种面向对象的编程语言，广泛用于企业级应用开发。",
        "人工智能是计算机科学的一个分支，研究如何让机器模拟人类智能。"
    ]

    for doc in documents:
        store.add_document(doc, {"source": "knowledge_base"})

    # 搜索
    query = "什么是 Python？"
    results = store.search(query, top_k=2)

    print(f"查询: {query}\n")
    for i, result in enumerate(results, 1):
        print(f"结果 {i} (相似度: {result['similarity']:.4f}):")
        print(f"{result['document']}\n")

2. RAG 系统实现

# examples/stage4/rag_agent.py
from openai import OpenAI
from typing import List, Dict
import json

client = OpenAI()

class RAGAgent:
    """检索增强生成 Agent"""

    def __init__(self, vector_store):
        self.vector_store = vector_store

    def retrieve(self, query: str, top_k: int = 3) -> List[str]:
        """检索相关文档"""
        results = self.vector_store.search(query, top_k)
        return [r["document"] for r in results]

    def generate_answer(self, query: str, context_docs: List[str]) -> str:
        """基于检索到的文档生成答案"""
        context = "\n\n".join([f"文档 {i+1}: {doc}" 
                               for i, doc in enumerate(context_docs)])

        prompt = f"""
        基于以下文档回答问题。如果文档中没有相关信息，请明确说明。

        文档:
        {context}

        问题: {query}

        请给出准确、简洁的答案。
        """

        response = client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "你是一个知识问答助手，基于提供的文档回答问题。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3
        )

        return response.choices[0].message.content

    def answer(self, query: str) -> Dict:
        """完整的 RAG 流程"""
        # 1. 检索
        retrieved_docs = self.retrieve(query, top_k=3)

        # 2. 生成
        answer = self.generate_answer(query, retrieved_docs)

        return {
            "query": query,
            "retrieved_documents": retrieved_docs,
            "answer": answer
        }

# 使用示例
if __name__ == "__main__":
    # 初始化向量存储并添加文档
    from vector_store import SimpleVectorStore

    store = SimpleVectorStore()

    # 添加知识库文档
    knowledge_base = [
        "Python 是一种解释型、面向对象的高级编程语言，由 Guido van Rossum 于 1991 年创建。它以简洁的语法和强大的功能而闻名。",
        "Python 的主要特点包括：简洁易读的语法、丰富的标准库、跨平台兼容性、强大的第三方生态系统。",
        "Python 常用于 Web 开发、数据分析、人工智能、自动化脚本等领域。",
        "机器学习是人工智能的一个分支，它使计算机能够从数据中学习，而无需显式编程。",
        "深度学习是机器学习的一个子领域，使用多层神经网络来处理复杂的数据。",
    ]

    for doc in knowledge_base:
        store.add_document(doc)

    # 创建 RAG Agent
    rag_agent = RAGAgent(store)

    # 提问
    questions = [
        "Python 是什么时候创建的？",
        "Python 有哪些特点？",
        "什么是深度学习？"
    ]

    for question in questions:
        print("=" * 60)
        print(f"问题: {question}")
        print("-" * 60)

        result = rag_agent.answer(question)

        print("检索到的文档:")
        for i, doc in enumerate(result["retrieved_documents"], 1):
            print(f"{i}. {doc[:100]}...")

        print(f"\n答案:\n{result['answer']}")
        print()

📚 实践项目：个人笔记问答 Agent

# examples/stage4/note_agent.py
from openai import OpenAI
import os
from datetime import datetime
from typing import List, Dict

client = OpenAI()

class NoteAgent:
    """个人笔记管理和问答 Agent"""

    def __init__(self, vector_store):
        self.vector_store = vector_store
        self.notes = []

    def add_note(self, title: str, content: str, tags: List[str] = None):
        """添加笔记"""
        note = {
            "id": len(self.notes) + 1,
            "title": title,
            "content": content,
            "tags": tags or [],
            "created_at": datetime.now().isoformat()
        }

        self.notes.append(note)

        # 添加到向量存储
        full_text = f"{title}\n{content}"
        self.vector_store.add_document(
            full_text,
            metadata={
                "note_id": note["id"],
                "title": title,
                "tags": tags
            }
        )

        return note

    def search_notes(self, query: str, top_k: int = 5) -> List[Dict]:
        """搜索笔记"""
        results = self.vector_store.search(query, top_k)

        # 获取完整笔记信息
        note_results = []
        for result in results:
            note_id = result["metadata"]["note_id"]
            note = self.notes[note_id - 1]
            note_results.append({
                "note": note,
                "similarity": result["similarity"]
            })

        return note_results

    def answer_from_notes(self, question: str) -> Dict:
        """从笔记中回答问题"""
        # 搜索相关笔记
        relevant_notes = self.search_notes(question, top_k=3)

        if not relevant_notes:
            return {
                "question": question,
                "answer": "抱歉，我在你的笔记中没有找到相关信息。",
                "sources": []
            }

        # 构建上下文
        context = "\n\n".join([
            f"笔记：{n['note']['title']}\n{n['note']['content']}"
            for n in relevant_notes
        ])

        # 生成答案
        prompt = f"""
        基于我的以下笔记内容回答问题：

        {context}

        问题：{question}

        请基于笔记内容给出答案。如果笔记中没有相关信息，请说明。
        """

        response = client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "你是一个个人笔记助手，帮助用户从笔记中找到信息。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3
        )

        return {
            "question": question,
            "answer": response.choices[0].message.content,
            "sources": [n["note"]["title"] for n in relevant_notes]
        }

    def summarize_notes_by_tag(self, tag: str) -> str:
        """总结某个标签的所有笔记"""
        tagged_notes = [n for n in self.notes if tag in n.get("tags", [])]

        if not tagged_notes:
            return f"没有找到标签为 '{tag}' 的笔记。"

        notes_content = "\n\n".join([
            f"标题：{n['title']}\n内容：{n['content']}"
            for n in tagged_notes
        ])

        prompt = f"""
        请总结以下关于 '{tag}' 的笔记：

        {notes_content}

        生成一个简洁的总结，提取主要观点和关键信息。
        """

        response = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.5
        )

        return response.choices[0].message.content

# 使用示例
if __name__ == "__main__":
    from vector_store import SimpleVectorStore

    # 创建笔记 Agent
    store = SimpleVectorStore()
    agent = NoteAgent(store)

    # 添加笔记
    agent.add_note(
        "Python 基础",
        "Python 是一种解释型语言，语法简洁。主要特点：动态类型、自动内存管理、丰富的库支持。",
        tags=["编程", "Python"]
    )

    agent.add_note(
        "机器学习入门",
        "机器学习是 AI 的子领域。主要分类：监督学习、无监督学习、强化学习。常用库：scikit-learn, TensorFlow, PyTorch。",
        tags=["AI", "机器学习"]
    )

    agent.add_note(
        "Web 开发框架",
        "Python Web 框架：Flask（轻量级）、Django（全栈）、FastAPI（异步高性能）。",
        tags=["编程", "Python", "Web"]
    )

    # 问答
    print("=== 笔记问答 ===\n")

    result = agent.answer_from_notes("Python 有哪些 Web 框架？")
    print(f"问题：{result['question']}")
    print(f"答案：{result['answer']}")
    print(f"来源：{', '.join(result['sources'])}")
    print()

    # 标签总结
    print("=== 标签总结 ===\n")
    summary = agent.summarize_notes_by_tag("Python")
    print(f"'Python' 标签总结：\n{summary}")

🔧 高级技术

1. 分块策略

def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> List[str]:
    """将长文本分块"""
    chunks = []
    start = 0

    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        chunks.append(chunk)
        start = end - overlap  # 重叠部分

    return chunks

2. 混合检索

def hybrid_search(query: str, vector_store, keyword_index):
    """结合向量搜索和关键词搜索"""
    # 向量检索
    vector_results = vector_store.search(query, top_k=10)

    # 关键词检索
    keyword_results = keyword_index.search(query, top_k=10)

    # 合并结果（可以使用 RRF 等算法）
    combined_results = merge_results(vector_results, keyword_results)

    return combined_results

✅ 阶段完成检查清单

• [ ] 理解 Agent 记忆系统的层次结构
• [ ] 实现了对话历史管理
• [ ] 掌握向量数据库的基本使用
• [ ] 理解 RAG 的工作原理
• [ ] 实现了基于 RAG 的问答系统
• [ ] 完成个人笔记 Agent 项目
• [ ] 了解文本分块和混合检索策略

🎯 下一步

完成本阶段后，进入 阶段 5：多 Agent 协同与角色体系，学习如何构建多 Agent 协作系统。

---

💡 小贴士：记忆是 Agent 的"经验积累"，好的记忆系统让 Agent 越用越智能。推荐使用专业向量数据库如 Pinecone、Weaviate 或 Qdrant。