Policy and FAQ Chatbot

"How do I return this?" "When will my order arrive?" "Do you ship internationally?"

These questions make up 60-70% of customer support tickets. A well-built policy chatbot can handle them instantly, 24/7, freeing human agents for complex issues.

Structuring Policy Documents for RAG

from langchain.text_splitter import RecursiveCharacterTextSplitter
from typing import List, Dict
import hashlib

def chunk_policy_documents(policies: List[Dict]) -> List[Dict]:
    """
    Chunk policy documents with proper metadata for retrieval.

    Each policy dict should have:
    - name: "Return Policy", "Shipping Policy", etc.
    - content: The full policy text
    - effective_date: When this version became effective
    - category: "returns", "shipping", "payment", etc.
    """

    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=100,
        separators=["\n## ", "\n### ", "\n\n", "\n", ". ", " "]
    )

    all_chunks = []

    for policy in policies:
        chunks = splitter.split_text(policy["content"])

        for i, chunk in enumerate(chunks):
            # Create unique ID based on content
            chunk_id = hashlib.md5(f"{policy['name']}_{chunk[:100]}".encode()).hexdigest()

            all_chunks.append({
                "id": chunk_id,
                "text": chunk,
                "metadata": {
                    "policy_name": policy["name"],
                    "category": policy["category"],
                    "effective_date": policy["effective_date"],
                    "chunk_index": i,
                    "total_chunks": len(chunks)
                }
            })

    return all_chunks


# Example policy document
return_policy = {
    "name": "Return Policy",
    "category": "returns",
    "effective_date": "2024-01-01",
    "content": """
## Return Policy

### Standard Return Window
Most items can be returned within 30 days of delivery for a full refund.
Items must be in original condition with tags attached.

### Extended Holiday Returns
Items purchased between November 1 and December 31 can be returned
until January 31 of the following year.

### Non-Returnable Items
The following items cannot be returned:
- Personalized or custom items
- Intimate apparel and swimwear
- Perishable goods
- Digital downloads
- Gift cards

### How to Start a Return
1. Log into your account
2. Go to Order History
3. Select the item to return
4. Print the prepaid shipping label
5. Drop off at any UPS location

### Refund Processing
Refunds are processed within 5-7 business days after we receive your return.
Original payment method will be credited.
"""
}

chunks = chunk_policy_documents([return_policy])
print(f"Created {len(chunks)} chunks from return policy")

Embedding and Indexing Policies

from openai import OpenAI
from pinecone import Pinecone

client = OpenAI()
pc = Pinecone(api_key="your-api-key")
index = pc.Index("ecommerce-policies")

def index_policy_chunks(chunks: List[Dict]):
    """Embed and index policy chunks"""

    vectors = []

    for chunk in chunks:
        # Create embedding
        embedding = client.embeddings.create(
            model="text-embedding-3-small",
            input=chunk["text"]
        ).data[0].embedding

        vectors.append({
            "id": chunk["id"],
            "values": embedding,
            "metadata": {
                **chunk["metadata"],
                "text": chunk["text"]
            }
        })

    # Upsert to Pinecone
    index.upsert(vectors=vectors, batch_size=100)
    print(f"Indexed {len(vectors)} policy chunks")


# Index all policies
policies = load_all_policies()  # Your function to load policies
chunks = chunk_policy_documents(policies)
index_policy_chunks(chunks)

The Policy FAQ Agent

from openai import OpenAI
from typing import Optional, Dict
import json

