# Without multimodal: Convert image to description separately
def handle_image_query(user_text, image_url):
# Old approach: separate vision model call
vision_response = client.vision_model.describe_image(image_url)
description = vision_response.text
# Then feed to GPT for reasoning
chat_response = client.chat.completions.create(
model=”gpt-4-turbo”,
messages=[
{“role”: “user”, “content”: f”{user_text}\n\nImage: {description}”}
]
)
return chat_response.choices[0].message.content

# With GPT-4o: Native multimodal
def handle_image_query_native(user_text, image_url):
chat_response = client.chat.completions.create(
model=”gpt-4o”,
messages=[
{
“role”: “user”,
“content”: [
{“type”: “text”, “text”: user_text},
{“type”: “image_url”, “image_url”: {“url”: image_url}}
]
}
]
)
return chat_response.choices[0].message.content

User Query
|
v
[Input Validation & Sanitization]
|
v
[Conversation History Retrieval] -> PostgreSQL
|
v
[RAG: Retrieve Relevant Docs] -> Vector DB (Pinecone/Weaviate)
|
v
[Build Messages Array]
– System prompt
– Relevant context from RAG
– Conversation history (last N turns)
– Current user message
|
v
[Call GPT-4o API]
|
v
[Post-processing & Safety Checks]
|
v
[Store Response in History]
|
v
[Return to User]

You are Sam, a customer support agent for CloudFormula
(a data analytics platform).

Your responsibilities:
1. Answer questions about CloudFormula features, pricing,
and documentation
2. Help users troubleshoot data pipeline issues
3. Escalate to human agents when appropriate
4. Never make up features or pricing information

Constraints:
– Keep responses under 300 words
– Never claim to have access to customer billing data
(you cannot, and claiming so is a liability)
– If a user asks about refunds, say:
“Refund requests are handled by our billing team.
Please contact [email protected]”
– If you are not sure, say so explicitly and suggest
escalation
– Current date: April 2026

If the user asks about:
– Technical setup: refer to
https://docs.cloudformula.com
– Billing questions: escalate to [email protected]
– Security/compliance: escalate to [email protected]
– Product roadmap: escalate to [email protected]

import os
from openai import OpenAI

client = OpenAI(api_key=os.getenv(“OPENAI_API_KEY”))

def chat_with_gpt4o(user_message, conversation_history):
“””
Send a message to GPT-4o and get a response.

Args:
user_message: The current user input
conversation_history: List of previous messages

Returns:
The model’s response text
“””

# Build the messages list
system_prompt = “””You are a helpful customer support agent.
Keep responses concise and friendly.”””

messages = [{“role”: “system”, “content”: system_prompt}]
messages.extend(conversation_history)
messages.append({“role”: “user”, “content”: user_message})

# Call GPT-4o
response = client.chat.completions.create(
model=”gpt-4o”,
messages=messages,
temperature=0.7,
max_tokens=500
)

assistant_message = response.choices[0].message.content

# Return response and updated history
return assistant_message

# Example usage
conversation = []
user_input = “How do I reset my password?”
response = chat_with_gpt4o(user_input, conversation)
print(response)

# Add to history for next turn
conversation.append({“role”: “user”, “content”: user_input})
conversation.append({“role”: “assistant”, “content”: response})

import os
import json
from datetime import datetime
from openai import OpenAI
import psycopg2

client = OpenAI(api_key=os.getenv(“OPENAI_API_KEY”))

class ConversationManager:
def __init__(self, db_connection_string):
self.db = psycopg2.connect(db_connection_string)
self.cursor = self.db.cursor()

def save_message(self, conversation_id, role, content, tokens_used=0):
“””Store a message in PostgreSQL”””
self.cursor.execute(“””
INSERT INTO conversation_messages
(conversation_id, role, content, tokens_used, timestamp)
VALUES (%s, %s, %s, %s, %s)
“””, (conversation_id, role, content, tokens_used, datetime.now()))
self.db.commit()

def get_conversation_history(self, conversation_id, last_n_turns=10):
“””Retrieve the last N turns of conversation”””
self.cursor.execute(“””
SELECT role, content FROM conversation_messages
WHERE conversation_id = %s
ORDER BY timestamp DESC
LIMIT %s
“””, (conversation_id, last_n_turns * 2)) # *2 because each turn is user + assistant

rows = self.cursor.fetchall()
# Reverse to get chronological order
history = [{“role”: row[0], “content”: row[1]} for row in reversed(rows)]
return history

def count_tokens_in_history(self, history):
“””Rough estimate of tokens (4 chars per token)”””
total_chars = sum(len(msg[“content”]) for msg in history)
return total_chars // 4

