参考mem0深度解析：开源长期记忆体_哔哩哔哩_bilibili

Langchain - Mem0

Dashboard | Mem0

pip install langchain langchain_openai mem0ai python-dotenv

import os
from typing import List, Dict
from httpx._transports import base
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from mem0 import MemoryClient
from dotenv import load_dotenv

load_dotenv()

# Configuration
# os.environ["OPENAI_API_KEY"] = "your-openai-api-key"
# os.environ["MEM0_API_KEY"] = "your-mem0-api-key"

# Initialize LangChain and Mem0
llm = ChatOpenAI(model="qwen3:4b", base_url=os.environ["ollama_OPENAI_API_URL"],api_key="empty")
mem0 = MemoryClient( api_key=os.environ["mem0_api_key"])
prompt = ChatPromptTemplate.from_messages([
    SystemMessage(content=os.environ["system_prompt"]),
    MessagesPlaceholder(variable_name="context"),
    HumanMessage(content="{input}")
])

def retrieve_context(query: str, user_id: str) -> List[Dict]:
    """Retrieve relevant context from Mem0"""
    try:
        filters = {
 "OR":[
    {
       "user_id":user_id
    }
 ]
}
        memories = mem0.search(query, user_id=user_id, filters=filters)
        memory_list = memories['results']
        
        serialized_memories = ' '.join([mem["memory"] for mem in memory_list])
        context = [
            {
                "role": "system", 
                "content": f"Relevant information: {serialized_memories}"
            },
            {
                "role": "user",
                "content": query
            }
        ]
        return context
    except Exception as e:
        print(f"Error retrieving memories: {e}")
        # Return empty context if there's an error
        return [{"role": "user", "content": query}]

def generate_response(input: str, context: List[Dict]) -> str:
    """Generate a response using the language model"""
    chain = prompt | llm
    response = chain.invoke({
        "context": context,
        "input": input
    })
    return response.content

def save_interaction(user_id: str, user_input: str, assistant_response: str):
    """Save the interaction to Mem0"""
    try:
        interaction = [
            {
              "role": "user",
              "content": user_input
            },
            {
                "role": "assistant",
                "content": assistant_response
            }
        ]
        result = mem0.add(interaction, user_id=user_id)
        print(f"Memory saved successfully: {len(result.get('results', []))} memories added")
    except Exception as e:
        print(f"Error saving interaction: {e}")

def chat_turn(user_input: str, user_id: str) -> str:
    # Retrieve context
    context = retrieve_context(user_input, user_id)
    
    # Generate response
    response = generate_response(user_input, context)
    
    # Save interaction
    save_interaction(user_id, user_input, response)
    
    return response

if __name__ == "__main__":
    print("Welcome to your personal Travel Agent Planner! How can I assist you with your travel plans today?")
    user_id = "plana"
    
    while True:
        user_input = input("You: ")
        if user_input.lower() in ['quit', 'exit', 'bye']:
            print("Travel Agent: Thank you for using our travel planning service. Have a great trip!")
            break
        
        response = chat_turn(user_input, user_id)
        print(f"Travel Agent: {response}")