基于 LLM 的商城智能客服助理开发实战

参考LLM开源文档 Datawhale LLM教程🌐📚

💡实现思路

在商城智能客服助理回答用户关于某个或某类商品的询问时，可基于事先准备好的商品数据进行回答。初始 prompt 如下：

system_prompt = f"""你一个大型商城的客服助理，对于用户询问，你需要做如下步骤，1. 从用户询问中提取商品关键词2. 浏览商品数据列表，检索商品详细信息3. 根据商品信息进行回答商品数据列表:{products_list}
"""

初看这个 prompt 似乎没问题，但可能存在products_list数据量过大导致超过输入max_tokens的问题。因此，我们通过拆解 prompt 链的方式拆分任务，同时增加对输入和输出的评估机制：

• 📝 评估输入，防止违规内容
• 🔍 提取产品和类别关键词
• 🔍 检索商品详细信息
• 📝 生成并评估回答内容
• ✅ 根据评估结果返回最终回复

🚀实现步骤

📊 数据集介绍

本次使用从小米商城爬取的商品数据，包含商品名称、描述、价格等字段，数据结构如下

⚙️ 数据处理

新建processon.py文件，实现 Excel 数据转 JSON 并提取关键信息

import pandas as pd  
import json  
def processonExcel():  """  Excel -> JSON    """  # 读取 Excel 文件  excel_file = pd.ExcelFile('data/爬虫_小米商城数据_分类筛选后.xlsx')  # 获取所有表名  sheet_names = excel_file.sheet_names  # 创建一个空字典来存储结果  result = {}  # 遍历每个工作表  for sheet_name in sheet_names:  df = excel_file.parse(sheet_name)  # 获取数据的基本信息  print(f'sheet表名为{sheet_name}的基本信息：')  df.info()  # 查看数据集行数和列数  rows, columns = df.shape  if rows < 100 and columns < 20:  # 短表数据（行数少于100且列数少于20）查看全量数据信息  print(f'sheet表名为{sheet_name}的全部内容信息：')  print(df.to_csv(sep='\t', na_rep='nan'))  else:  # 长表数据查看数据前几行信息  print(f'sheet表名为{sheet_name}的前几行内容信息：')  print(df.head().to_csv(sep='\t', na_rep='nan'))  # 遍历每个工作表  for sheet_name in sheet_names:  df = excel_file.parse(sheet_name)  # 创建当前工作表的字典  sheet_dict = {}  for index, row in df.iterrows():  item_dict = {  '商品名': row['名字'],  '描述': row['描述'],  '价格': row['price']  }  sheet_dict[index] = item_dict  # 将当前工作表的字典添加到结果字典中  result[sheet_name] = sheet_dict  # 将结果保存为 JSON 文件  json_path = 'data/爬虫_小米商城数据_分类筛选后.json'  with open(json_path, 'w', encoding='utf-8') as json_file:  json.dump(result, json_file, ensure_ascii=False, indent=4)  processonExcel()def processonJsonData():  """  处理JSON数据  :return:商品名列表、商品列表  """    # 读取 JSON 文件  with open('data/爬虫_小米商城数据_分类筛选后.json', 'r', encoding='utf-8') as file:  data = json.load(file)  # print(data)  all_productNames = []  all_products = []  # 遍历不同分类的数据  for category_products in data.values():  all_products.extend(list(category_products.values()))  for product in all_products:  all_productNames.append(product['商品名'])  return all_productNames,all_products

📝 评估输入

通过独立 prompt 判断用户输入是否包含违规内容，避免主 prompt 冗长：

# 检查输入  
def get_safe_input(prompt):  check_msg = """你是一个严格的内容安全过滤器，请判断以下内容是否包含：  - 违法/违规内容  - 仇恨/歧视言论  - NSFW内容  返回JSON格式：{"safe": bool, "reason": str}"""  response = client.chat.completions.create(  model="deepseek-chat",  messages=[  {"role": "system", "content": check_msg},  {"role": "user", "content": prompt}  ],  response_format={"type": "json_object"}  )  return json.loads(response.choices[0].message.content)check_input = get_safe_input(user_input)  
if not check_input['safe']:  print(f"----第一步：输入被拒绝: {check_input['reason']}")  return "输入违规，请注意用词！"  
print("----第一步：输入通过检查")

🔍 提取商品关键词

结合商品名列表和对话上下文（最近 3 条）提取关键词，未识别则返回提示：

