摘要

随着大语言模型技术落地产业，传统固定问答式客服存在拓展性差、无法对接真实业务数据的痛点。本文基于硅基流动大模型 API，自主实现一套具备思考 - 行动循环的智能客服 Agent 系统，支持三大核心工具调用：订单物流查询、商品库存统计、对话记录自动推送 QQ 邮箱。项目采用 Flask 搭建本地业务后端读取 CSV 订单数据集，Agent 依靠大模型自主识别用户口语化需求、分发对应工具执行，完整记录每一轮思考决策链并可视化输出，解决传统客服无法联动真实业务数据、交互生硬、会话记录无法自动归档的行业痛点。全文附带完整可运行代码、环境部署教程、功能测试效果与排错方案，适合大模型 Agent 入门、Python 后端开发、智能客服实训项目参考。 关键词：大模型 Agent；硅基流动；思考行动循环；工具调用；智能客服；Flask；Python；邮件自动归档

一、项目背景与需求分析

1.1 传统客服系统现存缺陷

当前电商平台客服分为两种模式，均存在明显短板： 第一类是规则关键词客服，依靠固定 if-else 匹配关键词，仅能应对预设标准提问，用户口语化、多样化提问极易匹配失败，新增业务功能需要大量修改规则代码，维护成本极高； 第二类是通用大模型对话客服，仅具备闲聊问答能力，无法对接企业真实订单、库存业务数据库，不能返回真实物流、库存数据，不具备业务落地能力； 同时两类客服均缺少会话自动归档能力，人工导出聊天记录效率低下，无法实现物流信息一键推送至客户邮箱。

1.2 项目核心需求

本项目依托大模型 Agent 的思考 - 行动循环架构，实现自主规划、工具调用、结果反馈全流程自动化，明确四大核心需求：

1. 自主意图识别：无需硬编码关键词，硅基大模型根据用户自然口语自主判断需求，区分查询订单、查询库存、发送邮件、闲聊引导四类场景；
2. 真实业务数据联动：搭建本地 Flask HTTP 后端，读取标准化订单 CSV 数据集，提供订单物流、商品库存查询接口，返回真实业务数据；
3. 多工具链式调用：内置三大可执行工具，模型自主选择工具、提取对应参数完成业务操作；
4. 全流程决策日志留存：记录每一轮对话的 AI 思考内容、执行工具、入参、返回结果，程序退出后完整可视化打印决策链；
5. 真实邮件自动推送：集成 yagmail 实现 QQ SMTP 邮件发送，一键将全量对话 + 物流记录归档发送至指定 QQ 邮箱；
6. 高容错稳定运行：解决 Windows 中文编码、大模型接口超时、模型输出 JSON 解析异常等常见问题，提供完整异常捕获机制。

1.3 整体架构设计

项目分为三层架构，分层解耦便于后期扩展商品新增、支付查询、售后工单等工具：

1. 大模型 Agent 交互层：核心调度层，负责接收用户输入、调用硅基流动 API、清洗模型输出 JSON、分发工具、记录决策日志、控制台对话交互；
2. 业务工具层：封装三大业务工具，包含订单查询、库存查询、QQ 邮件发送，统一入口分发执行；
3. 后端数据服务层：Flask 搭建本地 HTTP 接口，读取 orders.csv 订单数据集，提供带 Token 鉴权的查询接口，隔离数据层与 Agent 交互层。

数据流完整流程：用户输入口语提问 → Agent 转发至硅基大模型思考 → 模型输出标准化 JSON（思考文本 + 工具名称 + 参数）→ Agent 解析 JSON 并分发对应工具 → 工具调用后端接口获取真实数据 → 结果返回用户控制台 → 本轮完整流程存入决策日志。

二、项目环境与依赖部署

2.1 开发环境

• 操作系统：Windows 10/11
• Python 版本：3.10 及以上
• 大模型服务：硅基流动 DeepSeek-R1-0528-Qwen3-8B
• 数据存储：CSV 本地订单数据集（轻量化，无需部署数据库）

2.2 依赖库一键安装命令

打开 CMD 执行，一次性安装全部所需第三方库：