class PolicyFAQAgent:
    def __init__(self, vector_index, openai_client):
        self.index = vector_index
        self.client = openai_client

    def answer(self, question: str, context: Optional[Dict] = None) -> Dict:
        """
        Answer a policy question.

        Args:
            question: Customer's question
            context: Optional context like order_id, product_type, etc.

        Returns:
            Dict with answer, sources, and escalation flag
        """

        # Step 1: Retrieve relevant policy chunks
        relevant_policies = self._retrieve_policies(question)

        if not relevant_policies:
            return self._no_answer_response(question)

        # Step 2: Generate answer with citations
        answer = self._generate_answer(question, relevant_policies, context)

        # Step 3: Check if escalation needed
        needs_escalation = self._check_escalation(question, answer)

        return {
            "answer": answer,
            "sources": [p["metadata"]["policy_name"] for p in relevant_policies],
            "needs_escalation": needs_escalation,
            "confidence": self._calculate_confidence(relevant_policies)
        }

    def _retrieve_policies(self, question: str, top_k: int = 5) -> List[Dict]:
        """Retrieve relevant policy sections"""

        query_embedding = self.client.embeddings.create(
            model="text-embedding-3-small",
            input=question
        ).data[0].embedding

        results = self.index.query(
            vector=query_embedding,
            top_k=top_k,
            include_metadata=True
        )

        # Filter by relevance threshold
        return [
            {"score": m.score, "metadata": m.metadata}
            for m in results.matches
            if m.score > 0.7
        ]

    def _generate_answer(
        self,
        question: str,
        policies: List[Dict],
        context: Optional[Dict]
    ) -> str:
        """Generate helpful answer from retrieved policies"""

        policy_context = "\n\n---\n\n".join([
            f"From {p['metadata']['policy_name']} (effective {p['metadata']['effective_date']}):\n{p['metadata']['text']}"
            for p in policies
        ])

        context_str = f"\nCustomer context: {json.dumps(context)}" if context else ""

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "system",
                    "content": """You are a helpful customer support assistant. Answer the customer's question based ONLY on the provided policy information.

Guidelines:
1. Be friendly and helpful
2. Cite which policy you're referencing
3. If the policy has specific steps, list them clearly
4. If there are exceptions, mention them
5. If you're not sure or the policy doesn't cover this exactly, say so
6. For complex situations, suggest contacting support for clarification"""
                },
                {
                    "role": "user",
                    "content": f"""Customer question: {question}
{context_str}

Relevant policies:
{policy_context}"""
                }
            ],
            temperature=0.3  # Lower temperature for accuracy
        )

        return response.choices[0].message.content

    def _check_escalation(self, question: str, answer: str) -> bool:
        """Check if this should be escalated to human support"""

        escalation_triggers = [
            "complaint",
            "supervisor",
            "manager",
            "fraud",
            "discrimination",
            "lawsuit",
            "attorney",
            "exception",
            "special circumstance",
            "not satisfied"
        ]

        question_lower = question.lower()
        for trigger in escalation_triggers:
            if trigger in question_lower:
                return True

        # Also check if the answer indicates uncertainty
        uncertainty_phrases = [
            "I'm not sure",
            "please contact",
            "for clarification",
            "may vary",
            "depends on"
        ]
        for phrase in uncertainty_phrases:
            if phrase.lower() in answer.lower():
                return True

        return False

    def _calculate_confidence(self, policies: List[Dict]) -> float:
        """Calculate confidence based on retrieval scores"""
        if not policies:
            return 0.0
        return sum(p["score"] for p in policies) / len(policies)

    def _no_answer_response(self, question: str) -> Dict:
        """Response when no relevant policy found"""
        return {
            "answer": """I couldn't find specific policy information for your question.
This might be a unique situation that our standard policies don't cover.

I'd recommend:
1. Checking our Help Center at help.example.com
2. Contacting our support team who can help with your specific case

Is there anything else I can help you with?""",
            "sources": [],
            "needs_escalation": True,
            "confidence": 0.0
        }


# Usage
agent = PolicyFAQAgent(index, client)

# Simple question
result = agent.answer("How do I return an item?")
print(result["answer"])

# Question with context
result = agent.answer(
    "Can I return this swimsuit?",
    context={"product_category": "swimwear", "days_since_delivery": 15}
)
print(result["answer"])

Example Conversations

Handling Complex Scenarios

def handle_policy_exception_request(
    agent: PolicyFAQAgent,
    question: str,
    order_id: str = None
) -> Dict:
    """Handle requests that might need exceptions"""

    # First, get the standard policy answer
    result = agent.answer(question, context={"order_id": order_id})

    # Check if customer is asking for an exception
    exception_keywords = ["exception", "special case", "but", "however", "please"]
    asks_for_exception = any(kw in question.lower() for kw in exception_keywords)

    if asks_for_exception or result["confidence"] < 0.8:
        # Add escalation message
        result["answer"] += """

---

I want to make sure you get the help you need. Since your situation might require special consideration, I'm flagging this for our customer care team.

A team member will review your case and reach out within 24 hours. If you need immediate assistance, you can also call us at 1-800-XXX-XXXX."""

        result["needs_escalation"] = True

    return result

Keeping Policies Up to Date

import hashlib
from datetime import datetime

class PolicyManager:
    def __init__(self, vector_index, openai_client):
        self.index = vector_index
        self.client = openai_client

    def update_policy(self, policy: Dict):
        """Update a policy with change tracking"""

        # Archive old version
        old_chunks = self._get_policy_chunks(policy["name"])
        self._archive_chunks(old_chunks)

        # Create new chunks
        new_chunks = chunk_policy_documents([policy])

        # Index new version
        index_policy_chunks(new_chunks)

        # Log the change
        self._log_policy_change(
            policy_name=policy["name"],
            effective_date=policy["effective_date"],
            change_summary=policy.get("change_summary", "Policy updated")
        )

    def _get_policy_chunks(self, policy_name: str) -> List[str]:
        """Get IDs of existing chunks for a policy"""
        # Query by metadata filter
        # Implementation depends on your vector DB
        pass

    def _archive_chunks(self, chunk_ids: List[str]):
        """Archive old policy chunks (don't delete, for audit trail)"""
        for chunk_id in chunk_ids:
            # Rename with archive prefix
            pass

    def _log_policy_change(self, policy_name: str, effective_date: str, change_summary: str):
        """Log policy change for audit and notifications"""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "policy_name": policy_name,
            "effective_date": effective_date,
            "change_summary": change_summary
        }
        # Store in database
        print(f"Policy updated: {policy_name} effective {effective_date}")

Key Takeaways

Next up: Order History Agent - Build an agent that helps customers track and understand their orders.