Strona główna / Artykuły

groqpythonlangchainmem0ai-agentagents

Stwórz agenta AI z prawdziwą pamięcią przy użyciu Mem0, LangChain i Groq

Agent sztucznej inteligencji z prawdziwą pamięcią wykorzystujący Mem0, LangChain i Groq: dlaczego modele językowe zapominają między sesjami, embeddingi przechowywane w ChromaDB oraz pełny proces budowy krok po kroku.

Wydawca: REACTAPP.TOP12 min czytania2680 słów

Źródła: **Build an AI Agent with Real Memory Using Mem0, LangChain & Groq**

Poniższe notatki przedstawiają praktyczny plan działania w ramach tematu „Stworzenie agenta AI z prawdziwą pamięcią przy użyciu Mem0, LangChain i Groq”. Nacisk kładziony jest na umowy, sprawdzania oraz miejsca zastępcze dla kodu, a nie na motywacyjne aspekty.

Dlaczego modele LLM zapominają — i dlaczego to ma znaczenie

Podczas prace nad etapem „Dlaczego modele LLM zapominają”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zdokumentuj zarówno prawidłowy przebieg działania, jak i ścieżkę naprawczą. Próby ponowne, kontrola przez człowieka oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później. Zachowuj w pamięci tymczasowej stabilne instrukcje systemu oraz schematy narzędzi. Ponowne wysyłanie identycznego wstępu jest częstą przyczyną marnotrawstwa zasobów.

To, co budujemy

Gdy przechodzisz przez etap „Co budujemy”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Ustalaj punkty kontrolne po kosztownych krokach. System powinien unikać ponownego pobierania opłat za tę samą funkcję LLM, gdy operator próbuje ponownie wykonać późniejszy element.

Tech Stack w pigułce

Gdy analizujesz stack technologiczny na danym etapie, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć możliwość cichego, częściowego ukończenia zadania. Ustaw punkty kontrolne po kosztownych krokach. System powrotu nie powinien ponownie naliczać opłat za tę samą funkcję LLM, gdy operator próbuje ponownie wykonać późniejszy element.

Mem0 — warstwa pamięci

Podczas pracy nad etapem warstwy pamięci Mem0, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Obok wyników funkcjonalnych zapisz czasy wykonywania oraz koszt tokenów lub zapytań. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Ustal punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.

Groq — inferencja

Podczas pracy nad etapem inferencji w Groq najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Ustaw punkty kontrolne po kosztownych krokach. Funkcja wznowienia nie powinna ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł. Podczas pracy nad etapem inferencji w Groq najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolij małe, testowalne jednostki nad rozbudowane skrypty. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną ścieżkę przetwarzania.

LangChain — formatowanie wiadomości

Faza formatowania wiadomości w LangChain działa najlepiej, gdy traktuje się ją jako mierzalną powierzchnię do analizy. Zapisz jeden idealny przykład transkrypcji, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres pracy. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzucaj ciche, częściowe ukończenie zadań. Utrzymuj stan grafu w prostej formie i określ jego typ. Wplecione elementy ukrywają informację o tym, który węzeł zapisał dane w danym polu, co utrudnia kontynuację pracy po przerwach.

ChromaDB — magazyn wektorowy

Etap przechowywania wektorowego ChromaDB funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zapisz jeden idealny przekaz, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres działania. Zapisuj czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy przechodzi się z środowiska demonstracyjnego do współdzielonych środowisk. Utrzymuj stan grafu w formie prostych, spójnie skategoryzowanych elementów. Wplecione błędy utrudniają określenie, który węzeł zapisał dane w danym polu, i powodują przerwę w kontynuacji pracy po zakłóceniach.

sentence-transformers/all-MiniLM-L6-v2 — embeddingi

Faza embeddingów Sentence-Transformers all-MiniLM-L6-v2 działa najlepiej, gdy jest traktowana jako mierzalna powierzchnia. Zapisz jeden idealny przepis, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań przed rozszerzaniem zakresu. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, składysek tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całego grafu. Oddziel zasadę dzielenia na fragmenty od zasady wyszukiwania. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej, gdy zmieniają się metryki jakości. Faza embeddingów Sentence-Transformers all-MiniLM-L6-v2 działa najlepiej, gdy jest traktowana jako mierzalna powierzchnia. Zapisz jeden idealny przepis, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań przed rozszerzaniem zakresu. Wolimy małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji.

Kod — szczegółowe omówienie

W fazie analizy kodu należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nazwij poszczególne elementy, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania. Konieczna jest ludzka akceptacja w przypadkach, gdy dochodzi do wydawania pieniędzy lub modyfikacji danych produkcyjnych. Podłączenia realizowane w czasie kompilacji nie równają się pełnej kompletności procesu biznesowego.

1. Centralna konfiguracja

W fazie 1 „Centralna konfiguracja” należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy rejestrować czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy ścieżka przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Konieczne jest ludzkie zatwierdzenie w przypadkach, gdy dochodzi do wydatków lub zmian w danych produkcyjnych. Podłączenia realizowane w czasie kompilacji nie gwarantują pełnej kompletności rozwiązania biznesowego.

@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. Narzędzia pomocnicze pamięci

W fazie 2 Memory Helpers należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Konfigurację należy przechowywać poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Wprowadź ludzką aprobatę dla operacji, które wiążą się z wydawaniem pieniędzy lub modyfikacją danych produkcyjnych. Połączenia skompilowane w czasie kompilacji nie równają się pełnej kompletności biznesowej. W fazie 2 Memory Helpers należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, testowalne jednostki nad rozbudowanymi skryptami. Gdy dany krok zawiedzie, powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną ścieżkę przetwarzania.

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. Inteligentniejszy wyzwalacz pamięci

Podczas przechodzenia przez etap 3 Inteligentniejszego wyzwalacza pamięci najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć możliwość cichego, częściowego ukończenia zadania. Ustal punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłat za tę samą próbę wywołania LLM, gdy operator ponawia działanie w późniejszym etapie.

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

4. Ustrukturyzowany prompt do czatu

Gdy przechodzisz przez 4 etapy Grounded Chat Prompt, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co się dzieje w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zapisz czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy przechodzi się z środowiska demonstracyjnego do współdzielonych środowisk. Zachowuj w pamięci tymczasowej stabilne instrukcje systemu oraz schematy narzędzi. Ponowne wysyłanie identycznego wstępu to częsty powód marnotrawstwa zasobów.

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. Dwa tryby: REPL i Demo

Gdy przechodzisz przez 5 etapów REPL w trybie Dual Modes, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Ustaw punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł. Gdy przechodzisz przez 5 etapów REPL w trybie Dual Modes, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolij małe, testowalne jednostki nad rozbudowane skrypty. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną sekwencję operacji.

# Interactive session
python memory_agent.py

# Scripted demo (no terminal needed)
python memory_agent.py demo

Ustawienie w mniej niż 5 minut

Etap „Ustawienie w mniej niż 5 krokach” działa najlepiej, gdy traktuje się go jako mierzalną powierzchnię do analizy. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Traktuj ten etap jako umowę pomiędzy wprowadzanymi danymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć przypadki częściowego ukończenia pracy bez informacji.

Zainstaluj zależności

!pip install mem0ai langchain langchain-groq chromadb sentence-transformers

Pobierz bezpłatny klucz API Groq

Zrób test

export GROQ_API_KEY=gsk_...
python memory_agent.py

Jak wygląda demo

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.

Pełny kod:

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

Wniosek

Listwa kontrolna operacyjna

Pozycje pokrewne do przeczytania

← Wszystkie artykułyWyślij brief