Python AI Agent 零基础教程 | 第9篇：实战项目——构建智能客服Agent
前言
经过前 8 期的学习，我们已经掌握了 AI Agent 的核心技能。今天我们将综合运用这些知识，从零构建一个完整的智能客服系统！
这个客服 Agent 将能够：
- 🤖 自动回答常见问题
- 📦 查询订单状态
- 💡 推荐产品
- 📝 收集用户反馈
一、项目需求分析
1.1 客服场景描述
我们为一家「优品数码专营店」构建智能客服：
Python 代码
┌─────────────────────────────────────────────────────────────┐
│ 优品数码专营店 智能客服 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 🏪 店铺信息： │
│ - 营业时间：9:00 - 22:00 │
│ - 退换货政策：7天无理由退换 │
│ - 快递说明：全场包邮，48小时内发货 │
│ │
│ 📋 主要业务： │
│ - 商品咨询 │
│ - 订单查询 │
│ - 退换货处理 │
│ - 售后问题 │
│ │
│ 💡 核心功能： │
│ - 自动问答 │
│ - 订单管理 │
│ - 投诉建议 │
│ - 转人工服务 │
│ │
└─────────────────────────────────────────────────────────────┘
1.2 功能模块
| 模块 | 功能 | 优先级 |
|---|
| 基础问答 | 回答常见问题 | ⭐⭐⭐ 高 |
| 订单查询 | 查单、改地址、取消 | ⭐⭐⭐ 高 |
| 商品推荐 | 根据需求推荐产品 | ⭐⭐ 中 |
| 投诉处理 | 记录投诉、跟进 | ⭐⭐ 中 |
| 转人工 | 无法解答时转人工 | ⭐⭐⭐ 高 |
---
二、系统架构设计
2.1 整体架构
Python 代码
┌─────────────────────────────────────────────────────────────┐
│ 智能客服系统架构 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 用户 ──▶ 接入层 │
│ │ │
│ ▼ │
│ ┌────────────────────────────────────────────┐ │
│ │ 意图识别层 │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │ 问答 │ │ 订单 │ │ 转人工 │ │ │
│ │ │ 意图 │ │ 意图 │ │ 意图 │ │ │
│ │ └────┬────┘ └────┬────┘ └────┬────┘ │ │
│ └───────┼──────────┼──────────┼────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌────────────────────────────────────────────┐ │
│ │ 知识库层 │ │
│ │ ┌────────┐ ┌────────┐ ┌────────┐ │ │
│ │ │ FAQ │ │ 订单 │ │ 商品 │ │ │
│ │ │ 知识库 │ │ 数据库 │ │ 数据库 │ │ │
│ │ └────────┘ └────────┘ └────────┘ │ │
│ └────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ 返回结果 │
│ │
└─────────────────────────────────────────────────────────────┘
2.2 核心组件
Python 代码
系统组件概览
COMPONENTS = {
"意图识别": "识别用户想做什么（查单/咨询/投诉/转人工）",
"知识库": "存储常见问题和答案",
"订单系统": "模拟订单查询和管理",
"对话管理": "管理对话流程和状态",
"响应生成": "生成自然语言回复"
}
三、数据模型设计
3.1 知识库数据
Python 代码
FAQ 知识库
FAQ_KNOWLEDGE = {
"配送问题": [
{
"question": ["什么时候发货", "几天到", "快递", "物流"],
"answer": "您好！我们全场包邮，48小时内发货。正常情况下2-5天可达，具体以物流信息为准。"
},
{
"question": ["可以改地址吗", "修改地址"],
"answer": "您好！订单未发货前可以修改地址。请提供订单号，我来帮您处理。"
}
],
"退换货": [
{
"question": ["怎么退货", "可以退吗", "退换货"],
"answer": "您好！我们支持7天无理由退换货（不影响二次销售）。请提供订单号，我会帮您申请。"
}
],
"支付问题": [
{
"question": ["支持哪些支付", "怎么付款", "支付方式"],
"answer": "您好！我们支持支付宝、微信支付、银行卡支付等多种方式。"
}
]
}
商品数据
PRODUCTS = [
{"id": "P001", "name": "iPhone 15 Pro", "price": 7999, "category": "手机", "stock": 100},
{"id": "P002", "name": "MacBook Pro 14", "price": 14999, "category": "电脑", "stock": 50},
{"id": "P003", "name": "AirPods Pro 2", "price": 1899, "category": "耳机", "stock": 200},
{"id": "P004", "name": "iPad Air", "price": 4799, "category": "平板", "stock": 80},
]
订单数据（模拟）
ORDERS = [
{"order_id": "ORD20240101001", "product": "iPhone 15 Pro", "status": "配送中", "address": "北京市朝阳区", "phone": "138**1234"},
{"order_id": "ORD20240101002", "product": "AirPods Pro 2", "status": "已签收", "address": "上海市浦东新区", "phone": "139**5678"},
{"order_id": "ORD20240101003", "product": "MacBook Pro 14", "status": "待发货", "address": "广州市天河区", "phone": "137**9012"},
]
3.2 对话状态
Python 代码
class ConversationState:
"""对话状态"""
def __init__(self, session_id):
self.session_id = session_id
self.intent = None # 当前意图
self.context = {} # 上下文信息
self.waiting_for = None # 等待用户输入什么
self.history = [] # 对话历史
def set_intent(self, intent):
"""设置意图"""
self.intent = intent
self.context = {}
def add_context(self, key, value):
"""添加上下文"""
self.context[key] = value
def set_waiting(self, field):
"""设置等待输入"""
self.waiting_for = field
def clear(self):
"""清除状态"""
self.intent = None
self.context = {}
self.waiting_for = None
四、核心代码实现
4.1 知识库问答
Python 代码
class KnowledgeBase:
"""知识库"""
def __init__(self):
self.faq = self._load_faq()
def _load_faq(self):
"""加载 FAQ"""
return {
"配送": "您好！我们全场包邮，48小时内发货。正常2-5天可达。",
"退货": "您好！我们支持7天无理由退换货，请提供订单号申请。",
"支付": "您好！我们支持支付宝、微信、银行卡支付。",
"保修": "您好！正品保障，提供官方保修服务。",
"发票": "您好！可以开具电子发票，请联系客服提供开票信息。",
"优惠": "您好！我们经常有优惠活动，关注店铺享更多福利！",
"店铺": "您好！我们是优品数码专营店，正品保障，服务至上！",
"营业": "您好！我们的营业时间是每天9:00-22:00。"
}
def search(self, query):
"""搜索答案"""
query = query.lower()
# 关键词匹配
for keyword, answer in self.faq.items():
if keyword in query:
return answer
return None
def get_greeting(self):
"""问候语"""
return """👋 您好！欢迎来到优品数码专营店！
我是您的智能客服小e，请问有什么可以帮您？
📦 您可以咨询：
• 配送物流问题
• 退换货流程
• 产品推荐
• 订单查询
或者直接告诉我您的需求哦~"""
def get_goodbye(self):
"""结束语"""
return """感谢您的咨询，祝您购物愉快！🛒
如有其他问题，随时联系我~
📞 人工客服热线：400-888-8888
⏰ 营业时间：9:00-22:00"""
4.2 订单管理
Python 代码
class OrderManager:
"""订单管理器"""
def __init__(self):
# 模拟订单数据
self.orders = {
"ORD20240101001": {
"product": "iPhone 15 Pro 256G 钛金色",
"status": "配送中",
"address": "北京市朝阳区XX路XX号",
"phone": "138**1234",
"create_time": "2024-01-01 10:30",
"delivery_time": "预计2-3天送达"
},
"ORD20240101002": {
"product": "AirPods Pro 2",
"status": "已签收",
"address": "上海市浦东新区XX路XX号",
"phone": "139**5678",
"create_time": "2023-12-28 15:20",
"delivery_time": "已签收"
},
"ORD20240101003": {
"product": "MacBook Pro 14寸",
"status": "待发货",
"address": "广州市天河区XX路XX号",
"phone": "137**9012",
"create_time": "2024-01-02 09:15",
"delivery_time": "48小时内发货"
}
}
def query_by_order_id(self, order_id):
"""按订单号查询"""
order_id = order_id.upper().strip()
if order_id in self.orders:
return self.orders[order_id]
return None
def query_by_phone(self, phone):
"""按手机号查询"""
results = []
for oid, order in self.orders.items():
if phone in order["phone"]:
results.append((oid, order))
return results
def format_order(self, order_id, order):
"""格式化订单信息"""
return f"""
📦 订单号：{order_id}
🛍️ 商品：{order['product']}
📍 地址：{order['address']}
📱 联系方式：{order['phone']}
🚚 状态：{order['status']}
⏰ {order['delivery_time']}
"""
def modify_address(self, order_id, new_address):
"""修改地址"""
if order_id in self.orders:
self.orders[order_id]["address"] = new_address
return True
return False
4.3 产品推荐
Python 代码
class ProductRecommender:
"""产品推荐"""
def __init__(self):
self.products = [
{"name": "iPhone 15 Pro", "price": 7999, "tags": ["手机", "高端", "苹果"]},
{"name": "iPhone 15", "price": 5999, "tags": ["手机", "性价比", "苹果"]},
{"name": "MacBook Pro 14", "price": 14999, "tags": ["电脑", "高端", "苹果"]},
{"name": "MacBook Air", "price": 8999, "tags": ["电脑", "轻薄", "苹果"]},
{"name": "AirPods Pro 2", "price": 1899, "tags": ["耳机", "降噪", "苹果"]},
{"name": "AirPods 3", "price": 1399, "tags": ["耳机", "入门", "苹果"]},
{"name": "iPad Pro", "price": 6999, "tags": ["平板", "高端", "苹果"]},
{"name": "iPad Air", "price": 4799, "tags": ["平板", "性价比", "苹果"]},
{"name": "Apple Watch", "price": 2999, "tags": ["手表", "健康", "苹果"]},
]
def recommend(self, budget=None, category=None):
"""推荐产品"""
results = self.products
# 按类别筛选
if category:
results = [p for p in results if category in p["tags"]]
# 按预算筛选
if budget:
if isinstance(budget, tuple):
results = [p for p in results if budget[0] <= p["price"] <= budget[1]]
else:
results = [p for p in results if p["price"] <= budget]
# 格式化输出
if not results:
return "抱歉，暂未找到符合条件的产品"
response = "根据您的需求，为您推荐：\n\n"
for i, p in enumerate(results[:5], 1):
response += f"{i}. {p['name']} - ¥{p['price']}\n"
return response
def search(self, keyword):
"""搜索产品"""
results = [p for p in self.products if keyword.lower() in p["name"].lower()]
if not results:
return f"抱歉，未找到「{keyword}」相关产品"
response = f"为您找到 {len(results)} 个相关产品：\n\n"
for p in results:
response += f"📱 {p['name']}\n 价格：¥{p['price']}\n 标签：{' '.join(p['tags'])}\n\n"
return response
五、智能客服 Agent
5.1 完整 Agent 代码
Python 代码
import requests
import json
import re
class CustomerServiceAgent:
"""智能客服 Agent"""
def __init__(self, api_key):
self.api_key = api_key
self.knowledge_base = KnowledgeBase()
self.order_manager = OrderManager()
self.product_recommender = ProductRecommender()
self.messages = []
self.conversation_state = None
# 系统提示词
self.system_prompt = """你是一个专业的电商客服，名字叫小e，服务于「优品数码专营店」。
【店铺信息】
- 营业时间：9:00-22:00
- 退换货政策：7天无理由退换
- 快递说明：全场包邮，48小时内发货
【服务原则】
- 保持礼貌和耐心
- 回答专业、准确
- 无法解答时转人工
- 结束时询问是否还有其他问题
【回复格式】
- 使用 emoji 增加亲切感
- 回答简洁明了
- 必要时分点说明
【转人工条件】
- 涉及投诉处理
- 需要人工确认的重要事项
- 连续3次无法理解用户意图"""
def reset(self):
"""重置对话"""
self.messages = [{"role": "system", "content": self.system_prompt}]
self.conversation_state = None
def think(self, user_input):
"""处理用户输入"""
# 检查是否需要转人工
if "转人工" in user_input or "人工客服" in user_input:
return self._transfer_to_human()
# 先尝试知识库匹配
kb_answer = self.knowledge_base.search(user_input)
if kb_answer:
return kb_answer
# 检查是否包含订单号
order_id = self._extract_order_id(user_input)
if order_id:
return self._handle_order_query(order_id)
# 检查是否需要产品推荐
if any(kw in user_input for kw in ["推荐", "想要", "想买", "有什么"]):
category = self._extract_category(user_input)
budget = self._extract_budget(user_input)
return self.product_recommender.recommend(budget, category)
# 检查是否是产品搜索
if "有" in user_input and ("吗" in user_input or "卖" in user_input):
keyword = self._extract_product_keyword(user_input)
if keyword:
return self.product_recommender.search(keyword)
# 使用 AI 生成回复
return self._generate_response(user_input)
def _extract_order_id(self, text):
"""提取订单号"""
pattern = r'ORD\d{12}'
match = re.search(pattern, text.upper())
return match.group(0) if match else None
def _extract_category(self, text):
"""提取商品类别"""
categories = ["手机", "电脑", "平板", "耳机", "手表"]
for cat in categories:
if cat in text:
return cat
return None
def _extract_budget(self, text):
"""提取预算"""
# 简化处理
if "预算" in text:
numbers = re.findall(r'\d+', text)
if numbers:
return int(numbers[0])
return None
def _extract_product_keyword(self, text):
"""提取产品关键词"""
# 去除问号和语气词
text = text.replace("吗", "").replace("有", "").replace("卖", "").replace("?", "").strip()
if text:
return text
return None
def _handle_order_query(self, order_id):
"""处理订单查询"""
order = self.order_manager.query_by_order_id(order_id)
if order:
return f"为您查询到订单信息：\n{self.order_manager.format_order(order_id, order)}"
return f"未找到订单 {order_id}，请核对订单号是否正确。"
def _transfer_to_human(self):
"""转人工"""
return """🤝 正在为您转接人工客服...
请稍候，人工客服将尽快为您服务。
如有紧急问题，可拨打客服热线：📞 400-888-8888
⏰ 人工服务时间：9:00-22:00"""
def _generate_response(self, user_input):
"""使用 AI 生成回复"""
self.messages.append({"role": "user", "content": user_input})
# 构建 prompt
context = ""
if self.conversation_state and self.conversation_state.context:
context = f"\n当前上下文：{self.conversation_state.context}"
# 调用 API
response = self._call_api(context)
self.messages.append({"role": "assistant", "content": response})
return response
def _call_api(self, context=""):
"""调用 API"""
url = "https://api.openai.com/v1/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}"
}
# 添加上下文到系统提示
messages = [{"role": "system", "content": self.system_prompt + context}]
messages.extend(self.messages[-6:]) # 只保留最近几条
data = {
"model": "gpt-3.5-turbo",
"messages": messages,
"temperature": 0.7
}
try:
response = requests.post(url, headers=headers, json=data, timeout=30)
result = response.json()
return result["choices"][0]["message"]["content"]
except Exception as e:
return f"抱歉，系统暂时繁忙，请稍后再试。错误：{str(e)}"
def run(self):
"""运行客服"""
print("="*60)
print("🏪 优品数码专营店 智能客服")
print("="*60)
print(self.knowledge_base.get_greeting())
print("\n（输入 'quit' 退出，'reset' 重置对话）")
print("="*60)
while True:
try:
user_input = input("\n👤 您：").strip()
if not user_input:
continue
if user_input.lower() == "quit":
print("\n" + self.knowledge_base.get_goodbye())
break
elif user_input.lower() == "reset":
self.reset()
print("\n✅ 对话已重置")
print(self.knowledge_base.get_greeting())
else:
response = self.think(user_input)
print(f"\n🤖 小e：{response}")
except KeyboardInterrupt:
print("\n\n感谢使用，再见！")
break
5.2 运行效果
Python 代码
============================================================
🏪 优品数码专营店 智能客服
============================================================
👋 您好！欢迎来到优品数码专营店！
我是您的智能客服小e，请问有什么可以帮您？
📦 您可以咨询：
• 配送物流问题
• 退换货流程
• 产品推荐
• 订单查询
或者直接告诉我您的需求哦~
（输入 'quit' 退出，'reset' 重置对话）
============================================================
👤 您：你们的发货时间是几点？
🤖 小e：您好！我们全场包邮，48小时内发货。正常情况下2-5天可达，具体以物流信息为准。
👤 您：ORD20240101001 这个订单到哪了？
🤖 小e：为您查询到订单信息：
📦 订单号：ORD20240101001
🛍️ 商品：iPhone 15 Pro 256G 钛金色
📍 地址：北京市朝阳区XX路XX号
📱 联系方式：138**1234
🚚 状态：配送中
⏰ 预计2-3天送达
👤 您：我想买一个平板，预算5000左右
🤖 小e：根据您的需求，为您推荐：
- iPad Air - ¥4799
- iPad Pro - ¥6999
👤 您：quit
感谢您的咨询，祝您购物愉快！🛒
如有其他问题，随时联系我~
📞 人工客服热线：400-888-8888
⏰ 营业时间：9:00-22:00
六、部署与扩展
6.1 Web 接口部署
Python 代码
from flask import Flask, request, jsonify
app = Flask(__name__)
agent = CustomerServiceAgent("your-api-key")
@app.route("/chat", methods=["POST"])
def chat():
"""对话接口"""
data = request.json
user_input = data.get("message", "")
if not user_input:
return jsonify({"error": "请输入消息"})
response = agent.think(user_input)
return jsonify({
"code": 0,
"message": "success",
"data": {
"reply": response
}
})
@app.route("/reset", methods=["POST"])
def reset():
"""重置对话"""
agent.reset()
return jsonify({"code": 0, "message": "success"})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000)
6.2 微信小程序接入
Python 代码
微信公众号自动回复示例
@app.route("/wechat", methods=["GET", "POST"])
def wechat():
if request.method == "GET":
# 验证微信服务器
return request.args.get("echostr", "")
# 处理消息
xml_data = request.data
# 解析并处理...
# 调用 Agent
reply = agent.think(user_message)
# 返回回复
return render_template("reply.xml", reply=reply)
七、本章小结
今天我们构建了：
| 模块 | 功能 |
|---|
| ✅ 知识库 | FAQ 问答、关键词匹配 |
| ✅ 订单管理 | 订单查询、状态跟踪 |
| ✅ 产品推荐 | 按类别、预算推荐 |
| ✅ 智能 Agent | 意图识别、自动分流 |
| ✅ Web 接口 | API 部署方案 |
---
下期预告
第10篇：调试优化与常见问题解决
最后一期我们将学习：
- 常见错误排查
- 性能优化技巧
- 安全性考虑
- 进阶学习路径
👨💻 作者：鹏鹏 | 专注于 AI + 编程教育
📱 关注公众号「跟着鹏鹏学技术」
💬 动手练习：运行客服系统，测试各种功能！
往期精选
- 📖 [第8篇：工具调用]()
- 📖 [第7篇：记忆系统]()
- 📖 [第6篇：提示词工程]()
👨💻 作者：鹏鹏
📱 关注公众号「跟着鹏鹏学技术」
🔔 点赞 + 在看，让更多人看到！