bash

运行

pip install flask pandas yagmail requests -i https://pypi.tuna.tsinghua.edu.cn/simple

各库作用说明：

1. flask：搭建本地业务 HTTP 后端接口；
2. pandas：读取、解析订单 CSV 数据集；
3. yagmail：封装 QQ 邮箱 SMTP 服务，实现邮件发送；
4. requests：发起硅基流动大模型 API 请求、本地后端接口请求。

2.3 前置资源准备

1. 硅基流动平台密钥：登录硅基流动官网，在 API 密钥页面复制 sk 开头密钥，填入代码配置区；
2. QQ 邮箱 SMTP 授权码：QQ 邮箱网页端 - 设置 - 账户，开启 POP3/SMTP 服务，短信验证获取 16 位授权码；
3. 订单数据集 orders.csv：标准化 7 字段订单文件，存放订单、商品、物流、客户信息。

三、项目完整代码实现

项目共三份核心文件，放置同一文件夹运行，分别为后端接口backend_api.py、智能客服 Agent 主程序smart_customer_agent.py、订单数据集orders.csv。

3.1 后端接口文件 backend_api.py

作用：提供带鉴权的订单、库存查询 HTTP 接口，读取本地 CSV 订单数据，隔离业务数据与对话层。

python

运行

from flask import Flask, request, jsonify
import pandas as pd

app = Flask(__name__)
# 接口统一鉴权令牌，主程序需保持完全一致
VALID_API_TOKEN = "shop20260705"
# 读取订单数据集，UTF-8编码避免中文乱码
df = pd.read_csv("./orders.csv", encoding="utf-8")

# Token鉴权校验函数
def check_token():
    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("Bearer "):
        return False
    token = auth_header.replace("Bearer ", "")
    return token == VALID_API_TOKEN

# 订单物流查询接口
@app.route("/api/order", methods=["GET"])
def query_order():
    if not check_token():
        return jsonify({"code": 401, "data": None, "msg": "接口鉴权失败，请核对Token"})
    order_id = request.args.get("order_no", "")
    target_data = df[df["order_id"] == order_id]
    if len(target_data) == 0:
        return jsonify({"code": 404, "data": None, "msg": "未查询到该订单信息"})
    res_data = target_data.iloc[0].to_dict()
    return jsonify({"code": 200, "data": res_data})

# 商品库存统计接口（同商品行数即为库存数量）
@app.route("/api/stock", methods=["GET"])
def query_stock():
    if not check_token():
        return jsonify({"code": 401, "data": None, "msg": "接口鉴权失败，请核对Token"})
    product_name = request.args.get("product_name", "")
    target_data = df[df["product_name"] == product_name]
    stock_num = len(target_data)
    return jsonify({"code": 200, "data": {"stock": stock_num}})

if __name__ == "__main__":
    app.run(host="127.0.0.1", port=8080, debug=True)

3.2 订单数据集 orders.csv

新建文本文件修改后缀为 csv，粘贴以下内容，UTF-8 编码保存：

csv

order_id,phone,product_name,product_id,shipping_status,logistics_no,receiver_email
01001,13800138000,无线鼠标,P001,已发货,SF20260705001,user1@qq.com
01002,13900139000,机械键盘,P002,待出库,YT20260705002,user2@163.com
01003,13700137000,电竞耳机,P003,运输中,ZTO20260705003,user3@gmail.com
01004,13600136000,27英寸显示器,P004,已签收,SF20260705004,user4@outlook.com
01005,13500135000,电脑音箱,P005,待付款,JD20260705005,user5@qq.com

3.3 智能客服 Agent 主程序 smart_customer_agent.py

核心代码，实现编码修复、大模型调用、JSON 清洗、工具分发、决策日志、邮件发送全功能，已将接口超时优化至 20 秒解决网络超时问题：

python

运行

# 全局UTF-8编码修复Windows系统latin-1中文报错
import sys
sys.stdout.reconfigure(encoding='utf-8')

import time
import yagmail
import requests
import json
import re
from typing import Dict, List, Any, Optional
from dataclasses import dataclass

