تقویت پشتیبانی مشتری: عوامل هوش مصنوعی، LangGraph و RAG برای اتوماسیون ایمیل

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_google_genai import ChatGoogleGenerativeAI, GoogleGenerativeAIEmbeddings
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# Loading Docs
loader = TextLoader("./data/agency.txt")
docs = loader.load()

# Split document into chuncks
doc_splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=50)
doc_chunks = doc_splitter.split_documents(docs)

# Creating vector embeddings
embeddings = GoogleGenerativeAIEmbeddings(model="models/embedding-001") # Updated model name
vectorstore = Chroma.from_documents(doc_chunks, embeddings, persist_directory="db")

# Semantic vector search retriever
vectorstore_retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# load inbox emails
workflow.set_entry_point("load_emails")

# check if there are new emails to process
workflow.add_edge("load_emails", "is_email_inbox_empty")
workflow.add_conditional_edges(
    "is_email_inbox_empty",
    nodes.check_new_emails,
    {
        "process": "categorize_email",
        "empty": END
    }
)

# route email based on category
workflow.add_conditional_edges(
    "categorize_email",
    nodes.route_email_based_on_category,
    {
        "product related": "construct_rag_queries",
        "not product related": "email_writer", # Feedback or Complaint
        "unrelated": "skip_unrelated_email"
    }
)

# pass constructed queries to RAG chain to get information
workflow.add_edge("construct_rag_queries", "retrieve_from_rag")
# give information to writer agent to write the email
workflow.add_edge("retrieve_from_rag", "email_writer")
# proofread the generated email
workflow.add_edge("email_writer", "email_proofreader")
# check if email is sendable or not, if not rewrite the email
workflow.add_conditional_edges(
    "email_proofreader",
    nodes.must_rewrite,
    {
        "send": "send_email",
        "rewrite": "email_writer"
    }
)

# check if there are still emails to be processed
workflow.add_edge("send_email", "is_email_inbox_empty")
workflow.add_edge("skip_unrelated_email", "is_email_inbox_empty" )

class GmailToolsClass:
    def __init__(self):
        self.service = self._get_gmail_service()

    def fetch_unanswered_emails(self, max_results=50):
        try:
            # 1. Get recent emails
            recent_emails = self.fetch_recent_emails(max_results)
            if not recent_emails: return []

            # 2. Get draft replies to filter out already answered threads
            drafts = self.fetch_draft_replies()
            threads_with_drafts = {draft['threadId'] for draft in drafts}

            # 3. Process each new email
            seen_threads = set()
            unanswered_emails = []
            for email in recent_emails:
                thread_id = email['threadId']
                # Skip if thread was seen before or has a draft reply
                if thread_id not in seen_threads and thread_id not in threads_with_drafts:
                    seen_threads.add(thread_id)
                    email_info = self._get_email_info(email['id'])
                    # Skip if email is sent from the agency (not received)
                    if self._should_skip_email(email_info):
                        continue
                    unanswered_emails.append(email_info)
            return unanswered_emails

        except Exception as e:
            print(f"An error occurred: {e}")
            return []

    def fetch_recent_emails(self, max_results=50):
        """Fetches emails from the last 8 hours."""
        try:
            # Query emails from the last 8 hours
            now = datetime.now()
            delay = now - timedelta(hours=8)
            after_timestamp = int(delay.timestamp())
            before_timestamp = int(now.timestamp())
            query = f"after:{after_timestamp} before:{before_timestamp}"

            results = self.service.users().messages().list(
                userId="me", q=query, maxResults=max_results
            ).execute()
            messages = results.get("messages", [])
            return messages

        except Exception as error:
            print(f"An error occurred while fetching emails: {error}")
            return []

    # ... other methods ...

    def _get_email_info(self, msg_id):
        message = self.service.users().messages().get(
            userId="me", id=msg_id, format="full"
        ).execute()

        payload = message.get('payload', {})
        headers = {header["name"].lower(): header["value"] for header in payload.get("headers", [])}

        return {
            "id": msg_id,
            "threadId": message.get("threadId"),
            "messageId": headers.get("message-id"),
            "references": headers.get("references", ""),
            "sender": headers.get("from", "Unknown"),
            "subject": headers.get("subject", "No Subject"),
            "body": self._get_email_body(payload),
        }

