Building a Customer Support Agent with AgentDbg: Complete Tutorial

# Create project directory
mkdir support_agent && cd support_agent

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install agentdbg[langchain]
pip install langchain-openai langchain-community

# Create project structure
mkdir -p agents tools knowledge_base logs
touch requirements.txt

agentdbg[langchain]>=0.1.0
langchain-openai>=0.0.5
langchain-community>=0.0.10
openai>=1.0.0
pydantic>=2.0.0
python-dotenv>=1.0.0

# OpenAI API Key
OPENAI_API_KEY=sk-your-key-here

# Agent Configuration
AGENTDBG_MAX_LLM_CALLS=50
AGENTDBG_MAX_TOOL_CALLS=100
AGENTDBG_MAX_DURATION_S=300  # 5 minutes max
AGENTDBG_STOP_ON_LOOP=true

{
  "pricing": {
    "plans": [
      {
        "name": "Starter",
        "price": "$29/month",
        "features": ["5 users", "Basic analytics", "Email support"]
      },
      {
        "name": "Professional", 
        "price": "$99/month",
        "features": ["20 users", "Advanced analytics", "Priority support"]
      },
      {
        "name": "Enterprise",
        "price": "Custom",
        "features": ["Unlimited users", "Custom analytics", "24/7 support"]
      }
    ],
    "refund_policy": "30-day money-back guarantee for all plans",
    "payment_methods": ["Credit Card", "PayPal", "Wire Transfer"]
  },
  "features": {
    "analytics": "Real-time dashboards, custom reports, data export",
    "collaboration": "Team workspaces, commenting, file sharing",
    "integrations": "Slack, GitHub, Google Workspace, 100+ more"
  },
  "account": {
    "trial": "14-day free trial, no credit card required",
    "cancellation": "Cancel anytime from account settings",
    "data_export": "Export all data in JSON, CSV, or PDF formats"
  }
}

from langchain.tools import StructuredTool
from pydantic import BaseModel, Field
import json
import os

class PasswordResetInput(BaseModel):
    email: str = Field(description="User email address")

def reset_password(email: str) -> str:
    """Reset user password and send reset link."""
    # In production: Call your API
    return f"Password reset link sent to {email}. Check your inbox."

class SubscriptionInput(BaseModel):
    email: str = Field(description="User email address")
    new_plan: str = Field(description="New plan name")

def change_subscription(email: str, new_plan: str) -> str:
    """Change user subscription plan."""
    available_plans = ["Starter", "Professional", "Enterprise"]
    if new_plan not in available_plans:
        return f"Invalid plan. Available: {', '.join(available_plans)}"
    
    # In production: Call your billing API
    return f"Subscription changed to {new_plan} for {email}. Check email for confirmation."

class EscalationInput(BaseModel):
    issue: str = Field(description="Detailed issue description")
    user_email: str = Field(description="User email address")
    urgency: str = Field(description="Urgency level: low, medium, high")

def escalate_to_human(issue: str, user_email: str, urgency: str) -> str:
    """Escalate complex issue to human support agent."""
    # In production: Create support ticket in your system
    ticket_id = f"TKT-{hash(issue) % 10000:04d}"
    
    return f"Ticket {ticket_id} created. Our team will respond within {'1 hour' if urgency == 'high' else '4 hours'}."

def search_faq(query: str) -> str:
    """Search FAQ knowledge base."""
    # Load FAQ
    with open('knowledge_base/faq.json', 'r') as f:
        faq = json.load(f)
    
    # Simple keyword search (in production: use vector search)
    query_lower = query.lower()
    results = []
    
    for category, items in faq.items():
        for key, value in (items.items() if isinstance(items, dict) else items):
            if query_lower in str(value).lower():
                results.append(f"{category}.{key}: {value}")
    
    if results:
        return "\n".join(results[:3])  # Top 3 results
    else:
        return "No exact match found in FAQ. Let me transfer you to a human agent."

# Create tools
password_reset_tool = StructuredTool.from_function(
    func=reset_password,
    name="reset_password",
    description="Reset user password and send reset link to email",
    args_schema=PasswordResetInput
)

subscription_tool = StructuredTool.from_function(
    func=change_subscription,
    name="change_subscription",
    description="Change user subscription plan",
    args_schema=SubscriptionInput
)

escalation_tool = StructuredTool.from_function(
    func=escalate_to_human,
    name="escalate_to_human",
    description="Escalate complex issues to human support agent",
    args_schema=EscalationInput
)

