test / chatbot_langgraph_validated.py
SGaleshchuk's picture
add
e983a89 verified
Raw
History Blame Contribute Delete
31.3 kB
"""
LangGraph-based Car Finder Chatbot with Validation Agents
This implementation includes:
- Template-based SQL queries (no SQL generation)
- LangGraph for agent orchestration
- Tool-based architecture for security
- Content moderation agent (validates user input)
- Response quality agent (validates assistant responses)
- State management for conversation flow
"""
from typing import TypedDict, Annotated, Optional, Literal
from operator import add
import sqlite3
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from pydantic import BaseModel, Field
# Load environment variables
load_dotenv()
# Validate API key
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key:
raise ValueError("OPENAI_API_KEY environment variable is not set")
# Initialize LLMs
llm = ChatOpenAI(model="gpt-4o", temperature=0.7, api_key=api_key)
llm_validator = ChatOpenAI(model="gpt-4o-mini", temperature=0, api_key=api_key) # Cheaper for validation
# Load database schema with error handling
try:
with open('database_schema.txt', 'r') as f:
SCHEMA_DESCRIPTION = f.read()
except FileNotFoundError:
raise FileNotFoundError("database_schema.txt not found. Please ensure it exists in the current directory.")
# Constants
MIN_RESULTS = 1
MAX_RESULTS = 20
DB_PATH = 'cars.db'
# ============================================================================
# PYDANTIC MODELS FOR VALIDATION
# ============================================================================
class ContentModerationResult(BaseModel):
"""Result from content moderation validation"""
is_safe: bool = Field(description="True if content is safe, False if harmful")
violation_type: Optional[Literal["hate_speech", "harassment", "profanity", "spam", "inappropriate", "off_topic"]] = None
severity: Optional[Literal["low", "medium", "high"]] = None
explanation: str = Field(description="Brief explanation of the decision")
should_block: bool = Field(description="True if the message should be blocked from processing")
class ResponseQualityResult(BaseModel):
"""Result from response quality validation"""
is_relevant: bool = Field(description="True if response is relevant to user query")
stays_in_role: bool = Field(description="True if response stays in car shopping assistant role")
is_helpful: bool = Field(description="True if response is helpful and actionable")
issues_found: list[str] = Field(default_factory=list, description="List of quality issues")
severity: Optional[Literal["minor", "major", "critical"]] = None
explanation: str = Field(description="Brief explanation of quality assessment")
should_regenerate: bool = Field(description="True if response should be regenerated")
class SearchParameters(BaseModel):
"""Parameters for searching cars using template-based SQL"""
min_price: Optional[int] = Field(None, description="Minimum price in USD", ge=0, le=100000)
max_price: Optional[int] = Field(None, description="Maximum price in USD", ge=0, le=100000)
fuel_type: Optional[Literal["Gasoline", "Diesel", "Electric", "Hybrid", "Plug-in Hybrid"]] = Field(None, description="Type of fuel")
is_suv: Optional[bool] = Field(None, description="True for SUVs, False for sedans/coupes")
min_seating: Optional[int] = Field(None, description="Minimum seating capacity", ge=4, le=8)
max_seating: Optional[int] = Field(None, description="Maximum seating capacity", ge=4, le=8)
drivetrain: Optional[Literal["FWD", "RWD", "AWD", "4WD"]] = Field(None, description="Drive system")
min_fuel_efficiency_city: Optional[float] = Field(None, description="Minimum city MPG", ge=0)
min_cargo_space: Optional[int] = Field(None, description="Minimum cargo space in cubic feet", ge=0)
has_sunroof: Optional[bool] = Field(None, description="Must have sunroof")
has_leather_seats: Optional[bool] = Field(None, description="Must have leather seats")
has_navigation: Optional[bool] = Field(None, description="Must have navigation system")
has_backup_camera: Optional[bool] = Field(None, description="Must have backup camera")
min_safety_rating: Optional[float] = Field(None, description="Minimum safety rating", ge=0, le=5)
# ============================================================================
# VALIDATION TOOLS
# ============================================================================
@tool
def moderate_user_input(user_message: str) -> dict:
"""
Validates user input for harmful speech, spam, or off-topic content.
Checks for:
- Hate speech, harassment, discrimination
- Profanity or vulgar language
- Spam or scam attempts
- Off-topic requests (not about cars)
- Incoherent or nonsensical input
Returns moderation result with safety verdict.
"""
system_prompt = """Moderate user input for a car chatbot. Flag only serious issues.
Block if: hate speech, harassment, spam, sexual content, threats, illegal requests, complete gibberish.
Allow: greetings, car questions, normal conversation.
Return is_safe (true/false) and should_block (true only if harmful)."""
try:
response = llm_validator.with_structured_output(ContentModerationResult).invoke([
SystemMessage(content=system_prompt),
HumanMessage(content=f"Moderate this user message:\n\n{user_message}")
])
return {
"is_safe": response.is_safe,
"violation_type": response.violation_type,
"severity": response.severity,
"explanation": response.explanation,
"should_block": response.should_block
}
except Exception as e:
# On error, default to safe (don't block)
return {
"is_safe": True,
"violation_type": None,
"severity": None,
"explanation": f"Moderation check failed: {str(e)}",
"should_block": False
}
@tool
def validate_assistant_response(user_message: str, assistant_response: str) -> dict:
"""
Validates that the assistant's response is relevant, helpful, and stays in role.
Checks:
- Relevance to user's question
- Stays in car shopping assistant role
- Provides actionable information
- No hallucinations or false claims
- Professional and helpful tone
Returns quality assessment with regeneration recommendation.
"""
system_prompt = """Check if assistant response is relevant, helpful, and stays in car assistant role.
Regenerate only if seriously flawed (off-topic, unhelpful, incoherent).
Minor issues are OK."""
try:
response = llm_validator.with_structured_output(ResponseQualityResult).invoke([
SystemMessage(content=system_prompt),
HumanMessage(content=f"""Validate this conversation:
USER: {user_message}
ASSISTANT: {assistant_response}
Assess the assistant's response quality.""")
])
return {
"is_relevant": response.is_relevant,
"stays_in_role": response.stays_in_role,
"is_helpful": response.is_helpful,
"issues_found": response.issues_found,
"severity": response.severity,
"explanation": response.explanation,
"should_regenerate": response.should_regenerate
}
except Exception as e:
# On error, assume response is acceptable
return {
"is_relevant": True,
"stays_in_role": True,
"is_helpful": True,
"issues_found": [],
"severity": None,
"explanation": f"Validation check failed: {str(e)}",
"should_regenerate": False
}
# ============================================================================
# SQL TEMPLATE-BASED TOOLS (Same as before)
# ============================================================================
@tool
def search_cars(
min_price: Optional[int] = None,
max_price: Optional[int] = None,
fuel_type: Optional[str] = None,
is_suv: Optional[bool] = None,
min_seating: Optional[int] = None,
max_seating: Optional[int] = None,
drivetrain: Optional[str] = None,
min_fuel_efficiency_city: Optional[float] = None,
min_cargo_space: Optional[int] = None,
has_sunroof: Optional[bool] = None,
has_leather_seats: Optional[bool] = None,
has_navigation: Optional[bool] = None,
has_backup_camera: Optional[bool] = None,
min_safety_rating: Optional[float] = None,
) -> dict:
"""Search for cars using a secure template-based SQL query."""
conditions = []
params = []
if min_price is not None:
conditions.append("price >= ?")
params.append(min_price)
if max_price is not None:
conditions.append("price <= ?")
params.append(max_price)
if fuel_type is not None:
conditions.append("fuel_type = ?")
params.append(fuel_type)
if is_suv is not None:
conditions.append("is_suv = ?")
params.append(1 if is_suv else 0)
if min_seating is not None:
conditions.append("seating_capacity >= ?")
params.append(min_seating)
if max_seating is not None:
conditions.append("seating_capacity <= ?")
params.append(max_seating)
if drivetrain is not None:
conditions.append("drivetrain = ?")
params.append(drivetrain)
if min_fuel_efficiency_city is not None:
conditions.append("fuel_efficiency_city >= ?")
params.append(min_fuel_efficiency_city)
if min_cargo_space is not None:
conditions.append("cargo_space >= ?")
params.append(min_cargo_space)
if has_sunroof is not None:
conditions.append("has_sunroof = ?")
params.append(1 if has_sunroof else 0)
if has_leather_seats is not None:
conditions.append("has_leather_seats = ?")
params.append(1 if has_leather_seats else 0)
if has_navigation is not None:
conditions.append("has_navigation = ?")
params.append(1 if has_navigation else 0)
if has_backup_camera is not None:
conditions.append("has_backup_camera = ?")
params.append(1 if has_backup_camera else 0)
if min_safety_rating is not None:
conditions.append("safety_rating >= ?")
params.append(min_safety_rating)
where_clause = " AND ".join(conditions) if conditions else "1=1"
query = f"SELECT * FROM cars WHERE {where_clause} ORDER BY price LIMIT 21"
try:
with sqlite3.connect(DB_PATH) as conn:
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute(query, params)
results = cursor.fetchall()
cars = [dict(row) for row in results]
count = len(cars)
if count < MIN_RESULTS:
status = "too_few"
elif count > MAX_RESULTS:
status = "too_many"
cars = cars[:MAX_RESULTS]
else:
status = "good"
return {
"count": count,
"cars": cars,
"status": status,
"params_used": {k: v for k, v in [
("min_price", min_price), ("max_price", max_price),
("fuel_type", fuel_type), ("is_suv", is_suv),
("min_seating", min_seating), ("max_seating", max_seating),
("drivetrain", drivetrain), ("min_fuel_efficiency_city", min_fuel_efficiency_city),
("min_cargo_space", min_cargo_space), ("has_sunroof", has_sunroof),
("has_leather_seats", has_leather_seats), ("has_navigation", has_navigation),
("has_backup_camera", has_backup_camera), ("min_safety_rating", min_safety_rating),
] if v is not None}
}
except sqlite3.Error as e:
return {"count": 0, "cars": [], "status": "error", "error": str(e)}
@tool
def count_cars_only(
min_price: Optional[int] = None,
max_price: Optional[int] = None,
fuel_type: Optional[str] = None,
is_suv: Optional[bool] = None,
min_seating: Optional[int] = None,
max_seating: Optional[int] = None,
drivetrain: Optional[str] = None,
min_fuel_efficiency_city: Optional[float] = None,
min_cargo_space: Optional[int] = None,
has_sunroof: Optional[bool] = None,
has_leather_seats: Optional[bool] = None,
has_navigation: Optional[bool] = None,
has_backup_camera: Optional[bool] = None,
min_safety_rating: Optional[float] = None,
) -> dict:
"""Count how many cars match the criteria without returning full results."""
conditions = []
params = []
if min_price is not None:
conditions.append("price >= ?")
params.append(min_price)
if max_price is not None:
conditions.append("price <= ?")
params.append(max_price)
if fuel_type is not None:
conditions.append("fuel_type = ?")
params.append(fuel_type)
if is_suv is not None:
conditions.append("is_suv = ?")
params.append(1 if is_suv else 0)
if min_seating is not None:
conditions.append("seating_capacity >= ?")
params.append(min_seating)
if max_seating is not None:
conditions.append("seating_capacity <= ?")
params.append(max_seating)
if drivetrain is not None:
conditions.append("drivetrain = ?")
params.append(drivetrain)
if min_fuel_efficiency_city is not None:
conditions.append("fuel_efficiency_city >= ?")
params.append(min_fuel_efficiency_city)
if min_cargo_space is not None:
conditions.append("cargo_space >= ?")
params.append(min_cargo_space)
if has_sunroof is not None:
conditions.append("has_sunroof = ?")
params.append(1 if has_sunroof else 0)
if has_leather_seats is not None:
conditions.append("has_leather_seats = ?")
params.append(1 if has_leather_seats else 0)
if has_navigation is not None:
conditions.append("has_navigation = ?")
params.append(1 if has_navigation else 0)
if has_backup_camera is not None:
conditions.append("has_backup_camera = ?")
params.append(1 if has_backup_camera else 0)
if min_safety_rating is not None:
conditions.append("safety_rating >= ?")
params.append(min_safety_rating)
where_clause = " AND ".join(conditions) if conditions else "1=1"
query = f"SELECT COUNT(*) as count FROM cars WHERE {where_clause}"
try:
with sqlite3.connect(DB_PATH) as conn:
cursor = conn.cursor()
cursor.execute(query, params)
count = cursor.fetchone()[0]
return {
"count": count,
"status": "too_few" if count < MIN_RESULTS else "too_many" if count > MAX_RESULTS else "good"
}
except sqlite3.Error as e:
return {"count": 0, "status": "error", "error": str(e)}
# ============================================================================
# LANGGRAPH STATE DEFINITION
# ============================================================================
class ConversationState(TypedDict):
"""State for the conversation graph with validation tracking"""
messages: Annotated[list, add]
search_params: Optional[dict]
search_results: Optional[dict]
iteration_count: int
user_satisfied: bool
requires_search: bool
moderation_result: Optional[dict] # NEW: Track moderation results
quality_result: Optional[dict] # NEW: Track quality validation results
regenerate_count: int # NEW: Track regeneration attempts
# ============================================================================
# LANGGRAPH NODES WITH VALIDATION
# ============================================================================
def moderate_input(state: ConversationState) -> ConversationState:
"""
Node: Validate user input for harmful or inappropriate content.
This runs BEFORE processing the user's message.
"""
messages = state["messages"]
# Get last user message
user_messages = [m for m in messages if isinstance(m, HumanMessage)]
if not user_messages:
return {"moderation_result": {"is_safe": True, "should_block": False}}
last_user_msg = user_messages[-1].content
# Call moderation tool
moderation_result = moderate_user_input.invoke({"user_message": last_user_msg})
return {
"moderation_result": moderation_result
}
def gather_requirements(state: ConversationState) -> ConversationState:
"""
Node: Gather requirements from user and determine search parameters.
Only runs if moderation passed.
"""
messages = state["messages"]
system_prompt = f"""You are a car shopping advisor. Be concise and helpful.
{SCHEMA_DESCRIPTION}
CRITICAL - You must provide text explanation WITH your tool call:
When responding, ALWAYS include BOTH:
1. **Text content explaining your recommendation:**
Example: "For driving in Paris, I'd recommend a compact sedan or hybrid with good fuel efficiency (25+ MPG) since city parking is tight and gas is expensive. Let me search our inventory..."
2. **Tool call to search:**
Call search_cars with appropriate parameters based on user needs:
Common mappings:
- City driving/parking → max_price=35000, min_fuel_efficiency_city=25 (compact, efficient)
- Family/kids → is_suv=True, min_seating=5, min_safety_rating=4.0
- Fuel efficiency → min_fuel_efficiency_city=25 or fuel_type="Hybrid"/"Electric"
- Budget conscious → max_price=30000
- Luxury → has_leather_seats=True, has_navigation=True, min_price=35000
- Outdoor/adventure → is_suv=True, drivetrain="AWD" or "4WD"
- Long commute → min_fuel_efficiency_city=28, max_price=35000
IMPORTANT: Your response must have BOTH text content (recommendation) AND a tool_call (search). Never tool call without explanation text."""
llm_with_tools = llm.bind_tools([search_cars, count_cars_only])
response = llm_with_tools.invoke([SystemMessage(content=system_prompt)] + messages)
new_state = {
"messages": [response],
"requires_search": bool(response.tool_calls),
"iteration_count": state.get("iteration_count", 0)
}
if "satisfied" in response.content.lower() or "perfect" in response.content.lower():
new_state["user_satisfied"] = True
return new_state
def execute_search(state: ConversationState) -> ConversationState:
"""Node: Execute the search using tool calls from the LLM."""
messages = state["messages"]
last_message = messages[-1]
if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
tool_node = ToolNode([search_cars, count_cars_only])
result = tool_node.invoke(state)
tool_messages = result["messages"]
if tool_messages:
tool_response = tool_messages[-1]
if hasattr(tool_response, 'content'):
import json
try:
search_results = json.loads(tool_response.content)
except:
search_results = {"error": "Failed to parse tool response"}
else:
search_results = {}
else:
search_results = {}
return {
"messages": tool_messages,
"search_results": search_results,
"iteration_count": state.get("iteration_count", 0) + 1
}
return {"iteration_count": state.get("iteration_count", 0)}
def present_results(state: ConversationState) -> ConversationState:
"""Node: Present search results to user and provide guidance."""
search_results = state.get("search_results", {})
messages = state["messages"]
system_prompt = """Present search results very briefly. DO NOT re-explain recommendations.
STATUS RESPONSES:
**good (1-20 cars):**
"Found X vehicles matching these criteria. See the full list below."
**too_few (<1):**
"No matches. Try: broaden budget, relax features, or consider more fuel types."
**too_many (>20):**
"Over 20 matches. Let's narrow down - what's your priority: budget, efficiency, space, or safety?"
**error:**
"Search error. Please rephrase."
Keep it very short - 1 sentence. Don't describe the cars, just confirm results are ready."""
context = f"\nSearch Results: {search_results}"
response = llm.invoke([
SystemMessage(content=system_prompt),
*messages,
HumanMessage(content=context)
])
return {"messages": [response]}
def validate_response_quality(state: ConversationState) -> ConversationState:
"""
Node: Validate the quality of the assistant's response.
Checks relevance, role adherence, and helpfulness.
"""
messages = state.get("messages", [])
# Get last user and assistant messages
user_messages = [m for m in messages if isinstance(m, HumanMessage)]
ai_messages = [m for m in messages if isinstance(m, AIMessage)]
if not user_messages or not ai_messages:
return {"quality_result": {"should_regenerate": False, "is_relevant": True, "stays_in_role": True, "is_helpful": True}}
last_user_msg = user_messages[-1].content if hasattr(user_messages[-1], 'content') else str(user_messages[-1])
last_ai_msg = ai_messages[-1].content if hasattr(ai_messages[-1], 'content') else str(ai_messages[-1])
# Call quality validation tool
try:
quality_result = validate_assistant_response.invoke({
"user_message": last_user_msg,
"assistant_response": last_ai_msg
})
except Exception as e:
# If validation fails, assume response is acceptable
quality_result = {
"should_regenerate": False,
"is_relevant": True,
"stays_in_role": True,
"is_helpful": True,
"explanation": f"Validation error: {str(e)}"
}
return {
"quality_result": quality_result
}
# ============================================================================
# CONDITIONAL EDGE LOGIC
# ============================================================================
def should_moderate(state: ConversationState) -> Literal["moderate_input", "gather_requirements"]:
"""Decide if we need to moderate user input"""
messages = state["messages"]
# Check if last message is from user
if messages and isinstance(messages[-1], HumanMessage):
return "moderate_input"
return "gather_requirements"
def check_moderation(state: ConversationState) -> Literal["gather_requirements", "end"]:
"""Check if moderation passed or blocked content"""
moderation = state.get("moderation_result", {})
if moderation.get("should_block", False):
return "end" # Block harmful content
return "gather_requirements"
def should_continue(state: ConversationState) -> Literal["execute_search", "present_results", "validate_response", "end"]:
"""Determine next step in the graph"""
if state.get("user_satisfied", False):
return "end"
if state.get("requires_search", False) and not state.get("search_results"):
return "execute_search"
if state.get("search_results"):
return "present_results"
# If we have an AI response, validate it
messages = state.get("messages", [])
if messages and isinstance(messages[-1], AIMessage):
return "validate_response"
return "end"
def check_quality(state: ConversationState) -> Literal["gather_requirements", "end"]:
"""Check if response needs regeneration"""
quality = state.get("quality_result", {})
regenerate_count = state.get("regenerate_count", 0)
# Only regenerate once to avoid loops
if quality.get("should_regenerate", False) and regenerate_count < 1:
return "gather_requirements"
return "end"
# ============================================================================
# BUILD LANGGRAPH WITH VALIDATION
# ============================================================================
def build_graph() -> StateGraph:
"""Build the LangGraph workflow with validation nodes"""
workflow = StateGraph(ConversationState)
# Add validation nodes
workflow.add_node("moderate_input", moderate_input)
workflow.add_node("validate_response", validate_response_quality)
# Add core nodes
workflow.add_node("gather_requirements", gather_requirements)
workflow.add_node("execute_search", execute_search)
workflow.add_node("present_results", present_results)
# Set entry point - start with moderation
workflow.set_entry_point("moderate_input")
# Flow: moderate -> gather -> search -> present -> validate
workflow.add_conditional_edges(
"moderate_input",
check_moderation,
{
"gather_requirements": "gather_requirements",
"end": END
}
)
workflow.add_conditional_edges(
"gather_requirements",
should_continue,
{
"execute_search": "execute_search",
"present_results": "present_results",
"validate_response": "validate_response",
"end": END
}
)
workflow.add_edge("execute_search", "present_results")
workflow.add_edge("present_results", "validate_response")
workflow.add_conditional_edges(
"validate_response",
check_quality,
{
"gather_requirements": "gather_requirements",
"end": END
}
)
return workflow.compile()
# ============================================================================
# HELPER FUNCTIONS
# ============================================================================
def format_car_display(car: dict) -> str:
"""Format a single car for display"""
return f"""
{car['brand']} {car['model_name']} ({car['year']})
Price: ${car['price']:,}
Type: {'SUV' if car['is_suv'] else 'Sedan/Coupe'}
Fuel: {car['fuel_type']}
Seats: {car['seating_capacity']} | Cargo: {car['cargo_space']} cu ft
Drivetrain: {car['drivetrain']} | Transmission: {car['transmission']}
Features: {'Sunroof, ' if car['has_sunroof'] else ''}{'Leather, ' if car['has_leather_seats'] else ''}{'Navigation, ' if car['has_navigation'] else ''}{'Backup Camera' if car['has_backup_camera'] else ''}
"""
# ============================================================================
# MAIN CHATBOT LOOP
# ============================================================================
def main():
"""Main chatbot loop using LangGraph with validation"""
print("=" * 60)
print("Car Finder Chatbot (LangGraph + Validation Agents)")
print("=" * 60)
print("Features: Content Moderation + Response Quality Validation")
print("Tell me what kind of car you're looking for!")
print("Type 'quit' to exit.\n")
app = build_graph()
conversation_state = {
"messages": [],
"search_params": None,
"search_results": None,
"iteration_count": 0,
"user_satisfied": False,
"requires_search": False,
"moderation_result": None,
"quality_result": None,
"regenerate_count": 0
}
while True:
user_input = input("You: ").strip()
if user_input.lower() in ['quit', 'exit', 'bye']:
print("\nThanks for using Car Finder! Goodbye!")
break
if not user_input:
continue
# Add user message to state
conversation_state["messages"].append(HumanMessage(content=user_input))
try:
result = app.invoke(conversation_state)
conversation_state = result
# Check if moderation blocked the message
moderation = result.get("moderation_result", {})
if moderation.get("should_block", False):
print(f"\n[Content Warning] Your message was flagged as potentially {moderation.get('violation_type', 'inappropriate')}.")
print(f"Reason: {moderation.get('explanation')}")
print("Please keep the conversation professional and car-related.\n")
# Remove the blocked message from history
conversation_state["messages"] = conversation_state["messages"][:-1]
continue
# Display assistant's responses - show all new AI messages
if result.get("messages"):
# Get all AI messages that weren't in the previous state
new_messages = []
for msg in result["messages"]:
if isinstance(msg, AIMessage):
new_messages.append(msg)
# Display all new AI messages with content
for msg in new_messages:
if hasattr(msg, 'content') and msg.content:
print(f"\nAssistant: {msg.content}\n")
# Show quality validation feedback (for debugging)
quality = result.get("quality_result")
if quality and not quality.get("is_relevant", True):
print(f"[Quality Warning] Response quality issues detected: {quality.get('explanation')}\n")
# Display search results if available
search_results = result.get("search_results")
if search_results and search_results.get("status") == "good":
cars = search_results.get("cars", [])
if cars:
print("=" * 60)
print("MATCHING CARS:")
print("=" * 60)
for car in cars:
print(format_car_display(car))
print("=" * 60)
print()
# Check if conversation should end
if result.get("user_satisfied", False):
print("\nThank you for using Car Finder! Have a great day!")
break
except Exception as e:
print(f"\nError: {str(e)}")
print("Let's try again. Please rephrase your request.\n")
conversation_state["messages"] = conversation_state["messages"][:-1]
if __name__ == "__main__":
main()