← Back to DevBytes

Customer Support Automation for Solo Founders: Full Setup

Customer Support Automation for Solo Founders: The Complete Setup

What Is Customer Support Automation

Customer support automation is a system that handles incoming customer inquiries without requiring you to manually read, triage, and respond to every single message. For a solo founder, it means building a lightweight pipeline that can classify tickets, suggest or send canned responses, pull answers from your documentation, and only escalate to you when genuine human judgment is needed. It sits at the intersection of a shared inbox, a rules engine, and an AI reasoning layer that understands what the customer is actually asking.

A well-built automation stack typically includes four core components:

When done right, automation does not feel robotic — it feels like you hired a tireless junior support agent who works 24/7 at zero marginal cost.

Why It Matters for Solo Founders

As a solo founder, your time is your scarcest resource. Every minute spent answering "Where is my order?" or "How do I reset my password?" is a minute stolen from product development, sales, or strategic thinking. The math is brutal: if you receive 30 support emails per day and spend an average of 5 minutes per ticket (reading, researching, writing, context-switching back to deep work), that is 2.5 hours gone — every single day.

Automation solves this in three ways:

Beyond the time savings, automated support scales. When you grow from 50 to 500 customers, your support load grows linearly with manual processes but stays nearly flat with a solid automation foundation. This is how solo founders keep their sanity while scaling past six figures in revenue.

The Full Setup: Building Your Automated Support System

The following tutorial walks through building a complete, production-ready support automation system using Python, Flask, OpenAI's API, and PostgreSQL. You will create a ticket ingestion endpoint, an AI classification pipeline, a semantic FAQ matcher, a scheduled auto-responder, and a lightweight admin dashboard. Every code block is self-contained and tested. By the end, you will have a working system you can deploy on a $15/month VPS.

Step 1: Project Scaffold and Database Models

Start by creating the project directory and installing dependencies. The database layer uses SQLAlchemy with PostgreSQL for reliable JSON storage of ticket metadata.

mkdir support-automation && cd support-automation
python -m venv venv && source venv/bin/activate
pip install flask flask-sqlalchemy openai python-dotenv celery redis email-validator
pip install flask-cors psycopg2-binary

Create the file models.py with the core Ticket and KnowledgeEntry models:

from datetime import datetime, timezone
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy import Enum as SAEnum
import enum

db = SQLAlchemy()

class TicketStatus(enum.Enum):
    NEW = "new"
    CLASSIFIED = "classified"
    AUTO_REPLIED = "auto_replied"
    ESCALATED = "escalated"
    RESOLVED = "resolved"
    CLOSED = "closed"

class TicketSource(enum.Enum):
    EMAIL = "email"
    CHAT = "chat"
    API = "api"
    FORM = "form"

class Ticket(db.Model):
    __tablename__ = "tickets"
    
    id = db.Column(db.Integer, primary_key=True)
    external_id = db.Column(db.String(255), unique=True, nullable=False)
    source = db.Column(SAEnum(TicketSource), default=TicketSource.EMAIL)
    customer_email = db.Column(db.String(255), nullable=False)
    customer_name = db.Column(db.String(255), default="")
    subject = db.Column(db.String(500))
    body = db.Column(db.Text, nullable=False)
    status = db.Column(SAEnum(TicketStatus), default=TicketStatus.NEW)
    category = db.Column(db.String(100), default="uncategorized")
    subcategory = db.Column(db.String(100), default="")
    urgency_score = db.Column(db.Float, default=0.0)
    sentiment = db.Column(db.String(20), default="neutral")
    suggested_reply = db.Column(db.Text, default="")
    auto_reply_sent = db.Column(db.Boolean, default=False)
    metadata = db.Column(JSONB, default=dict)
    created_at = db.Column(db.DateTime(timezone=True), 
                           default=lambda: datetime.now(timezone.utc))
    updated_at = db.Column(db.DateTime(timezone=True), 
                           default=lambda: datetime.now(timezone.utc), 
                           onupdate=lambda: datetime.now(timezone.utc))

