Building Your First AI Agent

# Check if Python is installed (need 3.8+)
python --version

# Create virtual environment
python -m venv ai_agent_env

# Activate environment
# On Windows:
ai_agent_env\Scripts\activate
# On Mac/Linux:
source ai_agent_env/bin/activate

# Install required packages
pip install langchain openai chromadb tiktoken faiss-cpu streamlit python-dotenv

# agent.py
import os
from typing import List, Dict, Any
from dotenv import load_dotenv
from langchain.llms import OpenAI
from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory
from langchain.agents import Tool, AgentExecutor, LLMSingleActionAgent
from langchain.prompts import StringPromptTemplate
from langchain.schema import AgentAction, AgentFinish
import re

# Load environment variables
load_dotenv()

class CustomerSupportAgent:
    """AI Agent for customer support with memory and tool usage"""
    
    def __init__(self, use_ollama=False):
        """Initialise the agent with specified LLM"""
        self.use_ollama = use_ollama
        self.setup_llm()
        self.setup_memory()
        self.setup_tools()
        self.setup_knowledge_base()
        
    def setup_llm(self):
        """Configure the language model"""
        if self.use_ollama:
            from langchain.llms import Ollama
            self.llm = Ollama(model="llama2")
            self.embeddings = None  # Use default embeddings
        else:
            api_key = os.getenv("OPENAI_API_KEY")
            if not api_key:
                raise ValueError("OpenAI API key not found in .env file")
            
            self.llm = ChatOpenAI(
                temperature=0.7,
                model_name="gpt-3.5-turbo",
                openai_api_key=api_key
            )
            self.embeddings = OpenAIEmbeddings(openai_api_key=api_key)
    
    def setup_memory(self):
        """Initialise conversation memory"""
        self.memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True,
            output_key="answer"
        )
    
    def setup_tools(self):
        """Define tools the agent can use"""
        self.tools = [
            Tool(
                name="Search Knowledge Base",
                func=self.search_knowledge_base,
                description="Search internal documentation for answers"
            ),
            Tool(
                name="Escalate to Human",
                func=self.escalate_to_human,
                description="Escalate complex issues to human support"
            ),
            Tool(
                name="Check Order Status",
                func=self.check_order_status,
                description="Check the status of a customer order"
            ),
            Tool(
                name="Process Refund",
                func=self.process_refund,
                description="Initiate a refund process"
            )
        ]
    
    def setup_knowledge_base(self):
        """Load and index documentation"""
        # Create vector store for semantic search
        self.vectorstore = Chroma(
            persist_directory="./data/vectorstore",
            embedding_function=self.embeddings
        )
        
    def search_knowledge_base(self, query: str) -> str:
        """Search documentation for relevant information"""
        try:
            docs = self.vectorstore.similarity_search(query, k=3)
            if docs:
                return "\n".join([doc.page_content for doc in docs])
            return "No relevant information found in knowledge base."
        except:
            return "Knowledge base search is currently unavailable."
    
    def escalate_to_human(self, issue: str) -> str:
        """Escalate to human support"""
        # In production, this would create a ticket
        return f"I've escalated your issue to our human support team. Ticket #SUP-{hash(issue) % 10000:04d} created. They'll contact you within 24 hours."
    
    def check_order_status(self, order_id: str) -> str:
        """Mock order status check"""
        # In production, this would query your order database
        statuses = ["Processing", "Shipped", "Delivered", "In Transit"]
        import random
        status = random.choice(statuses)
        return f"Order {order_id} status: {status}"
    
    def process_refund(self, order_id: str) -> str:
        """Mock refund processing"""
        # In production, this would integrate with payment system
        return f"Refund initiated for order {order_id}. Processing time: 3-5 business days."
    
    def respond(self, user_input: str) -> str:
        """Generate response to user input"""
        # System prompt defining agent behaviour
        system_prompt = """You are a helpful customer support AI agent.
        Your goal is to assist customers with their queries professionally and efficiently.
        
        Guidelines:
        1. Be polite, empathetic, and professional
        2. Use available tools when appropriate
        3. Ask clarifying questions if needed
        4. Admit when you don't know something
        5. Never make up information
        
        Available tools:
        - Search Knowledge Base: For product/service information
        - Check Order Status: For order inquiries (need order ID)
        - Process Refund: For refund requests (need order ID)
        - Escalate to Human: For complex issues you can't resolve
        
        Current conversation:
        {chat_history}
        
        Human: {human_input}
        Assistant: Let me help you with that."""
        
        # Create prompt template
        from langchain.prompts import ChatPromptTemplate
        prompt = ChatPromptTemplate.from_template(system_prompt)
        
        # Create conversation chain
        chain = ConversationalRetrievalChain.from_llm(
            llm=self.llm,
            retriever=self.vectorstore.as_retriever() if self.vectorstore else None,
            memory=self.memory,
            verbose=True
        )
        
        # Get response
        response = chain({"question": user_input})
        return response["answer"]

