Build a Chatbot with Memory

from memvid_sdk import use
import os

# Create or open memory file
mem = use('langchain', 'chatbot-memory.mv2', mode='auto')

# Add some initial knowledge
mem.put(
    "Product Overview",
    "knowledge",
    {},
    text="""Our product is an AI-powered analytics platform that helps businesses
understand their data. Key features include:
- Real-time dashboards
- Automated insights
- Custom reports
- API access for developers"""
)

mem.put(
    "Pricing Information",
    "knowledge",
    {},
    text="""Pricing tiers:
- Starter: $29/month - Up to 10,000 events
- Pro: $99/month - Up to 100,000 events
- Enterprise: Custom pricing - Unlimited events
All plans include 14-day free trial."""
)

print("Memory initialized with knowledge base")

from datetime import datetime
from typing import List, Dict
import json

class ConversationMemory:
    def __init__(self, memory_path: str):
        self.mem = use('langchain', memory_path, mode='auto')
        self.session_id = datetime.now().strftime("%Y%m%d_%H%M%S")

    def add_message(self, role: str, content: str):
        """Store a message in memory."""
        self.mem.put(
            f"Conversation - {self.session_id}",
            "conversation",
            {
                "session_id": self.session_id,
                "role": role,
                "timestamp": datetime.now().isoformat()
            },
            text=f"[{role.upper()}]: {content}"
        )

    def get_relevant_context(self, query: str, k: int = 5) -> str:
        """Retrieve relevant information for the query."""
        results = self.mem.find(query, k=k)

        context_parts = []
        for hit in results.hits:
            if hit.label == "knowledge":
                context_parts.append(f"[Knowledge] {hit.text}")
            elif hit.label == "conversation":
                context_parts.append(f"[Previous conversation] {hit.text}")

        return "\n\n".join(context_parts)

    def get_recent_messages(self, limit: int = 10) -> List[Dict]:
        """Get recent conversation messages."""
        timeline = self.mem.timeline(limit=limit)

        messages = []
        for entry in timeline.entries:
            if entry.label == "conversation":
                messages.append({
                    "role": entry.metadata.get("role", "unknown"),
                    "content": entry.text,
                    "timestamp": entry.metadata.get("timestamp")
                })

        return messages

from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, AIMessage, SystemMessage

class MemvidChatbot:
    def __init__(self, memory_path: str = "chatbot-memory.mv2"):
        self.memory = ConversationMemory(memory_path)
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

        self.system_prompt = """You are a helpful AI assistant with access to a knowledge base and conversation history.

When answering questions:
1. Use the provided context to give accurate, relevant answers
2. Reference specific information from the knowledge base when applicable
3. Remember details from previous conversations
4. Be conversational and helpful

If you don't have enough information, say so honestly."""

    def chat(self, user_message: str) -> str:
        """Process a user message and return a response."""

        # Store the user message
        self.memory.add_message("user", user_message)

        # Get relevant context from memory
        context = self.memory.get_relevant_context(user_message)

        # Build the prompt with context
        messages = [
            SystemMessage(content=self.system_prompt),
            HumanMessage(content=f"""Context from knowledge base and previous conversations:
{context}

---

User message: {user_message}

Please respond helpfully based on the context above.""")
        ]

        # Get response from LLM
        response = self.llm.invoke(messages)
        assistant_message = response.content

        # Store the assistant's response
        self.memory.add_message("assistant", assistant_message)

        return assistant_message

    def add_knowledge(self, title: str, content: str):
        """Add new knowledge to the chatbot's memory."""
        self.memory.mem.put(title, "knowledge", {}, text=content)
        print(f"Added knowledge: {title}")

def main():
    print("Memvid Chatbot initialized")
    print("Commands: /add (add knowledge), /history (view recent), /quit (exit)")
    print("-" * 50)

    bot = MemvidChatbot()

    while True:
        user_input = input("\nYou: ").strip()

        if not user_input:
            continue

        if user_input.lower() == "/quit":
            print("Goodbye!")
            break

        elif user_input.lower() == "/history":
            messages = bot.memory.get_recent_messages(10)
            print("\nRecent conversation:")
            for msg in messages:
                print(f"  [{msg['role']}]: {msg['content'][:100]}...")

        elif user_input.lower().startswith("/add "):
            # Format: /add Title | Content
            parts = user_input[5:].split("|", 1)
            if len(parts) == 2:
                bot.add_knowledge(parts[0].strip(), parts[1].strip())
            else:
                print("Usage: /add Title | Content")

        else:
            response = bot.chat(user_input)
            print(f"\nBot: {response}")

if __name__ == "__main__":
    main()

def summarize_conversation(self) -> str:
    """Generate a summary of the current conversation."""
    recent = self.memory.get_recent_messages(20)

    if not recent:
        return "No conversation history yet."

    conversation_text = "\n".join([
        f"{m['role']}: {m['content']}" for m in recent
    ])

    summary_prompt = f"""Summarize this conversation in 2-3 sentences:

{conversation_text}

Summary:"""

    response = self.llm.invoke([HumanMessage(content=summary_prompt)])
    return response.content

def detect_topics(self, message: str) -> List[str]:
    """Detect topics in a message for better retrieval."""
    prompt = f"""Extract 1-3 topic tags from this message. Return only comma-separated tags.

Message: {message}

Tags:"""

    response = self.llm.invoke([HumanMessage(content=prompt)])
    tags = [t.strip() for t in response.content.split(",")]
    return tags

class MultiUserChatbot:
    def __init__(self, memory_path: str):
        self.mem = use('langchain', memory_path, mode='auto')
        self.sessions = {}

    def get_session(self, user_id: str) -> MemvidChatbot:
        """Get or create a session for a user."""
        if user_id not in self.sessions:
            self.sessions[user_id] = MemvidChatbot(self.mem)
            self.sessions[user_id].session_id = user_id
        return self.sessions[user_id]

    def chat(self, user_id: str, message: str) -> str:
        """Chat with session isolation."""
        session = self.get_session(user_id)
        return session.chat(message)

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()
bot = MemvidChatbot()

class ChatRequest(BaseModel):
    message: str
    user_id: str = "default"

class ChatResponse(BaseModel):
    response: str

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    response = bot.chat(request.message)
    return ChatResponse(response=response)

@app.post("/knowledge")
async def add_knowledge(title: str, content: str):
    bot.add_knowledge(title, content)
    return {"status": "added"}

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]