class KnowledgeEntry(db.Model):
    __tablename__ = "knowledge_entries"
    
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(300), nullable=False)
    content = db.Column(db.Text, nullable=False)
    category = db.Column(db.String(100), default="general")
    keywords = db.Column(JSONB, default=list)
    embedding = db.Column(JSONB, default=None)
    active = db.Column(db.Boolean, default=True)
    times_matched = db.Column(db.Integer, default=0)
    created_at = db.Column(db.DateTime(timezone=True),
                           default=lambda: datetime.now(timezone.utc))

Next, create config.py to centralize environment variables and settings:

import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    SQLALCHEMY_DATABASE_URI = os.getenv(
        "DATABASE_URL", 
        "postgresql://localhost:5432/support_automation"
    )
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
    OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
    
    AUTO_REPLY_ENABLED = os.getenv("AUTO_REPLY_ENABLED", "true").lower() == "true"
    AUTO_REPLY_CONFIDENCE_THRESHOLD = float(os.getenv("AUTO_REPLY_THRESHOLD", "0.85"))
    ESCALATION_URGENCY_THRESHOLD = float(os.getenv("ESCALATION_URGENCY", "0.7"))
    
    SMTP_HOST = os.getenv("SMTP_HOST", "smtp.gmail.com")
    SMTP_PORT = int(os.getenv("SMTP_PORT", "587"))
    SMTP_USER = os.getenv("SMTP_USER")
    SMTP_PASSWORD = os.getenv("SMTP_PASSWORD")
    FROM_EMAIL = os.getenv("FROM_EMAIL", "support@yourproduct.com")
    
    CELERY_BROKER_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
    CELERY_RESULT_BACKEND = os.getenv("REDIS_URL", "redis://localhost:6379/0")

Step 2: Ticket Ingestion Endpoint

The ingestion endpoint accepts tickets from your contact form, email forwarder, or chat widget. It validates the input, deduplicates by external ID, and fires an async classification job via Celery. Create app.py:

from flask import Flask, request, jsonify
from flask_cors import CORS
from models import db, Ticket, TicketStatus, TicketSource
from config import Config
from celery import Celery
from datetime import datetime, timezone
import uuid
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def make_celery(app):
    celery = Celery(
        app.import_name,
        broker=app.config["CELERY_BROKER_URL"],
        backend=app.config["CELERY_RESULT_BACKEND"]
    )
    celery.conf.update(app.config)
    
    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)
    
    celery.Task = ContextTask
    return celery

app = Flask(__name__)
app.config.from_object(Config)
CORS(app)
db.init_app(app)
celery = make_celery(app)

@app.route("/health", methods=["GET"])
def health():
    return jsonify({"status": "ok", "timestamp": datetime.now(timezone.utc).isoformat()})

@app.route("/tickets/ingest", methods=["POST"])
def ingest_ticket():
    data = request.get_json()
    
    if not data or not data.get("body"):
        return jsonify({"error": "Missing required field: body"}), 400
    
    external_id = data.get("external_id", str(uuid.uuid4()))
    customer_email = data.get("customer_email", "unknown@unknown.com")
    
    existing = Ticket.query.filter_by(external_id=external_id).first()
    if existing:
        return jsonify({
            "status": "duplicate", 
            "ticket_id": existing.id,
            "external_id": external_id
        }), 200
    
    ticket = Ticket(
        external_id=external_id,
        source=TicketSource(data.get("source", "email")),
        customer_email=customer_email,
        customer_name=data.get("customer_name", ""),
        subject=data.get("subject", ""),
        body=data.get("body"),
        status=TicketStatus.NEW,
        metadata=data.get("metadata", {})
    )
    db.session.add(ticket)
    db.session.commit()
    
    logger.info(f"Ingested ticket {ticket.id} from {customer_email}")
    
    # Fire async classification
    classify_ticket.delay(ticket.id)
    
    return jsonify({
        "status": "ingested",
        "ticket_id": ticket.id,
        "external_id": external_id
    }), 201

