当前位置：首页>python>Python AI Agent 零基础教程 | 第7篇:记忆系统——让AI Agent记住上下文

Python AI Agent 零基础教程 | 第7篇:记忆系统——让AI Agent记住上下文

2026-07-02 12:57:04

Python AI Agent 零基础教程 | 第7篇：记忆系统——让AI Agent记住上下文

前言

你有没有遇到过这种情况：和 AI 聊了半天后，第二天再问它，它完全不记得你们之前聊过什么？这就是因为 AI 缺少"记忆系统"！今天我们就来学习如何给 AI Agent 装上"记忆"，让它能够记住上下文，甚至长期学习。

一、为什么 AI Agent 需要记忆？

1.1 对比：传统 AI vs 有记忆的 AI

对比项	传统 AI	有记忆的 AI Agent
记住之前说的	❌ 每次都是新对话	✅ 可以记住历史
了解用户偏好	❌ 不了解	✅ 可以记住用户习惯
持续学习	❌ 不学习	✅ 可以积累知识
对话连贯性	❌ 可能前后矛盾	✅ 保持一致

1.2 记忆的类型

Python 代码

┌─────────────────────────────────────────────────────────────┐
│                    AI Agent 记忆系统                        │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │                    长期记忆                          │   │
│  │  ┌─────────┐  ┌─────────┐  ┌─────────┐            │   │
│  │  │ 知识库  │  │用户画像 │  │ 经验库  │            │   │
│  │  └─────────┘  └─────────┘  └─────────┘            │   │
│  └─────────────────────────────────────────────────────┘   │
│                           ▲                                 │
│                           │                                 │
│  ┌─────────────────────────────────────────────────────┐   │
│  │                    短期记忆                          │   │
│  │  ┌─────────────────────────────────────────────┐   │   │
│  │  │              对话历史（上下文）               │   │   │
│  │  └─────────────────────────────────────────────┘   │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

1.3 记忆系统的作用

-上下文理解：理解"它"指代什么

-个性化服务：记住用户偏好，提供更好的服务

-知识积累：从交互中学习，不断进步

-连续对话：实现真正的多轮对话

二、短期记忆：对话历史管理

2.1 最简单的记忆实现

Python 代码

class SimpleMemoryAgent:
    """带简单记忆的 Agent"""
    
    def __init__(self, api_key):
        self.api_key = api_key
        self.messages = []  # 对话历史就是记忆
    
    def set_system_prompt(self, prompt):
        """设置系统提示词"""
        self.messages = [{"role": "system", "content": prompt}]
    
    def think(self, user_input):
        """带记忆的思考"""
        
        # 1. 添加用户消息到记忆
        self.messages.append({"role": "user", "content": user_input})
        
        # 2. 调用 API
        response = self._call_api()
        
        # 3. 添加 AI 回复到记忆
        ai_message = response["choices"][0]["message"]
        self.messages.append(ai_message)
        
        return ai_message["content"]
    
    def _call_api(self):
        # API 调用...
        pass
    
    def show_memory(self):
        """查看记忆内容"""
        print("\n📝 当前记忆：")
        for msg in self.messages:
            role = msg["role"]
            content = msg["content"][:50] + "..." if len(msg["content"]) > 50 else msg["content"]
            print(f"  {role}: {content}")

2.2 记忆容量管理

AI 有上下文长度限制（通常是 4K-128K tokens），当对话太长时需要管理记忆。

Python 代码

class ManagedMemoryAgent:
    """带记忆管理的 Agent"""
    
    def __init__(self, api_key, max_tokens=3000):
        self.api_key = api_key
        self.max_tokens = max_tokens
        self.messages = []
    
    def think(self, user_input):
        """思考（自动管理记忆）"""
        
        # 添加用户消息
        self.messages.append({"role": "user", "content": user_input})
        
        # 检查是否需要压缩记忆
        if self._estimate_tokens() > self.max_tokens:
            self._compress_memory()
        
        # 调用 API
        response = self._call_api()
        
        # 添加 AI 回复
        ai_message = response["choices"][0]["message"]
        self.messages.append(ai_message)
        
        return ai_message["content"]
    
    def _estimate_tokens(self):
        """估算当前 tokens 数量"""
        total = 0
        for msg in self.messages:
            # 粗略估算：中文约 2 tokens/字
            total += len(msg["content"]) * 1.3
        return int(total)
    
    def _compress_memory(self):
        """压缩记忆（保留重要信息）"""
        
        # 策略：保留 system 消息 + 最近 10 条对话
        if len(self.messages) > 11:
            system_msg = self.messages[0] if self.messages[0]["role"] == "system" else None
            recent_msgs = self.messages[-10:]
            
            if system_msg:
                self.messages = [system_msg] + recent_msgs
            else:
                self.messages = recent_msgs
        
        print(f"📦 记忆已压缩，当前 {len(self.messages)} 条消息")