all_productNames, all_products = processonJsonData()  
delimiter = "```"  
extraction_prompt = f"""  您需要执行商品关键词提取任务，请严格按以下规则处理：  1. 直接提取规则：  - 当用户输入中明确包含商品名称时  - 提取格式示例：["小米手环8", "Redmi Book 14"]  2. 上下文推断规则：  - 当用户输入为"它的续航怎么样？"类问题时  - 需结合最近3条对话历史推断商品  - 示例上下文：  User: "推荐小米笔记本"  Assistant: "Redmi Book 14和16..."  User: "它的续航怎么样？"  → 应返回["Redmi Book 14", "Redmi Book 16"]  3. 强制格式要求：  - 必须返回Python列表格式  - 空结果返回["未识别"]  接下来用户的输入将用{delimiter} {delimiter}包裹  可用商品名列表: {', '.join(all_productNames)}  最新对话上下文：  {delimiter}  {json.dumps([msg['content'] for msg in all_message[-3:] if msg['role'] != 'system'])}  {delimiter}  
"""  
messages = [  {'role': 'system', 'content': extraction_prompt},  {'role': 'user', 'content': f"{delimiter}{user_input}{delimiter}"}  
]  
res = client.chat.completions.create(  model="deepseek-chat",  messages= messages,  
)  
product_names_key = res.choices[0].message.content  
if "未识别" in product_names_key:  print("----第二步：商品未识别，终止处理")  return "抱歉，没有找到您询问的商品"  
print(f"----第二步：提取到商品关键词 - {product_names_key}")  

🔍 检索商品信息

通过上述的商品关键词，利用模糊匹配商品的完整名称，进而检索出商品的详细信息，同时，为防止查询到的同类商品过多，导致token太长，这里简单地限制最多只返回10条商品信息。

# 模糊匹配用户询问关键词，返回符合条件的商品列表  
def processon_query(custom_query_keys, all_products):  closed = set()  matched_products = []  for product in all_products:  product_name = product['商品名']  for query_key in custom_query_keys:  # 使用模糊匹配计算相似度得分  similarity_score = fuzz.partial_ratio(query_key.lower(), product_name.lower())  # 可以根据实际情况调整这个阈值  if similarity_score >= 90 and product_name not in closed:  closed.add(product_name)  matched_products.append(product)  continue  print(f"----匹配的商品列表：{matched_products}")  # 若数据太多，则返回前10条  return matched_products if len(matched_products)<10 else matched_products[:10]

📝 生成并评估回答

通过独立 prompt 评估回答是否准确，不符合要求则转接人工客服：

# 评估输出  
def check_response(user_input, response, matched_products):  check_msg = f"""  请评估以下代理回复是否很好地回答了用户问题：  用户问题:：{user_input}  代理回复：{response}  可供参考的商品数据：{matched_products}  返回JSON：{{"valid": bool, "reason": str}}  """  response = client.chat.completions.create(  model="deepseek-chat",  messages=[  {"role": "system", "content": system_message},  {"role": "user", "content": check_msg},  ],  response_format={"type": "json_object"}  )  return json.loads(response.choices[0].message.content)response_prompt = f"""  关于这个问题:{delimiter}{user_input}{delimiter}，  请结合商品列表{matched_products}，  简洁地解答疑惑"""  
messages = [  {'role': 'user', 'content': response_prompt}  
]  
all_message.extend(messages)  
res = client.chat.completions.create(  model="deepseek-chat",  messages= all_message,  
)  
response = res.choices[0].message.content  
check_output = check_response(user_input, response, matched_products)  
if not check_output["valid"]:  print(f"----第三步：输出被拒绝: {check_output['reason']}")  return "很抱歉，我无法提供您所需的信息。我将为您转接到一位人工客服代表以获取进一步帮助。"  
print("----第三步：输出通过模型评估")

✨ 生成最终回复

实现流式输出效果，提升用户体验：

print("AI: ", end="", flush=True)  
for char in response:  # 模拟流式输出  print(char, end="", flush=True)  time.sleep(0.02)  # 控制输出速度  
print()

🖥️ 完整代码