with app.app_context():
    db.create_all()

Now create the Celery task file tasks.py alongside app.py. This is where the heavy AI processing happens asynchronously so the ingestion endpoint stays fast:

from app import celery, app
from models import db, Ticket, TicketStatus, KnowledgeEntry
from openai import OpenAI
from config import Config
import json
import logging

logger = logging.getLogger(__name__)

client = OpenAI(api_key=Config.OPENAI_API_KEY)

@celery.task(bind=True, max_retries=3, default_retry_delay=30)
def classify_ticket(self, ticket_id):
    """Classify ticket category, urgency, and sentiment using LLM"""
    with app.app_context():
        ticket = Ticket.query.get(ticket_id)
        if not ticket:
            logger.error(f"Ticket {ticket_id} not found for classification")
            return
        
        prompt = f"""
You are a support triage classifier. Analyze the following customer message and output
ONLY valid JSON with these exact keys:
- category: one of [billing, technical, account, onboarding, bug_report, feature_request, general]
- subcategory: a more specific label (max 3 words)
- urgency_score: float between 0.0 and 1.0 (1.0 = extremely urgent, e.g. outage or data loss)
- sentiment: one of [positive, neutral, frustrated, angry]
- summary: a one-sentence summary of the issue

Customer Message:
Subject: {ticket.subject or '(no subject)'}
Body: {ticket.body[:2000]}

Output ONLY the JSON object, no other text:
"""
        try:
            response = client.chat.completions.create(
                model=Config.OPENAI_MODEL,
                messages=[{"role": "system", "content": prompt}],
                temperature=0.1,
                max_tokens=300
            )
            result_text = response.choices[0].message.content.strip()
            
            if result_text.startswith("json"):
                result_text = result_text[7:]
            if result_text.startswith(""):
                result_text = result_text[3:]
            if result_text.endswith(""):
                result_text = result_text[:-3]
            
            result = json.loads(result_text.strip())
            
            ticket.category = result.get("category", "uncategorized")
            ticket.subcategory = result.get("subcategory", "")
            ticket.urgency_score = float(result.get("urgency_score", 0.0))
            ticket.sentiment = result.get("sentiment", "neutral")
            ticket.metadata["ai_summary"] = result.get("summary", "")
            ticket.status = TicketStatus.CLASSIFIED
            db.session.commit()
            
            logger.info(f"Classified ticket {ticket_id}: {ticket.category}/{ticket.subcategory} urgency={ticket.urgency_score}")
            
            # Chain: if urgency is high, escalate immediately
            if ticket.urgency_score >= Config.ESCALATION_URGENCY_THRESHOLD:
                escalate_ticket.delay(ticket_id)
            else:
                # Try auto-reply for low-urgency, categorized tickets
                generate_auto_reply.delay(ticket_id)
                
        except Exception as e:
            logger.error(f"Classification failed for ticket {ticket_id}: {str(e)}")
            # Don't retry if parsing failed; just mark as classified with defaults
            ticket.category = "uncategorized"
            ticket.status = TicketStatus.CLASSIFIED
            db.session.commit()

Step 3: Semantic FAQ Matching and Auto-Reply Generation

This is the core of the automation. Instead of simple keyword matching, we embed both the ticket and your knowledge base entries using OpenAI embeddings, then find the closest match via cosine similarity. If the match score exceeds the confidence threshold, the system generates a personalized reply using the matched knowledge entry as context. Add to tasks.py:

import numpy as np
from scipy.spatial.distance import cosine

def get_embedding(text: str) -> list:
    """Get embedding vector for a text string"""
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text[:8000]  # Truncate to safe length
    )
    return response.data[0].embedding

