使用 Mem0、LangChain 和 Groq 构建具有真实记忆功能的 AI 智能体
使用Mem0、LangChain和Groq构建具备真实记忆功能的AI智能体：为何大语言模型会在会话间遗忘信息、嵌入数据如何存储在ChromaDB中，以及完整的构建流程详解。
来源: **Build an AI Agent with Real Memory Using Mem0, LangChain & Groq**
以下笔记为“使用Mem0、LangChain和Groq构建具备真实记忆功能的AI智能体”提供了实用的实施路径。重点在于契约定义、校验机制以及可直接插入的代码占位符，而非激励性表述。
为什么大语言模型会遗忘——以及这为何重要
在处理“为什么大语言模型会遗忘”这一阶段时，首先需明确契约内容：所需输入、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及错误处理都属于产品功能的一部分，而非后续需要补充的内容。 将稳定的系统指令和工具结构进行缓存。重复发送相同的开头信息是导致资源浪费的常见原因。
我们正在构建什么
在“我们正在构建什么”阶段工作时，首先写下相关约定：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 优先选择小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一的责任模块，而非复杂的流程链。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
技术栈概览
在处理技术栈的某个阶段时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与验证后输出之间的契约。为相关成果命名，明确成功判定标准，杜绝无声的半完成状态。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
Mem0 — 内存层
在处理 Mem0 的内存层阶段时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及代币或查询成本。提前明确成本信息，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次收取相同的 LLM 调用费用。
Groq — 推理
在处理 Groq 推理阶段时，首先需明确相关规范：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 将配置信息置于应用程序代码之外。环境文件、密钥存储和功能开关应集中存放，以便操作员无需查看整个流程即可进行审计。 在耗时较高的步骤之后设置检查点。当操作员重新执行后续节点时，恢复流程不应再次计费相同的 LLM 调用。 在处理 Groq 推理阶段时，首先需明确相关规范：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 相较于庞大的脚本，应优先使用小型且可测试的单元。当某个步骤失败时，故障应指向单一责任模块，而非复杂的流程链。
LangChain — 消息格式化
将 LangChain 的消息格式化阶段视为可度量的对象，其效果最佳。在扩大范围之前，先记录一个理想的处理结果、一个失败案例以及回滚说明。 把这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声的、不完整的处理结果。 保持图结构的状态简洁且具有类型约束。嵌套的数据块会掩盖是哪个节点写了哪个字段，还会在处理中断后导致无法继续。
ChromaDB — 向量存储
将 ChromaDB 向量存储层视为可度量的对象使用效果最佳。在扩大应用范围之前，先记录一份理想的转换结果、一个故障案例以及回滚说明。在功能结果旁同时记录处理时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。需保持图结构简洁且类型明确，嵌套的二进制数据会掩盖具体是哪个节点修改了哪一字段，还会在进程中断后导致无法继续处理。
sentence-transformers/all-MiniLM-L6-v2 — 嵌入向量
将 sentence-transformers all-MiniLM-L6-v2 嵌入阶段视为可测量的对象时，其效果最佳。在扩大范围之前，需记录一份理想样本、一个故障案例以及回滚说明。 配置应置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放，以便操作人员无需查看整个系统结构即可进行审计。 应将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。 将 sentence-transformers all-MiniLM-L6-v2 嵌入阶段视为可测量的对象时，其效果最佳。在扩大范围之前，需记录一份理想样本、一个故障案例以及回滚说明。 相比庞大的脚本，应优先使用小型且可测试的单元。当某个步骤出现故障时，故障点应指向单一责任模块，而非复杂的流程链。
代码——逐步解析
在代码逐步演示阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，杜绝默许的半完成状态。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务上的完整性。
1. 中央配置
在“1 中央配置”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。对于会产生支出或更改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。
@dataclass
class AgentConfig:
groq_model: str = "llama-3.1-8b-instant"
temperature: float = 0.7
max_tokens: int = 1024
chroma_collection: str = "dev_assistant_memories"
chroma_path: str = "./mem0_db"
embedding_model: str = "sentence-transformers/all-MiniLM-L6-v2"
memory_trigger_phrases: tuple = (
"my name is", "i am building", "i am working on",
"my stack is", "i prefer", "my team", ...
)
2. 内存辅助工具
在“2个内存辅助工具”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个位置，以便操作人员无需查看整个流程即可进行审计。 对于涉及资金支出或修改生产数据的操作，需经过人工审批。编译时的连接方式并不等同于业务流程的完整性。 在“2个内存辅助工具”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 相较于庞大的脚本，应优先使用小型且可测试的单元。当某个步骤出现故障时，故障原因应能明确指向某个特定职责，而非整个复杂的流程。
def store_memory(memory, text, user_id) -> None:
try:
memory.add(text, user_id=user_id, infer=False)
except Exception as exc:
logger.warning("Could not store memory: %s", exc)
def retrieve_memories(memory, query, user_id, top_k=5) -> list[str]:
try:
results = memory.search(query=query,
filters={'user_id': user_id},
limit=top_k)
return [item['memory'] for item in results.get('results', [])
if 'memory' in item]
except Exception as exc:
logger.warning("Memory retrieval failed: %s", exc)
return []
3. 更智能的记忆触发机制
在实施“更智能的记忆触发机制”的三个阶段时，首先需写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与验证后输出之间的契约。为相关组件命名，明确成功判定标准，杜绝无声的半完成状态。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
def should_store(text: str, triggers: tuple) -> bool:
lowered = text.lower()
return any(phrase in lowered for phrase in triggers)
4. 立足实际的聊天提示词
在完成4个“基于现实的聊天提示”阶段时，首先写下相关规范：所需输入、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 缓存稳定的系统指令和工具结构。重复发送相同的开头信息是导致资源浪费的常见原因。
system = SystemMessage(content=(
"You are a memory-aware developer assistant. "
"Answer ONLY from the provided memory context. "
"If memory is empty or irrelevant, say: "
"'I don't have that in memory yet — tell me and I'll remember it.' "
"Never fabricate projects, names, or technical choices."
))
human = HumanMessage(content=(
f"### Memory context\n{memory_block}\n\n"
f"### User question\n{user_query}"
))
5. 双模式：REPL与演示
在完成5个双模式REPL阶段时，首先需明确规范：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 将配置信息置于应用程序代码之外。环境文件、密钥存储和功能开关应集中存放，以便操作员无需查看整个系统结构即可进行审计。 在耗时较高的步骤之后设置检查点。当操作员重新执行后续节点时，恢复流程不应再次调用相同的LLM接口。 在完成5个双模式REPL阶段时，首先需明确规范：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 优先选择小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一责任模块，而非复杂的流程链。
# Interactive session
python memory_agent.py
# Scripted demo (no terminal needed)
python memory_agent.py demo
5分钟内完成设置
在“5步内完成设置”阶段中，若将其视为可度量的工作面，则效果最佳。在扩大范围之前，先记录一份成功的示例、一个失败案例以及回滚说明。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许默许部分完成的情况。
安装依赖项
!pip install mem0ai langchain langchain-groq chromadb sentence-transformers
获取免费的Groq API密钥
运行
export GROQ_API_KEY=gsk_...
python memory_agent.py
演示效果如何
User: My name is Naveen and I am building a multimodal RAG pipeline using Groq and LangChain.
AI: Got it, Aditya! I'll remember your multimodal RAG project.
User: I recently switched from OpenAI embeddings to HuggingFace sentence-transformers.
AI: Noted — HuggingFace sentence-transformers for embeddings.
User: What AI project am I working on?
AI: You are building a multimodal RAG pipeline using Groq and LangChain.
User: Which embedding provider did I switch to?
AI: You switched from OpenAI embeddings to HuggingFace sentence-transformers.
完整代码：
import os
import logging
from dataclasses import dataclass, field
from typing import Optional
from getpass import getpass
from mem0 import Memory
from langchain_groq import ChatGroq
from langchain_core.messages import SystemMessage, HumanMessage
# ── Logging ──────────────────────────────────────────────────
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger(__name__)
# ── Configuration ─────────────────────────────────────────────
@dataclass
class AgentConfig:
"""Central config – tweak here, nowhere else."""
groq_model: str = "llama-3.1-8b-instant"
temperature: float = 0.7
max_tokens: int = 1024
chroma_collection: str = "dev_assistant_memories"
chroma_path: str = "./mem0_db"
embedding_model: str = "sentence-transformers/all-MiniLM-L6-v2"
# Phrases that signal the user is sharing personal context
memory_trigger_phrases: tuple = (
"my name is", "i am building", "i am researching",
"i am working on", "my favorite", "i like", "i study",
"i use", "i prefer", "my stack is", "i switched to",
"i recently", "my team", "my company",
)
# ── Mem0 + LLM bootstrap ──────────────────────────────────────
def build_memory(cfg: AgentConfig) -> Memory:
mem0_config = {
"vector_store": {
"provider": "chroma",
"config": {
"collection_name": cfg.chroma_collection,
"path": cfg.chroma_path,
},
},
"embedder": {
"provider": "huggingface",
"config": {"model": cfg.embedding_model},
},
}
memory = Memory.from_config(mem0_config)
logger.info("Mem0 memory store ready (collection=%s)", cfg.chroma_collection)
return memory
def build_llm(cfg: AgentConfig) -> ChatGroq:
llm = ChatGroq(model=cfg.groq_model, temperature=cfg.temperature, max_tokens=cfg.max_tokens)
logger.info("Groq LLM ready (model=%s)", cfg.groq_model)
return llm
# ── Memory helpers ────────────────────────────────────────────
def store_memory(memory: Memory, text: str, user_id: str) -> None:
"""Persist a memory fragment for the given user."""
try:
memory.add(text, user_id=user_id, infer=False)
logger.debug("Stored memory for %s: %s", user_id, text[:80])
except Exception as exc:
logger.warning("Could not store memory: %s", exc)
def retrieve_memories(memory: Memory, query: str, user_id: str, top_k: int = 5) -> list[str]:
"""Return the top-k relevant memory strings for a query."""
try:
results = memory.search(query=query, filters={"user_id": user_id}, limit=top_k)
memories = [
item["memory"]
for item in results.get("results", [])
if "memory" in item
]
logger.debug("Retrieved %d memories for query: %s", len(memories), query[:60])
return memories
except Exception as exc:
logger.warning("Memory retrieval failed: %s", exc)
return []
def should_store(text: str, triggers: tuple) -> bool:
"""Return True if the utterance carries information worth storing."""
lowered = text.lower()
return any(phrase in lowered for phrase in triggers)
# ── Core chat function ────────────────────────────────────────
def chat(
llm: ChatGroq,
memory: Memory,
user_query: str,
user_id: str,
cfg: AgentConfig,
) -> str:
"""
Full turn:
1. Retrieve relevant memories
2. Build a grounded prompt
3. Call LLM
4. Conditionally store new memories
5. Return response text
"""
# 1. Retrieve
memories = retrieve_memories(memory, user_query, user_id)
memory_block = "\n".join(f"- {m}" for m in memories) if memories else "(none yet)"
# 2. Build prompt
system = SystemMessage(content=(
"You are a memory-aware developer assistant. "
"Answer ONLY from the provided memory context. "
"If memory is empty or irrelevant, say: "
"'I don't have that in memory yet — tell me and I'll remember it.' "
"Never fabricate projects, names, or technical choices."
))
human = HumanMessage(content=(
f"### Memory context\n{memory_block}\n\n"
f"### User question\n{user_query}"
))
# 3. Call LLM
response = llm.invoke([system, human])
answer = response.content
# 4. Store if the query carries new personal context
if should_store(user_query, cfg.memory_trigger_phrases):
store_memory(memory, user_query, user_id)
return answer
# ── Interactive REPL ──────────────────────────────────────────
def run_repl(user_id: str = "dev_user") -> None:
"""
A simple command-line loop – type 'quit' or 'exit' to stop.
Special commands:
/memories – dump all stored memories
/clear – (future) clear memories for this user
"""
cfg = AgentConfig()
memory = build_memory(cfg)
llm = build_llm(cfg)
print("\nDeveloper Assistant (type 'exit' to quit)\n")
while True:
try:
user_input = input("You: ").strip()
except (EOFError, KeyboardInterrupt):
print("\nGoodbye!")
break
if not user_input:
continue
if user_input.lower() in {"exit", "quit"}:
print("Goodbye!")
break
if user_input == "/memories":
mems = retrieve_memories(memory, "", user_id, top_k=20)
if mems:
print("\nStored memories:")
for i, m in enumerate(mems, 1):
print(f" {i}. {m}")
else:
print(" (no memories stored yet)")
print()
continue
answer = chat(llm, memory, user_input, user_id, cfg)
print(f"\nAssistant: {answer}\n")
# ── Demo (non-interactive) ────────────────────────────────────
def run_demo() -> None:
"""
Scripted demo that shows the memory round-trip without
needing a terminal. Useful for notebooks or CI checks.
"""
cfg = AgentConfig()
memory = build_memory(cfg)
llm = build_llm(cfg)
USER = "demo_user"
turns = [
"My name is Naveen and I am building a multimodal RAG pipeline using Groq and LangChain.",
"I recently switched from OpenAI embeddings to HuggingFace sentence-transformers.",
"My team prefers async FastAPI endpoints for all our inference services.",
# Recall queries
"What AI project am I working on?",
"Which embedding provider did I switch to and why?",
"What's my team's preferred API framework?",
]
for query in turns:
print(f"\n{query}")
response = chat(llm, memory, query, USER, cfg)
print(f"{response}")
# ── Entry point ───────────────────────────────────────────────
if __name__ == "__main__":
import sys
# Set keys
if not os.getenv("GROQ_API_KEY"):
os.environ["GROQ_API_KEY"] = getpass("Enter GROQ API Key: ")
# Mem0 needs OPENAI_API_KEY present even when using local embeddings
os.environ.setdefault("OPENAI_API_KEY", "sk-dummy")
mode = sys.argv[1] if len(sys.argv) > 1 else "repl"
if mode == "demo":
run_demo()
else:
run_repl()