2.3 智能记忆管理

Python 代码

class SmartMemoryAgent:
    """智能记忆管理 Agent"""
    
    def __init__(self, api_key):
        self.api_key = api_key
        self.messages = []
        self.important_info = []  # 重要信息单独存储
    
    def think(self, user_input):
        """思考（提取重要信息）"""
        
        self.messages.append({"role": "user", "content": user_input})
        
        response = self._call_api()
        ai_message = response["choices"][0]["message"]
        self.messages.append(ai_message)
        
        # 尝试提取重要信息
        self._extract_important_info(user_input, ai_message["content"])
        
        return ai_message["content"]
    
    def _extract_important_info(self, user_input, ai_response):
        """从对话中提取重要信息"""
        
        # 这里可以用关键词检测来识别重要信息
        important_keywords = ["名字", "生日", "喜好", "地址", "电话", "邮箱"]
        
        for keyword in important_keywords:
            if keyword in user_input or keyword in ai_response:
                # 提取包含关键词的句子
                if keyword in user_input:
                    self.important_info.append(f"用户提到：{keyword}")
                    print(f"💡 记住重要信息：{keyword}")
    
    def get_user_profile(self):
        """获取用户画像"""
        return {
            "important_info": self.important_info,
            "conversation_count": len(self.messages) // 2
        }

三、长期记忆：用户画像

3.1 用户画像结构

Python 代码

class UserProfile:
    """用户画像"""
    
    def __init__(self, user_id):
        self.user_id = user_id
        self.name = None
        self.preferences = {}  # 偏好
        self.history = []       # 历史交互
        self.knowledge = {}     # 已知知识
    
    def update(self, key, value):
        """更新用户信息"""
        if key in ["name", "age", "city"]:
            setattr(self, key, value)
        else:
            self.preferences[key] = value
        
        print(f"✅ 已更新用户画像：{key} = {value}")
    
    def get_summary(self):
        """获取用户摘要"""
        return f"""
用户ID：{self.user_id}
姓名：{self.name or "未知"}
偏好：{self.preferences}
"""
    
    def to_dict(self):
        """转换为字典"""
        return {
            "user_id": self.user_id,
            "name": self.name,
            "preferences": self.preferences,
            "history": self.history,
            "knowledge": self.knowledge
        }
    
    @classmethod
    def from_dict(cls, data):
        """从字典创建"""
        profile = cls(data["user_id"])
        profile.name = data.get("name")
        profile.preferences = data.get("preferences", {})
        profile.history = data.get("history", [])
        profile.knowledge = data.get("knowledge", {})
        return profile

3.2 带用户画像的 Agent

Python 代码

class PersonalizedAgent:
    """个性化 Agent"""
    
    def __init__(self, api_key, user_id):
        self.api_key = api_key
        self.messages = []
        self.user_profile = UserProfile(user_id)
    
    def set_system_prompt(self):
        """根据用户画像设置系统提示词"""
        
        base_prompt = """你是一个智能助手，能够记住用户信息并提供个性化服务。"""
        
        # 如果有用户信息，添加到系统提示词
        if self.user_profile.name:
            base_prompt += f"\n当前用户名叫 {self.user_profile.name}。"
        
        if self.user_profile.preferences:
            prefs = ", ".join([f"{k}={v}" for k, v in self.user_profile.preferences.items()])
            base_prompt += f"\n用户偏好：{prefs}"
        
        self.messages = [{"role": "system", "content": base_prompt}]
    
    def think(self, user_input):
        """思考"""
        
        # 更新用户画像
        self._update_profile_from_input(user_input)
        
        # 正常对话流程
        self.messages.append({"role": "user", "content": user_input})
        
        response = self._call_api()
        ai_message = response["choices"][0]["message"]
        self.messages.append(ai_message)
        
        return ai_message["content"]
    
    def _update_profile_from_input(self, user_input):
        """从输入中更新用户画像"""
        
        # 简单的关键词检测
        if "我叫" in user_input:
            name = user_input.split("我叫")[1].split()[0]
            self.user_profile.update("name", name)
        
        if "我喜欢" in user_input:
            like = user_input.split("我喜欢")[1].split()[0]
            self.user_profile.update("hobby", like)
    
    def save_profile(self, filename):
        """保存用户画像"""
        import json
        with open(filename, "w", encoding="utf-8") as f:
            json.dump(self.user_profile.to_dict(), f, ensure_ascii=False, indent=2)
        print(f"✅ 用户画像已保存到 {filename}")
    
    def load_profile(self, filename):
        """加载用户画像"""
        import json
        with open(filename, "r", encoding="utf-8") as f:
            data = json.load(f)
            self.user_profile = UserProfile.from_dict(data)
        print(f"✅ 用户画像已加载")

