Build an AI Chatbot in Python

This lesson builds a production-quality AI chatbot from scratch — one that maintains conversation history, uses system prompts effectively, persists conversations to disk, and can be extended with tools. Everything integrates into a clean CLI or web interface.

Architecture Overview

User Input
    ↓
Message Manager (history + context window management)
    ↓
AI Client (OpenAI or Anthropic)
    ↓
Tool Handler (if tool calls needed)
    ↓
Response → User
    ↓
Storage (save conversation to disk)

The Core Chatbot

import os
import json
from datetime import datetime
from pathlib import Path
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

class ChatMessage:
    def __init__(self, role: str, content: str):
        self.role = role
        self.content = content
        self.timestamp = datetime.now().isoformat()
    
    def to_dict(self):
        return {"role": self.role, "content": self.content}
    
    def to_storage_dict(self):
        return {"role": self.role, "content": self.content, "timestamp": self.timestamp}

class Chatbot:
    def __init__(
        self, 
        name: str = "Assistant",
        system_prompt: str = "You are a helpful, friendly assistant.",
        model: str = "gpt-4o",
        max_context_messages: int = 20,
        save_path: str = "conversations"
    ):
        self.name = name
        self.system_prompt = system_prompt
        self.model = model
        self.max_context_messages = max_context_messages
        self.save_path = Path(save_path)
        self.save_path.mkdir(exist_ok=True)
        
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.conversation_id = datetime.now().strftime("%Y%m%d_%H%M%S")
        self.history: list[ChatMessage] = []
    
    def _get_context_messages(self) -> list[dict]:
        """Get recent messages for API call (manages context window)."""
        recent = self.history[-self.max_context_messages:]
        return [msg.to_dict() for msg in recent]
    
    def _build_messages(self) -> list[dict]:
        messages = [{"role": "system", "content": self.system_prompt}]
        messages.extend(self._get_context_messages())
        return messages
    
    def chat(self, user_input: str) -> str:
        # Add user message
        self.history.append(ChatMessage("user", user_input))
        
        # Call API
        response = self.client.chat.completions.create(
            model=self.model,
            messages=self._build_messages(),
            temperature=0.7
        )
        
        assistant_reply = response.choices[0].message.content
        
        # Save assistant response
        self.history.append(ChatMessage("assistant", assistant_reply))
        
        return assistant_reply
    
    def save_conversation(self):
        """Save conversation history to JSON file."""
        filepath = self.save_path / f"conversation_{self.conversation_id}.json"
        data = {
            "id": self.conversation_id,
            "name": self.name,
            "model": self.model,
            "system_prompt": self.system_prompt,
            "messages": [msg.to_storage_dict() for msg in self.history]
        }
        with open(filepath, 'w', encoding='utf-8') as f:
            json.dump(data, f, indent=2, ensure_ascii=False)
        return filepath
    
    def load_conversation(self, filepath: str):
        """Load a previous conversation."""
        with open(filepath, 'r', encoding='utf-8') as f:
            data = json.load(f)
        self.history = [
            ChatMessage(msg['role'], msg['content'])
            for msg in data['messages']
        ]
        self.conversation_id = data['id']
    
    def reset(self):
        self.history = []
        self.conversation_id = datetime.now().strftime("%Y%m%d_%H%M%S")
    
    def get_summary(self) -> str:
        """Get an AI-generated summary of the conversation."""
        if not self.history:
            return "No conversation yet."
        
        context = "\n".join([
            f"{msg.role.upper()}: {msg.content}"
            for msg in self.history
        ])
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{
                "role": "user",
                "content": f"Summarize this conversation in 2-3 sentences:\n\n{context}"
            }],
            temperature=0,
            max_tokens=200
        )
        return response.choices[0].message.content

CLI Interface

import readline  # Enables arrow key history in terminal

