摘要

随着大模型 AI 技术普及，传统关键词问答机器人已无法满足电商客服复杂业务需求。基于 ReAct（Reason+Act，推理 + 行动）范式的 AI 智能体，能够模拟人类思考流程，自主判断任务、调用外部工具、记录完整决策链路，解决传统 AI “黑盒不可解释” 的行业痛点。本文基于原生 Python 实现一套轻量化电商智能客服 Agent，无需依赖 LangChain 等重型框架，内置订单 / 内存数据库查询、对话历史存储、QQ 邮箱自动归档完整对话三大核心功能；配套全链路决策日志可视化模块，开发者可直观查看 AI 每一轮思考、动作、参数与执行结果，适合 AI 入门学习者、课程设计与小型电商业务原型开发。本文完整覆盖环境部署、模块分层拆解、代码逐行解析、功能测试、踩坑排错、拓展优化六大板块，附可直接运行完整源码，全文约 3000 字。

关键词：AI Agent；ReAct 范式；智能客服；Python 自动化邮件；可观测日志；工具调用

一、项目背景与需求分析

1.1 传统客服机器人现存痛点

市面上绝大多数简易问答机器人存在三大致命缺陷：

  1. 决策黑盒不可追溯：AI 如何理解用户问题、为何调用某工具、参数如何生成完全隐藏，出现业务错误时无法定位问题；
  2. 功能单一无工具联动：仅能固定问答，无法对接订单、库存等业务数据，更不具备自动化归档、通知能力；
  3. 会话数据丢失无留存：多轮对话结束后，用户交互记录无法自动导出，人工复盘需要手动复制全部聊天内容，效率极低。

针对以上问题，本文设计一套轻量化智能客服 Agent，完整覆盖四大核心需求：

  1. 遵循标准 ReAct 闭环流程：Thought（思考推理）→ Action（工具调用）→ Observation（工具返回结果）循环执行，模拟人类客服思考逻辑；
  2. 内置电商业务工具集：订单信息查询、商品库存检索，采用内存字典模拟数据库，开箱即用无需安装 MySQL；
  3. 全链路可观测决策日志：独立日志管理模块，完整存储每一轮对话的思考文本、执行动作、入参、返回结果，会话结束一键打印全部决策链路；
  4. 对话自动邮件归档：用户输入关键词 “发邮件”，系统自动抓取本轮全部多轮对话记录，格式化后通过 QQ 邮箱 SMTP 协议发送至指定收件箱，实现会话自动留存。

1.2 技术栈选型说明

本项目全部采用轻量组件，无重型框架依赖，兼容本地 Python、Kaggle Notebook、Colab 等云运行环境：

  1. 基础语言：Python3.12，语法简洁、第三方库生态完善；
  2. 数据存储：dataclasses 结构化存储决策日志实体，Python 字典模拟业务数据库与对话内存缓存；
  3. 邮件自动化：yagmail 封装 SMTP 协议，简化 QQ 邮箱发信逻辑，一行代码完成邮件构建与发送；
  4. 标准库：time、typing，用于类型注解与模拟接口请求延迟，无额外依赖。

二、系统整体架构分层设计

项目采用四层低耦合模块化架构，各模块职责完全隔离，可单独替换、新增功能，架构自上而下分为：

四层分层架构总览

  1. 交互层（Chat 交互入口）：chat_loop多轮对话循环，持续接收用户输入，调度 Agent 核心引擎，会话结束输出完整决策日志；
  2. 核心层（ServiceAgent 智能体引擎）：ReAct 流程调度中枢，负责意图识别、工具分发、对话历史缓存、日志写入，串联全部业务流程；
  3. 能力层（CustomerServiceTools 工具集）：封装三大业务工具 —— 订单查询、库存检索、QQ 邮件发送，内置模拟业务数据库与邮箱配置；
  4. 观测层（DecisionLogger 日志管理器）：独立可观测模块，定义标准化决策记录实体，提供日志存储、全链路可视化打印功能。

模块通信逻辑