四、持久化存储

4.1 文件存储

Python 代码

import json
from datetime import datetime

class PersistentAgent:
    """持久化 Agent"""
    
    def __init__(self, api_key, user_id):
        self.api_key = api_key
        self.user_id = user_id
        self.conversation_file = f"conversation_{user_id}.json"
        self.profile_file = f"profile_{user_id}.json"
        
        # 加载已有数据
        self.messages = self._load_conversation()
        self.user_profile = self._load_profile()
    
    def think(self, user_input):
        """思考并自动保存"""
        
        self.messages.append({"role": "user", "content": user_input})
        
        response = self._call_api()
        ai_message = response["choices"][0]["message"]
        self.messages.append(ai_message)
        
        # 自动保存
        self._save_conversation()
        
        return ai_message["content"]
    
    def _load_conversation(self):
        """加载对话历史"""
        try:
            with open(self.conversation_file, "r", encoding="utf-8") as f:
                return json.load(f)
        except FileNotFoundError:
            return []
    
    def _save_conversation(self):
        """保存对话历史"""
        with open(self.conversation_file, "w", encoding="utf-8") as f:
            json.dump(self.messages, f, ensure_ascii=False, indent=2)
    
    def _load_profile(self):
        """加载用户画像"""
        try:
            with open(self.profile_file, "r", encoding="utf-8") as f:
                return UserProfile.from_dict(json.load(f))
        except FileNotFoundError:
            return UserProfile(self.user_id)
    
    def _save_profile(self):
        """保存用户画像"""
        with open(self.profile_file, "w", encoding="utf-8") as f:
            json.dump(self.user_profile.to_dict(), f, ensure_ascii=False, indent=2)

4.2 SQLite 数据库存储

Python 代码

import sqlite3
from datetime import datetime

