Project Guide
Build a Production Customer Support Agent with LangGraph
Source: mortalapps.comThis guide provides a comprehensive, end-to-end tutorial on building a production-ready customer support agent AI with LangChain (specifically, LangGraph for orchestration). You will learn to construct an intelligent agent capable of handling customer inquiries, leveraging a Retrieval-Augmented Generation (RAG) system over a knowledge base, performing intent classification, and maintaining multi-turn conversation memory. Crucially, this agent incorporates a Human-in-the-Loop (HITL) escalation path, ensuring that complex or low-confidence queries are seamlessly handed off to a human support queue.
This project is designed for AI engineers, developers, and solution architects looking to deploy robust, reliable, and scalable AI agents in real-world customer service environments. It addresses the common business problem of reducing support ticket volume, improving response times, and ensuring consistent, accurate information delivery, while maintaining a safety net for complex issues.
By the end of this guide, you will have a fully functional, deployable customer support agent system, complete with a FastAPI interface, containerization via Docker, and an understanding of how to manage its lifecycle from development to production. We will explore an architecture that prioritizes modularity, testability, and resilience, making it suitable for enterprise applications.
What You Will Build
You will build a sophisticated customer support agent that intelligently processes incoming customer queries. The agent will first classify the user's intent (e.g., 'product inquiry', 'billing issue', 'technical support'). Based on the classified intent, it will either directly answer the query using its internal knowledge or perform a RAG lookup against a simulated product knowledge base. The agent maintains conversational context across multiple turns, allowing for fluid and natural interactions.
A key feature of this system is its human-in-the-loop (HITL) escalation mechanism. If the agent's confidence in its answer is below a predefined threshold, or if the intent is classified as requiring human intervention, the conversation will be flagged for human review and handed off to a simulated human support queue. The system will communicate primarily via a RESTful API endpoint, accepting user messages and returning agent responses. It will be packaged as a Docker container, ready for deployment on cloud platforms. The final architecture involves a LangGraph state machine orchestrating LLM calls, tool usage, and conditional routing based on intent and confidence scores, all exposed through a FastAPI application.
Technology Stack
| Component | Technology | Why This Choice |
|---|---|---|
| LLM Provider | OpenAI (GPT-4o) |
Industry-leading models for robust reasoning, intent classification, and generation capabilities. Offers a good balance of performance and cost. |
| Orchestration Framework | LangGraph |
Provides a powerful, flexible framework for building stateful, multi-actor applications. Its graph-based approach is ideal for complex conversational flows, conditional routing, and human-in-the-loop logic. |
| RAG/Vector Store | FAISS (in-memory) + OpenAI Embeddings |
FAISS offers efficient similarity search for retrieval, suitable for smaller knowledge bases or local development. OpenAI Embeddings provide high-quality vector representations of text for effective semantic search. |
| Tool Execution | LangChain Tools |
Seamless integration with LangGraph, providing a standardized way to define and execute external functionalities (like RAG lookups). |
| Memory Management | LangGraph Checkpointers (SQLite) |
Enables persistent state across turns, crucial for multi-turn conversations. SQLite is used for simplicity in local development, but easily swappable for production-grade databases like PostgreSQL. |
| API Interface | FastAPI |
A modern, fast (high-performance) web framework for building APIs with Python, offering automatic interactive API documentation (Swagger UI). |
| Observability (Basic) | Python Logging |
Provides essential visibility into agent execution flow and potential issues. Can be extended with OpenTelemetry for production monitoring. |
| Deployment | Docker |
Ensures consistent environments from development to production, simplifying deployment and scaling across various infrastructure types. |
The choice of OpenAI's GPT-4o as the core LLM is driven by its advanced reasoning capabilities, multimodal understanding, and strong performance in tasks like intent classification and nuanced response generation. While alternatives like Anthropic's Claude or open-source models like Llama 3 could be considered, GPT-4o offers a robust starting point for a production-grade agent, balancing quality and cost-effectiveness. Its API is also widely supported, simplifying integration.
LangGraph is selected for orchestration due to its explicit state management and graph-based approach, which is particularly well-suited for complex, multi-step agentic workflows. Unlike simpler sequential chains in LangChain, LangGraph allows for dynamic routing, conditional logic, and robust human-in-the-loop (HITL) patterns, which are critical for this customer support agent. Alternatives like standard LangChain agents or CrewAI are powerful but might offer less fine-grained control over state transitions and conditional logic, which are essential for our escalation path.
For RAG, we're combining FAISS with OpenAI Embeddings. FAISS provides extremely fast similarity search, making it suitable for retrieving relevant information from our knowledge base efficiently. OpenAI Embeddings are chosen for their quality and consistency with the main LLM, ensuring good semantic alignment between queries and documents. For production, FAISS would be replaced by a scalable vector database like Pinecone, Weaviate, or Qdrant, but for this guide, FAISS demonstrates the core RAG principle effectively without external dependencies.
FastAPI is the preferred choice for the API interface due to its asynchronous capabilities, Pydantic-based data validation, and automatic API documentation. This allows for rapid development of a robust and well-documented endpoint for interacting with the agent. Other frameworks like Flask or Django REST Framework are viable but FastAPI offers a modern, performant, and developer-friendly experience for this type of application.
Project Structure
customer_support_agent/
├── .env.example
├── Dockerfile
├── README.md
├── pyproject.toml
├── poetry.lock
├── src/
│ ├── __init__.py
│ ├── main.py
│ ├── agent.py
│ ├── tools.py
│ ├── config.py
│ └── knowledge_base.py
└── tests/
├── __init__.py
└── test_agent.py
The customer_support_agent/ is the root directory for our project. The .env.example file provides a template for environment variables, ensuring no secrets are hardcoded. The Dockerfile defines how to containerize our application. pyproject.toml and poetry.lock manage Python dependencies using Poetry, promoting reproducible builds.
The src/ directory contains all the core application logic. main.py will host our FastAPI application and expose the agent's API endpoint. agent.py encapsulates the LangGraph definition, including the state machine, nodes, and edges that define our agent's behavior. tools.py defines the external tools our agent can use, such as the RAG knowledge base lookup. config.py centralizes application-wide settings and environment variable loading. knowledge_base.py handles the loading and indexing of our simulated knowledge base, providing the data for our RAG tool. Finally, the tests/ directory will contain unit and integration tests to ensure our agent functions correctly and robustly.
Implementation
Phase 1: Environment Setup and Project Scaffold
This phase focuses on setting up the project environment, defining dependencies, and creating the basic file structure. We'll ensure all necessary libraries are installed and environment variables are properly configured for secure access to LLM APIs.
import os
import logging
# Configure basic logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# customer_support_agent/.env.example
# OPENAI_API_KEY="your_openai_api_key_here"
# KNOWLEDGE_BASE_PATH="./data/knowledge_base.txt"
# customer_support_agent/pyproject.toml
# [tool.poetry]
# name = "customer-support-agent"
# version = "0.1.0"
# description = "Production-ready customer support agent with LangGraph"
# authors = ["Your Name <[email protected]>"]
# readme = "README.md"
# packages = [{include = "src"}]
#
# [tool.poetry.dependencies]
# python = "^3.9"
# langchain = "^0.2.0"
# langchain-openai = "^0.1.0"
# langgraph = "^0.2.0"
# faiss-cpu = "^1.7.4"
# pydantic = "^2.7.1"
# python-dotenv = "^1.0.0"
# uvicorn = "^0.29.0"
# fastapi = "^0.111.0"
# numpy = "^1.26.4"
# tiktoken = "^0.7.0"
#
# [tool.poetry.group.dev.dependencies]
# pytest = "^8.2.0"
#
# [build-system]
# requires = ["poetry-core"]
# build-backend = "poetry.core.masonry.api"]
# src/config.py
import os
import logging
from dotenv import load_dotenv
logger = logging.getLogger(__name__)
class Settings:
def __init__(self):
load_dotenv() # Load environment variables from .env file
self.OPENAI_API_KEY: str = os.environ.get("OPENAI_API_KEY", "")
if not self.OPENAI_API_KEY:
logger.warning("OPENAI_API_KEY not found. LLM operations will fail.")
self.KNOWLEDGE_BASE_PATH: str = os.environ.get("KNOWLEDGE_BASE_PATH", "./data/knowledge_base.txt")
self.CONFIDENCE_THRESHOLD: float = float(os.environ.get("CONFIDENCE_THRESHOLD", "0.7"))
self.MAX_CONVERSATION_TURNS: int = int(os.environ.get("MAX_CONVERSATION_TURNS", "10"))
settings = Settings()
# src/__init__.py
# (empty file to mark src as a Python package)
# src/main.py
import logging
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from src.config import settings
logger = logging.getLogger(__name__)
app = FastAPI(
title="Customer Support Agent API",
description="API for interacting with the AI-powered customer support agent.",
version="0.1.0",
)
class ChatRequest(BaseModel):
user_id: str
message: str
class ChatResponse(BaseModel):
agent_response: str
escalated_to_human: bool = False
confidence: float | None = None
@app.post("/chat", response_model=ChatResponse)
async def chat_with_agent(request: ChatRequest):
try:
# The actual agent logic will be integrated here in later phases
logger.info(f"Received chat request for user {request.user_id}: {request.message}")
# Placeholder for now
return ChatResponse(
agent_response=f"Hello {request.user_id}, I received your message: '{request.message}'. I'm still in development!",
escalated_to_human=False,
confidence=1.0
)
except Exception as e:
logger.error(f"Error processing chat request for user {request.user_id}: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal Server Error")
@app.get("/health")
async def health_check():
return {"status": "healthy", "message": "Customer Support Agent is running."}
We begin by setting up our
pyproject.toml with necessary dependencies like langgraph, langchain-openai, faiss-cpu, fastapi, and uvicorn. Using Poetry ensures dependency isolation and reproducible builds. The .env.example file serves as a blueprint for environment variables, crucial for securely managing API keys and configuration settings. In src/config.py, we implement a Settings class that loads environment variables using python-dotenv. This centralizes configuration and ensures that sensitive information like OPENAI_API_KEY is never hardcoded. We also define a CONFIDENCE_THRESHOLD and MAX_CONVERSATION_TURNS here, which will be used later for agent decision-making.
The src/main.py file initializes a FastAPI application. We define two endpoints: /chat for interacting with the agent and /health for basic health checks. The ChatRequest and ChatResponse Pydantic models ensure strong typing for API payloads and responses, improving data validation and developer experience. Initially, the /chat endpoint contains a simple placeholder response, which will be replaced by the actual LangGraph agent execution in subsequent phases. Logging is configured at the module level to provide basic operational insights, capturing information about requests and errors.
Expected outcome
A runnable FastAPI application that responds to `/health` checks and a placeholder response for `/chat`. The project structure will be established with all initial dependencies installed and environment variables configured.
How to test
From the project root, run poetry install to install dependencies. Create a .env file from .env.example and populate OPENAI_API_KEY. Then, run poetry run uvicorn src.main:app --reload. Access http://127.0.0.1:8000/health in your browser to confirm a 'healthy' status. Use a tool like curl or Postman to send a POST request to http://127.0.0.1:8000/chat with a JSON body like {"user_id": "test_user", "message": "Hello there!"} and verify the placeholder response.
Phase 2: Core Agent Graph and State Definition
In this phase, we will define the core components of our LangGraph agent: the state, the nodes (representing distinct agent actions or decisions), and the edges (representing transitions between nodes). This sets up the foundational conversational flow including intent classification, RAG retrieval, and the initial human escalation check.
import os
import logging
from typing import Literal, TypedDict, List, Annotated
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
from src.config import settings
logger = logging.getLogger(__name__)
class AgentState(TypedDict):
chat_history: List[BaseMessage]
user_input: str
intent: Literal["product_inquiry", "billing_issue", "technical_support", "escalate_to_human", "general_chat"] | None
confidence: float | None
knowledge_base_query: str | None
knowledge_base_result: str | None
escalated_to_human: bool
class Agent:
def __init__(self, llm_model_name: str = "gpt-4o"):
if not settings.OPENAI_API_KEY:
raise ValueError("OPENAI_API_KEY is not set in environment variables.")
self.llm = ChatOpenAI(model=llm_model_name, temperature=0, api_key=settings.OPENAI_API_KEY)
self.memory = SqliteSaver(conn=sqlite3.connect(":memory:", check_same_thread=False)) # In-memory for simplicity, use file path for persistence
def _get_llm_with_retries(self, prompt: ChatPromptTemplate, max_retries: int = 3) -> ChatOpenAI:
for i in range(max_retries):
try:
return self.llm.with_structured_output(prompt)
except Exception as e:
logger.warning(f"LLM initialization failed (attempt {i+1}/{max_retries}): {e}")
if i == max_retries - 1:
raise RuntimeError(f"Failed to initialize LLM after {max_retries} attempts: {e}")
return self.llm.with_structured_output(prompt) # Should not be reached
def classify_intent(self, state: AgentState) -> AgentState:
logger.info("Node: classify_intent")
user_message = state["user_input"]
chat_history_str = "
".join([msg.content for msg in state["chat_history"][-2:]]) # Last 2 messages for context
prompt = ChatPromptTemplate.from_messages([
("system", "You are an intent classifier for a customer support agent. "
"Classify the user's intent based on the following conversation history and current message. "
"Output one of: product_inquiry, billing_issue, technical_support, escalate_to_human, general_chat. "
"If the user expresses frustration, anger, or explicitly asks for a human, choose 'escalate_to_human'."),
("user", f"Conversation history: {chat_history_str}
User message: {user_message}")
])
try:
# For now, we'll just get a string output and parse it. Later, we'll use structured output.
response = self.llm.invoke(prompt.format(conversation_history=chat_history_str, user_message=user_message))
intent_str = response.content.strip().lower()
# Simple parsing for initial phase, will improve with Pydantic in later phases
valid_intents = ["product_inquiry", "billing_issue", "technical_support", "escalate_to_human", "general_chat"]
intent = intent_str if intent_str in valid_intents else "general_chat"
# Simulate confidence for now
confidence = 0.9 if intent != "escalate_to_human" else 0.5
logger.info(f"Intent classified as: {intent} with confidence: {confidence}")
return {"intent": intent, "confidence": confidence, "user_input": user_message}
except Exception as e:
logger.error(f"Error classifying intent: {e}", exc_info=True)
return {"intent": "escalate_to_human", "confidence": 0.1, "user_input": user_message}
def determine_action(self, state: AgentState) -> Literal["retrieve_kb", "respond_general", "escalate"]:
logger.info(f"Node: determine_action. Current intent: {state['intent']}, confidence: {state['confidence']}")
if state["escalated_to_human"] or (state["confidence"] is not None and state["confidence"] < settings.CONFIDENCE_THRESHOLD) or state["intent"] == "escalate_to_human":
logger.info("Action: escalate to human due to low confidence or explicit request.")
return "escalate"
elif state["intent"] in ["product_inquiry", "technical_support", "billing_issue"]:
logger.info("Action: retrieve_kb for specific inquiry.")
return "retrieve_kb"
else:
logger.info("Action: respond_general for general chat.")
return "respond_general"
def build_graph(self):
workflow = StateGraph(AgentState)
workflow.add_node("classify_intent", self.classify_intent)
# Placeholder nodes for now, will be implemented in later phases
workflow.add_node("retrieve_kb", lambda state: {"knowledge_base_query": state["user_input"], "knowledge_base_result": "Placeholder KB result."})
workflow.add_node("generate_response", lambda state: {"chat_history": state["chat_history"] + [AIMessage(content="Placeholder AI response.")]})
workflow.add_node("escalate_to_human_node", lambda state: {"escalated_to_human": True, "chat_history": state["chat_history"] + [AIMessage(content="I need to escalate this to a human agent.")]})
workflow.add_node("update_chat_history", lambda state: {"chat_history": state["chat_history"] + [HumanMessage(content=state["user_input"])], "user_input": ""})
workflow.set_entry_point("classify_intent")
# Conditional routing from classify_intent
workflow.add_conditional_edges(
"classify_intent",
self.determine_action,
{
"retrieve_kb": "retrieve_kb",
"respond_general": "generate_response",
"escalate": "escalate_to_human_node",
},
)
workflow.add_edge("retrieve_kb", "generate_response")
workflow.add_edge("generate_response", "update_chat_history")
workflow.add_edge("update_chat_history", END)
workflow.add_edge("escalate_to_human_node", END)
return workflow.compile(checkpointer=self.memory)
agent_instance = Agent()
agent_executor = agent_instance.build_graph()
In
src/agent.py, we define the AgentState using TypedDict, which rigorously types the data that flows through our LangGraph. This state includes chat_history, user_input, intent, confidence, and flags for RAG and escalation. This explicit state definition is a cornerstone of LangGraph, making the agent's memory and current context transparent and manageable.
The Agent class initializes the ChatOpenAI model with our API key and sets up an SqliteSaver as a checkpointer for memory persistence. For this phase, the checkpointer uses a file-based SQLite database (memory.sqlite in the project root), which persists state across restarts and allows multi-turn conversations to resume after process restart. The classify_intent method uses a simple prompt to determine the user's intent, returning a raw string for now. We also introduce a determine_action method which acts as a conditional router: it decides the next step based on the classified intent and confidence score, including the crucial check for settings.CONFIDENCE_THRESHOLD to trigger escalation.
The build_graph method constructs the StateGraph. We add nodes for classify_intent, retrieve_kb (placeholder), generate_response (placeholder), escalate_to_human_node (placeholder), and update_chat_history (placeholder). The set_entry_point defines where the graph starts. Crucially, add_conditional_edges links classify_intent to determine_action, allowing dynamic transitions based on the intent and confidence. This creates the branching logic for our agent. Simple add_edge calls connect sequential steps, leading to END for termination. The graph is compiled with the checkpointer to enable state persistence.
Expected outcome
A foundational LangGraph agent capable of classifying user intent and routing the conversation flow based on that intent and a simulated confidence score. While responses are still placeholders, the core logic for decision-making and graph transitions is in place.
How to test
Modify src/main.py to import agent_executor and then create a simple test function within main.py to invoke the agent_executor with an initial AgentState dictionary. You can print the final state after execution to verify the intent classification and the determined next action. Remember that agent_executor.invoke will require a config argument with a configurable key specifying a thread_id for the checkpointer, e.g., {"configurable": {"thread_id": "test-123"}}.
Phase 3: Tool Integration - RAG Knowledge Base
This phase integrates a Retrieval-Augmented Generation (RAG) tool into our agent. We will create a simulated knowledge base, implement a tool to query it, and update the LangGraph to use this tool for specific inquiry types, enabling the agent to provide informed answers.
import os
import logging
from typing import List, Dict, Any
import json
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.tools import tool
from src.config import settings
logger = logging.getLogger(__name__)
# src/knowledge_base.py
class KnowledgeBase:
def __init__(self, path: str = settings.KNOWLEDGE_BASE_PATH):
self.path = path
self.vector_store = None
self._load_knowledge_base()
def _load_knowledge_base(self):
if not os.path.exists(self.path):
logger.warning(f"Knowledge base file not found at {self.path}. Creating a dummy one.")
os.makedirs(os.path.dirname(self.path), exist_ok=True)
with open(self.path, "w") as f:
f.write("Product A: A versatile gadget for daily use. Costs $99. Features include long battery life and waterproof design.
")
f.write("Product B: Our premium offering for professionals. Costs $299. Features include advanced processing and modular design.
")
f.write("Billing Issue: For any billing discrepancies, please check your invoice or contact our billing department directly.
")
f.write("Technical Support: For technical issues, please visit our troubleshooting guide or open a support ticket.
")
f.write("Return Policy: Items can be returned within 30 days of purchase for a full refund, provided they are in original condition.
")
f.write("Shipping: Standard shipping takes 3-5 business days. Express shipping is available for an extra charge.
")
try:
loader = TextLoader(self.path)
documents = loader.load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
texts = text_splitter.split_documents(documents)
logger.info(f"Loaded {len(texts)} documents from knowledge base.")
embeddings = OpenAIEmbeddings(api_key=settings.OPENAI_API_KEY)
self.vector_store = FAISS.from_documents(texts, embeddings)
logger.info("Knowledge base indexed successfully.")
except Exception as e:
logger.error(f"Failed to load or index knowledge base: {e}", exc_info=True)
self.vector_store = None # Ensure it's None on failure
def search(self, query: str, k: int = 3) -> List[str]:
if not self.vector_store:
logger.error("Vector store not initialized. Cannot perform search.")
return ["Error: Knowledge base unavailable."]
try:
docs = self.vector_store.similarity_search(query, k=k)
return [doc.page_content for doc in docs]
except Exception as e:
logger.error(f"Error searching knowledge base for query '{query}': {e}", exc_info=True)
return ["Error during knowledge base search."]
knowledge_base_instance = KnowledgeBase()
# src/tools.py
from src.knowledge_base import knowledge_base_instance
@tool
def retrieve_from_knowledge_base(query: str) -> str:
"""Searches the product knowledge base for information relevant to the user's query.
Input should be a standalone question or keywords related to the inquiry.
Returns a summary of relevant information from the knowledge base."""
logger.info(f"Tool: retrieve_from_knowledge_base with query: {query}")
try:
results = knowledge_base_instance.search(query)
if results:
return "
".join(results)
return "No relevant information found in the knowledge base."
except Exception as e:
logger.error(f"Error in retrieve_from_knowledge_base tool: {e}", exc_info=True)
return f"An error occurred while searching the knowledge base: {e}"
# src/agent.py (modifications)
import os
import logging
from typing import Literal, TypedDict, List, Annotated
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
from src.config import settings
from src.tools import retrieve_from_knowledge_base
logger = logging.getLogger(__name__)
class AgentState(TypedDict):
chat_history: List[BaseMessage]
user_input: str
intent: Literal["product_inquiry", "billing_issue", "technical_support", "escalate_to_human", "general_chat"] | None
confidence: float | None
knowledge_base_query: str | None
knowledge_base_result: str | None
escalated_to_human: bool
tool_calls: List[Dict[str, Any]] # Added for tool calls
tool_output: List[str] # Added for tool output
class Agent:
def __init__(self, llm_model_name: str = "gpt-4o"):
if not settings.OPENAI_API_KEY:
raise ValueError("OPENAI_API_KEY is not set in environment variables.")
self.llm = ChatOpenAI(model=llm_model_name, temperature=0, api_key=settings.OPENAI_API_KEY)
self.tools = [retrieve_from_knowledge_base]
self.llm_with_tools = self.llm.bind_tools(self.tools)
self.memory = SqliteSaver(conn=sqlite3.connect("file:memory.sqlite?mode=rwc", check_same_thread=False)) # Using file for persistence now
def _get_llm_with_retries(self, prompt: ChatPromptTemplate, max_retries: int = 3) -> ChatOpenAI:
for i in range(max_retries):
try:
# This method is for basic LLM calls, not tool-bound ones
return self.llm
except Exception as e:
logger.warning(f"LLM initialization failed (attempt {i+1}/{max_retries}): {e}")
if i == max_retries - 1:
raise RuntimeError(f"Failed to initialize LLM after {max_retries} attempts: {e}")
return self.llm # Should not be reached
def classify_intent(self, state: AgentState) -> AgentState:
logger.info("Node: classify_intent")
user_message = state["user_input"]
chat_history_str = "
".join([f"{msg.type}: {msg.content}" for msg in state["chat_history"][-2:]]) if state["chat_history"] else ""
prompt = ChatPromptTemplate.from_messages([
("system", "You are an intent classifier for a customer support agent. "
"Classify the user's intent based on the following conversation history and current message. "
"Output one of: product_inquiry, billing_issue, technical_support, escalate_to_human, general_chat. "
"If the user expresses frustration, anger, or explicitly asks for a human, choose 'escalate_to_human'."),
("user", f"Conversation history: {chat_history_str}
User message: {user_message}")
])
try:
response = self.llm.invoke(prompt)
intent_str = response.content.strip().lower()
valid_intents = ["product_inquiry", "billing_issue", "technical_support", "escalate_to_human", "general_chat"]
intent = intent_str if intent_str in valid_intents else "general_chat"
confidence = 0.9 if intent not in ["escalate_to_human", "general_chat"] else 0.6
if "frustrated" in user_message.lower() or "angry" in user_message.lower(): # Simple keyword check for demo
intent = "escalate_to_human"
confidence = 0.4
logger.info(f"Intent classified as: {intent} with confidence: {confidence}")
return {"intent": intent, "confidence": confidence, "user_input": user_message}
except Exception as e:
logger.error(f"Error classifying intent: {e}", exc_info=True)
return {"intent": "escalate_to_human", "confidence": 0.1, "user_input": user_message}
def call_tool(self, state: AgentState) -> AgentState:
logger.info("Node: call_tool")
tool_query = state["knowledge_base_query"]
if not tool_query:
logger.warning("No knowledge_base_query found in state for tool call.")
return {"knowledge_base_result": "No query provided for knowledge base.", "tool_output": ["No query provided"]}
try:
tool_result = retrieve_from_knowledge_base.invoke({"query": tool_query})
logger.info(f"Tool 'retrieve_from_knowledge_base' executed. Result length: {len(tool_result)}")
return {"knowledge_base_result": tool_result, "tool_output": [tool_result]}
except Exception as e:
logger.error(f"Error calling retrieve_from_knowledge_base tool: {e}", exc_info=True)
return {"knowledge_base_result": f"Error retrieving information: {e}", "tool_output": [f"Error: {e}"]}
def generate_response(self, state: AgentState) -> AgentState:
logger.info("Node: generate_response")
user_message = state["user_input"]
chat_history = state["chat_history"]
intent = state["intent"]
kb_result = state["knowledge_base_result"]
system_prompt_parts = [
"You are a helpful customer support agent.",
"Always provide clear, concise, and polite answers.",
"If you don't know the answer, state that you don't have enough information."
]
if intent == "product_inquiry":
system_prompt_parts.append("Focus on product details and features.")
elif intent == "billing_issue":
system_prompt_parts.append("Address billing concerns, but advise contacting billing department for specific account details.")
elif intent == "technical_support":
system_prompt_parts.append("Provide technical guidance or suggest troubleshooting steps.")
if kb_result and kb_result != "No relevant information found in the knowledge base.":
system_prompt_parts.append(f"Use the following knowledge base information to answer the user's query: {kb_result}")
full_system_prompt = " ".join(system_prompt_parts)
prompt = ChatPromptTemplate.from_messages([
("system", full_system_prompt),
*chat_history,
("human", user_message)
])
try:
response = self.llm.invoke(prompt)
ai_message = AIMessage(content=response.content)
logger.info(f"Generated response: {ai_message.content[:50]}...")
return {"chat_history": state["chat_history"] + [HumanMessage(content=user_message), ai_message]}
except Exception as e:
logger.error(f"Error generating response: {e}", exc_info=True)
return {"chat_history": state["chat_history"] + [HumanMessage(content=user_message), AIMessage(content="I apologize, but I encountered an error while trying to generate a response. Please try again or ask for human assistance.")]}
def update_chat_history(self, state: AgentState) -> AgentState:
logger.info("Node: update_chat_history")
# This node is primarily for updating the state with the user's input before processing
# and for ensuring the chat history is consistently managed.
return {"chat_history": state["chat_history"] + [HumanMessage(content=state["user_input"])], "user_input": ""}
def escalate_to_human_node(self, state: AgentState) -> AgentState:
logger.info("Node: escalate_to_human_node")
ai_message = AIMessage(content="I'm escalating your request to a human agent who will contact you shortly. Please provide your contact details if you haven't already.")
return {"escalated_to_human": True, "chat_history": state["chat_history"] + [ai_message]}
def build_graph(self):
workflow = StateGraph(AgentState)
workflow.add_node("classify_intent", self.classify_intent)
workflow.add_node("call_tool", self.call_tool)
workflow.add_node("generate_response", self.generate_response)
workflow.add_node("escalate_to_human_node", self.escalate_to_human_node)
# The update_chat_history node is implicitly handled by how we add messages in generate_response
# and classify_intent, but we'll keep the conceptual flow clear.
workflow.set_entry_point("classify_intent")
workflow.add_conditional_edges(
"classify_intent",
self.determine_action,
{
"retrieve_kb": "call_tool", # Route to call_tool for RAG
"respond_general": "generate_response",
"escalate": "escalate_to_human_node",
},
)
workflow.add_edge("call_tool", "generate_response") # After tool call, generate response
workflow.add_edge("generate_response", END)
workflow.add_edge("escalate_to_human_node", END)
return workflow.compile(checkpointer=self.memory)
agent_instance = Agent()
agent_executor = agent_instance.build_graph()
# src/main.py (modifications)
import logging
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from src.agent import agent_executor, AgentState
from src.config import settings
logger = logging.getLogger(__name__)
app = FastAPI(
title="Customer Support Agent API",
description="API for interacting with the AI-powered customer support agent.",
version="0.1.0",
)
class ChatRequest(BaseModel):
user_id: str
message: str
class ChatResponse(BaseModel):
agent_response: str
escalated_to_human: bool = False
confidence: float | None = None
thread_id: str
@app.post("/chat", response_model=ChatResponse)
async def chat_with_agent(request: ChatRequest):
try:
# Initial state or retrieve from checkpointer (handled by agent_executor automatically)
# LangGraph automatically loads/saves state via the checkpointer based on thread_id
# Prepare the input for the agent
initial_state = {
"user_input": request.message,
"chat_history": [HumanMessage(content=request.message)], # Initial message is part of history
"escalated_to_human": False,
"confidence": None,
"intent": None,
"knowledge_base_query": None,
"knowledge_base_result": None,
"tool_calls": [],
"tool_output": []
}
# Invoke the agent executor with the user's message and thread_id
# LangGraph will manage loading and saving the state for this thread_id
final_state: AgentState = agent_executor.invoke(
input=initial_state,
config={
"configurable": {"thread_id": request.user_id}
}
)
# Extract the last AI message from the chat history
ai_response = "An unexpected error occurred."
escalated = final_state.get("escalated_to_human", False)
confidence = final_state.get("confidence")
if final_state.get("chat_history"):
for msg in reversed(final_state["chat_history"]):
if isinstance(msg, AIMessage):
ai_response = msg.content
break
logger.info(f"Chat request for user {request.user_id} processed. Escalated: {escalated}, Confidence: {confidence}")
return ChatResponse(
agent_response=ai_response,
escalated_to_human=escalated,
confidence=confidence,
thread_id=request.user_id
)
except Exception as e:
logger.error(f"Error processing chat request for user {request.user_id}: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal Server Error")
@app.get("/health")
async def health_check():
# Check if LLM can be initialized or perform a dummy call for readiness
try:
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, api_key=settings.OPENAI_API_KEY)
llm.invoke("Hello") # Small dummy call to check connectivity
return {"status": "healthy", "message": "Customer Support Agent is running and LLM is accessible."}
except Exception as e:
logger.error(f"Health check failed: LLM not accessible. {e}", exc_info=True)
raise HTTPException(status_code=503, detail=f"Service Unavailable: LLM connectivity issue. {e}")
We introduce
src/knowledge_base.py to manage our RAG data. This class loads a simple text file, splits it into documents, and indexes them into an in-memory FAISS vector store using OpenAIEmbeddings. This setup simulates a real knowledge base, allowing the agent to retrieve relevant information. For demonstration, if the file doesn't exist, it creates a dummy knowledge base. Error handling is included for file operations and vector store creation.
In src/tools.py, we define retrieve_from_knowledge_base as a LangChain tool. The @tool decorator makes it discoverable by LangChain agents. This tool takes a query and uses our KnowledgeBase instance to search for information, returning the relevant document content. Robust error handling is present to catch issues during tool execution.
src/agent.py is updated to integrate this tool. The AgentState now includes tool_calls and tool_output to track tool interactions. The Agent constructor binds the retrieve_from_knowledge_base tool to the LLM. A new node, call_tool, is added to the LangGraph, responsible for executing the RAG tool. The classify_intent node is enhanced to refine intent classification and a basic keyword check for immediate escalation. The generate_response node is updated to dynamically construct its system prompt, incorporating the knowledge_base_result when available, making the agent's answers contextually rich. The graph's conditional edges are modified to route to call_tool when a specific inquiry intent is detected.
Finally, src/main.py is updated to properly invoke the agent_executor with the user's message and a thread_id (derived from user_id) for state persistence. The response extraction now correctly retrieves the last AIMessage and communicates escalation status and confidence back to the user.
Expected outcome
The agent can now classify intents, and for specific inquiry types (product, technical, billing), it will attempt to retrieve information from the simulated knowledge base using the RAG tool before generating a response. General queries and explicit escalation requests will follow their respective paths. Multi-turn conversation state will be persisted using the SQLite checkpointer.
How to test
Ensure your .env has OPENAI_API_KEY set. Run poetry run uvicorn src.main:app --reload. Send POST requests to /chat with various messages. Try: "What is Product A?", "How much does Product B cost?", "I have a billing issue.", "I am very angry, I want to talk to a human!", "Tell me a joke.". Observe the logs for intent classification, tool calls, and the generated responses. Verify that consecutive messages from the same user_id maintain conversational context.
Phase 4: Multi-turn Conversation Memory and Robust Response Generation
This phase refines the agent's ability to maintain coherent multi-turn conversations by properly managing chat history and ensuring robust, context-aware response generation. We will also enhance the generate_response node to produce more nuanced and helpful answers, taking into account all available state information.
import os
import logging
from typing import Literal, TypedDict, List, Annotated, Any, Dict
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.pydantic_v1 import BaseModel, Field
from src.config import settings
from src.tools import retrieve_from_knowledge_base
logger = logging.getLogger(__name__)
# src/agent.py (modifications for generate_response and State)
# Pydantic model for structured intent classification output
class IntentClassifierOutput(BaseModel):
intent: Literal["product_inquiry", "billing_issue", "technical_support", "escalate_to_human", "general_chat"]
confidence: float = Field(..., ge=0.0, le=1.0)
reasoning: str
class AgentState(TypedDict):
chat_history: Annotated[List[BaseMessage], add_messages] # Ensure chat history persists
user_input: str
intent: Literal["product_inquiry", "billing_issue", "technical_support", "escalate_to_human", "general_chat"] | None
confidence: float | None
knowledge_base_query: str | None
knowledge_base_result: str | None
escalated_to_human: bool
tool_calls: List[Dict[str, Any]]
tool_output: List[str]
class Agent:
def __init__(self, llm_model_name: str = "gpt-4o"):
if not settings.OPENAI_API_KEY:
raise ValueError("OPENAI_API_KEY is not set in environment variables.")
self.llm = ChatOpenAI(model=llm_model_name, temperature=0.2, api_key=settings.OPENAI_API_KEY) # Slightly higher temperature for generation
self.llm_classifier = ChatOpenAI(model=llm_model_name, temperature=0, api_key=settings.OPENAI_API_KEY).with_structured_output(IntentClassifierOutput)
self.tools = [retrieve_from_knowledge_base]
self.llm_with_tools = self.llm.bind_tools(self.tools)
self.memory = SqliteSaver(conn=sqlite3.connect("file:memory.sqlite?mode=rwc", check_same_thread=False))
def classify_intent(self, state: AgentState) -> AgentState:
logger.info("Node: classify_intent")
user_message = state["user_input"]
# Use entire chat history for better context in classification
full_chat_history_for_classification = "
".join([f"{msg.type}: {msg.content}" for msg in state["chat_history"]]) if state["chat_history"] else ""
prompt = ChatPromptTemplate.from_messages([
("system", "You are an intent classifier for a customer support agent. "
"Classify the user's intent based on the following conversation history and current message. "
"Output one of: product_inquiry, billing_issue, technical_support, escalate_to_human, general_chat. "
"If the user expresses frustration, anger, or explicitly asks for a human, or if the intent is ambiguous, choose 'escalate_to_human'. "
"Provide a confidence score (0.0-1.0) and a brief reasoning."),
("user", f"Conversation history: {full_chat_history_for_classification}
User message: {user_message}")
])
try:
# Use the structured output LLM for classification
classification_output: IntentClassifierOutput = self.llm_classifier.invoke(prompt)
intent = classification_output.intent
confidence = classification_output.confidence
# Apply business logic for escalation threshold override
if confidence < settings.CONFIDENCE_THRESHOLD or intent == "escalate_to_human":
intent = "escalate_to_human"
confidence = min(confidence, 0.4) # Lower confidence for explicit escalation
logger.info(f"Intent classified as: {intent} with confidence: {confidence} (Reasoning: {classification_output.reasoning})")
return {"intent": intent, "confidence": confidence, "user_input": user_message}
except Exception as e:
logger.error(f"Error classifying intent: {e}", exc_info=True)
# Fallback to escalation on classification failure
return {"intent": "escalate_to_human", "confidence": 0.1, "user_input": user_message}
def determine_action(self, state: AgentState) -> Literal["retrieve_kb", "respond_general", "escalate"]:
logger.info(f"Node: determine_action. Current intent: {state['intent']}, confidence: {state['confidence']}")
if state["escalated_to_human"] or (state["confidence"] is not None and state["confidence"] < settings.CONFIDENCE_THRESHOLD) or state["intent"] == "escalate_to_human":
logger.info("Action: escalate to human due to low confidence or explicit request.")
return "escalate"
elif state["intent"] in ["product_inquiry", "technical_support", "billing_issue"]:
logger.info("Action: retrieve_kb for specific inquiry.")
# If intent is for KB, the KB query should be the user_input itself for initial retrieval
return "retrieve_kb"
else:
logger.info("Action: respond_general for general chat.")
return "respond_general"
def call_tool(self, state: AgentState) -> AgentState:
logger.info("Node: call_tool")
# For RAG, the query is typically the user's last input or a refined version of it.
tool_query = state["user_input"]
if not tool_query:
logger.warning("No user_input found in state for tool call.")
return {"knowledge_base_result": "No query provided for knowledge base.", "tool_output": ["No query provided"]}
try:
tool_result = retrieve_from_knowledge_base.invoke({"query": tool_query})
logger.info(f"Tool 'retrieve_from_knowledge_base' executed. Result length: {len(tool_result)}")
return {"knowledge_base_result": tool_result, "tool_output": [tool_result]}
except Exception as e:
logger.error(f"Error calling retrieve_from_knowledge_base tool: {e}", exc_info=True)
return {"knowledge_base_result": f"Error retrieving information: {e}", "tool_output": [f"Error: {e}"]}
def generate_response(self, state: AgentState) -> AgentState:
logger.info("Node: generate_response")
user_message = state["user_input"]
chat_history = state["chat_history"]
intent = state["intent"]
kb_result = state["knowledge_base_result"]
# Build a more sophisticated system prompt based on context and KB results
system_prompt_parts = [
"You are a helpful and empathetic customer support agent.",
"Your goal is to provide accurate, concise, and polite answers.",
"If you cannot find a definitive answer, politely state that you lack sufficient information and offer to escalate."
]
if intent == "product_inquiry":
system_prompt_parts.append("Focus on providing detailed product information, features, and pricing if available.")
elif intent == "billing_issue":
system_prompt_parts.append("Address billing concerns generally, but always advise the user to contact the dedicated billing department for account-specific details.")
elif intent == "technical_support":
system_prompt_parts.append("Offer relevant troubleshooting steps, links to support articles (if you were given any), or guide them to technical support channels.")
elif intent == "general_chat":
system_prompt_parts.append("Engage in friendly conversation or redirect to relevant topics if possible.")
if kb_result and "No relevant information found" not in kb_result and "Error" not in kb_result:
system_prompt_parts.append(f"\
\
IMPORTANT KNOWLEDGE BASE INFORMATION:
{kb_result}\
\
Use the above information to formulate your response. Do not explicitly mention 'knowledge base' unless necessary. Integrate the information naturally.")
else:
system_prompt_parts.append("\
\
No specific knowledge base information was found for this query, or an error occurred during retrieval. Rely on your general knowledge.")
full_system_prompt = " ".join(system_prompt_parts)
# Construct messages for the LLM, including all relevant chat history
messages_for_llm = [
("system", full_system_prompt),
]
# Add actual chat history, excluding the current user_input which is handled separately
for msg in chat_history:
if isinstance(msg, HumanMessage):
messages_for_llm.append(("human", msg.content))
elif isinstance(msg, AIMessage):
messages_for_llm.append(("ai", msg.content))
elif isinstance(msg, ToolMessage):
messages_for_llm.append(("tool", msg.content))
# Add the current user input as the last human message
messages_for_llm.append(("human", user_message))
prompt = ChatPromptTemplate.from_messages(messages_for_llm)
try:
response = self.llm.invoke(prompt)
ai_message = AIMessage(content=response.content)
logger.info(f"Generated response: {ai_message.content[:80]}...")
# Update chat history with current user message and AI response
new_chat_history = state["chat_history"] + [HumanMessage(content=user_message), ai_message]
# Truncate chat history if it exceeds MAX_CONVERSATION_TURNS
if len(new_chat_history) > settings.MAX_CONVERSATION_TURNS * 2: # *2 because each turn has human and AI message
logger.info(f"Truncating chat history. Current length: {len(new_chat_history)}")
new_chat_history = new_chat_history[-(settings.MAX_CONVERSATION_TURNS * 2):]
return {"chat_history": new_chat_history}
except Exception as e:
logger.error(f"Error generating response: {e}", exc_info=True)
# Fallback response on generation failure
fallback_message = "I apologize, but I encountered an error while trying to generate a response. Please try again or ask for human assistance."
return {"chat_history": state["chat_history"] + [HumanMessage(content=user_message), AIMessage(content=fallback_message)]}
def escalate_to_human_node(self, state: AgentState) -> AgentState:
logger.info("Node: escalate_to_human_node")
ai_message = AIMessage(content="I'm escalating your request to a human agent who will contact you shortly. To help them, please provide your contact details and a brief summary of your issue if you haven't already.")
# Ensure the escalation message is added to history
return {"escalated_to_human": True, "chat_history": state["chat_history"] + [HumanMessage(content=state["user_input"]), ai_message]}
def build_graph(self):
workflow = StateGraph(AgentState)
workflow.add_node("classify_intent", self.classify_intent)
workflow.add_node("call_tool", self.call_tool)
workflow.add_node("generate_response", self.generate_response)
workflow.add_node("escalate_to_human_node", self.escalate_to_human_node)
workflow.set_entry_point("classify_intent")
workflow.add_conditional_edges(
"classify_intent",
self.determine_action,
{
"retrieve_kb": "call_tool",
"respond_general": "generate_response",
"escalate": "escalate_to_human_node",
},
)
workflow.add_edge("call_tool", "generate_response")
workflow.add_edge("generate_response", END)
workflow.add_edge("escalate_to_human_node", END)
return workflow.compile(checkpointer=self.memory)
agent_instance = Agent()
agent_executor = agent_instance.build_graph()
# src/main.py (minor adjustment to initial state in /chat endpoint)
import logging
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from src.agent import agent_executor, AgentState
from src.config import settings
logger = logging.getLogger(__name__)
app = FastAPI(
title="Customer Support Agent API",
description="API for interacting with the AI-powered customer support agent.",
version="0.1.0",
)
class ChatRequest(BaseModel):
user_id: str
message: str
class ChatResponse(BaseModel):
agent_response: str
escalated_to_human: bool = False
confidence: float | None = None
thread_id: str
@app.post("/chat", response_model=ChatResponse)
async def chat_with_agent(request: ChatRequest):
try:
# LangGraph automatically loads/saves state via the checkpointer based on thread_id
# The initial state passed to invoke is merged with the loaded state from the checkpointer.
# Only provide the *new* user message. Chat history is managed by the checkpointer.
initial_input = {
"user_input": request.message,
"chat_history": [], # This will be populated by the checkpointer if thread_id exists
"escalated_to_human": False,
"confidence": None,
"intent": None,
"knowledge_base_query": None,
"knowledge_base_result": None,
"tool_calls": [],
"tool_output": []
}
final_state: AgentState = agent_executor.invoke(
input=initial_input,
config={
"configurable": {"thread_id": request.user_id}
}
)
ai_response = "An unexpected error occurred."
escalated = final_state.get("escalated_to_human", False)
confidence = final_state.get("confidence")
if final_state.get("chat_history"):
for msg in reversed(final_state["chat_history"]):
if isinstance(msg, AIMessage):
ai_response = msg.content
break
logger.info(f"Chat request for user {request.user_id} processed. Escalated: {escalated}, Confidence: {confidence}")
return ChatResponse(
agent_response=ai_response,
escalated_to_human=escalated,
confidence=confidence,
thread_id=request.user_id
)
except Exception as e:
logger.error(f"Error processing chat request for user {request.user_id}: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal Server Error")
This phase significantly enhances the agent's conversational capabilities. In
src/agent.py, we introduce Annotated[List[BaseMessage], {"persist": True}] for chat_history in AgentState. The add_messages reducer ensures messages are appended rather than replaced; the SqliteSaver checkpointer handles persistence automatically, ensuring that the entire conversation history is loaded and saved across turns for a given thread_id.
We refine classify_intent to use a structured output Pydantic model (IntentClassifierOutput) for intent and confidence. This improves the reliability and parseability of the LLM's classification. The LLM for classification is now bound with with_structured_output, ensuring a predictable output format. The generate_response node is significantly improved: it now dynamically constructs a more detailed system prompt based on the classified intent and the presence of RAG results. It includes the full chat_history (from AgentState) in the prompt to the LLM, enabling truly multi-turn, context-aware responses. We also add logic to truncate the chat_history to settings.MAX_CONVERSATION_TURNS to prevent context window exhaustion, a critical production consideration.
The call_tool node is updated to ensure the knowledge_base_query is always derived from the user_input, making the RAG process more direct. The escalate_to_human_node also ensures the current user input is recorded before the escalation message. In src/main.py, the chat_history in the initial input to agent_executor.invoke is now an empty list; LangGraph's checkpointer is responsible for loading the actual history based on the thread_id, then merging the user_input into the state. This simplifies the API interaction and correctly leverages LangGraph's state management.
Expected outcome
The agent now maintains coherent multi-turn conversations, leveraging the full chat history for intent classification and response generation. RAG results are seamlessly integrated into more nuanced responses, and the system robustly handles context window management and escalation based on confidence and explicit user requests.
How to test
Run the application and test multi-turn conversations. Ask about Product A, then follow up with "And what about Product B?" or "Can I return it?". Observe how the agent maintains context. Test the escalation path by expressing strong dissatisfaction or explicitly asking for a human. Verify that the agent's responses are more natural and context-aware. Check the memory.sqlite file (it should be created in the root directory) to confirm state persistence across restarts (kill uvicorn, restart, and continue a conversation).
Phase 5: API Endpoint Integration and Error Handling
This phase focuses on robustly integrating the LangGraph agent into the FastAPI application, ensuring proper error handling, logging, and a clear API contract for external interaction. We will also refine the FastAPI endpoint to reflect the agent's capabilities accurately.
import os
import logging
from typing import Literal, TypedDict, List, Annotated, Any, Dict
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.pydantic_v1 import BaseModel, Field
from langgraph.graph.graph import CompiledGraph
from src.config import settings
from src.tools import retrieve_from_knowledge_base
logger = logging.getLogger(__name__)
# src/agent.py (no changes, just re-listing for completeness)
# Pydantic model for structured intent classification output
class IntentClassifierOutput(BaseModel):
intent: Literal["product_inquiry", "billing_issue", "technical_support", "escalate_to_human", "general_chat"]
confidence: float = Field(..., ge=0.0, le=1.0)
reasoning: str
class AgentState(TypedDict):
chat_history: Annotated[List[BaseMessage], add_messages]
user_input: str
intent: Literal["product_inquiry", "billing_issue", "technical_support", "escalate_to_human", "general_chat"] | None
confidence: float | None
knowledge_base_query: str | None
knowledge_base_result: str | None
escalated_to_human: bool
tool_calls: List[Dict[str, Any]]
tool_output: List[str]
class Agent:
def __init__(self, llm_model_name: str = "gpt-4o"):
if not settings.OPENAI_API_KEY:
raise ValueError("OPENAI_API_KEY is not set in environment variables.")
self.llm = ChatOpenAI(model=llm_model_name, temperature=0.2, api_key=settings.OPENAI_API_KEY)
self.llm_classifier = ChatOpenAI(model=llm_model_name, temperature=0, api_key=settings.OPENAI_API_KEY).with_structured_output(IntentClassifierOutput)
self.tools = [retrieve_from_knowledge_base]
self.llm_with_tools = self.llm.bind_tools(self.tools)
self.memory = SqliteSaver(conn=sqlite3.connect("file:memory.sqlite?mode=rwc", check_same_thread=False))
def classify_intent(self, state: AgentState) -> AgentState:
logger.info("Node: classify_intent")
user_message = state["user_input"]
full_chat_history_for_classification = "
".join([f"{msg.type}: {msg.content}" for msg in state["chat_history"]]) if state["chat_history"] else ""
prompt = ChatPromptTemplate.from_messages([
("system", "You are an intent classifier for a customer support agent. "
"Classify the user's intent based on the following conversation history and current message. "
"Output one of: product_inquiry, billing_issue, technical_support, escalate_to_human, general_chat. "
"If the user expresses frustration, anger, or explicitly asks for a human, or if the intent is ambiguous, choose 'escalate_to_human'. "
"Provide a confidence score (0.0-1.0) and a brief reasoning."),
("user", f"Conversation history: {full_chat_history_for_classification}
User message: {user_message}")
])
try:
classification_output: IntentClassifierOutput = self.llm_classifier.invoke(prompt)
intent = classification_output.intent
confidence = classification_output.confidence
if confidence < settings.CONFIDENCE_THRESHOLD or intent == "escalate_to_human":
intent = "escalate_to_human"
confidence = min(confidence, 0.4)
logger.info(f"Intent classified as: {intent} with confidence: {confidence} (Reasoning: {classification_output.reasoning})")
return {"intent": intent, "confidence": confidence, "user_input": user_message}
except Exception as e:
logger.error(f"Error classifying intent: {e}", exc_info=True)
return {"intent": "escalate_to_human", "confidence": 0.1, "user_input": user_message}
def determine_action(self, state: AgentState) -> Literal["retrieve_kb", "respond_general", "escalate"]:
logger.info(f"Node: determine_action. Current intent: {state['intent']}, confidence: {state['confidence']}")
if state["escalated_to_human"] or (state["confidence"] is not None and state["confidence"] < settings.CONFIDENCE_THRESHOLD) or state["intent"] == "escalate_to_human":
logger.info("Action: escalate to human due to low confidence or explicit request.")
return "escalate"
elif state["intent"] in ["product_inquiry", "technical_support", "billing_issue"]:
logger.info("Action: retrieve_kb for specific inquiry.")
return "retrieve_kb"
else:
logger.info("Action: respond_general for general chat.")
return "respond_general"
def call_tool(self, state: AgentState) -> AgentState:
logger.info("Node: call_tool")
tool_query = state["user_input"]
if not tool_query:
logger.warning("No user_input found in state for tool call.")
return {"knowledge_base_result": "No query provided for knowledge base.", "tool_output": ["No query provided"]}
try:
tool_result = retrieve_from_knowledge_base.invoke({"query": tool_query})
logger.info(f"Tool 'retrieve_from_knowledge_base' executed. Result length: {len(tool_result)}")
return {"knowledge_base_result": tool_result, "tool_output": [tool_result]}
except Exception as e:
logger.error(f"Error calling retrieve_from_knowledge_base tool: {e}", exc_info=True)
return {"knowledge_base_result": f"Error retrieving information: {e}", "tool_output": [f"Error: {e}"]}
def generate_response(self, state: AgentState) -> AgentState:
logger.info("Node: generate_response")
user_message = state["user_input"]
chat_history = state["chat_history"]
intent = state["intent"]
kb_result = state["knowledge_base_result"]
system_prompt_parts = [
"You are a helpful and empathetic customer support agent.",
"Your goal is to provide accurate, concise, and polite answers.",
"If you cannot find a definitive answer, politely state that you lack sufficient information and offer to escalate."
]
if intent == "product_inquiry":
system_prompt_parts.append("Focus on providing detailed product information, features, and pricing if available.")
elif intent == "billing_issue":
system_prompt_parts.append("Address billing concerns generally, but always advise the user to contact the dedicated billing department for account-specific details.")
elif intent == "technical_support":
system_prompt_parts.append("Offer relevant troubleshooting steps, links to support articles (if you were given any), or guide them to technical support channels.")
elif intent == "general_chat":
system_prompt_parts.append("Engage in friendly conversation or redirect to relevant topics if possible.")
if kb_result and "No relevant information found" not in kb_result and "Error" not in kb_result:
system_prompt_parts.append(f"\
\
IMPORTANT KNOWLEDGE BASE INFORMATION:
{kb_result}\
\
Use the above information to formulate your response. Do not explicitly mention 'knowledge base' unless necessary. Integrate the information naturally.")
else:
system_prompt_parts.append("\
\
No specific knowledge base information was found for this query, or an error occurred during retrieval. Rely on your general knowledge.")
full_system_prompt = " ".join(system_prompt_parts)
messages_for_llm = [
("system", full_system_prompt),
]
for msg in chat_history:
if isinstance(msg, HumanMessage):
messages_for_llm.append(("human", msg.content))
elif isinstance(msg, AIMessage):
messages_for_llm.append(("ai", msg.content))
elif isinstance(msg, ToolMessage):
messages_for_llm.append(("tool", msg.content))
messages_for_llm.append(("human", user_message))
prompt = ChatPromptTemplate.from_messages(messages_for_llm)
try:
response = self.llm.invoke(prompt)
ai_message = AIMessage(content=response.content)
logger.info(f"Generated response: {ai_message.content[:80]}...")
new_chat_history = state["chat_history"] + [HumanMessage(content=user_message), ai_message]
if len(new_chat_history) > settings.MAX_CONVERSATION_TURNS * 2:
logger.info(f"Truncating chat history. Current length: {len(new_chat_history)}")
new_chat_history = new_chat_history[-(settings.MAX_CONVERSATION_TURNS * 2):]
return {"chat_history": new_chat_history}
except Exception as e:
logger.error(f"Error generating response: {e}", exc_info=True)
fallback_message = "I apologize, but I encountered an error while trying to generate a response. Please try again or ask for human assistance."
return {"chat_history": state["chat_history"] + [HumanMessage(content=user_message), AIMessage(content=fallback_message)]}
def escalate_to_human_node(self, state: AgentState) -> AgentState:
logger.info("Node: escalate_to_human_node")
ai_message = AIMessage(content="I'm escalating your request to a human agent who will contact you shortly. To help them, please provide your contact details and a brief summary of your issue if you haven't already.")
return {"escalated_to_human": True, "chat_history": state["chat_history"] + [HumanMessage(content=state["user_input"]), ai_message]}
def build_graph(self) -> CompiledGraph:
workflow = StateGraph(AgentState)
workflow.add_node("classify_intent", self.classify_intent)
workflow.add_node("call_tool", self.call_tool)
workflow.add_node("generate_response", self.generate_response)
workflow.add_node("escalate_to_human_node", self.escalate_to_human_node)
workflow.set_entry_point("classify_intent")
workflow.add_conditional_edges(
"classify_intent",
self.determine_action,
{
"retrieve_kb": "call_tool",
"respond_general": "generate_response",
"escalate": "escalate_to_human_node",
},
)
workflow.add_edge("call_tool", "generate_response")
workflow.add_edge("generate_response", END)
workflow.add_edge("escalate_to_human_node", END)
return workflow.compile(checkpointer=self.memory)
agent_instance = Agent()
agent_executor = agent_instance.build_graph()
# src/main.py (final version)
import logging
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from typing import List, Optional
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from src.agent import agent_executor, AgentState
from src.config import settings
logger = logging.getLogger(__name__)
app = FastAPI(
title="Customer Support Agent API",
description="API for interacting with the AI-powered customer support agent.",
version="0.1.0",
contact={
"name": "MortalAI Support",
"url": "https://mortalapps.com/support",
"email": "[email protected]",
},
license_info={
"name": "Apache 2.0",
"url": "https://www.apache.org/licenses/LICENSE-2.0.html",
},
)
class ChatRequest(BaseModel):
user_id: str = Field(..., description="Unique identifier for the user/conversation thread.")
message: str = Field(..., description="The user's message to the agent.")
class ChatResponse(BaseModel):
agent_response: str = Field(..., description="The agent's response to the user.")
escalated_to_human: bool = Field(False, description="True if the conversation was escalated to a human agent.")
confidence: Optional[float] = Field(None, description="The agent's confidence in its last response (0.0-1.0).")
thread_id: str = Field(..., description="The ID of the conversation thread.")
@app.post("/chat", response_model=ChatResponse, summary="Send a message to the customer support agent.")
async def chat_with_agent(request: ChatRequest):
"""Sends a user message to the AI customer support agent and receives a response.
The agent maintains conversation context based on `user_id`. If the agent's
confidence is low or an explicit escalation is requested, the `escalated_to_human`
flag will be set to `true` in the response.
"""
logger.info(f"Received chat request for user {request.user_id}: {request.message[:100]}...")
try:
# The initial state passed to invoke is merged with the loaded state from the checkpointer.
# Only provide the *new* user message. Chat history is managed by the checkpointer.
initial_input = {
"user_input": request.message,
"chat_history": [],
"escalated_to_human": False,
"confidence": None,
"intent": None,
"knowledge_base_query": None,
"knowledge_base_result": None,
"tool_calls": [],
"tool_output": []
}
final_state: AgentState = agent_executor.invoke(
input=initial_input,
config={
"configurable": {"thread_id": request.user_id}
}
)
ai_response = "I apologize, but I encountered an unexpected issue. Please try again or contact support."
escalated = final_state.get("escalated_to_human", False)
confidence = final_state.get("confidence")
# Extract the last AI message from the chat history
if final_state.get("chat_history"):
for msg in reversed(final_state["chat_history"]):
if isinstance(msg, AIMessage):
ai_response = msg.content
break
logger.info(f"Chat request for user {request.user_id} processed. Escalated: {escalated}, Confidence: {confidence}. Response: {ai_response[:100]}...")
return ChatResponse(
agent_response=ai_response,
escalated_to_human=escalated,
confidence=confidence,
thread_id=request.user_id
)
except Exception as e:
logger.error(f"Unhandled error processing chat request for user {request.user_id}: {e}", exc_info=True)
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Internal Server Error: An unexpected error occurred.")
@app.get("/health", summary="Health check endpoint.")
async def health_check():
"""Checks the health of the application and LLM connectivity."""
try:
# Perform a small dummy call to check LLM connectivity
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, api_key=settings.OPENAI_API_KEY)
_ = llm.invoke("Hello", max_tokens=5)
logger.info("Health check successful: LLM accessible.")
return {"status": "healthy", "message": "Customer Support Agent is running and LLM is accessible."}
except Exception as e:
logger.error(f"Health check failed: LLM not accessible. {e}", exc_info=True)
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=f"Service Unavailable: LLM connectivity issue. {e}")
This phase finalizes the
src/main.py FastAPI application. We enhance the /chat and /health endpoints with more descriptive summary and description fields, which automatically populate the OpenAPI (Swagger UI) documentation. Pydantic Field objects are used within ChatRequest and ChatResponse to add detailed descriptions and validation, improving the API's usability and clarity.
Crucially, robust error handling is implemented. The /chat endpoint now wraps the agent_executor.invoke call in a try...except block, catching any unexpected exceptions during agent execution and returning a HTTPException with a 500 Internal Server Error status. This prevents unhandled server crashes and provides a consistent error response to API consumers. Similarly, the /health endpoint performs a dummy LLM call to verify connectivity to the OpenAI API. If this call fails, it raises a 503 Service Unavailable error, providing a clear signal of downstream dependency issues. This ensures that our health check is not just superficial but actively verifies critical components. Logging is maintained throughout to provide visibility into request processing, agent decisions, and error occurrences.
Expected outcome
A fully functional FastAPI application with well-documented endpoints. The `/chat` endpoint now robustly handles agent interactions, including multi-turn conversations and escalation, with comprehensive error handling. The `/health` endpoint accurately reflects the service's readiness, including LLM connectivity. The agent will be ready for containerization and deployment.
How to test
Ensure your .env is correctly set up. Run poetry run uvicorn src.main:app --reload. Navigate to http://127.0.0.1:8000/docs to view the automatically generated API documentation. Test the /chat endpoint with various inputs, including valid queries, escalation requests, and potentially invalid API keys (by temporarily removing OPENAI_API_KEY from .env) to observe the error handling. Test the /health endpoint to ensure it correctly reports service status and LLM connectivity.
Testing & Evaluation
Testing and evaluation are critical for ensuring the reliability and effectiveness of a production customer support agent. We'll cover functional testing, LLM-as-a-Judge evaluation, failure scenario testing, edge case coverage, performance validation, and regression testing.
Functional Testing
Functional tests verify that the agent behaves as expected for common user queries and scenarios. This includes:
- Intent Classification Accuracy: Test with a diverse set of messages covering all defined intents (product inquiry, billing, technical, general, escalation) to ensure correct classification.
- RAG Effectiveness: Query for information known to be in the knowledge base and verify the agent retrieves and synthesizes the correct information. Test for queries that require multi-hop reasoning if applicable.
- Multi-turn Coherence: Engage in multi-turn conversations to ensure the agent maintains context, refers to previous turns, and provides relevant follow-up information.
- Escalation Path: Verify that explicit requests for a human, expressions of frustration, or low-confidence scenarios correctly trigger the escalation flag.
LLM-as-a-Judge Evaluation
For qualitative aspects like helpfulness, politeness, and relevance, an LLM-as-a-Judge approach is highly effective. This involves using a powerful LLM to evaluate the agent's responses against a rubric.
import pytest
import requests
import os
from typing import Dict, Any
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage, SystemMessage
# Assuming FastAPI app is running on http://localhost:8000
BASE_URL = "http://localhost:8000"
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")
if not OPENAI_API_KEY:
pytest.skip("OPENAI_API_KEY not set, skipping LLM-as-a-Judge tests", allow_module_level=True)
llm_judge = ChatOpenAI(model="gpt-4o", temperature=0, api_key=OPENAI_API_KEY)
def evaluate_response(query: str, agent_response: str, expected_behavior: str) -> Dict[str, Any]:
prompt = ChatPromptTemplate.from_messages([
SystemMessage(
"You are an impartial judge evaluating a customer support agent's response. "
"Your task is to assess if the agent's response is helpful, polite, relevant, and meets the expected behavior. "
"Provide a score from 1 to 5 (1=poor, 5=excellent) for each criterion: 'Helpfulness', 'Politeness', 'Relevance'. "
"Also, provide a 'Overall Score' from 1 to 5 and a 'Reasoning'.
"
f"Expected behavior for this query: {expected_behavior}"
),
HumanMessage(content=f"User query: {query}
Agent response: {agent_response}")
])
try:
response = llm_judge.invoke(prompt)
# A more robust parser would be needed for production, but this demonstrates the concept.
# For simplicity, we'll try to parse key scores from content.
content = response.content
scores = {}
for criterion in ['Helpfulness', 'Politeness', 'Relevance', 'Overall Score']:
if f"{criterion}:" in content:
try:
score_str = content.split(f"{criterion}:")[1].split('
')[0].strip()
scores[criterion.lower().replace(' ', '_')] = int(score_str.split('/')[0].strip())
except (ValueError, IndexError):
scores[criterion.lower().replace(' ', '_')] = 0 # Default to 0 on parse error
reasoning = content.split("Reasoning:")[-1].strip() if "Reasoning:" in content else content
scores['reasoning'] = reasoning
return scores
except Exception as e:
print(f"Error during LLM-as-a-Judge evaluation: {e}")
return {"error": str(e)}
@pytest.mark.parametrize("user_id, query, expected_intent, expected_escalation, expected_judge_behavior", [
("user1", "What are the features of Product A?", "product_inquiry", False, "Provide detailed features of Product A."),
("user2", "How do I return a faulty item?", "technical_support", False, "Explain the return policy or guide to support."),
("user3", "I'm very upset, I want to talk to a manager!", "escalate_to_human", True, "Politely escalate to a human."),
("user4", "Tell me about your company.", "general_chat", False, "Provide a general overview of the company.")
])
def test_agent_with_llm_judge(user_id, query, expected_intent, expected_escalation, expected_judge_behavior):
response = requests.post(f"{BASE_URL}/chat", json={"user_id": user_id, "message": query})
assert response.status_code == 200
data = response.json()
assert data['thread_id'] == user_id
assert data['escalated_to_human'] == expected_escalation
# Evaluate agent's response using LLM-as-a-Judge
judge_scores = evaluate_response(query, data['agent_response'], expected_judge_behavior)
print(f"
--- LLM Judge for Query: '{query}' ---")
print(f"Agent Response: {data['agent_response']}")
print(f"Judge Scores: {judge_scores}")
# Assert minimum acceptable scores for non-escalated cases
if not expected_escalation:
assert judge_scores.get('helpfulness', 0) >= 3
assert judge_scores.get('politeness', 0) >= 4
assert judge_scores.get('relevance', 0) >= 4
assert judge_scores.get('overall_score', 0) >= 3
else:
# For escalation, we primarily care that it escalated and the message is appropriate
assert "escalating" in data['agent_response'].lower() or "human agent" in data['agent_response'].lower()
assert judge_scores.get('politeness', 0) >= 4 # Still expect politeness even when escalatingFailure Scenario Testing
- LLM API Downtime: Simulate by invalidating
OPENAI_API_KEY. Verify the agent gracefully handles the error, logs it, and returns a generic error message or escalates. - Knowledge Base Failure: Simulate by corrupting
knowledge_base.txtor makingFAISSfail. Verify the RAG tool returns an error message, and the agent responds appropriately (e.g., states it couldn't find information). - Invalid Input: Send malformed JSON requests to the API to test Pydantic validation and FastAPI's error handling.
- Context Window Exhaustion: Send very long conversations or extremely detailed messages to test the
MAX_CONVERSATION_TURNStruncation logic and ensure the agent doesn't crash.
Edge Case Coverage
- Ambiguous Intents: Queries like "I have a problem" should ideally lead to follow-up questions or escalation if the confidence is low.
- Empty Messages: What happens if an empty string is sent? The agent should handle this gracefully.
- Rapid-fire Messages: Test concurrent requests from the same
user_idto ensure checkpointer consistency.
Performance Validation
- Latency: Measure the P99 latency of the
/chatendpoint under various loads. Identify bottlenecks (e.g., LLM inference time, RAG retrieval time). - Throughput: Determine how many requests per second the agent can handle before degradation.
- Cost Analysis: Track token usage and API costs for typical conversations to understand operational expenses.
Regression Testing
Automated test suites (like the pytest example above) should be integrated into a CI/CD pipeline. Any new feature or bug fix must not break existing functionality. This includes running both functional and LLM-as-a-Judge tests against a baseline to detect regressions in agent behavior or quality.
Deployment
Deploying our customer support agent involves containerizing the application and orchestrating it in a production environment. We will use Docker for containerization and discuss deployment strategies for Kubernetes or cloud-native services.
Dockerfile
Our Dockerfile will create a lightweight image that includes all dependencies and our application code.
# Use an official Python runtime as a parent image
FROM python:3.9-slim-buster
# Set the working directory in the container
WORKDIR /app
# Install Poetry
RUN pip install poetry==1.8.2
# Copy pyproject.toml and poetry.lock to the working directory
COPY pyproject.toml poetry.lock ./
# Install project dependencies using Poetry
# The --no-root flag prevents installing the project itself as a package yet
# The --no-dev flag ensures dev dependencies are not installed in production
RUN poetry install --no-root --no-dev
# Copy the rest of the application code
COPY src/ ./src/
COPY data/ ./data/ # Assuming knowledge_base.txt is in a 'data' folder
# Expose the port the app runs on
EXPOSE 8000
# Create a non-root user for security
RUN adduser --system --group appuser
USER appuser
# Define environment variables for the application
ENV PYTHONUNBUFFERED=1 \
UVICORN_APP="src.main:app" \
UVICORN_HOST="0.0.0.0" \
UVICORN_PORT="8000"
# Command to run the application
CMD ["poetry", "run", "uvicorn", "${UVICORN_APP}", "--host", "${UVICORN_HOST}", "--port", "${UVICORN_PORT}"]To build the image: docker build -t customer-support-agent:latest .
To run locally: docker run -p 8000:8000 -e OPENAI_API_KEY="your_key_here" -e CONFIDENCE_THRESHOLD="0.7" customer-support-agent:latest
Kubernetes Manifest (Example)
For production, deploying on Kubernetes provides scalability, self-healing, and load balancing. Here's a basic deployment.yaml:
apiVersion: apps/v1
kind: Deployment
metadata:
name: customer-support-agent
labels:
app: customer-support-agent
spec:
replicas: 3 # Start with 3 replicas for high availability
selector:
matchLabels:
app: customer-support-agent
template:
metadata:
labels:
app: customer-support-agent
spec:
containers:
- name: agent
image: customer-support-agent:latest # Replace with your pushed image, e.g., yourrepo/customer-support-agent:latest
ports:
- containerPort: 8000
env:
- name: KNOWLEDGE_BASE_PATH
value: "/app/data/knowledge_base.txt" # Path inside the container
- name: CONFIDENCE_THRESHOLD
value: "0.7"
- name: MAX_CONVERSATION_TURNS
value: "10"
- name: OPENAI_API_KEY # Referencing a Kubernetes Secret
valueFrom:
secretKeyRef:
name: openai-api-key-secret
key: api_key
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 15
periodSeconds: 20
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 5
periodSeconds: 10
resources:
requests:
memory: "256Mi"
cpu: "200m"
limits:
memory: "1Gi"
cpu: "1"
# For persistent memory, you'd integrate a PersistentVolumeClaim here
# For SQLite checkpointer, consider using an external database like PostgreSQL
# or a shared volume for the SQLite file in single-replica scenarios.
# For multi-replica, external checkpointer (e.g., Redis, Postgres) is mandatory.
---
apiVersion: v1
kind: Service
metadata:
name: customer-support-agent-service
spec:
selector:
app: customer-support-agent
ports:
- protocol: TCP
port: 80
targetPort: 8000
type: LoadBalancer # Exposes the service externallySecrets Management
Never hardcode API keys. In Kubernetes, use Secrets. Create one: kubectl create secret generic openai-api-key-secret --from-literal=api_key='your_openai_api_key' This secret is referenced in the deployment.yaml using valueFrom.secretKeyRef.
Health Check and Readiness Probe
The livenessProbe ensures that if the container becomes unresponsive (e.g., due to an unhandled exception), Kubernetes restarts it. The readinessProbe ensures that traffic is only sent to the pod once it's truly ready to serve requests (e.g., after the LLM connectivity check passes). Both use our /health endpoint.
Scaling Considerations
- Horizontal Pod Autoscaler (HPA): Configure HPA to automatically scale the number of agent replicas based on CPU utilization or custom metrics (e.g., queue length for incoming requests).
- External Checkpointer: For multi-replica deployments, the
SqliteSaveris insufficient. Replace it with aPostgresSaverorRedisSaverto ensure all replicas share the same conversational state. This is crucial for seamless user experience across scaled instances. - Knowledge Base: For larger knowledge bases, replace FAISS with a managed vector database service (e.g., Pinecone, Weaviate, Qdrant, Azure AI Search, AWS OpenSearch) to handle scaling and persistence.
Rollback Strategy
Kubernetes deployments support easy rollbacks. If a new deployment introduces issues, you can revert to a previous stable version:
-
kubectl rollout history deployment/customer-support-agentto see revision history. -
kubectl rollout undo deployment/customer-support-agentto revert to the previous revision.
Ensure new deployments are thoroughly tested in staging environments before pushing to production. Implement blue/green or canary deployments for minimal downtime and risk.
Production Hardening
Hardening a production customer support agent involves addressing security, observability, reliability, and cost-efficiency. This section outlines key considerations and actions.
Security
- Input Validation and Sanitization: Implement strict input validation on all API endpoints (
FastAPIPydantic models help significantly). Sanitize user inputs before passing them to LLMs or tools to prevent prompt injection and data leakage. This agent already uses Pydantic for API requests, but further checks for malicious patterns inuser_inputare beneficial. - Access Control: Secure your API endpoints with authentication (e.g., API keys, OAuth2). Ensure only authorized applications or users can interact with the agent. The current FastAPI app is open; consider adding a dependency for API key validation.
- Least Privilege Principle: Ensure the container runs with the minimum necessary permissions. Our Dockerfile already creates a non-root
appuser. Further restrict network access and mounted volumes. - Dependency Scanning: Regularly scan your
poetry.lock(orrequirements.txt) for known vulnerabilities using tools like Snyk or Dependabot. - LLM Guardrails: Implement additional LLM-specific guardrails (e.g., using NeMo Guardrails, Microsoft's Guidance) to prevent the agent from generating harmful, toxic, or off-topic content, especially during general chat or unexpected inputs.
Observability
- Structured Logging: Enhance Python's
loggingto output structured logs (e.g., JSON format) that includethread_id,node_name,event_type,latency, andtoken_usage. Integrate with a centralized log management system (e.g., ELK Stack, Splunk, Datadog). - Distributed Tracing: Implement OpenTelemetry tracing to track requests across the entire agent workflow, from API ingress through LangGraph nodes, LLM calls, and tool executions. This is crucial for debugging complex multi-step interactions and performance analysis. Export traces to a compatible backend like Jaeger or Honeycomb. Refer to
/agents/protocols/tracing-multi-agent-workflows/. - Metrics and Alerting: Collect key metrics like request rate, error rate, LLM token usage, P99 latency, and escalation rate. Set up alerts for anomalies (e.g., sudden spikes in error rates, high latency, or unexpected increase in escalations).
Reliability
- Retry Mechanisms with Exponential Backoff: Implement retry logic for external API calls (LLM, vector DB, external tools) with exponential backoff to handle transient failures. Our
_get_llm_with_retriesis a basic example; extend this to all external calls. - Fallback LLM Providers: Configure a fallback LLM provider (e.g., a different OpenAI region or an entirely different vendor like Anthropic) in case the primary LLM becomes unavailable or hits rate limits. This enhances service resilience. Refer to
/agents/protocols/fallback-llm-provider-reliability/. - Circuit Breakers: Implement circuit breakers for flaky external services to prevent cascading failures.
- Graceful Degradation: If a non-critical component (e.g., a specific RAG source) fails, ensure the agent can still provide a graceful fallback response instead of crashing.
Rate Limiting
- API Rate Limiting: Implement rate limiting on the FastAPI endpoint to protect your service from abuse and ensure fair usage. Use libraries like
fastapi-limiter. - LLM Rate Limiting: Be aware of and manage rate limits imposed by your LLM provider. Implement client-side rate limiting or batching if necessary to avoid exceeding these limits, which can lead to
429 Too Many Requestserrors.
Authentication
- API Key Management: For client applications, use secure API key distribution and rotation. For internal services, consider mutual TLS or service mesh authentication.
- User Identity: If integrating with customer CRMs, ensure proper user identity management and authorization to access sensitive customer data.
Governance
- Audit Logging: Log all critical agent decisions, escalations, and human interventions. This creates an audit trail for compliance, debugging, and post-incident analysis. Refer to
/agents/memory-and-state/auditing-a2a-agent-communication-logs/. - Data Retention Policies: Define and enforce policies for retaining conversation history and associated data, especially if PII is involved, to comply with regulations like GDPR or CCPA. Ensure the SQLite checkpointer is configured for this, or migrate to a compliant external database.
Cost Optimisation
- Token Usage Monitoring: Continuously monitor and analyze token consumption per conversation, per user, and per intent. Identify areas where prompts can be optimized for conciseness without losing effectiveness. Refer to
/agents/foundations/ai-agent-token-economics/. - Model Tiering: Use cheaper, faster models (e.g.,
gpt-4o-mini) for simple tasks like general chat or initial intent classification, and reserve more expensive models (e.g.,gpt-4o) for complex reasoning or RAG synthesis. Our current setup usesgpt-4ofor both; consider agpt-4o-miniforclassify_intent. - Caching: Implement caching for frequently accessed RAG queries or LLM responses that are deterministic and non-sensitive. Refer to
/agents/protocols/mcp-caching-ttlms-cacheScope/.
Compliance
- Data Residency: If deploying globally, ensure customer data (especially conversation history) is stored and processed in the correct geographical regions to meet data residency requirements. This means replacing SQLite with a cloud-native, region-specific database. Refer to
/agents/production-engineering/data-residency-context-governance/. - Privacy by Design: Design the agent with privacy in mind from the outset, minimizing data collection, anonymizing sensitive information where possible, and providing clear user consent mechanisms.
Learning Objectives
- Design and implement a multi-turn conversational agent using LangGraph's state machine capabilities
- Integrate Retrieval-Augmented Generation (RAG) to provide context-aware answers from an external knowledge base
- Implement intent classification and confidence scoring for dynamic agent behavior and escalation
- Develop a Human-in-the-Loop (HITL) escalation path for complex or low-confidence queries
- Manage conversational memory and state persistence using LangGraph checkpointers
- Containerize an AI agent application using Docker for consistent deployment
- Apply best practices for logging, error handling, and secret management in production AI systems
Conceptual Prerequisites
What Is Agentic AI?
Understanding the core concepts of AI agents, their components, and how they differ from simple LLM wrappers is crucial for designing effective and autonomous systems.
ReAct Pattern: Reasoning and Acting in Agents
Familiarity with the ReAct pattern will help in understanding how the agent reasons, plans, and executes actions (tools) to achieve its goals.
AI Agent Memory Systems
A grasp of working, episodic, and procedural memory concepts is essential for managing conversational context and enabling multi-turn interactions.
State Machines for AI Agents
Understanding state machines is fundamental to working with LangGraph, which models agent workflows as stateful graphs with nodes and edges.
Technical Prerequisites
Python 3.9+
Ensure you have a modern Python environment installed. We'll be using features available in recent Python versions.
Poetry or Pip
A Python package manager (Poetry recommended for dependency management) to install project libraries.
OpenAI API Key
Access to the OpenAI API (or another LLM provider) is required for the agent's language model components.
Docker
Docker Desktop or Docker Engine is needed to build and run the containerized application.