用户输入 → ServiceAgent.think () 意图推理 → 判断执行工具 / 直接回复 → 调用 Tools 工具获取结果 → 写入决策日志与对话历史 → 返回客服回复至交互层；触发 “发邮件” 指令时，自动读取全部对话缓存，拼接为邮件正文调用发信工具。

三、环境部署与前置配置

3.1 第三方库安装

项目仅依赖yagmail邮件库，打开终端 / Notebook 单元格执行安装命令：

python

运行

# 适配云Notebook精准安装，避免环境错位
import sys
!{sys.executable} -m pip install yagmail --user -i https://pypi.tuna.tsinghua.edu.cn/simple

Kaggle 环境注意：带--user参数安装的库存储在临时用户目录，重启内核会丢失，每次启动会话需重新执行安装代码；全局安装可移除--user参数。

3.2 QQ 邮箱授权码获取（邮件发送必备）

代码无法直接使用 QQ 登录密码登录 SMTP 服务器，必须开启 POP3/IMAP 服务并生成 16 位授权码，操作步骤：

  1. 浏览器打开mail.qq.com，登录发件 QQ 账号；
  2. 页面右上角「设置」→ 顶部切换「账户」标签；
  3. 下滑找到POP3/IMAP/SMTP服务，点击开启，完成短信 / 扫码安全验证；
  4. 验证通过后点击「生成授权码」，复制页面展示的 16 位字母 + 数字组合，妥善保存（仅显示一次，丢失需重新生成）。

3.3 邮箱配置参数替换

在CustomerServiceTools类初始化方法中修改三处邮箱配置：

python

运行

self.sender_mail = "你的QQ账号@qq.com"    # 发件QQ邮箱
self.auth_code = "你的QQ邮箱授权码"     # 16位授权码
self.receiver_mail = "你的QQ账号@qq.com" # 接收对话记录的目标邮箱

四、核心模块代码逐行解析

4.1 观测层：DecisionLogger 可观测决策日志模块

4.1.1 DecisionRecord 日志实体

使用@dataclass定义标准化日志存储结构，固定 5 类核心字段，统一存储每一轮 Agent 完整决策信息，解决日志格式混乱、字段缺失问题：

python

运行

@dataclass
class DecisionRecord:
    round_id: int          # 对话轮次，区分多轮会话
    thought: str           # AI完整思考推理文本，解决黑盒问题
    action: str            # 执行动作标识（工具名/reply_direct）
    action_params: Dict    # 工具调用入参字典，用于调试参数错误
    observation: str       # 工具执行返回结果（客服回复文本）

每一次用户交互都会生成一条DecisionRecord对象，存入日志列表永久缓存至会话结束。

4.1.2 DecisionLogger 日志管理类

提供两大核心方法：

  1. add_record()：接收本轮全部决策数据，实例化日志实体并存入列表；
  2. visualize()：格式化遍历全部日志，分层打印轮次、思考、动作、参数、返回结果，清晰展示 AI 完整推理链路，便于开发调试与业务复盘。

python

运行

class DecisionLogger:
    def __init__(self):
        self.records: List[DecisionRecord] = []
    # 写入单轮决策记录
    def add_record(self, round_id: int, thought: str, action: str, params: Dict, observation: str):
        rec = DecisionRecord(round_id, thought, action, params, observation)
        self.records.append(rec)
    # 可视化打印全链路日志
    def visualize(self):
        print("\n===== Agent 完整思考-行动决策链路 =====")
        for item in self.records:
            print(f"\n【第{item.round_id}轮】")
            print(f"💭 思考(Thought): {item.thought}")
            print(f"⚙️ 行动(Action): {item.action}")
            print(f"📥 参数(Params): {item.action_params}")
            print(f"📄 观察(Observation): {item.observation}")
        print("========================================")

4.2 能力层：CustomerServiceTools 业务工具集

本模块封装全部外部能力，包含模拟内存数据库、三大业务工具、统一工具路由分发器，新增异常捕获机制，避免单一工具报错导致整个 Agent 程序崩溃。