# Example usage
if __name__ == "__main__":
    # Create agent
    agent = CustomerSupportAgent(use_ollama=False)
    
    # Test conversation
    print("Agent: Hello! I'm your AI support assistant. How can I help you today?")
    
    while True:
        user_input = input("\nYou: ")
        if user_input.lower() in ['exit', 'quit', 'bye']:
            print("Agent: Thank you for contacting us. Have a great day!")
            break
        
        response = agent.respond(user_input)
        print(f"\nAgent: {response}")

# app.py
import streamlit as st
from agent import CustomerSupportAgent
import time

# Page configuration
st.set_page_config(
    page_title="AI Support Agent",
    page_icon="🤖",
    layout="wide"
)

# Custom CSS
st.markdown("""

""", unsafe_allow_html=True)

# Initialise session state
if "messages" not in st.session_state:
    st.session_state.messages = []
    st.session_state.agent = CustomerSupportAgent(use_ollama=False)

# Header
st.title("🤖 AI Customer Support Agent")
st.markdown("---")

# Sidebar with information
with st.sidebar:
    st.header("About This Agent")
    st.info("""
    This AI agent can:
    - Answer product questions
    - Check order status
    - Process refunds
    - Escalate to human support
    
    Just type your question below!
    """)
    
    st.header("Example Queries")
    st.code("What's your return policy?")
    st.code("Check order #12345")
    st.code("I want a refund")
    st.code("Connect me to a human")
    
    # Settings
    st.header("Settings")
    use_ollama = st.checkbox("Use Local LLM (Ollama)", value=False)
    if use_ollama:
        st.warning("Make sure Ollama is running locally")

# Chat interface
chat_container = st.container()

# Display chat history
with chat_container:
    for message in st.session_state.messages:
        with st.chat_message(message["role"]):
            st.write(message["content"])

# Input field
user_input = st.chat_input("Type your message here...")

if user_input:
    # Add user message to history
    st.session_state.messages.append({"role": "user", "content": user_input})
    
    # Display user message
    with st.chat_message("user"):
        st.write(user_input)
    
    # Get agent response
    with st.chat_message("assistant"):
        message_placeholder = st.empty()
        
        # Show thinking animation
        with st.spinner("Thinking..."):
            response = st.session_state.agent.respond(user_input)
        
        # Display response with typing effect
        displayed_response = ""
        for char in response:
            displayed_response += char
            message_placeholder.markdown(displayed_response + "▌")
            time.sleep(0.01)
        
        message_placeholder.markdown(response)
    
    # Add agent response to history
    st.session_state.messages.append({"role": "assistant", "content": response})

# Footer
st.markdown("---")
st.caption("Built with LangChain and Streamlit | [View Tutorial](https://example.com)")

# Run with: streamlit run app.py

# load_knowledge.py
from langchain.document_loaders import TextLoader, PDFLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
import os
from dotenv import load_dotenv

load_dotenv()

def load_documents(directory="./data/docs"):
    """Load all documents from directory"""
    # Load different file types
    loaders = [
        DirectoryLoader(directory, glob="**/*.txt", loader_cls=TextLoader),
        DirectoryLoader(directory, glob="**/*.pdf", loader_cls=PDFLoader),
    ]
    
    documents = []
    for loader in loaders:
        try:
            documents.extend(loader.load())
        except:
            pass
    
    return documents

def create_vectorstore(documents):
    """Create searchable vector store from documents"""
    # Split documents into chunks
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200
    )
    chunks = text_splitter.split_documents(documents)
    
    # Create embeddings and vector store
    embeddings = OpenAIEmbeddings()
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory="./data/vectorstore"
    )
    
    vectorstore.persist()
    print(f"Created vector store with {len(chunks)} chunks")
    return vectorstore

# Example knowledge base content
sample_docs = """
RETURN POLICY
We offer a 30-day return policy for all products. Items must be unused and in original packaging.
To initiate a return, contact our support team with your order number.

SHIPPING INFORMATION
Standard shipping: 5-7 business days
Express shipping: 2-3 business days
International shipping: 10-15 business days

PRODUCT WARRANTY
All products come with a 1-year manufacturer warranty.
Extended warranty available for purchase within 30 days of delivery.

PAYMENT METHODS
We accept Visa, MasterCard, American Express, PayPal, and Apple Pay.
Payment is processed securely through our encrypted payment gateway.
"""

