用AI Agent自动处理客户邮件分类与初回复：一个真实客服场景的解决方案

# tools.py
from langchain.tools import BaseTool
from typing import Type, Optional
from pydantic import BaseModel, Field
import re

class ClassifyEmailInput(BaseModel):
    subject: str = Field(description="邮件主题")
    body: str = Field(description="邮件正文")

class ClassifyEmailTool(BaseTool):
    name = "classify_email"
    description = "根据主题和正文判断邮件类别，返回: tech_urgent, billing, sales, account, other"
    args_schema: Type[BaseModel] = ClassifyEmailInput
    
    def _run(self, subject: str, body: str) -> str:
        text = f"{subject} {body}".lower()
        if any(k in text for k in ["error", "500", "crash", "down", "not working"]):
            return "tech_urgent"
        if any(k in text for k in ["bill", "charge", "invoice", "refund", "$"]):
            return "billing"
        if any(k in text for k in ["pricing", "buy", "quote", "team plan"]):
            return "sales"
        if any(k in text for k in ["login", "invite", "access", "role"]):
            return "account"
        return "other"

class ExtractEntitiesInput(BaseModel):
    text: str = Field(description="邮件全文")

class ExtractEntitiesTool(BaseTool):
    name = "extract_entities"
    description = "提取订单号(ORD-数字)、错误码(数字)、邮箱地址"
    args_schema: Type[BaseModel] = ExtractEntitiesInput
    
    def _run(self, text: str) -> dict:
        order_id = re.search(r'ORD-(\d+)', text)
        error_code = re.search(r'error\s*code[:\s]*(\d+)', text, re.I)
        email = re.search(r'[\w\.-]+@[\w\.-]+\.\w+', text)
        return {
            "order_id": order_id.group(0) if order_id else None,
            "error_code": error_code.group(1) if error_code else None,
            "customer_email": email.group(0) if email else None
        }

# reply_generators.py
from langchain.prompts import ChatPromptTemplate
from langchain.chat_models import ChatOpenAI
import os

llm = ChatOpenAI(model="gpt-4", temperature=0.3)

# 技术问题专用prompt（可嫁接内部知识库）
tech_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是SaaS公司的技术支持专家。根据用户描述的错误，给出：
1. 可能原因分析（2-3条）
2. 排查步骤（具体命令或操作）
3. 如需要日志，说明要哪些字段
保持专业但友好，不超过150字。"""),
    ("human", "问题描述：{error_desc}\n提取到错误码：{error_code}")
])

billing_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是客服主管，处理账单咨询。先表达理解，然后引导用户提供订单号或邮箱以便核查。禁止承诺退款金额。"),
    ("human", "用户说：{billing_text}")
])

def gen_tech_reply(error_desc, error_code):
    chain = tech_prompt | llm
    return chain.invoke({"error_desc": error_desc, "error_code": error_code}).content

def gen_billing_reply(billing_text):
    chain = billing_prompt | llm
    return chain.invoke({"billing_text": billing_text}).content

def gen_sales_reply(question):
    return f"感谢您对定价的关注。当前团队版 $199/月，含10个席位。如需定制报价，请提供大致人数和功能需求。\n\n原问题：{question}"

def gen_account_reply(issue):
    return f"账号问题已收到。请您尝试：1) 登出后重新登录 2) 主账号在设置-成员中重新发送邀请。如仍失败，请提供您的账号邮箱和错误截图。\n\n{issue}"

# mail_agent.py
from langchain.agents import Tool, AgentExecutor, LLMSingleActionAgent
from langchain.agents import AgentOutputParser
from langchain.prompts import StringPromptTemplate
from langchain.chains import LLMChain
from typing import List, Union
import imaplib
import email
from email.header import decode_header

# 连接邮箱（示例用IMAP）
def fetch_unread_emails(imap_server, username, password):
    mail = imaplib.IMAP4_SSL(imap_server)
    mail.login(username, password)
    mail.select("INBOX")
    status, ids = mail.search(None, "UNSEEN")
    emails = []
    for num in ids[0].split():
        status, msg_data = mail.fetch(num, "(RFC822)")
        msg = email.message_from_bytes(msg_data[0][1])
        subject = decode_header(msg["Subject"])[0][0]
        if isinstance(subject, bytes):
            subject = subject.decode()
        body = ""
        if msg.is_multipart():
            for part in msg.walk():
                if part.get_content_type() == "text/plain":
                    body = part.get_payload(decode=True).decode()
                    break
        else:
            body = msg.get_payload(decode=True).decode()
        emails.append({"id": num, "subject": subject, "body": body})
    return emails

# Agent的prompt模板
class MailAgentPrompt(StringPromptTemplate):
    template = """你是一个邮件处理专家。你有以下工具：
{tools}

当前任务邮件：
主题：{subject}
正文：{body}

请按以下步骤思考：
1. 先分类（调用classify_email）
2. 提取实体（调用extract_entities） 
3. 根据类别选择回复策略：
   - tech_urgent → 生成技术回复
   - billing → 生成账单回复
   - sales → 生成销售回复
   - account → 生成账号回复
   - other → 转人工

输出格式：
Thought: 我需要做什么
Action: 工具名
Action Input: {{参数}}
... (可重复)
Final Answer: 最终的回复草稿 + 建议的下一步（人工发送/需补充信息）

开始！
{agent_scratchpad}"""
    
    def format(self, **kwargs) -> str:
        kwargs["tools"] = "\n".join([f"- {t.name}: {t.description}" for t in kwargs["tools"]])
        return self.template.format(**kwargs)

# 注册工具
from tools import ClassifyEmailTool, ExtractEntitiesTool
classify = ClassifyEmailTool()
extract = ExtractEntitiesTool()

tools = [
    Tool(name="classify_email", func=classify._run, description=classify.description),
    Tool(name="extract_entities", func=extract._run, description=extract.description)
]

# 初始化Agent
prompt = MailAgentPrompt()
llm_chain = LLMChain(llm=llm, prompt=prompt)
agent = LLMSingleActionAgent(
    llm_chain=llm_chain,
    output_parser=AgentOutputParser(),
    stop=["\nObservation:"],
    allowed_tools=[t.name for t in tools]
)
executor = AgentExecutor.from_agent_and_tools(agent=agent, tools=tools, verbose=True)

# main.py
def process_email(email_item, agent_executor):
    raw_input = f"Subject: {email_item['subject']}\nBody: {email_item['body']}"
    result = agent_executor.run(raw_input)
    
    # 简单后处理：如果是tech_urgent 且包含error_code，标记高优先级
    if "tech_urgent" in result and re.search(r'error_code', raw_input):
        priority = "HIGH"
    else:
        priority = "NORMAL"
    
    # 存储草稿到数据库（示意）
    draft_record = {
        "original_email_id": email_item["id"],
        "agent_reply": result,
        "priority": priority,
        "human_reviewed": False
    }
    # 实际开发存入PostgreSQL
    return draft_record

def main():
    # 配置
    IMAP_SERVER = "imap.example.com"
    USER = "support@yourcompany.com"
    PWD = "your_app_password"
    
    unread = fetch_unread_emails(IMAP_SERVER, USER, PWD)
    for email_item in unread:
        draft = process_email(email_item, executor)
        # 将草稿写入客服系统dashboard，等人点击“发送”
        print(f"草稿已生成：{draft['agent_reply'][:100]}...")
    print(f"处理完成，共 {len(unread)} 封")

if __name__ == "__main__":
    main()