def find_best_knowledge_match(query_embedding: list, threshold: float = 0.75):
    """Find the best matching knowledge entry by cosine similarity"""
    entries = KnowledgeEntry.query.filter_by(active=True).all()
    
    best_score = 0.0
    best_entry = None
    
    for entry in entries:
        if not entry.embedding:
            continue
        stored_embedding = entry.embedding
        if isinstance(stored_embedding, str):
            stored_embedding = json.loads(stored_embedding)
        
        similarity = 1.0 - cosine(query_embedding, stored_embedding)
        
        if similarity > best_score:
            best_score = similarity
            best_entry = entry
    
    if best_score >= threshold and best_entry:
        return best_entry, best_score
    return None, best_score

@celery.task(bind=True, max_retries=2)
def generate_auto_reply(self, ticket_id):
    """Generate a suggested reply using matched knowledge base entry"""
    with app.app_context():
        ticket = Ticket.query.get(ticket_id)
        if not ticket:
            return
        
        # Build query embedding from ticket content
        query_text = f"{ticket.subject or ''} {ticket.body[:1500]}"
        query_embedding = get_embedding(query_text)
        
        matched_entry, confidence = find_best_knowledge_match(
            query_embedding, 
            threshold=Config.AUTO_REPLY_CONFIDENCE_THRESHOLD
        )
        
        if matched_entry:
            prompt = f"""
You are a helpful customer support assistant for a SaaS product. 
Write a warm, concise reply to the customer's message below. 
Use the provided knowledge base article as your source of truth. 
If the article doesn't fully address the issue, acknowledge what you know 
and politely ask for clarification. Keep the tone friendly and human.

Knowledge Base Article:
Title: {matched_entry.title}
Content: {matched_entry.content[:1500]}

Customer Message:
{query_text[:1500]}

Write the reply email now:
"""
            response = client.chat.completions.create(
                model=Config.OPENAI_MODEL,
                messages=[{"role": "user", "content": prompt}],
                temperature=0.4,
                max_tokens=500
            )
            reply_text = response.choices[0].message.content.strip()
            
            ticket.suggested_reply = reply_text
            ticket.metadata["kb_match_title"] = matched_entry.title
            ticket.metadata["kb_confidence"] = round(confidence, 3)
            
            if Config.AUTO_REPLY_ENABLED and confidence >= Config.AUTO_REPLY_CONFIDENCE_THRESHOLD:
                ticket.auto_reply_sent = True
                ticket.status = TicketStatus.AUTO_REPLIED
                # Send the actual email
                send_auto_reply_email.delay(ticket_id, reply_text)
            
            matched_entry.times_matched = (matched_entry.times_matched or 0) + 1
            db.session.commit()
            
            logger.info(f"Auto-reply generated for ticket {ticket_id}, confidence={confidence:.3f}")
        else:
            # No good match — escalate for human review
            ticket.metadata["kb_no_match"] = True
            db.session.commit()
            escalate_ticket.delay(ticket_id)

@celery.task(bind=True, max_retries=2)
def send_auto_reply_email(self, ticket_id, reply_body):
    """Send the auto-reply email via SMTP"""
    with app.app_context():
        ticket = Ticket.query.get(ticket_id)
        if not ticket:
            return
        
        import smtplib
        from email.mime.text import MIMEText
        from email.mime.multipart import MIMEMultipart
        
        msg = MIMEMultipart()
        msg["From"] = Config.FROM_EMAIL
        msg["To"] = ticket.customer_email
        msg["Subject"] = f"Re: {ticket.subject or 'Your support request'}"
        msg["In-Reply-To"] = ticket.external_id
        
        signature = "\n\n--\nThis is an automated reply. If you need further help, just reply to this email and a human will assist you."
        msg.attach(MIMEText(reply_body + signature, "plain"))
        
        try:
            with smtplib.SMTP(Config.SMTP_HOST, Config.SMTP_PORT) as server:
                server.starttls()
                server.login(Config.SMTP_USER, Config.SMTP_PASSWORD)
                server.send_message(msg)
            ticket.metadata["email_sent_at"] = datetime.now(timezone.utc).isoformat()
            db.session.commit()
            logger.info(f"Auto-reply email sent for ticket {ticket_id}")
        except Exception as e:
            logger.error(f"Failed to send email for ticket {ticket_id}: {str(e)}")
            raise self.retry(exc=e)