class DatabaseAgent:
    """使用数据库的 Agent"""
    
    def __init__(self, api_key, db_name="agent_memory.db"):
        self.api_key = api_key
        self.conn = sqlite3.connect(db_name)
        self._init_database()
    
    def _init_database(self):
        """初始化数据库"""
        cursor = self.conn.cursor()
        
        # 创建对话表
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS conversations (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id TEXT,
                role TEXT,
                content TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        
        # 创建用户画像表
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS user_profiles (
                user_id TEXT PRIMARY KEY,
                name TEXT,
                preferences TEXT,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        
        self.conn.commit()
    
    def save_message(self, user_id, role, content):
        """保存消息"""
        cursor = self.conn.cursor()
        cursor.execute(
            "INSERT INTO conversations (user_id, role, content) VALUES (?, ?, ?)",
            (user_id, role, content)
        )
        self.conn.commit()
    
    def get_conversation_history(self, user_id, limit=20):
        """获取对话历史"""
        cursor = self.conn.cursor()
        cursor.execute(
            "SELECT role, content FROM conversations WHERE user_id = ? ORDER BY id DESC LIMIT ?",
            (user_id, limit)
        )
        results = cursor.fetchall()
        return [{"role": r[0], "content": r[1]} for r in reversed(results)]
    
    def save_profile(self, user_id, name, preferences):
        """保存用户画像"""
        cursor = self.conn.cursor()
        cursor.execute("""
            INSERT OR REPLACE INTO user_profiles (user_id, name, preferences, updated_at)
            VALUES (?, ?, ?, ?)
        """, (user_id, name, json.dumps(preferences), datetime.now()))
        self.conn.commit()
    
    def close(self):
        """关闭数据库连接"""
        self.conn.close()

五、完整记忆系统实现

5.1 综合记忆 Agent

Python 代码

class FullMemoryAgent:
    """完整记忆系统 Agent"""
    
    def __init__(self, api_key, user_id="default"):
        self.api_key = api_key
        self.user_id = user_id
        
        # 短期记忆
        self.short_term_memory = []
        
        # 长期记忆
        self.long_term_memory = {
            "user_profile": {},
            "knowledge_base": {},
            "conversation_summary": ""
        }
        
        # 配置
        self.max_short_term = 20  # 最多保留20条对话
    
    def set_system_prompt(self, base_prompt):
        """设置系统提示词"""
        
        # 构建完整的系统提示词
        full_prompt = base_prompt + "\n\n"
        
        # 添加用户画像
        if self.long_term_memory["user_profile"]:
            profile = self.long_term_memory["user_profile"]
            full_prompt += f"用户信息：{json.dumps(profile, ensure_ascii=False)}\n"
        
        # 添加对话摘要
        if self.long_term_memory["conversation_summary"]:
            full_prompt += f"对话摘要：{self.long_term_memory['conversation_summary']}\n"
        
        self.system_prompt = full_prompt
        self.short_term_memory = [{"role": "system", "content": full_prompt}]
    
    def think(self, user_input):
        """思考"""
        
        # 1. 添加到短期记忆
        self.short_term_memory.append({
            "role": "user",
            "content": user_input
        })
        
        # 2. 检查记忆容量
        if len(self.short_term_memory) > self.max_short_term:
            self._summarize_and_compress()
        
        # 3. 调用 API
        response = self._call_api()
        ai_message = response["choices"][0]["message"]
        
        # 4. 添加回复到短期记忆
        self.short_term_memory.append(ai_message)
        
        # 5. 提取重要信息到长期记忆
        self._extract_to_long_term(user_input, ai_message["content"])
        
        return ai_message["content"]
    
    def _call_api(self):
        # API 调用...
        pass
    
    def _summarize_and_compress(self):
        """总结并压缩记忆"""
        
        # 保留系统提示词 + 最近的对话
        system_msg = self.short_term_memory[0]
        recent_msgs = self.short_term_memory[-10:]
        
        # 更新短期记忆
        self.short_term_memory = [system_msg] + recent_msgs
        
        # 更新对话摘要
        self.long_term_memory["conversation_summary"] = "最近的对话涉及..."
        
        print("📦 记忆已压缩")
    
    def _extract_to_long_term(self, user_input, ai_response):
        """提取信息到长期记忆"""
        
        # 提取用户信息
        if "我叫" in user_input:
            name = user_input.split("我叫")[1].split()[0]
            self.long_term_memory["user_profile"]["name"] = name
        
        # 提取偏好
        if "喜欢" in user_input:
            self.long_term_memory["user_profile"]["preference"] = user_input
    
    def show_memory(self):
        """显示记忆状态"""
        print("\n" + "="*50)
        print("🧠 记忆状态")
        print("="*50)
        print(f"短期记忆：{len(self.short_term_memory)} 条")
        print(f"长期记忆：{json.dumps(self.long_term_memory, ensure_ascii=False, indent=2)}")
        print("="*50)

六、实战：记忆助手

6.1 完整代码

Python 代码

import requests
import json

class MemoryAssistant:
    """有记忆的个人助手"""
    
    def __init__(self, api_key, user_name="用户"):
        self.api_key = api_key
        self.user_name = user_name
        self.short_term = []  # 短期记忆
        self.long_term = {
            "name": user_name,
            "preferences": {},
            "todos": [],
            "facts": []
        }
        
        # 系统提示词
        self.system_prompt = f"""你是一个智能个人助手，名字叫小智。

你需要：
记住用户的名字和个人信息
记住用户的偏好和习惯
保持对话的连贯性
主动询问用户需求
当前用户：{user_name}"""
        
        self.short_term.append({"role": "system", "content": self.system_prompt})
    
    def think(self, user_input):
        """思考并回复"""
        
        # 记录用户消息
        self.short_term.append({"role": "user", "content": user_input})
        
        # 提取信息到长期记忆
        self._learn_from_input(user_input)
        
        # 调用 API
        response = self._call_api()
        ai_message = response["choices"][0]["message"]
        
        # 记录 AI 回复
        self.short_term.append(ai_message)
        
        # 管理记忆容量
        if len(self.short_term) > 20:
            self._compress_memory()
        
        return ai_message["content"]
    
    def _learn_from_input(self, user_input):
        """从输入中学习"""
        
        # 记住名字
        if "我叫" in user_input or "我是" in user_input:
            for keyword in ["我叫", "我是"]:
                if keyword in user_input:
                    parts = user_input.split(keyword)
                    if len(parts) > 1:
                        name = parts[1].split()[0].strip("，。")
                        self.long_term["name"] = name
                        self.long_term["facts"].append(f"用户名叫{name}")
                        break
        
        # 记住偏好
        if "我喜欢" in user_input:
            like = user_input.split("我喜欢")[1].split()[0]
            self.long_term["preferences"]["喜欢"] = like
        
        if "我不喜欢" in user_input:
            dislike = user_input.split("我不喜欢")[1].split()[0]
            self.long_term["preferences"]["不喜欢"] = dislike
    
    def _compress_memory(self):
        """压缩记忆"""
        # 保留系统消息 + 最近对话
        self.short_term = [self.short_term[0]] + self.short_term[-12:]
        print("📝 记忆已优化")
    
    def _call_api(self):
        url = "https://api.openai.com/v1/chat/completions"
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}"
        }
        data = {
            "model": "gpt-3.5-turbo",
            "messages": self.short_term
        }
        response = requests.post(url, headers=headers, json=data)
        return response.json()
    
    def show_knowledge(self):
        """展示已学到的知识"""
        print("\n📚 我记住的关于你的信息：")
        print(f"  名字：{self.long_term.get('name', '未知')}")
        if self.long_term.get("preferences"):
            print(f"  偏好：{self.long_term['preferences']}")
        if self.long_term.get("facts"):
            print(f"  事实：{self.long_term['facts']}")
    
    def run(self):
        """运行交互式对话"""
        print("="*50)
        print("🧠 记忆助手已启动！")
        print("输入 'quit' 退出，'memory' 查看记忆")
        print("="*50)
        
        while True:
            try:
                user_input = input("\n👤 你：").strip()
                
                if not user_input:
                    continue
                
                if user_input.lower() == "quit":
                    print("再见！")
                    break
                
                elif user_input.lower() == "memory":
                    self.show_knowledge()
                
                else:
                    response = self.think(user_input)
                    print(f"\n🤖 小智：{response}")
            
            except KeyboardInterrupt:
                print("\n再见！")
                break