# Save sample documentation
os.makedirs("./data/docs", exist_ok=True)
with open("./data/docs/policies.txt", "w") as f:
    f.write(sample_docs)

# Load and index documents
if __name__ == "__main__":
    print("Loading documents...")
    docs = load_documents()
    
    print("Creating vector store...")
    vectorstore = create_vectorstore(docs)
    
    print("Knowledge base ready!")

# custom_tools.py
from langchain.tools import Tool
import requests
import json

def weather_tool(location: str) -> str:
    """Get current weather for a location"""
    # Using free weather API
    api_url = f"https://wttr.in/{location}?format=j1"
    try:
        response = requests.get(api_url)
        data = response.json()
        current = data['current_condition'][0]
        return f"Weather in {location}: {current['weatherDesc'][0]['value']}, {current['temp_C']}°C"
    except:
        return f"Could not fetch weather for {location}"

def calculate_shipping(weight: float, distance: float) -> str:
    """Calculate shipping cost"""
    base_rate = 5.00
    weight_rate = 0.50 * weight
    distance_rate = 0.10 * distance
    total = base_rate + weight_rate + distance_rate
    return f"Estimated shipping cost: ${total:.2f}"

def sentiment_analysis(text: str) -> str:
    """Analyse customer sentiment"""
    # Simple sentiment analysis (use TextBlob or similar in production)
    positive_words = ['happy', 'great', 'excellent', 'good', 'amazing']
    negative_words = ['bad', 'terrible', 'awful', 'poor', 'disappointed']
    
    text_lower = text.lower()
    positive_count = sum(1 for word in positive_words if word in text_lower)
    negative_count = sum(1 for word in negative_words if word in text_lower)
    
    if positive_count > negative_count:
        return "Positive sentiment detected - customer seems satisfied"
    elif negative_count > positive_count:
        return "Negative sentiment detected - customer may need extra attention"
    else:
        return "Neutral sentiment detected"

# Create tool objects
custom_tools = [
    Tool(
        name="Weather",
        func=weather_tool,
        description="Get current weather for any location"
    ),
    Tool(
        name="Shipping Calculator",
        func=calculate_shipping,
        description="Calculate shipping costs based on weight and distance"
    ),
    Tool(
        name="Sentiment Analysis",
        func=sentiment_analysis,
        description="Analyse customer sentiment from their message"
    )
]

# Add to agent
# In agent.py, update setup_tools():
def setup_tools(self):
    self.tools.extend(custom_tools)

# analytics.py
import pandas as pd
import plotly.express as px
import streamlit as st
from datetime import datetime, timedelta

class ConversationAnalytics:
    """Track and analyse agent performance"""
    
    def __init__(self):
        self.conversations = []
        self.metrics = {
            'total_conversations': 0,
            'avg_response_time': 0,
            'escalation_rate': 0,
            'satisfaction_score': 0,
            'common_topics': {},
            'tool_usage': {}
        }
    
    def log_conversation(self, user_input, agent_response, response_time, tools_used):
        """Log conversation for analysis"""
        self.conversations.append({
            'timestamp': datetime.now(),
            'user_input': user_input,
            'agent_response': agent_response,
            'response_time': response_time,
            'tools_used': tools_used,
            'word_count': len(agent_response.split())
        })
        self.update_metrics()
    
    def update_metrics(self):
        """Calculate performance metrics"""
        if not self.conversations:
            return
        
        df = pd.DataFrame(self.conversations)
        
        self.metrics['total_conversations'] = len(df)
        self.metrics['avg_response_time'] = df['response_time'].mean()
        
        # Calculate escalation rate
        escalations = df['agent_response'].str.contains('escalate|human|support team', case=False).sum()
        self.metrics['escalation_rate'] = (escalations / len(df)) * 100
        
        # Tool usage statistics
        all_tools = []
        for tools in df['tools_used']:
            if tools:
                all_tools.extend(tools)
        
        from collections import Counter
        self.metrics['tool_usage'] = dict(Counter(all_tools))
    
    def display_dashboard(self):
        """Display analytics in Streamlit"""
        st.header("📊 Agent Analytics Dashboard")
        
        col1, col2, col3, col4 = st.columns(4)
        
        with col1:
            st.metric("Total Conversations", self.metrics['total_conversations'])
        
        with col2:
            st.metric("Avg Response Time", f"{self.metrics['avg_response_time']:.2f}s")
        
        with col3:
            st.metric("Escalation Rate", f"{self.metrics['escalation_rate']:.1f}%")
        
        with col4:
            st.metric("Satisfaction Score", "4.5/5.0")  # Mock score
        
        # Tool usage chart
        if self.metrics['tool_usage']:
            fig = px.bar(
                x=list(self.metrics['tool_usage'].keys()),
                y=list(self.metrics['tool_usage'].values()),
                title="Tool Usage Frequency"
            )
            st.plotly_chart(fig)
        
        # Response time trend
        if len(self.conversations) > 1:
            df = pd.DataFrame(self.conversations)
            fig = px.line(df, x='timestamp', y='response_time', title="Response Time Trend")
            st.plotly_chart(fig)