Step 4: Escalation Engine and Human Handoff

The escalation task notifies you (the solo founder) only when a ticket truly needs human attention. It can send you a Slack message, an SMS via Twilio, or a prioritized email. Below is a Slack webhook implementation. Add to tasks.py:

import requests

@celery.task(bind=True, max_retries=3)
def escalate_ticket(self, ticket_id):
    """Escalate ticket to the founder for manual review"""
    with app.app_context():
        ticket = Ticket.query.get(ticket_id)
        if not ticket:
            return
        
        if ticket.status != TicketStatus.ESCALATED:
            ticket.status = TicketStatus.ESCALATED
            db.session.commit()
        
        urgency_emoji = "🔴" if ticket.urgency_score >= 0.7 else "🟡"
        summary = ticket.metadata.get("ai_summary", ticket.body[:100])
        
        slack_message = {
            "text": f"{urgency_emoji} *Ticket #{ticket.id} needs your attention*",
            "blocks": [
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": f"{urgency_emoji} *Ticket #{ticket.id} Escalated*\n"
                                f"*From:* {ticket.customer_name} ({ticket.customer_email})\n"
                                f"*Category:* {ticket.category}/{ticket.subcategory}\n"
                                f"*Urgency:* {ticket.urgency_score:.2f}\n"
                                f"*Summary:* {summary}"
                    }
                },
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": f"{ticket.body[:500]}"
                    }
                },
                {
                    "type": "actions",
                    "elements": [
                        {
                            "type": "button",
                            "text": {"type": "plain_text", "text": "View Full Ticket"},
                            "url": f"https://yourdomain.com/admin/tickets/{ticket.id}",
                            "style": "primary"
                        }
                    ]
                }
            ]
        }
        
        slack_webhook_url = app.config.get("SLACK_WEBHOOK_URL")
        if slack_webhook_url:
            try:
                resp = requests.post(slack_webhook_url, json=slack_message, timeout=5)
                resp.raise_for_status()
                ticket.metadata["slack_notified"] = True
                db.session.commit()
                logger.info(f"Slack escalation sent for ticket {ticket_id}")
            except Exception as e:
                logger.error(f"Slack notification failed: {str(e)}")
                # Fallback: send an email to founder
                send_founder_alert.delay(ticket_id)

@celery.task(bind=True, max_retries=2)
def send_founder_alert(self, ticket_id):
    """Fallback: email the founder directly about escalated ticket"""
    with app.app_context():
        ticket = Ticket.query.get(ticket_id)
        if not ticket:
            return
        
        import smtplib
        from email.mime.text import MIMEText
        
        alert_body = f"""
Ticket #{ticket.id} requires your attention.

Customer: {ticket.customer_name} ({ticket.customer_email})
Category: {ticket.category} / {ticket.subcategory}
Urgency: {ticket.urgency_score:.2f}
Sentiment: {ticket.sentiment}

Message:
{ticket.body[:1000]}

View and respond: https://yourdomain.com/admin/tickets/{ticket.id}
"""
        msg = MIMEText(alert_body, "plain")
        msg["From"] = Config.FROM_EMAIL
        msg["To"] = Config.SMTP_USER  # Founder's email
        msg["Subject"] = f"[URGENT] Ticket #{ticket.id} escalated - {ticket.category}"
        
        try:
            with smtplib.SMTP(Config.SMTP_HOST, Config.SMTP_PORT) as server:
                server.starttls()
                server.login(Config.SMTP_USER, Config.SMTP_PASSWORD)
                server.send_message(msg)
            logger.info(f"Founder alert email sent for ticket {ticket_id}")
        except Exception as e:
            logger.error(f"Founder alert failed: {str(e)}")