def run_cli_chatbot():
    """Interactive command-line chatbot."""
    
    print("\n🤖 AI Chatbot")
    print("Commands: /quit, /save, /reset, /history, /summary")
    print("-" * 50)
    
    # Choose persona
    personas = {
        "1": ("Python Tutor", "You are an expert Python tutor. Explain concepts clearly with examples."),
        "2": ("Code Reviewer", "You are a senior software engineer doing code reviews. Be specific and constructive."),
        "3": ("General Assistant", "You are a helpful, friendly assistant. Be concise.")
    }
    
    print("Choose a persona:")
    for key, (name, _) in personas.items():
        print(f"  {key}: {name}")
    
    choice = input("\nSelect (1-3, default=3): ").strip() or "3"
    name, prompt = personas.get(choice, personas["3"])
    
    bot = Chatbot(name=name, system_prompt=prompt)
    print(f"\n{name} is ready. Start chatting!\n")
    
    while True:
        try:
            user_input = input("You: ").strip()
        except (KeyboardInterrupt, EOFError):
            print("\n\nSaving conversation...")
            bot.save_conversation()
            break
        
        if not user_input:
            continue
        
        # Handle commands
        if user_input.lower() == "/quit":
            filepath = bot.save_conversation()
            print(f"Conversation saved to {filepath}")
            break
        
        elif user_input.lower() == "/save":
            filepath = bot.save_conversation()
            print(f"Saved to: {filepath}")
            continue
        
        elif user_input.lower() == "/reset":
            bot.reset()
            print("Conversation reset.\n")
            continue
        
        elif user_input.lower() == "/history":
            for msg in bot.history:
                role = "You" if msg.role == "user" else bot.name
                print(f"\n{role}: {msg.content}")
            print()
            continue
        
        elif user_input.lower() == "/summary":
            print(f"\nSummary: {bot.get_summary()}\n")
            continue
        
        # Get AI response
        try:
            print(f"\n{bot.name}: ", end='', flush=True)
            response = bot.chat(user_input)
            print(response)
            print()
        except Exception as e:
            print(f"Error: {e}\n")

if __name__ == "__main__":
    run_cli_chatbot()

FastAPI Web Interface

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from typing import Optional

app = FastAPI(title="AI Chatbot API")

# In-memory session storage (use Redis in production)
sessions: dict[str, Chatbot] = {}

class ChatRequest(BaseModel):
    session_id: str
    message: str
    system_prompt: Optional[str] = None

@app.post("/chat")
async def chat(request: ChatRequest):
    # Get or create session
    if request.session_id not in sessions:
        system = request.system_prompt or "You are a helpful assistant."
        sessions[request.session_id] = Chatbot(system_prompt=system)
    
    bot = sessions[request.session_id]
    
    try:
        response = bot.chat(request.message)
        return {
            "response": response,
            "session_id": request.session_id,
            "message_count": len(bot.history)
        }
    except Exception as e:
        return {"error": str(e)}, 500

@app.delete("/chat/{session_id}")
async def reset_session(session_id: str):
    if session_id in sessions:
        del sessions[session_id]
    return {"status": "reset"}

@app.get("/chat/{session_id}/history")
async def get_history(session_id: str):
    if session_id not in sessions:
        return {"messages": []}
    bot = sessions[session_id]
    return {"messages": [msg.to_storage_dict() for msg in bot.history]}

Chatbot with Memory and Tools

import json

def create_smart_chatbot():
    """A chatbot with web search and calculator tools."""
    
    tools = [
        {
            "type": "function",
            "function": {
                "name": "calculate",
                "description": "Evaluate a mathematical expression",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "expression": {"type": "string", "description": "Math expression like '2 + 2' or 'sqrt(16)'"}
                    },
                    "required": ["expression"]
                }
            }
        }
    ]
    
    def calculate(expression: str) -> str:
        import math
        try:
            # Safe eval with only math functions
            allowed = {k: getattr(math, k) for k in dir(math) if not k.startswith('_')}
            result = eval(expression, {"__builtins__": {}}, allowed)
            return str(result)
        except Exception as e:
            return f"Error: {e}"
    
    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    messages = [{"role": "system", "content": "You are a helpful assistant with a calculator."}]
    
    def chat(user_input: str) -> str:
        messages.append({"role": "user", "content": user_input})
        
        while True:
            response = client.chat.completions.create(
                model="gpt-4o", messages=messages, tools=tools
            )
            choice = response.choices[0]
            messages.append(choice.message)
            
            if choice.finish_reason == "stop":
                return choice.message.content
            
            for tool_call in choice.message.tool_calls:
                args = json.loads(tool_call.function.arguments)
                result = calculate(**args)
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result
                })
    
    return chat

smart_chat = create_smart_chatbot()
print(smart_chat("What is 15% of 847?"))
print(smart_chat("And if I split that with 3 friends?"))

You now have a complete, production-quality chatbot architecture. The patterns here — session management, conversation persistence, tool integration — scale to real applications.

Next lesson: Automate Tasks with AI Agents — building agents that take actions autonomously.