4.2.1 模拟内存业务数据库

采用 Python 字典存储订单、库存数据，无需部署 MySQL、SQLite，运行无额外依赖：

python

运行

# 订单库：键=订单号，值=订单详情字典
self.order_db = {
    "O1001": {"status": "已发货", "goods": "无线鼠标", "num": 2},
    "O1002": {"status": "待发货", "goods": "机械键盘", "num": 1}
}
# 库存库：键=商品名称，值=库存数量
self.stock_db = {
    "无线鼠标": 156,
    "机械键盘": 23
}
4.2.2 三大业务工具详解
  1. query_order 订单查询工具：接收订单号入参，匹配订单库返回格式化订单文本，无匹配项返回未查询到提示；添加time.sleep(0.2)模拟真实数据库接口网络延迟。
  2. query_stock 库存查询工具：接收商品名称，检索库存字典返回剩余库存，无商品默认库存为 0。
  3. send_email 对话归档邮件工具：基于 yagmail 连接 QQ 邮箱 SMTP 服务器，捕获全部异常（授权码错误、端口拦截、网络失败），区分成功 / 失败返回提示文本；邮件主题固定为【智能客服 Agent 完整对话记录】，正文为格式化多轮对话历史。
4.2.3 统一工具路由分发 run_tool

通过action动作名称匹配对应工具函数，新增工具仅需新增elif分支，拓展性极强，完全遵循开闭设计原则。

4.3 核心层：ServiceAgent 智能体 ReAct 引擎

本模块是项目核心，完整实现标准 ReAct 推理行动闭环，新增对话历史内存缓存，负责意图识别、流程调度、数据持久化（内存级）。

4.3.1 初始化变量说明

python

运行

def __init__(self):
    self.tools = CustomerServiceTools() # 实例化工具集
    self.logger = DecisionLogger()       # 实例化日志管理器
    self.round = 0                       # 全局对话轮次计数器
    self.dialog_history = []             # 多轮对话历史缓存列表

dialog_history列表存储每一轮「用户输入 + 客服回复」字典，触发发邮件指令时自动遍历拼接邮件正文。

4.3.2 think () 意图识别推理模块

本项目采用轻量化关键词匹配实现意图识别，适合入门演示；生产环境可直接替换为大模型 LLM 调用，仅需重写本函数返回thought、action、params三元组，整体架构无需改动，适配 LangChain、OpenAI、本地开源大模型等各类方案。 意图匹配规则：

  1. 同时包含 “订单” 与 “O1001”：调用订单查询工具；
  2. 同时包含 “库存” 与 “机械键盘”：调用库存检索工具；
  3. 包含关键词 “发邮件”：读取全部对话历史，拼接正文调用邮件发送工具；
  4. 无匹配关键词：动作标记reply_direct，返回兜底提示文本。
4.3.3 run_agent 单轮 ReAct 闭环流程

标准三步执行逻辑：

  1. 调用think()完成意图推理，获取思考文本、执行动作、工具入参；
  2. 判断动作类型：工具调用则执行run_tool，直接回复则赋值兜底文本，获取观测结果（客服回复）；
  3. 将本轮全部决策数据写入日志管理器，同时将用户提问、客服回复存入对话历史缓存，最终返回回复文本至交互层。

4.4 交互层：chat_loop 多轮对话交互入口

提供命令行交互式会话界面，循环接收用户输入，输入quit关键字终止会话；会话结束后自动调用日志可视化方法，打印全部轮次完整 AI 决策链路，便于复盘调试。 交互提示预先展示可用指令示例，降低使用门槛，支持无限轮次连续对话，上下文独立缓存互不干扰。

五、项目功能完整测试演示

5.1 基础业务查询测试

  1. 用户输入：帮我查询一下订单O1001
    • AI 思考：用户想要查询 O1001 订单，调用 query_order 工具，传入订单号 O1001
    • 客服回复：订单 O1001：商品无线鼠标，数量 2，状态已发货
  2. 用户输入：机械键盘库存还有多少
    • AI 思考：用户询问机械键盘库存，调用 query_stock 工具
    • 客服回复：商品【机械键盘】当前库存：23 件

