Introduction to Onboarding Assistant Agents
An onboarding assistant agent is an intelligent conversational system designed to guide new users through the initial setup, configuration, and familiarization process of a product, platform, or service. Built with LangChain, these agents leverage large language models (LLMs) to provide context-aware, step-by-step assistance that adapts to each user's unique needs and progress level. Unlike static documentation or rigid wizards, an LLM-powered onboarding agent can answer questions dynamically, anticipate common stumbling blocks, and maintain a coherent, personalized conversation flow from initial login through to full productivity.
What Is an Onboarding Assistant Agent?
At its core, an onboarding assistant agent is a specialized conversational AI that combines several LangChain components into a unified system. It typically includes:
- Language Model: An LLM (such as GPT-4, Claude, or an open-source model) that generates natural, helpful responses
- Tools: Functions the agent can call to fetch user-specific data, check completion status, or perform actions like sending emails or updating records
- Memory: Conversation history that allows the agent to remember what steps the user has completed and what context has been established
- Knowledge Base: Product documentation, FAQs, and onboarding guides stored in a vector store for retrieval-augmented generation (RAG)
- Planner or Router: Logic that determines which onboarding step comes next and how to handle deviations from the expected flow
The agent orchestrates these components to deliver a seamless onboarding experience. It doesn't merely regurgitate documentation—it interprets the user's current state, answers specific questions, and proactively suggests next actions based on established onboarding milestones.
Why an Onboarding Assistant Agent Matters
Effective onboarding dramatically impacts user retention and satisfaction. Studies consistently show that users who receive guided onboarding are far more likely to become long-term, engaged customers. An AI-powered onboarding agent delivers several critical advantages:
- 24/7 Availability: Unlike human support teams, the agent is always ready to assist, regardless of time zone or business hours
- Scalability: The agent can simultaneously onboard thousands of users without degradation in response quality
- Consistency: Every user receives the same high-quality guidance, eliminating variability across different support representatives
- Personalization: The agent tailors its approach based on user behavior, role, and preferences—something static documentation cannot do
- Reduced Support Load: By handling routine onboarding questions autonomously, the agent frees human support teams to tackle complex, high-value issues
Architecture Overview
Before diving into code, let's map out the architecture. A well-designed onboarding assistant agent built with LangChain follows this general structure:
┌─────────────────────────────────────────────────────┐
│ User Interface │
│ (Web chat, Slack, email, etc.) │
└──────────────────────┬──────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ LangChain Agent Executor │
│ ┌─────────────┐ ┌──────────┐ ┌────────────────┐ │
│ │ Planner │ │ Router │ │ Response Gen │ │
│ └─────────────┘ └──────────┘ └────────────────┘ │
│ │
│ ┌──────────────────────────────────────────────┐ │
│ │ Tool Library │ │
│ │ ┌──────────┐ ┌───────────┐ ┌─────────────┐ │ │
│ │ │ User │ │ Knowledge │ │ Action │ │ │
│ │ │ Profile │ │ Base RAG │ │ Dispatcher │ │ │
│ │ │ Tool │ │ Tool │ │ Tool │ │ │
│ │ └──────────┘ └───────────┘ └─────────────┘ │ │
│ └──────────────────────────────────────────────┘ │
│ │
│ ┌──────────────────────────────────────────────┐ │
│ │ Memory Systems │ │
│ │ ┌──────────────┐ ┌──────────────────────┐ │ │
│ │ │ Conversation │ │ Onboarding State │ │ │
│ │ │ Memory │ │ Tracker │ │ │
│ │ └──────────────┘ └──────────────────────┘ │ │
│ └──────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────┘
Step-by-Step Implementation
Step 1: Setting Up the Environment
Begin by installing the required dependencies. You'll need LangChain, an LLM provider, a vector store for the knowledge base, and supporting libraries:
pip install langchain langchain-openai langchain-community chromadb tiktoken python-dotenvCreate a .env file to store your API keys securely:
OPENAI_API_KEY=your-openai-api-key-here
# Optional: for tracing and observability
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=your-langsmith-key-hereLoad environment variables and initialize the LLM:
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
# Initialize the primary LLM
llm = ChatOpenAI(
model="gpt-4-turbo-preview",
temperature=0.7,
max_tokens=1024
)
# A faster, cheaper model for simpler tasks like classification
fast_llm = ChatOpenAI(
model="gpt-3.5-turbo",
temperature=0.0,
max_tokens=256
)Step 2: Building the Knowledge Base with RAG
The agent needs access to your product's onboarding documentation, FAQs, and guides. We'll build a retrieval-augmented generation pipeline using ChromaDB as the vector store:
from langchain_community.document_loaders import TextLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
# Load onboarding documents
loader = DirectoryLoader(
"./onboarding_docs/",
glob="**/*.md",
loader_cls=TextLoader,
show_progress=True
)
documents = loader.load()
# Split documents into manageable chunks
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=100,
separators=["\n\n", "\n", ". ", " ", ""]
)
chunks = text_splitter.split_documents(documents)
# Create vector store
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
collection_name="onboarding_knowledge",
persist_directory="./chroma_db"
)
# Create a retriever with relevance threshold
retriever = vector_store.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={
"k": 5,
"score_threshold": 0.5
}
)Next, wrap the retriever in a tool that the agent can invoke:
from langchain.tools import tool
from langchain_core.documents import Document
from typing import List
@tool
def search_onboarding_knowledge(query: str) -> str:
"""Search the onboarding knowledge base for information about
features, setup steps, troubleshooting, and best practices.
Use this whenever a user asks a product-specific question."""
docs: List[Document] = retriever.invoke(query)
if not docs:
return "No relevant onboarding information found for this query."
formatted = []
for i, doc in enumerate(docs, 1):
source = doc.metadata.get("source", "Unknown")
formatted.append(f"Source {i} ({source}):\n{doc.page_content}")
return "\n\n---\n\n".join(formatted)Step 3: Creating User Profile and State Tracking Tools
An effective onboarding agent must know the user's current progress. Create tools that interface with your user database (here simulated with a dictionary, but in production you'd connect to your actual user store):
from typing import Dict, Any, Optional
from langchain.tools import tool
from datetime import datetime
# Simulated user database — replace with actual DB queries in production
user_database: Dict[str, Dict[str, Any]] = {
"user_123": {
"name": "Alice",
"email": "alice@example.com",
"role": "developer",
"onboarding_steps_completed": ["account_created", "email_verified"],
"onboarding_steps_remaining": [
"profile_setup",
"first_project_created",
"team_invitation",
"advanced_features_intro"
],
"days_since_signup": 3,
"last_action": "viewed_dashboard"
}
}
@tool
def get_user_onboarding_state(user_id: str) -> Dict[str, Any]:
"""Retrieve the current onboarding state for a user by their ID.
Returns completed steps, remaining steps, and user metadata."""
if user_id not in user_database:
return {"error": f"User {user_id} not found"}
user = user_database[user_id]
return {
"name": user["name"],
"role": user["role"],
"completed_steps": user["onboarding_steps_completed"],
"remaining_steps": user["onboarding_steps_remaining"],
"days_since_signup": user["days_since_signup"],
"last_action": user["last_action"]
}
@tool
def mark_onboarding_step_complete(user_id: str, step_name: str) -> str:
"""Mark an onboarding step as completed for a user.
Validates that the step exists in the user's remaining steps."""
if user_id not in user_database:
return f"Error: User {user_id} not found"
user = user_database[user_id]
if step_name in user["onboarding_steps_completed"]:
return f"Step '{step_name}' is already marked as completed for {user['name']}"
if step_name not in user["onboarding_steps_remaining"]:
return f"Error: '{step_name}' is not in {user['name']}'s remaining onboarding steps. Available steps: {user['onboarding_steps_remaining']}"
user["onboarding_steps_remaining"].remove(step_name)
user["onboarding_steps_completed"].append(step_name)
user["last_action"] = f"completed_{step_name}"
return f"Successfully marked '{step_name}' as completed for {user['name']}. {len(user['onboarding_steps_remaining'])} steps remaining."
@tool
def get_next_onboarding_step(user_id: str) -> str:
"""Determine the next recommended onboarding step for a user
based on their role and current progress."""
if user_id not in user_database:
return f"User {user_id} not found"
user = user_database[user_id]
if not user["onboarding_steps_remaining"]:
return f"All onboarding steps completed for {user['name']}! Consider suggesting advanced features."
next_step = user["onboarding_steps_remaining"][0]
# Role-specific guidance
guidance_map = {
"profile_setup": "Let's set up your profile. Upload a photo, add your bio, and configure your notification preferences.",
"first_project_created": "Create your first project. Choose a template that matches your needs and follow the guided setup.",
"team_invitation": "Invite your team members by sending email invitations or sharing an invite link from the Team settings page.",
"advanced_features_intro": "Explore advanced features like automations, integrations, and custom workflows tailored for your role."
}
guidance = guidance_map.get(next_step, f"Complete the '{next_step}' step.")
return f"Next step for {user['name']} ({user['role']}): {next_step}\n\nRecommended guidance: {guidance}"Step 4: Designing the System Prompt
The system prompt is the backbone of your agent's behavior. It establishes the agent's persona, boundaries, and operational rules:
ONBOARDING_SYSTEM_PROMPT = """You are an expert onboarding assistant for AcmeCloud, a cloud collaboration platform.
Your name is "Aria" and your purpose is to guide new users through their onboarding journey.
## Your Core Responsibilities:
1. **Progress Tracking:** Use the provided tools to check where users are in their onboarding journey
2. **Step Guidance:** Explain each onboarding step clearly with actionable instructions
3. **Question Answering:** Use the knowledge base tool to answer product-specific questions
4. **Motivation:** Celebrate completed steps and encourage continued progress
5. **Proactive Assistance:** Suggest the next logical step based on the user's role and progress
## Your Personality and Tone:
- Be warm, encouraging, and patient
- Use conversational language—avoid corporate jargon
- Celebrate milestones with genuine enthusiasm
- When users are stuck, empathize and offer alternative approaches
- Keep responses concise but thorough; aim for 2-4 paragraphs unless detailed instructions are needed
## Important Rules:
- ALWAYS check the user's onboarding state before giving guidance
- NEVER make up features or steps—rely on the knowledge base tool for factual information
- If a user asks about something outside onboarding scope, politely redirect to the appropriate support channel
- When a step is clearly completed based on user statements, use the mark_complete tool
- If the user seems confused, ask clarifying questions before proceeding
- Do not overwhelm users with too many steps at once—focus on one actionable item per response
## User Context:
When you have access to user data, reference it naturally in conversation. For example:
- "Since you're a developer, let me guide you through..."
- "I see you completed email verification—great job! Now let's tackle..."
- "You signed up 3 days ago, so let's make sure we're on track"
## Escalation Protocol:
If a user encounters a technical issue you cannot resolve, suggest they contact human support and provide a clear description of what to report.
"""Step 5: Assembling the Agent with LangChain's Agent Framework
Now we combine all components into a functional agent using LangChain's agent abstractions. We'll use the create_tool_calling_agent and AgentExecutor pattern, which provides robust tool handling and error recovery:
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationSummaryBufferMemory
from langchain_core.messages import SystemMessage
# Collect all tools
tools = [
search_onboarding_knowledge,
get_user_onboarding_state,
mark_onboarding_step_complete,
get_next_onboarding_step
]
# Build the prompt template
prompt = ChatPromptTemplate.from_messages([
("system", ONBOARDING_SYSTEM_PROMPT),
MessagesPlaceholder(variable_name="chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
# Set up conversation memory with summarization to handle long sessions
memory = ConversationSummaryBufferMemory(
llm=fast_llm,
max_token_limit=2000,
memory_key="chat_history",
return_messages=True
)
# Create the agent
agent = create_tool_calling_agent(llm=llm, tools=tools, prompt=prompt)
# Create the executor with error handling and retries
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
memory=memory,
verbose=True,
max_iterations=6,
max_execution_time=30,
early_stopping_method="generate",
handle_parsing_errors=True,
return_intermediate_steps=False
)Step 6: Building the Conversation Handler
Wrap the agent executor in a conversation handler that manages user sessions, initializes state, and formats responses for your frontend:
import uuid
from typing import Dict, Optional
class OnboardingSessionManager:
"""Manages onboarding conversations with persistent state."""
def __init__(self, agent_executor: AgentExecutor):
self.executor = agent_executor
self.sessions: Dict[str, Dict] = {}
def create_session(self, user_id: str) -> str:
"""Start a new onboarding session for a user."""
session_id = str(uuid.uuid4())
self.sessions[session_id] = {
"user_id": user_id,
"conversation_history": [],
"started_at": datetime.now().isoformat()
}
return session_id
def process_message(
self,
session_id: str,
user_message: str,
user_id: Optional[str] = None
) -> Dict[str, Any]:
"""Process a user message and return the agent's response."""
if session_id not in self.sessions:
raise ValueError(f"Session {session_id} not found")
session = self.sessions[session_id]
uid = user_id or session["user_id"]
# Prepend user ID context if not already in the message
enriched_input = user_message
if uid and "user_id" not in user_message.lower():
# Add user context subtly
enriched_input = f"[Current user_id: {uid}]\n\nUser message: {user_message}"
# Invoke the agent
try:
response = self.executor.invoke(
{"input": enriched_input}
)
except Exception as e:
return {
"error": str(e),
"message": "I encountered an issue processing your request. Could you rephrase or try again?"
}
output_text = response.get("output", "")
# Update session history
session["conversation_history"].append({
"user": user_message,
"assistant": output_text,
"timestamp": datetime.now().isoformat()
})
return {
"message": output_text,
"session_id": session_id,
"conversation_length": len(session["conversation_history"])
}
# Initialize the session manager
session_manager = OnboardingSessionManager(agent_executor)Step 7: Creating an Interactive Testing Loop
For development and demonstration, build a simple CLI interface to test the agent interactively:
def run_onboarding_demo():
"""Run an interactive onboarding session in the terminal."""
print("\n" + "="*60)
print(" Welcome to AcmeCloud Onboarding with Aria!")
print(" Type 'exit' at any time to end the session.")
print("="*60 + "\n")
user_id = input("Enter your user ID (or press Enter for 'user_123'): ").strip()
if not user_id:
user_id = "user_123"
session_id = session_manager.create_session(user_id)
# Start with a welcome message
initial_state = get_user_onboarding_state.invoke(user_id)
if isinstance(initial_state, dict) and "error" not in initial_state:
name = initial_state.get("name", "there")
print(f"\n🟢 Aria: Welcome, {name}! I'm Aria, your onboarding assistant.")
print(f"I see you've completed {len(initial_state.get('completed_steps', []))} steps so far.")
print("How can I help you today?\n")
while True:
user_input = input("\n👤 You: ").strip()
if user_input.lower() in ("exit", "quit", "bye"):
print("\n🟢 Aria: Thanks for chatting! You can return anytime by running this demo again. Goodbye! 👋\n")
break
if not user_input:
continue
print("\n🟢 Aria: ", end="", flush=True)
result = session_manager.process_message(session_id, user_input)
if "error" in result:
print(f"⚠️ {result['message']}")
else:
print(result["message"])
# Uncomment to run interactively:
# run_onboarding_demo()Step 8: Adding Observability and Monitoring
For production deployment, integrate LangSmith (or an equivalent observability platform) to trace agent decisions, tool calls, and response generation. This is essential for debugging and continuous improvement:
import os
from langchain.callbacks.tracers.langchain import LangChainTracer
# Configure LangSmith tracing
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "onboarding-agent"
# Create a traced executor
traced_executor = AgentExecutor(
agent=agent,
tools=tools,
memory=memory,
verbose=True,
max_iterations=6,
callbacks=[LangChainTracer(
project_name="onboarding-agent"
)]
)
# Log key metrics
def log_onboarding_metrics(session_id: str, step: str, duration_ms: float):
"""Log onboarding events for analytics."""
# In production, send to your metrics platform (Datadog, Prometheus, etc.)
print(f"[METRICS] session={session_id} step={step} duration_ms={duration_ms:.0f}")Step 9: Deploying as a REST API
To integrate the onboarding agent into a web application, wrap it in a FastAPI endpoint:
# This is a skeleton — expand for full production use
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn
app = FastAPI(title="Onboarding Assistant API")
class ChatRequest(BaseModel):
session_id: str
message: str
user_id: str
class ChatResponse(BaseModel):
message: str
session_id: str
conversation_length: int
@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
"""Process a chat message from an onboarding user."""
try:
result = session_manager.process_message(
session_id=request.session_id,
user_message=request.message,
user_id=request.user_id
)
if "error" in result:
raise HTTPException(status_code=500, detail=result["error"])
return ChatResponse(**result)
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
@app.post("/session/start")
async def start_session(user_id: str):
"""Initialize a new onboarding session."""
session_id = session_manager.create_session(user_id)
return {"session_id": session_id, "user_id": user_id}
# Run with: uvicorn app:app --host 0.0.0.0 --port 8000Advanced Enhancements
Adding Intent Classification
To handle diverse user inputs more gracefully, add an intent classifier that routes messages to specialized handlers:
from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import Literal
class OnboardingIntent(BaseModel):
"""Classified user intent for onboarding conversations."""
intent: Literal[
"ask_question",
"report_completion",
"seek_next_step",
"technical_issue",
"feedback",
"general_chat"
] = Field(description="The user's primary intent")
confidence: float = Field(description="Confidence score 0-1", ge=0, le=1)
detail: str = Field(description="Brief description of what the user wants")
intent_parser = PydanticOutputParser(pydantic_object=OnboardingIntent)
intent_prompt = ChatPromptTemplate.from_messages([
("system", """Classify the user's intent in the context of product onboarding.
Possible intents:
- ask_question: User wants information about features or how to do something
- report_completion: User indicates they've finished a step
- seek_next_step: User wants to know what to do next
- technical_issue: User encountered a bug or error
- feedback: User provides feedback about the product or onboarding
- general_chat: Greetings, small talk, or out-of-scope messages
{format_instructions}"""),
("human", "{message}")
])
# Use the fast LLM for classification
intent_chain = intent_prompt | fast_llm | intent_parser
def classify_intent(message: str) -> OnboardingIntent:
"""Classify user message intent before agent processing."""
return intent_chain.invoke({
"message": message,
"format_instructions": intent_parser.get_format_instructions()
})Implementing Conditional Routing
Based on intent, route to different agent configurations or handle certain intents with specialized logic:
def route_by_intent(intent: OnboardingIntent, user_id: str, message: str):
"""Route user messages based on classified intent."""
if intent.intent == "report_completion":
# Try to extract which step was completed
next_step_info = get_next_onboarding_step.invoke(user_id)
# Auto-mark common steps if user says they're done
return f"Great to hear you've made progress! {next_step_info}"
elif intent.intent == "technical_issue":
# Escalate or provide troubleshooting
return (
"I'm sorry you're experiencing a technical issue. "
"Let me search our knowledge base for solutions...\n\n"
f"{search_onboarding_knowledge.invoke(message)}\n\n"
"If the issue persists, please contact our support team at support@acmecloud.com "
"with a description of what happened."
)
elif intent.intent == "feedback":
# Log feedback and thank the user
print(f"[FEEDBACK] user={user_id}: {intent.detail}")
return (
"Thank you for your feedback! I've recorded it for our product team. "
"Is there anything else I can help with in your onboarding journey?"
)
else:
# Default: pass to the main agent executor
return None # Signal to use the standard agent flowBest Practices for Onboarding Assistant Agents
1. Design for Progressive Disclosure
Never dump all onboarding steps on a user at once. The agent should reveal information progressively, based on what the user has completed and what's immediately relevant. Structure your knowledge base documents to support this—write short, focused articles rather than monolithic guides. When the agent retrieves information, it should surface just enough for the current step.
2. Implement Robust Error Recovery
Onboarding agents interact with external systems (user databases, CRM, payment systems). These integrations can fail. Your agent must handle errors gracefully:
@tool
def safe_user_lookup(user_id: str) -> str:
"""Safely look up user data with comprehensive error handling."""
try:
state = get_user_onboarding_state.invoke(user_id)
if isinstance(state, dict) and "error" in state:
return f"Unable to retrieve your profile at the moment. Error: {state['error']}. Let's continue with what I know—what step are you currently on?"
return str(state)
except Exception as e:
return (
"I'm having trouble accessing your account data right now. "
"This is usually temporary. In the meantime, could you tell me "
"which onboarding step you're working on? I can guide you from there."
)3. Maintain State Across Sessions
Users may return after days or weeks. The agent should remember their progress without requiring them to repeat information. Use persistent memory (database-backed, not just in-memory) and always verify state freshness when a session resumes:
def resume_session(session_id: str, user_id: str):
"""Resume a previous session with state verification."""
session = session_manager.sessions.get(session_id)
if not session:
# Create new session but load historical context
new_session_id = session_manager.create_session(user_id)
# Fetch current user state to provide continuity
state = get_user_onboarding_state.invoke(user_id)
return {
"session_id": new_session_id,
"message": f"Welcome back! I see you're on step: {state}",
"is_resumed": True
}
# Verify stored state matches actual user state
stored_progress = session.get("last_known_progress", [])
current_state = get_user_onboarding_state.invoke(user_id)
if isinstance(current_state, dict):
actual_progress = current_state.get("completed_steps", [])
if set(stored_progress) != set(actual_progress):
# Sync the session state
session["last_known_progress"] = actual_progress
return {"session_id": session_id, "message": "Resuming your session...", "is_resumed": True}4. Test Extensively with Simulated Users
Before deployment, run your agent against a wide range of simulated user personas and scenarios. Create test harnesses that emulate different user types—fast adopters, confused beginners, skeptical evaluators, and power users:
# Example test scenarios
test_scenarios = [
{
"persona": "eager_beginner",
"user_id": "user_123",
"messages": [
"Hi! I just signed up and I'm excited to get started!",
"What should I do first?",
"Okay, I've uploaded my profile photo. What's next?",
],
"expected_intents": ["general_chat", "seek_next_step", "report_completion"]
},
{
"persona": "frustrated_user",
"user_id": "user_456",
"messages": [
"I can't figure out how to invite my team. This is confusing.",
"I've tried the invite button but it doesn't work.",
],
"expected_intents": ["ask_question", "technical_issue"]
},
{
"persona": "returning_user",
"user_id": "user_789",
"messages": [
"Hey, I'm back after a week. Where was I?",
"Yes, let's continue with the advanced features.",
],
"expected_intents": ["seek_next_step", "general_chat"]
}
]
def run_test_harness(scenarios: list):
"""Run automated tests against onboarding scenarios."""
results = []
for scenario in scenarios:
session_id = session_manager.create_session(scenario["user_id"])
for i, message in enumerate(scenario["messages"]):
intent = classify_intent(message)
result = session_manager.process_message(session_id, message, scenario["user_id"])
results.append({
"scenario": scenario["persona"],
"message_index": i,
"classified_intent": intent.intent,
"expected_intent": scenario["expected_intents"][i],
"match": intent.intent == scenario["expected_intents"][i],
"response": result.get("message", "")[:100]
})
# Calculate accuracy
matches = sum(1 for r in results if r["match"])
print(f"Intent classification accuracy: {matches}/{len(results)} ({matches/len(results)*100:.1f}%)")
return results5. Implement Guardrails and Safety Measures
Onboarding agents operate in a sensitive space—they interact with new users who are forming first impressions. Implement content moderation, prevent hallucinated features, and ensure the agent never makes promises about pricing or SLAs unless explicitly sourced from the knowledge base:
def validate_response(response: str, knowledge_base: Chroma) -> bool:
"""Check if the response contains claims not backed by the knowledge base."""
# Extract factual claims (simplified heuristic)
factual_markers = ["guarantee", "promise", "always", "never", "free", "price", "cost"]
has_claims = any(marker in response.lower() for marker in factual_markers)
if not has_claims:
return True
# Verify claims against knowledge base
verification_query = f"Verify factual accuracy: {response[:200]}"
docs = knowledge_base.similarity_search(verification_query, k=3)
# If no supporting documents found, flag for review
if not docs:
print(f"[GUARDRAIL] Potential unverified claims in response: {response[:150]}...")
return False
return True6. Continuously Improve Through Feedback Loops
Collect implicit and explicit feedback. Track when users disengage mid-onboarding, when they ask the same question multiple times (indicating poor initial guidance), and when they explicitly rate responses. Use this data to refine prompts, expand the knowledge base, and tune tool descriptions:
# Collect conversation analytics
conversation_metrics = {
"sessions_started": 0,
"sessions_completed": 0,
"average_steps_completed": 0.0,
"common_dropoff_step": None,
"user_satisfaction_ratings": []
}
def record_session_end(session_id: str, user_rating: Optional[int] = None):
"""Record end-of-session metrics for continuous improvement."""
session = session_manager.sessions.get(session_id)
if not session:
return
conversation_metrics["sessions_completed"] += 1
if user_rating is not None:
conversation_metrics["user_satisfaction_ratings"].append(user_rating)
# Analyze conversation for improvement signals
history = session.get("conversation_history", [])
# Detect repeated questions (signal of poor initial guidance)
user_questions = [msg["user"] for msg in history]
repeated = len(user_questions) - len(set(user_questions))
if repeated > 0:
print(f"[IMPROVEMENT] Session {session_id} had {repeated} repeated questions.")
print("Consider improving knowledge base coverage or initial explanations.")
# Clean up old sessions
del session_manager.sessions