import os  
import time  from openai import OpenAI  
from dotenv import load_dotenv, find_dotenv  import json  
from fuzzywuzzy import fuzz  from processon import processonJsonData  # 从env文件中读取API KEY  
def get_openai_key():  _ = load_dotenv(find_dotenv())  return os.environ['OPENAI_API_KEY']  # 检查输入  
def get_safe_input(prompt):  check_msg = """你是一个严格的内容安全过滤器，请判断以下内容是否包含：  - 违法/违规内容  - 仇恨/歧视言论  - NSFW内容  返回JSON格式：{"safe": bool, "reason": str}"""  response = client.chat.completions.create(  model="deepseek-chat",  messages=[  {"role": "system", "content": check_msg},  {"role": "user", "content": prompt}  ],  response_format={"type": "json_object"}  )  return json.loads(response.choices[0].message.content)  # 评估输出  
def check_response(user_input, response, matched_products):  check_msg = f"""  请评估以下代理回复是否很好地回答了用户问题：  用户问题:：{user_input}  代理回复：{response}  可供参考的商品数据：{matched_products}  返回JSON：{{"valid": bool, "reason": str}}  """  response = client.chat.completions.create(  model="deepseek-chat",  messages=[  {"role": "system", "content": system_message},  {"role": "user", "content": check_msg},  ],  response_format={"type": "json_object"}  )  return json.loads(response.choices[0].message.content)  # 模糊匹配用户询问关键词，返回符合条件的商品列表  
def processon_query(custom_query_keys, all_products):  closed = set()  matched_products = []  for product in all_products:  product_name = product['商品名']  for query_key in custom_query_keys:  # 使用模糊匹配计算相似度得分  similarity_score = fuzz.partial_ratio(query_key.lower(), product_name.lower())  # 可以根据实际情况调整这个阈值  if similarity_score >= 90 and product_name not in closed:  closed.add(product_name)  matched_products.append(product)  continue  print(f"----匹配的商品列表：{matched_products}")  # 若数据太多，则返回前10条  return matched_products if len(matched_products)<10 else matched_products[:10]  # AI 回复  
def response(user_input, all_message, debug=True):  # ==========================1 使用 OpenAI 的 Moderation API 检查用户输入是否合规或者是一个注入的 Prompt    check_input = get_safe_input(user_input)  # print(check)  if not check_input['safe']:  if debug:  print(f"----第一步：输入被拒绝: {check_input['reason']}")  return "输入违规，请注意用词！"  if debug: print("----第一步：输入通过检查")  # ==========================2 利用AI抽取出用户询问的有关的商品的商品名字关键词，方便后续查找对应商品  all_productNames, all_products = processonJsonData()  delimiter = "```"  extraction_prompt = f"""  您需要执行商品关键词提取任务，请严格按以下规则处理：  1. 直接提取规则：  - 当用户输入中明确包含商品名称时  - 提取格式示例：["小米手环8", "Redmi Book 14"]  2. 上下文推断规则：  - 当用户输入为"它的续航怎么样？"类问题时  - 需结合最近3条对话历史推断商品  - 示例上下文：  User: "推荐小米笔记本"  Assistant: "Redmi Book 14和16..."  User: "它的续航怎么样？"  → 应返回["Redmi Book 14", "Redmi Book 16"]  3. 强制格式要求：  - 必须返回Python列表格式  - 空结果返回["未识别"]  接下来用户的输入将用{delimiter} {delimiter}包裹  可用商品名列表: {', '.join(all_productNames)}  最新对话上下文：  {delimiter}  {json.dumps([msg['content'] for msg in all_message[-3:] if msg['role'] != 'system'])}  {delimiter}  """  messages = [  {'role': 'system', 'content': extraction_prompt},  {'role': 'user', 'content': f"{delimiter}{user_input}{delimiter}"}  ]  res = client.chat.completions.create(  model="deepseek-chat",  messages= messages,  )  product_names_key = res.choices[0].message.content  if "未识别" in product_names_key:  if debug: print("----第二步：商品未识别，终止处理")  return "抱歉，没有找到您询问的商品"  if debug: print(f"----第二步：提取到商品关键词 - {product_names_key}")  product_names_key = json.loads(product_names_key.replace("'", '"')) # 转为列表  matched_products = processon_query(product_names_key,all_products)  # ==========================3 评估输出  response_prompt = f"""  关于这个问题:{delimiter}{user_input}{delimiter}，  请结合商品列表{matched_products}，  简洁地解答疑惑    """  messages = [  {'role': 'user', 'content': response_prompt}  ]  all_message.extend(messages)  res = client.chat.completions.create(  model="deepseek-chat",  messages= all_message,  )  response = res.choices[0].message.content  check_output = check_response(user_input, response, matched_products)  if not check_output["valid"]:  if debug:  print(f"----第三步：输出被拒绝: {check_output['reason']}")  return "很抱歉，我无法提供您所需的信息。我将为您转接到一位人工客服代表以获取进一步帮助。"  if debug: print("----第三步：输出通过模型评估")  # ==========================4. 生成回答  print("AI: ", end="", flush=True)  for char in response:  # 模拟流式输出  print(char, end="", flush=True)  time.sleep(0.02)  # 控制输出速度  print()  all_message.append({'role': 'assistant', 'content': response})  return all_message  if __name__ == '__main__':  client = OpenAI(api_key=get_openai_key(), base_url="https://api.deepseek.com")  # 初始化对话上下文  system_message = f"""  您是一家大型电子商店的客户服务助理。\  请以友好和乐于助人的语气回答问题，并提供简洁明了的答案。\  请确保向用户提出相关的后续问题。  """  all_message = [  {'role': 'system', 'content': system_message}  ]  print("聊天机器人已启动！输入 'exit' 结束对话\n")  while True:  user_input = input("你: ")  if user_input.lower() == 'exit':  break  res = response(user_input,all_message)  if isinstance(res,str):  print(f"AI: {res}")  else:  all_message = res

📸 运行效果