# ====================== 全局配置区【必须自行修改为个人信息】 ======================
# 1.硅基流动大模型配置
SILICON_API_KEY = "sk-替换为你自己的硅基流动API密钥"
SILICON_API_URL = "https://api.siliconflow.cn/v1/chat/completions"
LLM_MODEL = "deepseek-ai/DeepSeek-R1-0528-Qwen3-8B"

# 2.后端接口配置，与backend_api.py内TOKEN保持一致
API_BASE_URL = "http://127.0.0.1:8080/api"
API_TOKEN = "shop20260705"
API_TIMEOUT = 5

# 3.QQ邮箱发送配置
SENDER_QQ_MAIL = "你自己的QQ账号@qq.com"    # 发件QQ邮箱
QQ_MAIL_AUTH_CODE = "替换为你的16位SMTP授权码" # QQ邮箱授权码
RECEIVE_MAIL = "你自己的QQ账号@qq.com"     # 接收归档邮件的邮箱
# ============================================================================

# 决策日志数据结构，存储每一轮对话完整链路
@dataclass
class DecisionRecord:
    round_id: int
    thought: str
    action: str
    params: Dict
    observation: str

class DecisionLogger:
    def __init__(self):
        self.records: List[DecisionRecord] = []

    def add_record(self, round_id: int, thought: str, action: str, params: Dict, observation: str):
        record = DecisionRecord(round_id, thought, action, params, observation)
        self.records.append(record)

    def show_all_log(self):
        """程序退出后可视化打印全部决策日志"""
        print("\n========== Agent完整思考-行动决策日志 ==========")
        for item in self.records:
            print(f"\n【第{item.round_id}轮对话】")
            print(f"AI思考内容：{item.thought}")
            print(f"执行工具动作：{item.action}")
            print(f"工具入参：{item.params}")
            print(f"工具返回结果：{item.observation}")
        print("===============================================\n")

# 后端HTTP接口请求封装类
class RealBusinessAPI:
    def __init__(self):
        self.headers = {
            "Authorization": f"Bearer {API_TOKEN}",
            "Content-Type": "application/json; charset=utf-8"
        }

    def get_order_info(self, order_no: str) -> Optional[Dict[str, Any]]:
        """查询订单完整物流信息"""
        try:
            resp = requests.get(
                url=f"{API_BASE_URL}/order",
                headers=self.headers,
                params={"order_no": order_no},
                timeout=API_TIMEOUT
            )
            resp.encoding = "utf-8"
            res_json = resp.json()
            if res_json["code"] == 200:
                return res_json["data"]
            return None
        except Exception as e:
            print(f"订单接口请求异常：{str(e)}")
            return None

    def get_product_stock(self, product_name: str) -> int:
        """查询商品库存数量"""
        try:
            resp = requests.get(
                url=f"{API_BASE_URL}/stock",
                headers=self.headers,
                params={"product_name": product_name},
                timeout=API_TIMEOUT
            )
            resp.encoding = "utf-8"
            res_json = resp.json()
            if res_json["code"] == 200:
                return res_json["data"]["stock"]
            return 0
        except Exception as e:
            print(f"库存接口请求异常：{str(e)}")
            return 0