# learning.py
import json
from datetime import datetime

class ContinuousLearning:
    """Enable agent to learn from interactions"""
    
    def __init__(self, feedback_file="feedback.json"):
        self.feedback_file = feedback_file
        self.load_feedback()
    
    def load_feedback(self):
        """Load existing feedback data"""
        try:
            with open(self.feedback_file, 'r') as f:
                self.feedback_data = json.load(f)
        except:
            self.feedback_data = {
                'successful_responses': [],
                'failed_responses': [],
                'corrections': {}
            }
    
    def record_feedback(self, query, response, success=True, correction=None):
        """Record user feedback on responses"""
        feedback_entry = {
            'timestamp': datetime.now().isoformat(),
            'query': query,
            'response': response,
            'success': success
        }
        
        if success:
            self.feedback_data['successful_responses'].append(feedback_entry)
        else:
            self.feedback_data['failed_responses'].append(feedback_entry)
            if correction:
                self.feedback_data['corrections'][query] = correction
        
        self.save_feedback()
    
    def save_feedback(self):
        """Save feedback to file"""
        with open(self.feedback_file, 'w') as f:
            json.dump(self.feedback_data, f, indent=2)
    
    def get_correction(self, query):
        """Check if we have a correction for this query"""
        # Simple similarity check - use embeddings in production
        for saved_query, correction in self.feedback_data['corrections'].items():
            if query.lower() in saved_query.lower() or saved_query.lower() in query.lower():
                return correction
        return None
    
    def generate_training_data(self):
        """Generate fine-tuning data from feedback"""
        training_data = []
        
        # Convert successful interactions to training examples
        for entry in self.feedback_data['successful_responses']:
            training_data.append({
                "prompt": entry['query'],
                "completion": entry['response']
            })
        
        # Add corrections as training examples
        for query, correction in self.feedback_data['corrections'].items():
            training_data.append({
                "prompt": query,
                "completion": correction
            })
        
        # Save as JSONL for fine-tuning
        with open('training_data.jsonl', 'w') as f:
            for item in training_data:
                f.write(json.dumps(item) + '\n')
        
        return training_data

# Integration with main agent
def enhance_response(self, user_input):
    """Check for corrections before generating response"""
    learner = ContinuousLearning()
    
    # Check if we have a correction for this query
    correction = learner.get_correction(user_input)
    if correction:
        return correction
    
    # Generate normal response
    response = self.respond(user_input)
    
    # Add feedback buttons in UI
    return response

# test_agent.py
import unittest
from agent import CustomerSupportAgent
import time

class TestAgent(unittest.TestCase):
    def setUp(self):
        """Set up test agent"""
        self.agent = CustomerSupportAgent(use_ollama=False)
    
    def test_basic_response(self):
        """Test basic query handling"""
        response = self.agent.respond("Hello")
        self.assertIsNotNone(response)
        self.assertIn("help", response.lower())
    
    def test_order_status(self):
        """Test order status checking"""
        response = self.agent.respond("Check order #12345")
        self.assertIn("order", response.lower())
        self.assertIn("12345", response)
    
    def test_knowledge_search(self):
        """Test knowledge base search"""
        response = self.agent.respond("What's your return policy?")
        self.assertIn("return", response.lower())
        self.assertIn("30", response)  # 30-day policy
    
    def test_escalation(self):
        """Test human escalation"""
        response = self.agent.respond("I need to speak to a human")
        self.assertIn("escalate", response.lower())
        self.assertIn("ticket", response.lower())
    
    def test_response_time(self):
        """Test response time is reasonable"""
        start = time.time()
        response = self.agent.respond("Simple question")
        elapsed = time.time() - start
        self.assertLess(elapsed, 5.0)  # Should respond within 5 seconds
    
    def test_memory(self):
        """Test conversation memory"""
        self.agent.respond("My name is John")
        response = self.agent.respond("What's my name?")
        self.assertIn("John", response)

if __name__ == "__main__":
    unittest.main()