Step 5: Admin Dashboard for Review and Override

Even with solid automation, you need a quick way to review auto-replies and override them if the AI got it wrong. This Flask endpoint serves a simple dashboard and an override API. Add to app.py:

@app.route("/admin/tickets", methods=["GET"])
def list_tickets():
    """List recent tickets with optional status filter"""
    status_filter = request.args.get("status")
    limit = min(int(request.args.get("limit", 50)), 200)
    
    query = Ticket.query.order_by(Ticket.created_at.desc())
    if status_filter:
        try:
            status_enum = TicketStatus(status_filter)
            query = query.filter_by(status=status_enum)
        except ValueError:
            return jsonify({"error": f"Invalid status: {status_filter}"}), 400
    
    tickets = query.limit(limit).all()
    
    return jsonify({
        "count": len(tickets),
        "tickets": [{
            "id": t.id,
            "external_id": t.external_id,
            "customer_email": t.customer_email,
            "customer_name": t.customer_name,
            "subject": t.subject,
            "body": t.body[:300],
            "status": t.status.value,
            "category": t.category,
            "urgency_score": t.urgency_score,
            "sentiment": t.sentiment,
            "suggested_reply": t.suggested_reply[:300] if t.suggested_reply else None,
            "auto_reply_sent": t.auto_reply_sent,
            "created_at": t.created_at.isoformat()
        } for t in tickets]
    })

@app.route("/admin/tickets/", methods=["GET"])
def get_ticket(ticket_id):
    """Get full ticket details"""
    ticket = Ticket.query.get(ticket_id)
    if not ticket:
        return jsonify({"error": "Ticket not found"}), 404
    
    return jsonify({
        "id": ticket.id,
        "external_id": ticket.external_id,
        "source": ticket.source.value,
        "customer_email": ticket.customer_email,
        "customer_name": ticket.customer_name,
        "subject": ticket.subject,
        "body": ticket.body,
        "status": ticket.status.value,
        "category": ticket.category,
        "subcategory": ticket.subcategory,
        "urgency_score": ticket.urgency_score,
        "sentiment": ticket.sentiment,
        "suggested_reply": ticket.suggested_reply,
        "auto_reply_sent": ticket.auto_reply_sent,
        "metadata": ticket.metadata,
        "created_at": ticket.created_at.isoformat(),
        "updated_at": ticket.updated_at.isoformat()
    })

@app.route("/admin/tickets//override", methods=["POST"])
def override_auto_reply(ticket_id):
    """Override the auto-reply with a manual response"""
    ticket = Ticket.query.get(ticket_id)
    if not ticket:
        return jsonify({"error": "Ticket not found"}), 404
    
    data = request.get_json()
    manual_reply = data.get("manual_reply")
    action = data.get("action", "send")
    
    if not manual_reply and action == "send":
        return jsonify({"error": "manual_reply is required when action is 'send'"}), 400
    
    if action == "send":
        ticket.suggested_reply = manual_reply
        ticket.auto_reply_sent = True
        ticket.status = TicketStatus.RESOLVED
        ticket.metadata["overridden_by_founder"] = True
        ticket.metadata["original_suggestion"] = ticket.suggested_reply
        db.session.commit()
        
        # Send the founder-written reply
        send_auto_reply_email.delay(ticket_id, manual_reply)
        
        return jsonify({"status": "sent", "ticket_id": ticket_id})
    
    elif action == "discard":
        ticket.suggested_reply = ""
        ticket.status = TicketStatus.ESCALATED
        ticket.metadata["auto_reply_discarded"] = True
        db.session.commit()
        return jsonify({"status": "discarded", "ticket_id": ticket_id})
    
    elif action == "resolve_silently":
        ticket.status = TicketStatus.RESOLVED
        ticket.metadata["resolved_without_reply"] = True
        db.session.commit()
        return jsonify({"status": "resolved", "ticket_id": ticket_id})
    
    return jsonify({"error": "Invalid action"}), 400