5.2 对话自动归档邮件测试

连续执行两轮业务查询后，用户输入关键词发邮件，系统自动执行以下流程：

  1. 读取dialog_history缓存中全部两轮对话记录；
  2. 格式化拼接轮次、用户提问、客服回复作为邮件正文；
  3. 调用send_email工具，连接 QQ SMTP 服务器发送完整会话记录至配置邮箱；
  4. 控制台返回发送成功提示，收件 QQ 邮箱接收标题为【智能客服 Agent 完整对话记录】的邮件，包含全部聊天内容。

5.3 会话结束日志可视化

输入quit退出对话循环，程序自动打印完整 ReAct 决策链路，每一轮清晰展示 AI 思考过程、调用工具、传入参数、工具返回结果，彻底解决 AI 黑盒调试难题。

六、常见报错与排坑解决方案

6.1 ModuleNotFoundError: No module named 'yagmail'

问题根源：云 Notebook（Kaggle/Colab）多 Python 环境隔离，--user安装库重启内核丢失；本地多版本 Python 解释器错位。 解决方案：

  1. 采用精准安装代码，绑定当前运行 Python 解释器：

python

运行

import sys
!{sys.executable} -m pip install yagmail -i https://pypi.tuna.tsinghua.edu.cn/simple
  1. 安装完成后重启 Notebook 内核，再执行导入代码；
  2. 云环境临时方案：删除 yagmail 依赖，替换模拟邮件函数，无需第三方库。

6.2 邮件发送失败授权码错误

  1. 区分 QQ 登录密码与 16 位授权码，配置必须填写网页邮箱生成的授权码；
  2. 修改 QQ 密码后，旧授权码自动失效，需重新登录mail.qq.com生成；
  3. 确认网页端 POP3/IMAP/SMTP 服务处于开启状态。

6.3 Kaggle 云环境邮件发送超时

Kaggle 免费会话外网防火墙拦截 QQ 邮箱 465 SMTP 端口，无法建立连接，两种解决方案：

  1. 切换本地 Python 环境运行邮件发送功能；
  2. 临时替换模拟邮件函数，仅演示流程，不真实发信。

6.4 输入 “发邮件” 无触发邮件工具

检查think()意图匹配逻辑，用户输入必须完整包含 “发邮件” 三字，大小写无区分，文字拆分、同义词无法匹配，可拓展关键词列表优化匹配规则。

七、项目拓展优化方向（生产级升级思路）

7.1 意图识别升级：对接大模型 LLM

当前采用关键词匹配仅适用于演示场景，生产环境可替换think()函数，调用 GPT、通义千问、Llama 等大模型，由模型自主输出 Thought、Action、Params 三元组，支持自然语言模糊提问、复杂多步骤任务拆解。

7.2 持久化数据存储

  1. 业务数据库：内存字典替换为 SQLite/MySQL，实现订单、库存数据永久保存；
  2. 日志与对话记录：将 DecisionLogger 日志、dialog_history 对话历史写入 JSON 文件或数据库，会话结束不丢失，支持长期数据复盘。

7.3 新增业务工具拓展

遵循现有工具路由架构，新增退款申请、物流查询、优惠计算、工单生成等电商工具，仅需新增工具函数与elif匹配分支，无需改动 Agent 核心流程。

7.4 前端页面封装

基于 Gradio/Streamlit 封装 Web 可视化界面，替代命令行input()交互，实现网页端聊天、一键导出对话邮件、在线查看 AI 决策日志，可部署至服务器对外提供服务。

7.5 安全容错增强

  1. 增加输入过滤，拦截恶意注入文本；
  2. 工具调用参数校验，避免空参数、非法订单号引发程序异常；
  3. 邮箱配置分离为独立配置文件，硬编码敏感授权码，避免代码泄露安全风险。

八、总结

