1. 项目背景
小陈是一名自学 Python 的转行程序员,他用 CrewAI 做了一个"个人学习教练 Crew":输入想学的知识点,Agent 帮他制定学习计划。第一版只能"一问一答"——你问"怎么学 FastAPI",Agent 给出学习路径。等你学完再问"接下来学什么",Agent 完全不记得你刚学过 FastAPI,又从零开始给你推荐 Python 基础。每次对话都是"第一次见面"。
更糟的是,当小陈连续学习了三周后,Agent 依然不知道他的学习进度——哪些章节已掌握、哪些薄弱、每天的学习时长偏好。没有记忆的 CrewAI,就像一个每天失忆的老师——态度认真,但跟学生之间没有任何积累。
核心痛点:默认的 CrewAI Agent 是无状态的——每次 kickoff 都是全新的上下文。但在需要持续交互的场景中(学习教练、客服助手、个人助理),"记住上一次发生了什么"是刚需。CrewAI 提供了三层记忆机制:短期记忆(当前会话上下文)、长期记忆(跨会话的用户偏好和历史)、实体记忆(关键实体之间的关系)。
本章的目标是理解三种记忆的工作方式、配置方法、风险和治理,并实现一个能记住学习进度的"个人学习教练 Crew"。
2. 项目设计
小胖(抱着一袋薯片):“大师,Memory 不就是让 Agent 记住之前说过的话吗?这有什么难的?”
大师:“记住不难,但记住什么、怎么用、什么时候该忘掉——这是三个完全不同层次的问题。短期记忆像便签纸——随手记、用完扔,成本最低。长期记忆像档案柜——持久保存、需要检索,成本中等。最难的是’知道什么该忘’——如果一个用户的学习偏好从’喜欢视频课程’变成了’更喜欢实战项目’,旧的偏好记忆如果不更新,Agent 就会一直推荐错误的资源。”
小白(认真地记笔记):“这三种记忆分别存在哪?用什么存储?”
大师:“短期记忆存在当前会话的上下文窗口里——每次 kickoff 时的上下文就是短期记忆的载体,会话结束就消失。长期记忆和实体记忆需要外部存储——默认用 SQLite,也可以配置成 Chroma(向量数据库)或其他存储后端。记忆不是无限累积的——CrewAI 有memory_config可以设置存储路径和 Provider。”
小胖:“那我怎么让一个 Crew 启用记忆?改个参数就行吗?”
大师:“启动很简单——Crew(memory=True)。但真正难的是记忆管理。开了记忆后你会发现:Agent 开始’记仇’了——之前犯过的错误会影响后续判断;Agent 开始’固执’了——最初的偏好一直延续,不愿意接受新信息。记忆污染是真实存在的,而且比内存泄漏更难排查。”
小白:“记忆污染具体指什么?能举个例子吗?”
大师:“比如你的学习教练在第一轮对话中,Agent 基于你的提问判断’你对数据库特别感兴趣’。这个判断被存入了长期记忆。但实际你只是随口一问。之后的所有学习计划,Agent 都偏向推荐数据库相关内容——这就是记忆污染:一个低质量的推断被固化成了事实。”
小白:“那怎么清理和隔离记忆?”
大师:“清理有三种方式:手动删除——直接操作 SQLite 或 Chroma 中的记忆表;TTL 策略——设置记忆的过期时间,超过 X 天后自动清除;版本隔离——每次启动新的学习周期时,用不同的 memory 存储路径,实现’新旧隔离’。生产环境中,建议组合使用:按用户隔离存储 + TTL 自动过期 + 定期人工审核。”
小胖:“那如果我既想用记忆的便利,又怕记忆污染,有没有折中方案?”
大师:“局部记忆。不是所有 Task 都需要记忆——你可以只在关键节点启用。比如学习教练中,'学习进度追踪’这个 Task 需要记忆(上周学到哪了),但’今日学习内容推荐’这个 Task 不需要(每天从零生成即可,避免被历史偏好局限)。这种方案平衡了记忆的价值和风险。”
技术映射总结:短期记忆 = 会话级别的上下文(当前 kickoff 内的 Task 传递),长期记忆 = 跨会话的用户画像和行为模式,实体记忆 = 关键名词之间的关系图谱。记忆是一把双刃剑——它让 Agent 从"陌生人"变成"老朋友",但也可能让它变成"固执的老师"。
3. 项目实战
3.1 实战目标
实现一个"个人学习教练 Crew"——能根据上一次学习记录调整下一次学习计划。对比有记忆和无记忆时 Agent 行为的差异,并演示记忆清理方法。
3.2 环境准备
learning-coach-crew/ ├── .env ├── main.py ├── agents.py ├── tasks.py ├── crew.py ├── memory_manager.py # 记忆管理和清理工具 └── outputs/依赖:crewai、python-dotenv
3.3 分步实现
步骤1:创建学习教练 Agent
目标:设计一个能利用记忆的个性化学习教练。
# agents.pyfromcrewaiimportAgentdefcreate_learning_coach(llm):"""创建学习教练 Agent——需配合 Memory 使用"""returnAgent(role="个性化学习教练",goal=("基于学员的学习历史、当前水平和学习偏好,制定最适合的下一步学习计划。""每次生成计划前,回顾学员的过往学习记录,确保计划有连续性而非从零开始。"),backstory=("你是一名经验丰富的编程导师,辅导过500+学员从零基础到入职。""你的教学理念是'因材施教、温故知新'——每次教学前必先了解学员的进度。""你关注三个维度:知识掌握度(哪些已掌握/薄弱)、学习偏好(视频/文档/实战)、""时间约束(每周可投入的时间)。你拒绝'一刀切'的推荐。"),llm=llm,verbose=True,allow_delegation=False,max_iter=10)步骤2:创建任务
目标:设计两个 session 的任务,验证记忆延续性。
# tasks.pyfromcrewaiimportTaskdefcreate_learning_task(agent,topic:str,session:int):"""创建学习计划任务"""returnTask(description=f""" 为学员制定关于"{topic}"的学习计划(第{session}次学习会话)。 【重要】请先回忆学员之前的回答和学习记录: - 如果之前学过相关内容,基于之前的进度继续,不要从零开始 - 如果学员表达过偏好(如"喜欢视频"或"喜欢实战"),尊重这些偏好 - 如果学员提到过薄弱环节,在本次计划中加强 【本次计划要求】 1. 推荐3个学习资源(注明类型:视频/文章/实战项目) 2. 安排1个练习任务(控制在2小时内可完成) 3. 提出2个自测问题(用于检查掌握程度) """,expected_output=(f"# 第{session}次学习计划:{topic}\n\n""## 学习资源推荐\n""| 序号 | 资源名称 | 类型 | 预计时长 |\n""|------|---------|------|--------|\n\n""## 练习任务\n""- 任务描述和验收标准\n\n""## 自测问题\n""1. ...\n""2. ...\n\n""## 基于历史进度的说明\n""- 本次计划的衔接点(与前次学习的关系)"),agent=agent,output_file=f"outputs/session_{session}_plan.md")步骤3:无记忆版 Crew(对照组)
目标:展示没有记忆时 Agent 的"失忆"行为。
# crew.pyfromcrewaiimportCrew,Process,LLMfromagentsimportcreate_learning_coachfromtasksimportcreate_learning_taskimportosdefcreate_learning_crew(use_memory:bool,memory_storage_path:str=None):"""创建学习教练 Crew(可选启用记忆)"""llm=LLM(model=os.getenv("MODEL_NAME","gpt-4o-mini"),temperature=0.5,timeout=120)agent=create_learning_coach(llm)crew_kwargs={"agents":[agent],"process":Process.sequential,"verbose":True,}ifuse_memory:crew_kwargs["memory"]=Truecrew_kwargs["memory_config"]={"provider":"mem0",# 使用 CrewAI 内置的记忆 Provider"config":{"user_id":"student_001",# 按用户隔离}}returnCrew(**crew_kwargs),agent# main.pyfromdotenvimportload_dotenv load_dotenv()fromcrewimportcreate_learning_crewfromtasksimportcreate_learning_taskimportosdefsimulate_learning_sessions(use_memory:bool):"""模拟多次学习会话,对比有无记忆的效果"""os.makedirs("outputs",exist_ok=True)print(f"\n{'='*60}")print(f"学习教练实验:记忆={'启用'ifuse_memoryelse'关闭'}")print(f"{'='*60}")crew,agent=create_learning_crew(use_memory=use_memory)# 模拟三次学习会话sessions=[{"topic":"Python 面向对象编程","interaction":"我刚学完Python基础语法,喜欢看视频教程,不太喜欢读文档。"},{"topic":"Python 装饰器进阶","interaction":"上周学了面向对象,但多继承部分有点懵。最近时间不多,每天只能学1小时。"},{"topic":"FastAPI 入门","interaction":"装饰器已经掌握了。这周时间充裕,想挑战一下实战项目。"},]fori,sessioninenumerate(sessions,1):print(f"\n--- 第{i}次学习会话:{session['topic']}---")print(f"学员输入:{session['interaction']}")task=create_learning_task(agent,session['topic'],i)task.description+=f"\n\n【学员最新反馈】{session['interaction']}"crew.tasks=[task]result=crew.kickoff()print(f"Agent 推荐预览:{result.raw[:300]}...\n")# 对比检查print("\n"+"-"*40)ifuse_memory:print("预期表现:Agent 应能引用前次学习内容,""体现学习偏好的延续性(视频优先、时间约束等)。")else:print("预期表现:Agent 每次输出的学习计划互不关联,""像'第一次见面'一样重新推荐。")returncrewif__name__=="__main__":# 实验1:无记忆crew_no_mem=simulate_learning_sessions(use_memory=False)print("\n\n")# 实验2:有记忆crew_with_mem=simulate_learning_sessions(use_memory=True)3.4 运行结果
python main.py关键差异对比:
============================================================ 学习教练实验:记忆=关闭 ============================================================ --- 第1次学习会话:Python 面向对象编程 --- 学员输入:我喜欢看视频教程 Agent 推荐:推荐Python OOP视频课程、LeetCode练习... --- 第2次学习会话:Python 装饰器进阶 --- 学员输入:上周学了面向对象,但多继承部分有点懵 Agent 推荐:【无记忆】从Python装饰器基础开始......(❌ 没提到OOP进度) --- 第3次学习会话:FastAPI 入门 --- 学员输入:装饰器已经掌握了,想挑战实战 Agent 推荐:【无记忆】FastAPI基础教程......(❌ 没提到之前的任何学习内容) ============================================================ 学习教练实验:记忆=启用 ============================================================ --- 第1次学习会话:Python 面向对象编程 --- Agent 推荐:推荐视频课程(✅ 记住了视频偏好)... --- 第2次学习会话:Python 装饰器进阶 --- Agent 推荐:【有记忆】基于你已掌握的基本OOP概念,但多继承仍需加强, 本次装饰器计划配合OOP复习。每天1小时安排......(✅ 引用了前次学习) --- 第3次学习会话:FastAPI 入门 --- Agent 推荐:【有记忆】鉴于你已掌握装饰器、熟悉视频学习, 推荐一个FastAPI实战项目。你有更多时间,可以挑战完整CRUD应用... (✅ 引用了之前所有学习记录,体现了学习路径的递进)3.5 记忆管理工具
# memory_manager.pyimportsqlite3importosfromdatetimeimportdatetime,timedeltaclassMemoryManager:"""CrewAI 记忆管理工具"""def__init__(self,db_path:str="crewai_memory.db"):self.db_path=db_pathdefget_memory_stats(self)->dict:"""获取记忆统计信息"""ifnotos.path.exists(self.db_path):return{"exists":False,"message":"记忆数据库不存在"}conn=sqlite3.connect(self.db_path)cursor=conn.cursor()# 获取表列表cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")tables=[row[0]forrowincursor.fetchall()]stats={"exists":True,"tables":tables,"row_counts":{}}fortableintables:try:cursor.execute(f"SELECT COUNT(*) FROM{table}")count=cursor.fetchone()[0]stats["row_counts"][table]=countexcept:passconn.close()returnstatsdefclean_old_memories(self,days:int=30):"""清理超过指定天数的旧记忆(需根据实际表结构调整)"""ifnotos.path.exists(self.db_path):print("记忆数据库不存在,无需清理。")returnconn=sqlite3.connect(self.db_path)cursor=conn.cursor()cutoff=(datetime.now()-timedelta(days=days)).isoformat()try:# 尝试清理(表名需根据实际 CrewAI 版本调整)cursor.execute("DELETE FROM memories WHERE created_at < ?",(cutoff,))deleted=cursor.rowcount conn.commit()print(f"已清理{deleted}条超过{days}天的旧记忆。")exceptExceptionase:print(f"清理时出错(可能是表结构不匹配):{e}")finally:conn.close()defreset_all_memories(self):"""重置所有记忆(仅用于开发/测试)"""ifos.path.exists(self.db_path):os.remove(self.db_path)print(f"已删除记忆数据库:{self.db_path}")else:print("记忆数据库不存在。")if__name__=="__main__":manager=MemoryManager()# 查看记忆状态stats=manager.get_memory_stats()print("记忆统计:")print(f" 数据库存在:{stats['exists']}")ifstats['exists']:print(f" 表:{stats['tables']}")fortable,countinstats['row_counts'].items():print(f"{table}:{count}条记录")# 清理30天前的旧记忆manager.clean_old_memories(days=30)3.6 测试验证
# test_memory.pyimportpytestfrommemory_managerimportMemoryManagerimportosimporttimedeftest_memory_manager_init():"""验证记忆管理器初始化"""manager=MemoryManager(db_path="test_memory.db")assertmanager.db_path=="test_memory.db"deftest_get_memory_stats_no_db():"""验证数据库不存在时的统计结果"""manager=MemoryManager(db_path="nonexistent_test.db")stats=manager.get_memory_stats()assertstats["exists"]==Falseassert"不存在"instats["message"]deftest_clean_nonexistent_db():"""验证清理不存在的数据库不报错"""manager=MemoryManager(db_path="nonexistent_test.db")manager.clean_old_memories(days=30)# 不应抛异常deftest_reset_nonexistent_db():"""验证重置不存在的数据库不报错"""manager=MemoryManager(db_path="nonexistent_test.db")manager.reset_all_memories()# 不应抛异常pytest test_memory.py-v# 4 passed4. 项目总结
4.1 优点与缺点
| 维度 | 启用 Memory | 关闭 Memory |
|---|---|---|
| 用户体验 | 高,个性化、有延续性 | 低,每次都是"新用户" |
| 学习效果 | 递进式,能针对性补强 | 重复式,可能返工 |
| 存储成本 | SQLite 或向量数据库存储 | 无需存储 |
| 风险 | 记忆污染、隐私泄漏 | 无记忆相关风险 |
| 开发复杂度 | 中,需设计记忆清理策略 | 低 |
| 适用场景 | 持续交互型应用 | 一次性分析任务 |
4.2 适用场景
推荐启用 Memory:
- 个人学习助手、健身教练等持续跟踪场景
- 客服系统(记住用户历史问题避免重复)
- 个人助理(记住偏好、习惯、日程)
- 长期项目协作 Agent(记住项目进展和决策历程)
- 多轮交互的对话系统
不推荐启用 Memory:
- 一次性报告生成(竞品分析、周报)
- 对记忆准确性要求极高的合规场景(法律、财务)
4.3 注意事项
- 按用户隔离存储:多用户场景下必须隔离 memory,避免跨用户信息泄漏。
- 记忆不是越多越好:记忆累积到一定程度后,检索效率和准确性都会下降,需要清理策略。
- 隐私合规:记忆可能包含用户个人信息,需要遵守 GDPR/个保法——提供记忆导出和删除功能。
- 记忆版本管理:当 Agent 的 backstory 或系统 prompt 发生重大变化时,旧记忆可能与新规则冲突,建议重置。
4.4 常见踩坑经验
案例1:记忆让 Agent 变得越来越"固执"
现象:用户改变偏好后,Agent 仍然按旧的偏好推荐。
根因:长期记忆中的旧偏好权重过高,新交互未覆盖旧记忆。
解决:在 Task description 中加上"用户可能改变了偏好,请优先参考最近的交互而非历史记录"。
案例2:记忆数据库无限增长
现象:运行几个月后,SQLite 文件从几 KB 涨到几百 MB。
根因:没有设置记忆清理策略,每次 kickoff 都在写入新记忆。
解决:实现 TTL 过期策略 + 定期压缩(合并相似记忆)+ 存储容量告警。
案例3:跨用户的记忆泄漏
现象:用户 A 的学习计划中出现了用户 B 的学习偏好。
根因:memory_config 中没有按 user_id 隔离,共用同一个存储。
解决:在 memory_config 中设置user_id参数,确保每个用户的记忆独立存储。
4.5 思考题
你有一个客服 Crew 需要同时服务 100 个客户。应该采用’全局记忆’(所有客户共享一个记忆库)还是’用户级记忆’(每个客户独立记忆)?各有什么利弊?
当用户的学习习惯发生根本性变化(比如从’喜欢视频’变为’只用文档’),你能设计什么机制让 Agent 识别这种变化并相应地降低旧记忆的权重?
答案将在后续章节揭晓。
延伸阅读与资源
10倍开发者的 Dify 魔法书:从零构建全栈 AI 应用
后端工程师转型AI第一课-Ollama 与私有化大模型实战
大型语言模型(LLM) vLLM 高性能推理落地实战
Agent开发之LlamaIndex 实战修炼与源码进阶
大语言模型Transformers 实战修炼与源码剖析