使用
if __name__ == "__main__":
    assistant = MemoryAssistant("sk-xxxxxx", "小明")
    assistant.run()

6.2 运行效果

Python 代码

==================================================
🧠 记忆助手已启动！
输入 'quit' 退出，'memory' 查看记忆
==================================================

👤 你：我叫张三，25岁，在北京工作

🤖 小智：你好张三！很高兴认识你。我是你的智能助手小智，有什么我可以帮你的吗？

👤 你：我喜欢编程，不喜欢加班

📝 记忆已优化

🤖 小智：明白了！你喜欢编程，不喜欢加班。作为程序员，我们都希望能更好地平衡工作和生活。

👤 你：memory
📚 我记住的关于你的信息：
  名字：张三
  偏好：{'喜欢': '编程', '不喜欢': '加班'}
  事实：['用户名叫张三']

👤 你：明天下午3点提醒我开会

🤖 小智：好的张三，我已经记住了！明天下午3点我会提醒你开会。还有什么需要我帮忙的吗？

👤 你：你还记得我叫什么吗？

🤖 小智：当然记得，你叫张三，今年25岁，在北京工作。喜欢编程，不喜欢加班。有什么需要我帮忙的吗？

七、本章小结

今天我们学习了：

记忆类型	实现方式
✅ 短期记忆	对话历史列表
✅ 记忆管理	Token 限制 + 压缩
✅ 长期记忆	用户画像 + 知识库
✅ 持久化	文件存储 + 数据库
✅ 智能学习	自动提取重要信息

---

下期预告

第8篇：工具调用——扩展 AI Agent 的能力

下一期我们将学习：

- 什么是工具调用

- 内置工具设计

- 动态工具注册

- 实战：天气查询 Agent

👨‍💻 作者：鹏鹏 | 专注于 AI + 编程教育

📱 关注公众号「跟着鹏鹏学技术」

💬 动手练习：给你的 Agent 添加记忆功能！

往期精选