# 业务工具封装层：订单查询、库存查询、QQ邮件发送
class ServiceTools:
    def __init__(self):
        self.api_client = RealBusinessAPI()
        # 初始化QQ邮箱SMTP客户端
        self.mail_client = yagmail.SMTP(
            user=SENDER_QQ_MAIL,
            password=QQ_MAIL_AUTH_CODE,
            host="smtp.qq.com"
        )

    def query_order(self, order_no: str) -> str:
        """物流查询工具"""
        time.sleep(0.2)
        order_data = self.api_client.get_order_info(order_no)
        if not order_data:
            return f"未查询到订单号 {order_no} 的物流信息（订单号为数字开头，直接输入即可）"
        info_text = (
            f"【订单号】{order_data['order_id']}\n"
            f"【商品名称】{order_data['product_name']}\n"
            f"【商品ID】{order_data['product_id']}\n"
            f"【收件手机号】{order_data['phone']}\n"
            f"【物流状态】{order_data['shipping_status']}\n"
            f"【物流单号】{order_data['logistics_no']}\n"
            f"【收件邮箱】{order_data['receiver_email']}"
        )
        return info_text

    def query_stock(self, product_name: str) -> str:
        """库存查询工具"""
        stock_num = self.api_client.get_product_stock(product_name)
        return f"商品【{product_name}】当前库存数量：{stock_num} 件"

    def send_mail_to_qq(self, receiver: str, mail_content: str) -> str:
        """发送对话记录至QQ邮箱"""
        try:
            self.mail_client.send(
                to=receiver,
                subject="【智能客服Agent】完整对话&物流记录归档",
                contents=mail_content
            )
            return "✅ 对话与物流信息已成功发送至QQ邮箱"
        except Exception as err:
            return f"❌ 邮件发送失败，错误详情：{str(err)}"

    def run_tool(self, tool_name: str, params: Dict[str, Any]) -> str:
        """工具统一分发入口"""
        if tool_name == "query_order":
            return self.query_order(params["order_no"])
        elif tool_name == "query_stock":
            return self.query_stock(params["product_name"])
        elif tool_name == "send_email":
            return self.send_mail_to_qq(params["receiver"], params["content"])
        else:
            return f"未知工具：{tool_name}"

# 核心Agent主体，实现思考-行动循环
class CustomerServiceAgent:
    def __init__(self):
        self.tools = ServiceTools()
        self.logger = DecisionLogger()
        self.round_count = 0
        self.dialogue_history = []
        # 请求头强制UTF-8编码，规避中文编码报错
        self.llm_header = {
            "Authorization": f"Bearer {SILICON_API_KEY}",
            "Content-Type": "application/json; charset=utf-8"
        }

    def clean_json_str(self, raw_text: str) -> str:
        """正则清洗模型输出，仅提取{}内标准JSON，过滤思考标签、换行、多余文字"""
        match = re.search(r"\{[\s\S]*\}", raw_text)
        if match:
            return match.group(0).strip()
        return raw_text

    def llm_thought_think(self, user_input: str) -> tuple[str, str, Dict]:
        """调用硅基流动大模型自主推理，输出标准化工具调用JSON"""
        self.round_count += 1
        # 系统提示词：约束模型仅输出JSON，定义工具规则
        system_prompt = """
你是电商智能客服大脑，根据用户自然语言自主判断需要执行的工具，仅输出标准JSON，禁止多余文字、思考标签。
可用工具列表：
1. query_order：查询订单完整物流信息，参数 {"order_no":"订单编号"}
2. query_stock：查询商品库存，参数 {"product_name":"商品名称"}
3. send_email：把全部对话+物流记录发送至邮箱，参数 {"receiver":"收件邮箱","content":"邮件正文"}
4. reply_direct：无法识别需求时直接回复用户，参数为空{}

输出固定JSON格式：
{
    "thought": "你的完整推理思考过程",
    "action": "工具名称",
    "params": {}
}
规则：
1. 用户询问订单、物流、发货状态 → action=query_order，提取订单号
2. 用户询问商品剩余库存 → action=query_stock，提取商品名称
3. 用户提出发邮件、保存聊天记录、归档记录 → action=send_email
4. 无关问题统一使用 reply_direct 输出引导话术
        """
        messages = [
            {"role": "system", "content": system_prompt.strip()},
            {"role": "user", "content": user_input}
        ]
        payload = {
            "model": LLM_MODEL,
            "messages": messages,
            "temperature": 0.1
        }
        try:
            # 超时优化至20秒，解决网络波动read timed out报错
            resp = requests.post(SILICON_API_URL, headers=self.llm_header, json=payload, timeout=20)
            resp.encoding = "utf-8"
            resp_json = resp.json()
            llm_output = resp_json["choices"][0]["message"]["content"].strip()
            # 清洗模型输出，剔除多余文本
            clean_str = self.clean_json_str(llm_output)
            # 安全解析JSON
            decision_data = json.loads(clean_str)
            thought = decision_data.get("thought", "")
            action = decision_data.get("action", "reply_direct")
            params = decision_data.get("params", {})
            return thought, action, params
        except json.JSONDecodeError as je:
            error_msg = f"模型输出JSON解析失败：原始输出={llm_output}，错误：{str(je)}"
            return error_msg, "reply_direct", {}
        except Exception as e:
            error_msg = f"硅基流动大模型调用失败：{str(e)}"
            return error_msg, "reply_direct", {}

    def run_single_chat(self, user_msg: str) -> str:
        """单轮对话完整思考-行动循环"""
        # 第一步：大模型思考决策
        thought, action, params = self.llm_thought_think(user_msg)
        # 第二步：执行对应工具
        if action == "reply_direct":
            if "大模型调用失败" in thought or "JSON解析失败" in thought:
                obs = thought
            else:
                obs = "你可以查询订单物流信息、商品库存，输入订单号即可查询物流，也可以发送全部对话记录至QQ邮箱"
        elif action == "send_email":
            # 拼接全部历史对话作为邮件正文
            mail_text = "===== 电商客服完整对话&物流记录 =====\n"
            for idx, item in enumerate(self.dialogue_history, 1):
                mail_text += f"\n【第{idx}轮】\n用户：{item['user']}\n客服：{item['bot']}\n"
            params["content"] = mail_text
            params["receiver"] = RECEIVE_MAIL
            obs = self.tools.run_tool(action, params)
        else:
            obs = self.tools.run_tool(action, params)
        # 第三步：保存本轮决策日志
        self.logger.add_record(self.round_count, thought, action, params, obs)
        # 第四步：存入对话历史，用于邮件归档
        self.dialogue_history.append({"user": user_msg, "bot": obs})
        return obs

    def start_chat_loop(self):
        """启动多轮交互式对话控制台"""
        print("===== 硅基流动大模型智能客服Agent已启动 =====")
        print("支持：查询订单物流、查询商品库存、发送全部记录至QQ邮箱")
        print("输入 quit 退出对话并打印完整AI决策日志\n")
        while True:
            user_input = input("用户：")
            if user_input.strip().lower() == "quit":
                print("\n对话结束，正在打印全部决策日志：")
                self.logger.show_all_log()
                break
            bot_reply = self.run_single_chat(user_input)
            print(f"客服：{bot_reply}\n")