CATEGORIZE_EMAIL_PROMPT = """
# **Role:**

You are a highly skilled customer support specialist working for a SaaS company specializing in AI agent design. Your expertise lies in understanding customer intent and meticulously categorizing emails to ensure they are handled efficiently.

# **Instructions:**

1. Review the provided email content thoroughly.
2. Use the following rules to assign the correct category:
   - **product_enquiry**: When the email seeks information about a product feature, benefit, service, or pricing.
   - **customer_complaint**: When the email communicates dissatisfaction or a complaint.
   - **customer_feedback**: When the email provides feedback or suggestions regarding a product or service.
   - **unrelated**: When the email content does not match any of the above categories.

# **EMAIL CONTENT:**
{email}

# **Notes:**

* Base your categorization strictly on the email content provided; avoid making assumptions or overgeneralizing.
"""

def categorize_email(self, state: GraphState) -> GraphState:
    """Categorizes the current email using the categorize_email agent."""
    # Get the last email
    current_email = state["emails"][-1]
    result = self.agents.categorize_email.invoke({"email": current_email.body})
    print(f"Email category: {result.category.value}")

    return {
        "email_category": result.category.value,
        "current_email": current_email
    }

EMAIL_WRITER_PROMPT = """
# **Role:**  

You are a professional email writer working as part of the customer support team at a SaaS company specializing in AI agent development. Your role is to draft thoughtful and friendly emails that effectively address customer queries based on the given category and relevant information.  

# **Tasks:**  

1. Use the provided email category, subject, content, and additional information to craft a professional and helpful response.  
2. Ensure the tone matches the email category, showing empathy, professionalism, and clarity.  
3. Write the email in a structured, polite, and engaging manner that addresses the customer’s needs.  

...

Write the email in the following format:  
   Dear [Customer Name],  

   [Email body responding to the query, based on the category and information provided.]  

   Best regards,  
   The Agentia Team    
   - Replace `[Customer Name]` with “Customer” if no name is provided.  
   - Ensure the email is friendly, concise, and matches the tone of the category.  

3. If a feedback is provided, use it to improve the email while ensuring it still aligns with the predefined guidelines.  

# **Notes:**  

* Return only the final email without any additional explanation or preamble.  
* Always maintain a professional and empathetic tone that aligns with the context of the email.  
* If the information provided is insufficient, politely request additional details from the customer.  
* Make sure to follow any feedback provided when crafting the email.  
"""

def write_draft_email(self, state: GraphState) -> GraphState:
    """Writes a draft email based on the current email and retrieved information."""
    # Get messages history for current email
    writer_messages = state.get('writer_messages', [])

    # Write email
    draft_result = self.agents.email_writer.invoke({
        "email_category": state["email_category"],
        "email_content": state["current_email"].body,
        "retrieved_documents": state["retrieved_documents"], # Empty for feedback or complaint
        "history": writer_messages
    })
    email = draft_result.email
    trials = state.get('trials', 0) + 1

    # Append writer's draft to the message list
    writer_messages.append(f"**Draft {trials}:**\n{email}")

    return {
        "generated_email": email,
        "trials": trials,
        "writer_messages": writer_messages
    }

GENERATE_RAG_QUERIES_PROMPT = """
# **Role:**

You are an expert at analyzing customer emails to extract their intent and construct the most relevant queries for internal knowledge sources.

# **Context:**

You will be given the text of an email from a customer. This email represents their specific query or concern. Your goal is to interpret their request and generate precise questions that capture the essence of their inquiry.

# **Instructions:**

1. Carefully read and analyze the email content provided.
2. Identify the main intent or problem expressed in the email.
3. Construct up to three concise, relevant questions that best represent the customer’s intent or information needs.
4. Include only relevant questions. Do not exceed three questions.
5. If a single question suffices, provide only that.

# **EMAIL CONTENT:**
{email}

# **Notes:**

* Focus exclusively on the email content to generate the questions; do not include unrelated or speculative information.
* Ensure the questions are specific and actionable for retrieving the most relevant answer.
* Use clear and professional language in your queries.
"""

