Створіть AI-агента з справжньою пам’яттю за допомогою Mem0, LangChain та Groq

Наступні примітки описують практичний підхід до реалізації проекту „Створення AI-агента з справжньою пам’яттю за допомогою Mem0, LangChain та Groq“. Основна увага приділяється контрактам, перевіркам та шаблонам коду, а не мотиваційним аспектам.

Чому LLMs забувають — і чому це важливо

Під час роботи над етапом „Чому LLMs забувають“ спочатку запишіть контракт: необхідні вхідні дані, сигнал про успіх та наслідки часткової невдачі. Цей перелік допоможе зберегти чесність у подальших змінах коду. Одночасно задокументуйте шлях успішної роботи та шлях відновлення. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації. Зберігайте у кеші стабільні інструкції системи та схеми інструментів. Повторна передача ідентичних даних є поширеною причиною витрат ресурсів.

Що ми створюємо

Під час роботи над етапом «Що ми створюємо» спочатку запишіть умови виконання: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Віддавайте перевагу невеликим, тестованим одиницям коду перед об’ємними скриптами. Якщо якийсь крок зазнає невдачі, причина має вказувати на конкретну функцію, а не на складну послідовність операцій. Робіть перевірки після дорогих кроків. Система повторного запуску не повинна знову оплачувати один і той самий виклик LLM, коли оператор намагається виконати наступний етап.

Технологічна стек у короткому огляді

Під час роботи над технологічною стеком на певному етапі спочатку запишіть умови контракту: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність пізніших змін у коді. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте безповідомного часткового виконання завдань. Робіть перевірки після дорогих кроків. Система повернення до виконання не повинна знову оплачувати однаковий виклик LLM, коли оператор намагається виконати наступний етап.

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 Memory Helpers необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Конфігурацію слід тримати окремо від коду додатку. Файли середовища, сховища конфіденційних даних та флаги функціоналу мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь алгоритм. Встановлюйте людське схвалення для кроків, які спричиняють витрати грошей чи змінюють дані у продакшені. Підключення на етапі компіляції не є гарантією повноти бізнес-функціоналу. На етапі 2 Memory Helpers необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Віддавайте перевагу малим, тестованим одиницям перед величезними скриптами. Коли крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану систему обробки даних.

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. Розумніший тригер пам’яті

Під час роботи над етапом 3 «Розумніший тригер пам’яті» спочатку запишіть умови договору: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допоможе зберегти чесність подальших змін у коді. Розглядайте цей етап як договір між вхідними даними та перевіреними результатами. Дайте назви елементам, визначте критерії успіху та не допускайте безпроблемного часткового виконання завдань. Робіть перевірки після дорогих кроків. Система відновлення не повинна знову оплачувати однаковий виклик LLM, коли оператор намагається виконати наступний етап.

def should_store(text: str, triggers: tuple) -> bool:
    lowered = text.lower()
    return any(phrase in lowered for phrase in triggers)

4. Чіткий запит для чату

Під час виконання 4 етапів Grounded Chat 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}"
))

5. Дві режими: REPL та Demo

Під час роботи над 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

Отримайте безкоштовний ключ API Groq

Запустіть

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()

Висновки

Чек-лист для експлуатації