def chat_with_memory(user_message, conversation_id, memory_manager):
“””
Chat with GPT-4o, managing conversation memory.
“””

system_prompt = “””You are a helpful customer support agent.
Keep responses concise and friendly.”””

# Retrieve conversation history
history = memory_manager.get_conversation_history(conversation_id)

# Build messages
messages = [{“role”: “system”, “content”: system_prompt}]
messages.extend(history)
messages.append({“role”: “user”, “content”: user_message})

# Call GPT-4o
response = client.chat.completions.create(
model=”gpt-4o”,
messages=messages,
temperature=0.7,
max_tokens=500
)

assistant_message = response.choices[0].message.content
tokens_used = response.usage.total_tokens

# Save to database
memory_manager.save_message(conversation_id, “user”, user_message)
memory_manager.save_message(conversation_id, “assistant”,
assistant_message, tokens_used)

return assistant_message

from pinecone import Pinecone
from openai import OpenAI

client = OpenAI(api_key=os.getenv(“OPENAI_API_KEY”))
pc = Pinecone(api_key=os.getenv(“PINECONE_API_KEY”))

def retrieve_relevant_docs(user_query, top_k=3):
“””
Retrieve relevant documents from vector database.
“””

# Embed the user query
embedding_response = client.embeddings.create(
model=”text-embedding-3-small”,
input=user_query
)
query_embedding = embedding_response.data[0].embedding

# Search Pinecone index
index = pc.Index(“chatbot-docs”)
search_results = index.query(
vector=query_embedding,
top_k=top_k,
include_metadata=True
)

# Extract document text from results
relevant_docs = []
for match in search_results[“matches”]:
doc_text = match[“metadata”].get(“text”, “”)
relevant_docs.append(doc_text)

return relevant_docs

def chat_with_rag(user_message, conversation_id, memory_manager):
“””
Chat with GPT-4o using RAG to ground responses.
“””

# Retrieve relevant documents
docs = retrieve_relevant_docs(user_message, top_k=3)

# Build context from docs
context = “Relevant information:\n”
for i, doc in enumerate(docs, 1):
context += f”{i}. {doc}\n”

system_prompt = f”””You are a helpful customer support agent.

{context}

Use the information above to answer questions accurately.
If the information does not contain an answer, say so explicitly.”””

# Get conversation history and call GPT-4o
history = memory_manager.get_conversation_history(conversation_id)

messages = [{“role”: “system”, “content”: system_prompt}]
messages.extend(history)
messages.append({“role”: “user”, “content”: user_message})

response = client.chat.completions.create(
model=”gpt-4o”,
messages=messages,
temperature=0.7,
max_tokens=500
)

assistant_message = response.choices[0].message.content

# Save to database
memory_manager.save_message(conversation_id, “user”, user_message)
memory_manager.save_message(conversation_id, “assistant”, assistant_message)

return assistant_message

import React, { useState } from ‘react’;

export default function ChatBot() {
const [messages, setMessages] = useState([]);
const [input, setInput] = useState(”);
const [loading, setLoading] = useState(false);

async function handleSend() {
if (!input.trim()) return;

setMessages(prev => […prev,
{ role: ‘user’, content: input }
]);
setInput(”);
setLoading(true);

try {
// Call your backend API (never expose API key)
const response = await fetch(‘/api/chat’, {
method: ‘POST’,
headers: { ‘Content-Type’: ‘application/json’ },
body: JSON.stringify({
message: input,
conversationId: localStorage.getItem(‘conversationId’)
})
});

if (!response.ok) throw new Error(‘API error’);

// Stream response
const reader = response.body.getReader();
const decoder = new TextDecoder();
let assistantMessage = ”;

while (true) {
const { done, value } = await reader.read();
if (done) break;

const chunk = decoder.decode(value);
assistantMessage += chunk;

// Update UI in real-time
setMessages(prev => {
const updated = […prev];
if (updated[updated.length – 1]?.role === ‘assistant’) {
updated[updated.length – 1].content = assistantMessage;
} else {
updated.push({ role: ‘assistant’, content: assistantMessage });
}
return updated;
});
}
} catch (error) {
console.error(‘API Error:’, error);
setMessages(prev => […prev, {
role: ‘assistant’,
content: ‘Sorry, I encountered an error. Please try again.’
}]);
} finally {
setLoading(false);
}
}

return (
<div className=”chat-container”>
<div className=”messages”>
{messages.map((msg, idx) => (
<div key={idx} className={`message ${msg.role}`}>
{msg.content}
</div>
))}
{loading && <div className=”message assistant”>Typing…</div>}
</div>
<div className=”input-area”>
<input
value={input}
onChange={(e) => setInput(e.target.value)}
onKeyDown={(e) => e.key === ‘Enter’ && handleSend()}
placeholder=”Type your question…”
disabled={loading}
/>
<button onClick={handleSend} disabled={loading}>Send</button>
</div>
</div>
);
}