def construct_rag_queries(self, state: GraphState) -> GraphState:
    """Constructs RAG queries based on the email content."""
    email_content = state["current_email"].body
    query_result = self.agents.design_rag_queries.invoke({"email": email_content})

    return {"rag_queries": query_result.queries}

GENERATE_RAG_ANSWER_PROMPT = """
# **Role:**

You are a highly knowledgeable and helpful assistant specializing in question-answering tasks.

# **Context:**

You will be provided with pieces of retrieved context relevant to the user's question. This context is your sole source of information for answering.

# **Instructions:**

1. Carefully read the question and the provided context.
2. Analyze the context to identify relevant information that directly addresses the question.
3. Formulate a clear and precise response based only on the context. Do not infer or assume information that is not explicitly stated.
4. If the context does not contain sufficient information to answer the question, respond with: "I don't know."
5. Use simple, professional language that is easy for users to understand.

# **Question:**
{question}

# **Context:**
{context}

# **Notes:**

* Stay within the boundaries of the provided context; avoid introducing external information.
* If multiple pieces of context are relevant, synthesize them into a cohesive and accurate response.
* Prioritize user clarity and ensure your answers directly address the question without unnecessary elaboration.
"""

def retrieve_from_rag(self, state: GraphState) -> GraphState:
    """Retrieves information from internal knowledge based on RAG questions."""
    final_answer = ""
    for query in state["rag_queries"]:
        rag_result = self.agents.generate_rag_answer.invoke(query)
        final_answer += query + "\n" + rag_result + "\n\n"

    return {"retrieved_documents": final_answer}

EMAIL_PROOFREADER_PROMPT = """
# **Role:**

You are an expert email proofreader working for the customer support team at a SaaS company specializing in AI agent development. Your role is to analyze and assess replies generated by the writer agent to ensure they accurately address the customer's inquiry, adhere to the company's tone and writing standards, and meet professional quality expectations.

# **Context:**

You are provided with the **initial email** content written by the customer and the **generated email** crafted by the our writer agent.

# **Instructions:**

1. Analyze the generated email for:
   - **Accuracy**: Does it appropriately address the customer’s inquiry based on the initial email and information provided?
   - **Tone and Style**: Does it align with the company’s tone, standards, and writing style?
   - **Quality**: Is it clear, concise, and professional?
2. Determine if the email is:
   - **Sendable**: The email meets all criteria and is ready to be sent.
   - **Not Sendable**: The email contains significant issues requiring a rewrite.
3. Only judge the email as "not sendable" (`send: false`) if lacks information or inversely contains irrelevant ones that would negatively impact customer satisfaction or professionalism.
4. Provide actionable and clear feedback for the writer agent if the email is deemed "not sendable."

---

# **INITIAL EMAIL:**
{initial_email}

# **GENERATED REPLY:**
{generated_email}

---

# **Notes:**

* Be objective and fair in your assessment. Only reject the email if necessary.
* Ensure feedback is clear, concise, and actionable.
"""

def verify_generated_email(self, state: GraphState) -> GraphState:
    """Verifies the generated email using the proofreader agent."""
    print(Fore.YELLOW + "Verifying generated email...\n" + Style.RESET_ALL)
    review = self.agents.email_proofreader.invoke({
        "initial_email": state["current_email"].body,
        "generated_email": state["generated_email"],
    })

    # Store feedback if email is not good
    if not review.send:
        writer_messages = state.get('writer_messages', [])
        writer_messages.append(f"**Proofreader Feedback:**\n{review.feedback}")

    return {
        "sendable": review.send,
        "writer_messages": writer_messages
    }

def create_draft_response(self, state: GraphState) -> GraphState:
    """Creates a draft response in Gmail."""
    print(Fore.YELLOW + "Creating draft email...\n" + Style.RESET_ALL)
    self.gmail_tools.create_draft_reply(state["current_email"], state["generated_email"])

    return {"retrieved_documents": "", "trials": 0}

def send_email_response(self, state: GraphState) -> GraphState:
    """Sends the email response directly using Gmail."""
    print(Fore.YELLOW + "Sending email...\n" + Style.RESET_ALL)
    self.gmail_tools.send_reply(state["current_email"], state["generated_email"])

    return {"retrieved_documents": "", "trials": 0}