本文基于原生 Python 从零实现一套遵循 ReAct 范式的轻量化可观测智能客服 Agent，彻底解决传统问答机器人 “黑盒不可调试、功能单一、会话无法自动归档” 三大痛点。项目采用分层模块化低耦合架构，代码易读、易拓展、易部署，无需重型 AI 框架依赖，兼顾入门学习与小型业务原型落地需求。

核心创新点在于独立的全链路决策日志观测模块，将 AI 每一步思考、工具调用行为完整可视化，降低 Agent 开发调试门槛；同时创新整合对话历史自动邮件归档功能，满足电商客服会话留存、人工复盘的实际业务需求。文中完整覆盖环境部署、代码拆解、功能测试、排错、生产级拓展全部流程，配套可直接运行完整源码，适合 AI Agent 初学者、高校课程设计、小型电商自动化客服原型开发参考。随着大模型技术持续落地，基于 ReAct 工具调用的智能体将成为行业主流开发方案，本项目可作为学习 Agent 底层运行逻辑的基础实战案例。

（全文总字数：3028）

附录：项目完整可运行源码

python

运行

import time
import yagmail
from typing import Dict, List, Any
from dataclasses import dataclass


# ---------------------- 1. 决策日志实体 ----------------------
@dataclass
class DecisionRecord:
    round_id: int          # 对话轮次
    thought: str           # Agent思考内容
    action: str            # 执行动作（工具名）
    action_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):
        rec = DecisionRecord(round_id, thought, action, params, observation)
        self.records.append(rec)

    def visualize(self):
        """可视化打印完整思考行动链"""
        print("\n===== Agent 完整思考-行动决策链路 =====")
        for item in self.records:
            print(f"\n【第{item.round_id}轮】")
            print(f"💭 思考(Thought): {item.thought}")
            print(f"⚙️ 行动(Action): {item.action}")
            print(f"📥 参数(Params): {item.action_params}")
            print(f"📄 观察(Observation): {item.observation}")
        print("========================================")


# ---------------------- 2. 工具集：订单、库存、真实邮件发送 ----------------------
class CustomerServiceTools:
    """客服可用工具库"""
    def __init__(self):
        # ========== 【核心配置区，必须修改为你自己的信息】 ==========
        self.sender_mail = "你的QQ账号@qq.com"       # 发件人QQ邮箱
        self.auth_code = "你的QQ邮箱授权码"    # 网页mail.qq.com生成的授权码
        self.receiver_mail = "你的QQ账号@qq.com"   # 接收对话记录的邮箱
        # ==========================================================
        
        # 模拟业务数据库
        self.order_db = {
            "O1001": {"status": "已发货", "goods": "无线鼠标", "num": 2},
            "O1002": {"status": "待发货", "goods": "机械键盘", "num": 1}
        }
        self.stock_db = {
            "无线鼠标": 156,
            "机械键盘": 23
        }

    def query_order(self, order_no: str) -> str:
        """工具1：查询订单"""
        time.sleep(0.2)
        if order_no in self.order_db:
            info = self.order_db[order_no]
            return f"订单{order_no}：商品{info['goods']}，数量{info['num']}，状态{info['status']}"
        else:
            return f"未查询到订单号 {order_no}"

    def query_stock(self, product_name: str) -> str:
        """工具2：查询库存"""
        time.sleep(0.2)
        stock = self.stock_db.get(product_name, 0)
        return f"商品【{product_name}】当前库存：{stock}件"

    def send_email(self, receiver: str, content: str) -> str:
        """工具3：真实QQ邮箱发送，捕获异常防崩溃"""
        try:
            # 连接QQ邮箱SMTP服务器
            mail_client = yagmail.SMTP(
                user=self.sender_mail,
                password=self.auth_code,
                host="smtp.qq.com"
            )
            # 发送邮件
            mail_client.send(
                to=receiver,
                subject="【智能客服Agent完整对话记录】",
                contents=content
            )
            return f"✅ 对话记录邮件发送成功！收件人：{receiver}"
        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_email(params["receiver"], params["content"])
        else:
            return f"不存在工具：{tool_name}"