faq_search_tool = StructuredTool.from_function(
    func=search_faq,
    name="search_faq",
    description="Search FAQ knowledge base for answers",
    args_schema=None
)

TOOLS = [password_reset_tool, subscription_tool, escalation_tool, faq_search_tool]

from langchain_openai import ChatOpenAI
from langchain.agents import initialize_agent, AgentType
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferMemory
from langchain.schema import SystemMessage
from agentdbg import trace, record_state, record_error
from agentdbg.integrations import AgentDbgLangChainCallbackHandler
import os
from dotenv import load_dotenv

load_dotenv()

@trace(
    max_llm_calls=50,
    max_tool_calls=100,
    max_duration_s=300,
    stop_on_loop=True
)
def create_support_agent():
    """Initialize the customer support agent with all tools and configurations."""
    
    # Record initial state
    record_state({
        "agent_type": "customer_support",
        "tools_available": ["password_reset", "subscription_change", "escalation", "faq_search"],
        "environment": os.getenv("ENVIRONMENT", "development")
    })
    
    # Initialize LLM
    llm = ChatOpenAI(
        model="gpt-3.5-turbo",
        temperature=0.3,  # Lower temperature for consistent responses
        max_tokens=500
    )
    
    # System prompt
    system_prompt = """You are a helpful customer support agent for a SaaS company. 

Your role:
- Answer questions about pricing, features, and account management
- Handle simple tasks like password resets and subscription changes  
- Escalate complex issues to human agents
- Always maintain a friendly, professional tone

Guidelines:
- Be concise but helpful
- If you're unsure, escalate to human
- For account changes, always confirm user email
- Never make up information - use the FAQ search tool
- If user seems frustrated, proactively escalate

Available tools:
- search_faq: Search our knowledge base
- reset_password: Help users reset passwords
- change_subscription: Modify subscription plans
- escalate_to_human: Create support tickets for complex issues"""

    # Create memory for conversation context
    memory = ConversationBufferMemory(
        memory_key="chat_history",
        return_messages=True
    )
    
    # Initialize AgentDbg callback handler
    agentdbg_handler = AgentDbgLangChainCallbackHandler()
    
    # Create agent
    agent = initialize_agent(
        tools=TOOLS,
        llm=llm,
        agent=AgentType.OPENAI_FUNCTIONS,
        verbose=True,
        memory=memory,
        agent_kwargs={
            "system_message": SystemMessage(content=system_prompt)
        },
        callbacks=[agentdbg_handler],
        handle_parsing_errors=True,
        max_iterations=5,  # Prevent infinite loops
        early_stopping_method="generate"
    )
    
    record_state({
        "agent_status": "initialized",
        "max_iterations": 5
    })
    
    return agent

@trace
def handle_support_query(user_query: str, user_email: str = None) -> dict:
    """Process a customer support query."""
    
    # Record query details
    record_state({
        "user_query": user_query,
        "user_email": user_email,
        "query_length": len(user_query)
    })
    
    try:
        # Initialize agent
        agent = create_support_agent()
        
        # Add user context to query
        if user_email:
            full_query = f"User email: {user_email}\nQuery: {user_query}"
        else:
            full_query = user_query
        
        # Process query
        response = agent.run(full_query)
        
        # Record successful response
        record_state({
            "query_status": "resolved",
            "response_length": len(response)
        })
        
        return {
            "status": "success",
            "response": response,
            "escalated": "escalate" in response.lower()
        }
        
    except Exception as e:
        # Record error details
        record_error(
            error_type=type(e).__name__,
            message=str(e),
            context={
                "user_query": user_query,
                "user_email": user_email
            }
        )
        
        # Fallback response
        return {
            "status": "error",
            "response": "I apologize, but I'm having trouble processing your request. Let me connect you with a human agent who can help right away.",
            "error": str(e)
        }

if __name__ == "__main__":
    # Test queries
    test_queries = [
        ("What are your pricing plans?", None),
        ("I need to reset my password", "user@example.com"),
        ("How do I cancel my subscription?", None),
        ("I found a bug in your analytics", "user@example.com"),
        ("Change me to the Professional plan", "user@example.com")
    ]
    
    for query, email in test_queries:
        print(f"\n{'='*60}")
        print(f"Query: {query}")
        print(f"Email: {email or 'Not provided'}")
        print(f"{'='*60}")
        
        result = handle_support_query(query, email)
        print(f"Response: {result['response']}")
        print(f"Status: {result['status']}")