if __name__ == "__main__":
    agent = CustomerServiceAgent()
    agent.start_chat_loop()

四、项目运行流程与功能测试

4.1 启动步骤

1. 三份文件放在同一文件夹，修改smart_customer_agent.py顶部配置：硅基密钥、QQ 邮箱授权码；
2. 打开第一个 CMD 窗口，进入项目目录，启动后端接口：

bash

运行

python backend_api.py

保持窗口常开，后端服务持续运行； 3. 新开第二个 CMD 窗口，先切换 UTF-8 编码避免中文乱码：

bash

运行

chcp 65001

1. 启动智能客服主程序：

bash

运行

python smart_customer_agent.py

控制台输出启动提示即可开始对话。

4.2 功能测试案例

测试 1：订单物流查询（调用 query_order 工具）

用户输入：帮我查 01001 订单的物流状态 程序自动提取订单号，调用后端接口返回完整商品、手机号、物流单号、物流状态信息。

测试 2：商品库存查询（调用 query_stock 工具）

用户输入：无线鼠标还有多少库存 模型提取商品名称，统计 CSV 内同款商品行数作为库存返回，截图实测返回 1 件。

测试 3：对话记录推送 QQ 邮箱（调用 send_email 工具）

用户输入：把本次所有聊天记录发送到我的邮箱 程序自动拼接全程所有问答，通过 QQ SMTP 发送完整归档邮件至配置邮箱。

测试 4：无关闲聊自动引导（reply_direct）

用户输入：桌面台灯的订单号是什么 模型识别无可用工具，输出标准化引导话术，不访问后端接口。

测试 5：退出打印完整决策链

输入quit，控制台可视化打印每一轮 AI 思考、工具、参数、返回结果，完整还原 Agent 思考 - 行动循环全过程。

示例：

4.3 项目核心运行效果截图解读