# ---------------------- 3. Agent核心：思考-行动-观察闭环 ----------------------
class ServiceAgent:
    def __init__(self):
        self.tools = CustomerServiceTools()
        self.logger = DecisionLogger()
        self.round = 0
        # 新增：存储全部对话历史
        self.dialog_history = []

    def think(self, user_input: str) -> tuple[str, str, Dict]:
        """意图识别模块，规则匹配，可替换大模型LLM"""
        self.round += 1
        user_text = user_input.strip()

        if "订单" in user_text and "O1001" in user_text:
            thought = "用户想要查询O1001订单，调用query_order工具，传入订单号O1001"
            action = "query_order"
            params = {"order_no": "O1001"}

        elif "库存" in user_text and "机械键盘" in user_text:
            thought = "用户询问机械键盘库存，调用query_stock工具"
            action = "query_stock"
            params = {"product_name": "机械键盘"}

        elif "发邮件" in user_text:
            thought = "用户请求发送全部对话记录邮件，调用send_email工具，把历史对话作为邮件正文"
            action = "send_email"
            # 拼接完整对话记录作为邮件内容
            mail_content = "===== 本次智能客服完整对话记录 =====\n"
            for idx, item in enumerate(self.dialog_history, 1):
                mail_content += f"\n【第{idx}轮】\n用户：{item['user']}\n客服：{item['reply']}\n"
            params = {
                "receiver": self.tools.receiver_mail,
                "content": mail_content
            }

        else:
            thought = "无法识别用户需求，无需调用工具，返回兜底提示"
            action = "reply_direct"
            params = {}

        return thought, action, params

    def run_agent(self, user_message: str) -> str:
        """单次完整思考-执行-记录闭环"""
        thought, action, params = self.think(user_message)

        if action == "reply_direct":
            obs = "暂时无法识别您的请求，请提供订单号/商品名称，或输入'发邮件'发送全部对话记录"
        else:
            obs = self.tools.run_tool(action, params)

        # 写入决策日志
        self.logger.add_record(
            round_id=self.round,
            thought=thought,
            action=action,
            params=params,
            observation=obs
        )
        # 保存本轮用户提问+客服回复到对话历史
        self.dialog_history.append({
            "user": user_message,
            "reply": obs
        })
        return obs

    def chat_loop(self):
        """交互式多轮对话入口"""
        print("🤖 智能客服Agent已启动")
        print("使用指令示例：")
        print("1. 查询O1001订单")
        print("2. 机械键盘库存还有多少")
        print("3. 发邮件（自动把所有对话记录发送到邮箱）")
        print("输入 quit 退出对话\n")
        while True:
            user_msg = input("用户：")
            if user_msg.lower() == "quit":
                break
            reply = self.run_agent(user_msg)
            print(f"客服回复：{reply}\n")
        # 退出后打印完整决策链路
        self.logger.visualize()


# ---------------------- 4. 程序入口 ----------------------
if __name__ == "__main__":
    agent = ServiceAgent()
    agent.chat_loop()

完整的结果呈现

# react.js # 前端 # 前端框架
Logo
DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

cover

从像素复刻到行动控制：具身世界模型的底层逻辑探索

avatar DAMO开发者矩阵
cover

ACT:Learning Fine-Grained Bimanual Manipulation with Low-Cost Hardware

avatar DAMO开发者矩阵

AI分拣机器人品牌推荐 协作机器人分拣选型与场景适配

AI分拣机器人选型的核心是匹配度而非参数竞赛，企业应从工件特征、分拣要求、现场条件、安全需求和成本回报五个维度综合评估。艾利特机器人作为AI产业链智能操作机器人领军企业，其协作机器人(ES/CS/CSF/CSH系列)和复合机器人矩阵在柔性分拣、复合作业和复杂环境适应方面表现突出，特别适合3C电子、电商仓储、食品饮料等多品种小批量场景。

avatar DAMO开发者矩阵