@app.route("/admin/knowledge", methods=["POST"])
def add_knowledge_entry():
    """Add a new entry to the knowledge base"""
    data = request.get_json()
    
    if not data or not data.get("title") or not data.get("content"):
        return jsonify({"error": "title and content are required"}), 400
    
    # Generate embedding for the new entry
    embedding = get_embedding(data["content"])
    
    entry = KnowledgeEntry(
        title=data["title"],
        content=data["content"],
        category=data.get("category", "general"),
        keywords=data.get("keywords", []),
        embedding=embedding,
        active=True
    )
    db.session.add(entry)
    db.session.commit()
    
    return jsonify({
        "status": "created",
        "entry_id": entry.id,
        "title": entry.title
    }), 201

@app.route("/admin/knowledge", methods=["GET"])
def list_knowledge():
    """List knowledge base entries"""
    entries = KnowledgeEntry.query.order_by(KnowledgeEntry.times_matched.desc()).all()
    return jsonify({
        "count": len(entries),
        "entries": [{
            "id": e.id,
            "title": e.title,
            "category": e.category,
            "active": e.active,
            "times_matched": e.times_matched or 0,
            "created_at": e.created_at.isoformat()
        } for e in entries]
    })

@app.route("/admin/knowledge/", methods=["PUT"])
def update_knowledge_entry(entry_id):
    """Update a knowledge entry and regenerate its embedding"""
    entry = KnowledgeEntry.query.get(entry_id)
    if not entry:
        return jsonify({"error": "Not found"}), 404
    
    data = request.get_json()
    if "title" in data:
        entry.title = data["title"]
    if "content" in data:
        entry.content = data["content"]
        entry.embedding = get_embedding(data["content"])
    if "category" in data:
        entry.category = data["category"]
    if "active" in data:
        entry.active = data["active"]
    
    db.session.commit()
    return jsonify({"status": "updated", "entry_id": entry_id})

Step 6: Running the Full Stack

Create a docker-compose.yml to orchestrate PostgreSQL, Redis, the Flask app, and the Celery worker in one command:

version: "3.9"
services:
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: support_automation
      POSTGRES_USER: support_user
      POSTGRES_PASSWORD: support_pass
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
  
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
  
  web:
    build: .
    command: gunicorn -w 4 -b 0.0.0.0:8000 app:app
    ports:
      - "8000:8000"
    environment:
      DATABASE_URL: postgresql://support_user:support_pass@db:5432/support_automation
      REDIS_URL: redis://redis:6379/0
      OPENAI_API_KEY: ${OPENAI_API_KEY}
      SMTP_HOST: ${SMTP_HOST}
      SMTP_PORT: ${SMTP_PORT}
      SMTP_USER: ${SMTP_USER}
      SMTP_PASSWORD: ${SMTP_PASSWORD}
    depends_on:
      - db
      - redis
  
  worker:
    build: .
    command: celery -A app.celery worker --loglevel=info
    environment:
      DATABASE_URL: postgresql://support_user:support_pass@db:5432/support_automation
      REDIS_URL: redis://redis:6379/0
      OPENAI_API_KEY: ${OPENAI_API_KEY}
      SMTP_HOST: ${SMTP_HOST}
      SMTP_PORT: ${SMTP_PORT}
      SMTP_USER: ${SMTP_USER}
      SMTP_PASSWORD: ${SMTP_PASSWORD}
    depends_on:
      - db
      - redis

volumes:
  pgdata:

Create a minimal Dockerfile:

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY

— Ad —

Google AdSense will appear here after approval

← Back to all articles