1. 库存查询：输入 “看看无线鼠标的库存还有多少”，客服返回库存 1 件，模型成功识别商品名、调用库存接口；
2. 无关提问：询问桌面台灯订单号，无有效工具可执行，输出引导提示，体现意图分类逻辑；
3. 异常容错：硅基接口网络波动出现超时，20 秒超时配置降低报错概率，重复提问即可正常请求；
4. 日志留存：退出后完整打印每一轮决策，直观展示 Agent 思考 - 行动闭环，适合课程实训展示。

五、项目关键技术难点与解决方案

5.1 Windows 中文 latin-1 编码报错

问题：Windows 默认编码 latin-1 无法解析中文，请求硅基接口抛出编码异常； 解决方案：程序最顶部添加sys.stdout.reconfigure(encoding='utf-8')，所有 requests 请求头强制携带charset=utf-8，接口返回手动指定编码，双重防护。

5.2 大模型输出夹带多余文本，JSON 解析崩溃

问题：DeepSeek-R1 模型会输出思考标签、换行、注释文本，直接 json.loads 解析报错； 解决方案：新增正则清洗函数clean_json_str，仅提取大模型输出中{}包裹的标准 JSON 片段，过滤无关字符，单独捕获 JSON 解析异常给出友好提示。

5.3 硅基流动 API 网络读取超时

问题：国内网络波动导致接口 12 秒超时抛出Read timed out； 解决方案：将 requests.post 超时参数从 12 调整至 20 秒，延长服务器响应等待时间，降低超时报错频率。

5.4 接口未授权数据泄露风险

问题：后端接口无鉴权，任意程序可访问订单、客户手机号隐私数据； 解决方案：后端增加全局 Token 鉴权，主程序与后端使用统一密钥，非法请求直接返回 401 拒绝访问，保护用户隐私信息。

5.5 会话记录无法自动归档

问题：传统客服需要人工复制聊天记录，效率低下； 解决方案：集成 yagmail 实现 QQ SMTP 邮件推送，自动拼接全轮对话内容，一键归档发送至邮箱，无需人工操作。

六、项目拓展优化方向

本项目基础架构具备极强拓展性，可基于现有代码快速新增业务工具，拓展方向如下：

1. 新增售后工单工具：新增售后 CSV 数据集，编写售后查询接口，在提示词新增售后工具定义，模型可自主识别 “退货、售后、维修” 类提问；
2. 替换数据库存储：将本地 CSV 改为 MySQL/PostgreSQL 数据库，适配海量订单数据，替换 pandas 读取逻辑为 SQL 查询；
3. 多轮记忆优化：增加对话窗口记忆，支持跨多轮上下文关联提问，例如 “这个订单物流更新了吗” 自动识别上一轮订单号；
4. 可视化 Web 前端：基于 Flask 搭建简单网页端，替代 CMD 控制台，实现网页对话、日志在线查看；
5. 多邮箱兼容：拓展 163、126、企业邮箱发送功能，适配不同客户邮箱需求；
6. 重试机制：给硅基接口请求增加自动重试逻辑，超时自动重发请求，进一步降低人工重复提问操作。

七、项目总结

本文完整实现一套基于硅基流动大模型的电商智能客服 Agent，彻底摆脱传统规则式客服的局限性，依靠大模型自主完成需求理解、工具选择、参数提取、业务查询、结果反馈全流程思考 - 行动循环。项目轻量化部署，无需复杂数据库，仅依靠 CSV 存储订单数据，配套完整异常处理、决策日志、邮件自动归档功能，代码分层清晰、注释完善，适合 Python 后端、大模型 Agent 初学者、高校人工智能实训课程作为完整实战项目。

项目所有代码开箱即用，仅需替换硅基流动 API 密钥与 QQ 邮箱授权码即可快速运行，同时针对 Windows 平台常见编码、网络超时、JSON 解析三大类高频报错提供针对性修复方案，运行稳定、容错性强，具备真实电商业务落地参考价值。通过本项目可深入理解大模型工具调用、Agent 循环架构、前后端接口交互、邮件自动化等核心技术，打通大模型从 API 调用到产业业务落地的完整开发链路。