# Run the agent with test queries
python agents/support_agent.py

# View the debug timeline
agentdbg view

@trace
def cost_optimized_agent():
    # Use cheaper model for simple queries
    if is_simple_query(query):
        llm = ChatOpenAI(model="gpt-3.5-turbo", max_tokens=150)
    else:
        llm = ChatOpenAI(model="gpt-4", max_tokens=500)
    
    # Record cost tracking
    record_state({
        "model_used": llm.model_name,
        "estimated_cost": estimate_cost(llm, query)
    })

from functools import lru_cache

@lru_cache(maxsize=100)
def cached_faq_search(query: str) -> str:
    """Cache FAQ responses to reduce LLM calls."""
    return search_faq(query)

# Integration
record_tool_call(
    name="cached_faq_search",
    args={"query": query},
    result=cached_faq_search(query)
)

import time
from agentdbg import record_tool_call

def resilient_tool_call(tool_func, *args, max_retries=3, **kwargs):
    """Execute tool with retry logic."""
    for attempt in range(max_retries):
        try:
            result = tool_func(*args, **kwargs)
            record_tool_call(
                name=tool_func.__name__,
                args=args,
                result=result,
                meta={"retry_attempts": attempt}
            )
            return result
            
        except Exception as e:
            if attempt == max_retries - 1:
                # Final attempt failed
                record_error(
                    error_type=type(e).__name__,
                    message=f"All {max_retries} attempts failed",
                    context={"last_error": str(e)}
                )
                raise
            
            # Exponential backoff
            wait_time = 2 ** attempt
            time.sleep(wait_time)
            
            record_state({
                "retry_attempt": attempt + 1,
                "wait_time_seconds": wait_time
            })

@trace
def monitored_support_agent():
    start_time = time.time()
    
    # ... agent logic ...
    
    # Record performance metrics
    record_state({
        "response_time_ms": (time.time() - start_time) * 1000,
        "llm_calls": count_llm_calls(),
        "tool_calls": count_tool_calls(),
        "total_tokens": count_tokens()
    })

# 1. Run comprehensive tests
python -m pytest tests/

# 2. Check guardrails are working
export AGENTDBG_MAX_LLM_CALLS=10  # Test with low limits
python agents/support_agent.py

# 3. Verify redaction is working
export AGENTDBG_REDACT=1
python agents/support_agent.py
grep -r "sk-" ~/.agentdbg/runs/  # Should find nothing

# 4. Test error scenarios
python -c "from agents.support_agent import handle_support_query; handle_support_query('', 'test@evil.com')"

# 5. Load testing
for i in {1..100}; do python agents/support_agent.py; done

# Production .env
OPENAI_API_KEY=sk-prod-key-here
ENVIRONMENT=production
AGENTDBG_MAX_LLM_CALLS=1000
AGENTDBG_MAX_TOOL_CALLS=2000
AGENTDBG_MAX_DURATION_S=600
AGENTDBG_STOP_ON_LOOP=true
AGENTDBG_REDACT=1
AGENTDBG_REDACT_KEYS="api_key,password,token,secret,credit_card"

# Add to production agent
import logging

@trace
def production_agent():
    logging.basicConfig(
        filename='logs/agent.log',
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    
    # Record production metrics
    record_state({
        "deployment": "production",
        "version": "1.0.0",
        "logging_enabled": True
    })

# Find most common issues
agentdbg list --json | jq '.[] | .error_count' | sort | uniq -c

# Export interesting runs for analysis
agentdbg export <run_id> --out analysis/

# Compare performance over time
agentdbg view <old_run_id>
agentdbg view <new_run_id>

@trace
def billing_agent():
    """Specialized agent for billing queries."""
    # More conservative, detail-oriented
    llm = ChatOpenAI(temperature=0, model="gpt-4")
    # ... specific billing tools ...

@trace
def technical_agent():
    """Specialized agent for technical issues."""
    # More detailed, troubleshooting-focused
    llm = ChatOpenAI(temperature=0.2, model="gpt-4")
    # ... technical debugging tools ...

@trace
def support_router():
    """Route queries to appropriate specialist."""
    # Simple classification logic
    if "billing" in query.lower() or "subscription" in query.lower():
        return billing_agent()
    elif "bug" in query.lower() or "error" in query.lower():
        return technical_agent()
    else:
        return general_support_agent()