import re
from typing import Tuple

class ResponseGuardrails:
def __init__(self):
self.blocked_phrases = [
r”i cannot access your account”,
r”i do not have authority”,
r”this requires human approval”
]
self.max_response_length = 1000

def validate_response(self, response: str) -> Tuple[bool, str]:
“””
Validate response before sending to user.

Returns:
(is_valid, potentially_modified_response)
“””

# Check length
if len(response) > self.max_response_length:
response = response[:self.max_response_length] + “…”

# Check for blocked patterns
for pattern in self.blocked_phrases:
if re.search(pattern, response, re.IGNORECASE):
return False, (“This requires human review. ”
“Escalating to support team.”)

# Check for potential PII exposure
if self._contains_pii(response):
return False, (“Response contains sensitive information. ”
“Escalating to human.”)

# Check for dangerous instructions
if self._contains_harmful_content(response):
return False, “I cannot provide that information.”

return True, response

def _contains_pii(self, text: str) -> bool:
“””Basic PII detection”””
# Credit card patterns
if re.search(r’\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}’, text):
return True
# Social security patterns
if re.search(r’\d{3}-\d{2}-\d{4}’, text):
return True
return False

def _contains_harmful_content(self, text: str) -> bool:
“””Detect harmful instruction patterns”””
harmful_patterns = [
r”delete.*database”,
r”drop table”,
r”rm -rf”,
r”format.*drive”
]
for pattern in harmful_patterns:
if re.search(pattern, text, re.IGNORECASE):
return True
return False

def validate_user_input(user_input: str) -> Tuple[bool, str]:
“””Check user input for injection attacks, spam, etc.”””

# Length check
if len(user_input) > 5000:
return False, (“Message too long. ”
“Keep it under 5000 characters.”)

# Empty check
if not user_input.strip():
return False, “Please enter a message.”

# SQL injection patterns (basic)
if any(pattern in user_input.lower()
for pattern in [“‘; drop”, “union select”, “exec(“]):
return False, “Invalid input detected.”

return True, user_input

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

ENV OPENAI_API_KEY=${OPENAI_API_KEY}
ENV PINECONE_API_KEY=${PINECONE_API_KEY}

EXPOSE 8000

CMD [“uvicorn”, “api:app”, “–host”, “0.0.0.0”,
“–port”, “8000”]

from opentelemetry import trace, metrics
from opentelemetry.exporter.prometheus import PrometheusMetricReader
from opentelemetry.sdk.resources import Resource

tracer = trace.get_tracer(__name__)
meter = metrics.get_meter(__name__)

# Track API latency
api_latency = meter.create_histogram(
name=”api.latency”,
unit=”ms”,
description=”Time to complete API call”
)

# Track token usage
tokens_used = meter.create_counter(
name=”tokens.used”,
unit=”1″,
description=”Total tokens processed”
)

# In your chat function
with tracer.start_as_current_span(“gpt4o_call”):
start_time = time.time()
response = client.chat.completions.create(…)
elapsed = (time.time() – start_time) * 1000

api_latency.record(elapsed)
tokens_used.add(response.usage.total_tokens)

def should_escalate(response_text: str, confidence_score: float) -> bool:
“””
Escalate if model is uncertain.

Note: GPT-4o does not return native confidence scores.
You must estimate based on response patterns.
“””

# Escalate if response contains uncertainty markers
uncertainty_markers = [
“i am not sure”,
“i believe”,
“it is possible that”,
“i am guessing”
]

if any(marker in response_text.lower()
for marker in uncertainty_markers):
return True

# Escalate if response is too short (likely incomplete thought)
if len(response_text) < 50:
return True

return False

async def summarize_old_context(conversation_id: str,
before_turn: int):
“””
Summarize conversation turns before a certain point.
Reduce old context to key facts.
“””
old_messages = get_messages(conversation_id,
before=before_turn)

summary_prompt = f”””
Summarize this conversation into 3-5 bullet points
of key facts. Focus on what the customer needs and
any decisions made.

{format_messages(old_messages)}
“””

summary = await client.chat.completions.create(
model=”gpt-4o”,
messages=[{“role”: “user”, “content”: summary_prompt}],
max_tokens=200
)

store_summary(conversation_id,
summary.choices[0].message.content)