目录二、项目整体架构与技术栈详解2.1 整体架构分层2.2 完整技术栈清单2.3 配套配置文件说明三、核心模块功能拆解与代码逻辑解析3.1 环境加载与全局配置模块3.2 业务工具集Agent 可自主调用三大工具3.3 LangGraph ReAct 智能体核心逻辑3.4 FastAPI 后端接口与邮件发送底层实现3.5 Streamlit 前端交互页面四、项目部署完整操作流程4.1 环境依赖安装4.2 私密配置文件.env 填写4.3 启动服务顺序4.4 功能测试示例五、项目优势与落地价值5.1 技术层面优势5.2 业务落地价值六、项目优化扩展方向七、总结电商、线上服务行业的客服体系长期存在难以根治的短板。传统规则式客服机器人依赖固定关键词匹配只能应对标准化问题面对用户复杂、多轮、交叉咨询场景时极易失效人工客服人力成本高、响应延迟、对话记录留存与追溯困难用户索要聊天记录时人工导出、发送流程繁琐低效。大模型与智能 Agent 技术的成熟为客服系统重构提供了全新思路。区别于普通对话机器人基于 LangGraph 构建的 ReAct 智能客服具备自主工具调用、逻辑推理、多轮记忆、业务操作联动能力不再局限于问答可自主完成订单查询、库存核验、邮件归档全流程业务闭环。本文将完整拆解一套可本地部署、开箱即用的自主客服 Agent 项目技术栈融合 FastAPI 后端、Streamlit 可视化前端、LangGraph 智能体框架、硅基流动开源大模型 API、QQ 邮箱 SMTP 邮件服务实现「用户对话 - 自主工具调用 - 自动归档对话记录邮件」完整链路全部代码开源可复现适配中小企业轻量化客服场景。二、项目整体架构与技术栈详解2.1 整体架构分层项目分为三层架构解耦清晰便于独立迭代、部署扩展前端交互层Streamlit轻量化网页聊天界面无需前端开发基础快速搭建对话页面提供邮箱配置、对话展示、Agent 思考过程可视化、邮件发送状态提示功能本地运行即可访问网页客户端。后端服务层FastAPI提供高性能 HTTP 接口负责接收前端对话请求、维护用户独立对话记忆、驱动 LangGraph 智能体执行推理、封装 SMTP 邮件发送逻辑处理全局环境配置与跨域请求。智能体核心层LangGraph 硅基流动大模型项目核心基于 ReAct 范式构建自主 Agent内置订单查询、库存查询、邮件发送三大业务工具大模型负责自主判断是否调用工具、选择对应工具、组装参数LangGraph 管理 Agent 思考、行动、观察循环流程实现自主决策。2.2 完整技术栈清单后端核心fastapiuvicorn高性能异步接口框架低延迟支撑多用户并发对话智能体框架langchain、langgraph、langchain-openai标准化大模型调用、工具定义、Agent 状态流转大模型服务硅基流动 SiliconFlow API兼容 OpenAI 调用格式搭载 Qwen2.5-32B-Instruct 开源大模型无需本地部署大模型降低硬件门槛前端界面streamlitPython 快速 Web 开发工具零 HTML/CSS/JS 编写交互式网页环境与工具依赖python-dotenv读取.env 私密配置文件、requests前端请求后端接口、内置smtplib实现邮件发送基础库Python 内置 os、email、pydantic 用于配置读取、邮件封装、请求参数校验。2.3 配套配置文件说明项目根目录.env文件统一管理所有私密、可变配置分离代码与密钥避免硬编码泄露敏感信息配置项分为四类硅基流动大模型配置SILICONFLOW_API_KEY接口密钥、MODEL_NAME指定调用模型QQ 邮箱 SMTP 服务配置SMTP_SERVER服务器地址、SMTP_PORTSSL 端口 465、SENDER_EMAIL发件邮箱、SMTP_AUTH_CODE邮箱授权码非登录密码 所有配置项可按需切换例如邮箱可替换 163 邮箱模型可更换硅基流动平台其他开源大模型。三、核心模块功能拆解与代码逻辑解析3.1 环境加载与全局配置模块项目通过python-dotenv加载.env文件使用os.getenv()读取配置并设置默认兜底值防止配置缺失导致程序崩溃。所有全局变量统一声明集中管理接口地址、邮箱参数、模型名称后续修改业务参数无需遍历全代码维护成本极低。同时实现分用户对话记忆存储使用字典conversation_history以用户邮箱为唯一 Key隔离不同用户对话记录配套get_history、add_to_history读写记忆函数format_dialogue_history格式化对话文本自动过滤「发邮件、发送记录」等指令类消息仅保留真实业务问答保证邮件归档内容干净整洁。3.2 业务工具集Agent 可自主调用三大工具基于 LangChaintool装饰器封装标准化工具大模型可自主识别工具用途、判断调用时机、自动填充入参无需人工手动触发功能订单查询工具 check_order内置模拟订单数据库覆盖已发货、运输中、待签收、待发货、已取消 5 种主流订单状态输入订单号即可返回物流商、运单号、预计送达时间、订单详情。用户咨询物流、订单进度时Agent 会自主调用工具查询数据而非笼统回复话术。库存查询工具 check_inventory存储商品库存、所属仓库数据用户咨询商品是否有货、备货情况时自动调取库存数量与仓库信息支撑售前咨询场景。邮件发送工具 send_email作为触发标记工具仅用于告知大模型「具备邮件发送能力」。当用户提出「把聊天记录发我邮箱」「发送对话记录」等需求时Agent 会调用该工具后端捕获工具调用事件后执行底层 SMTP 邮件函数自动读取该用户全部历史对话格式化后发送至用户填写的邮箱。3.3 LangGraph ReAct 智能体核心逻辑后端代码:import os import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from typing import List, Dict, Any from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langgraph.prebuilt import create_react_agent from dotenv import load_dotenv load_dotenv() # 配置 SILICONFLOW_API_KEY os.getenv(SILICONFLOW_API_KEY, 你的API密钥) MODEL_NAME os.getenv(MODEL_NAME, Qwen/Qwen2.5-32B-Instruct) SMTP_SERVER os.getenv(SMTP_SERVER, smtp.qq.com) SMTP_PORT int(os.getenv(SMTP_PORT, 465)) SENDER_EMAIL os.getenv(SENDER_EMAIL, 你的QQ账号qq.com) SMTP_AUTH_CODE os.getenv(SMTP_AUTH_CODE, 你的邮箱授权码) # 对话历史存储按邮箱 conversation_history: Dict[str, List[Dict[str, str]]] {} def get_history(email: str) - List[Dict[str, str]]: if email not in conversation_history: conversation_history[email] [] return conversation_history[email] def add_to_history(email: str, role: str, content: str): history get_history(email) history.append({role: role, content: content}) def format_dialogue_history(email: str) - str: 格式化纯对话历史过滤掉发邮件指令 history get_history(email) if not history: return 暂无对话记录。 # 过滤掉包含发邮件相关关键词的消息指令本身不记录到邮件 keywords_to_filter [发邮件, 发送到我的邮箱, 发送到邮箱, 把对话过程, 把记录发送, 全部对话记录, 发送邮件, 发我邮箱] filtered [] for msg in history: content msg[content] if any(kw in content for kw in keywords_to_filter): continue filtered.append(msg) if not filtered: return 暂无有效对话记录已过滤发邮件指令。 text 客服对话记录 \n\n for i, msg in enumerate(filtered, 1): role 用户 if msg[role] user else 客服 text f{i}. {role}{msg[content]}\n\n return text # 工具定义 tool def check_order(order_id: str) - str: 查询订单状态 orders { 1001: { product_name: 手机, status: 已发货, carrier: 顺丰速运, tracking_no: SF1234567890, estimate_delivery: 2026-07-12, detail: 包裹已从广州发出预计7月12日送达 }, 1002: { product_name: 耳机, status: 运输中, carrier: 中通快递, tracking_no: ZT9876543210, estimate_delivery: 2026-07-10, detail: 包裹已到达郑州中转站预计7月10日送达 }, 1003: { product_name: 键盘, status: 已签收, carrier: 京东物流, tracking_no: JD0011223344, estimate_delivery: 2026-07-05, detail: 包裹已于7月5日签收签收人为小区物业 }, 1004: { product_name: 鼠标, status: 待发货, carrier: 待分配, tracking_no: 暂无, estimate_delivery: 2026-07-15, detail: 订单已支付等待仓库发货预计7月15日前发出 }, 1005: { product_name: 电脑, status: 已取消, carrier: 无, tracking_no: 无, estimate_delivery: 无, detail: 用户已取消订单退款已原路返回 } } order_info orders.get(order_id) if order_info: return (f订单 {order_id}{order_info[product_name]}\n f状态{order_info[status]}\n f物流公司{order_info[carrier]}\n f运单号{order_info[tracking_no]}\n f预计送达{order_info[estimate_delivery]}\n f详情{order_info[detail]}) else: return f订单号 {order_id} 不存在请核对后重新查询。 tool def check_inventory(product_id: str) - str: 查询产品库存 inventory { 1001: {product_name: 手机, stock: 156, warehouse: 广州仓}, 1002: {product_name: 耳机, stock: 34, warehouse: 郑州仓}, 1003: {product_name: 键盘, stock: 89, warehouse: 上海仓}, 1004: {product_name: 鼠标, stock: 213, warehouse: 深圳仓}, 1005: {product_name: 电脑, stock: 7, warehouse: 北京仓} } product_info inventory.get(product_id) if product_info: return (f产品 {product_id}{product_info[product_name]}\n f库存{product_info[stock]} 件\n f仓库{product_info[warehouse]}) else: return f产品编号 {product_id} 不存在请核对后重新查询。 tool def send_email(recipient: str, subject: str, content: str) - str: 发送邮件给用户。当用户要求发送聊天记录时subject 写 聊天记录content 可留空。 return f邮件已准备发送至 {recipient}主题{subject} tools [check_order, check_inventory, send_email] # 初始化 Agent llm ChatOpenAI( base_urlhttps://api.siliconflow.cn/v1, api_keySILICONFLOW_API_KEY, modelMODEL_NAME, temperature0.1 ) agent create_react_agent(llm, tools) # FastAPI 应用 app FastAPI(title智能客服 Agent) app.add_middleware( CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*], ) class ChatRequest(BaseModel): message: str user_email: str class ChatResponse(BaseModel): response: str decision_chain: List[Dict[str, Any]] email_sent: bool # 邮件发送函数 def send_email_to_user(recipient: str, subject: str, body: str) - bool: try: msg MIMEMultipart() msg[From] SENDER_EMAIL msg[To] recipient msg[Subject] subject msg.attach(MIMEText(body, plain, utf-8)) with smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT) as server: server.login(SENDER_EMAIL, SMTP_AUTH_CODE) server.sendmail(SENDER_EMAIL, [recipient], msg.as_string()) return True except Exception as e: print(f邮件发送失败: {e}) return False def format_decision_chain(chain: List[Dict[str, Any]]) - str: text 决策链 \n\n for i, step in enumerate(chain, 1): text fStep {i} - {step[type].upper()}:\n if step[type] action: text f 调用工具: {step[tool]}\n text f 输入: {step[input]}\n else: text f {step[content]}\n text \n return text # 执行 Agent def run_agent(user_input: str, user_email: str) - tuple[str, List[Dict[str, Any]], bool]: # 获取历史记录 history get_history(user_email) # 构建消息列表包含历史 当前用户输入 messages [] for msg in history: messages.append({role: msg[role], content: msg[content]}) messages.append({role: user, content: user_input}) decision_chain [] final_answer email_sent False for step in agent.stream({messages: messages}): if agent in step: msg step[agent][messages][-1] thought msg.content if hasattr(msg, content) else str(msg) decision_chain.append({type: thought, content: thought}) final_answer thought if tools in step: msg step[tools][messages][-1] if hasattr(msg, name) and msg.name: tool_name msg.name tool_input msg.content if hasattr(msg, content) else str(msg) decision_chain.append({ type: action, tool: tool_name, input: tool_input }) if tool_name send_email: # 判断是否要求发送“聊天记录/对话” if any(kw in tool_input for kw in [聊天记录, 对话, 全部, 记录]): dialogue_text format_dialogue_history(user_email) if 暂无有效对话记录 in dialogue_text: email_body 您还没有有效的业务对话记录发邮件指令已自动过滤。 else: email_body dialogue_text email_sent send_email_to_user(user_email, 客服聊天记录, email_body) else: # 只发送当前回答不含决策链 email_body f回答{final_answer} email_sent send_email_to_user(user_email, 客服回答, email_body) else: obs msg.content if hasattr(msg, content) else str(msg) decision_chain.append({type: observation, content: obs}) # 兜底如果最终回答提到发邮件但未触发工具或用户明确要求发记录 if not email_sent and any(kw in final_answer for kw in [发邮件, 发送邮件]): dialogue_text format_dialogue_history(user_email) if 暂无有效对话记录 not in dialogue_text: email_body dialogue_text else: email_body f回答{final_answer} email_sent send_email_to_user(user_email, 客服聊天记录, email_body) # 将本次对话存入历史 add_to_history(user_email, user, user_input) add_to_history(user_email, assistant, final_answer) return final_answer, decision_chain, email_sent # API 路由 app.post(/chat, response_modelChatResponse) async def chat_endpoint(req: ChatRequest): try: answer, chain, sent run_agent(req.message, req.user_email) return ChatResponse( responseanswer, decision_chainchain, email_sentsent ) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/health) async def health(): return {status: ok}ReAct 范式是自主 Agent 的核心设计思路循环执行「思考 Thought - 行动 Action - 观察 Observation」三步Thought思考大模型接收用户问题与历史对话自主判断是否需要调用工具无需工具则直接生成回答需要工具则规划调用工具名称、入参Action行动执行对应业务工具传入模型生成参数Observation观察接收工具返回结果将业务数据返回给大模型结合历史对话生成最终完整回复。项目使用create_react_agent快速构建标准 ReAct 智能体接入硅基流动兼容 OpenAI 格式的接口base_url填写硅基官方 v1 接口地址temperature0.1降低模型随机性保证订单、库存查询结果严谨、无幻觉。后端完整记录 Agent 每一步决策过程存储decision_chain决策链数组前端通过折叠面板可视化展示思考、工具调用、返回结果全流程方便调试、向用户展示客服推理逻辑提升交互透明度。3.4 FastAPI 后端接口与邮件发送底层实现接口设计仅暴露两个核心接口/chat对话交互接口、/health健康检测接口。/chat接收用户消息、用户邮箱两个参数调用run_agent函数驱动智能体执行完整流程返回客服回答、完整决策链、邮件是否发送成功状态全局开启 CORS 跨域支持本地 Streamlit 前端跨端口访问后端服务。SMTP 邮件底层封装send_email_to_user函数基于 SSL 加密方式连接 QQ 邮箱服务器封装邮件标题、正文、收发件人捕获全部异常邮件发送失败时打印日志并返回布尔状态。区分两种邮件场景用户索要对话记录时发送格式化完整历史问答单纯触发邮件工具时仅发送单轮客服回答。同时增加兜底逻辑若模型输出文字提及发邮件但未调用工具程序自动补发对话记录规避模型漏调用工具的边界问题。对话记忆持久化逻辑每一轮用户提问、客服回复完成后自动写入对应邮箱的独立记忆列表保证多轮对话上下文连贯记忆过滤机制剔除邮件指令避免归档内容混杂操作指令提升邮件可读性。3.5 Streamlit 前端交互页面前端代码import streamlit as st import requests import json # 页面配置 st.set_page_config(page_title智能客服 Agent, layoutwide) st.title( 自主学习客服 Agent) # 侧边栏 - 用户邮箱 with st.sidebar: st.header( 设置) user_email st.text_input(请输入您的邮箱接收对话记录, placeholderexampleqq.com) st.markdown(---) st.caption(Powered by 硅基流动 LangGraph) # 主聊天区域 if messages not in st.session_state: st.session_state.messages [] # 显示历史消息 for msg in st.session_state.messages: with st.chat_message(msg[role]): st.markdown(msg[content]) # 用户输入 if prompt : st.chat_input(请输入您的问题...): if not user_email: st.warning(请先在侧边栏输入您的邮箱) st.stop() # 显示用户消息 with st.chat_message(user): st.markdown(prompt) st.session_state.messages.append({role: user, content: prompt}) # 调用后端 API with st.chat_message(assistant): placeholder st.empty() with st.spinner(Agent 思考中...): try: response requests.post( http://localhost:8000/chat, json{message: prompt, user_email: user_email}, timeout60 ) if response.status_code 200: data response.json() answer data[response] chain data[decision_chain] email_sent data[email_sent] placeholder.markdown(answer) # 显示决策链可折叠 with st.expander( 查看 Agent 思考过程, expandedFalse): for step in chain: if step[type] thought: st.info(f **思考**{step[content]}) elif step[type] action: st.warning(f **调用工具**{step[tool]} → 输入{step[input]}) else: st.success(f **观察结果**{step[content]}) if email_sent: st.success(f✅ 对话记录已发送至 {user_email}) else: st.info(ℹ️ 未触发邮件发送如需发送请明确告诉客服) # 保存到会话 st.session_state.messages.append({role: assistant, content: answer}) else: placeholder.error(fAPI 请求失败{response.status_code} - {response.text}) except requests.exceptions.ConnectionError: placeholder.error(❌ 无法连接到后端服务请确保后端已启动端口8000) except Exception as e: placeholder.error(f❌ 发生错误{str(e)})前端轻量化无部署成本一行命令即可启动网页客户端功能覆盖侧边栏配置用户接收邮箱未填写邮箱时拦截对话请求会话缓存存储网页端对话展示记录实时渲染用户、客服对话气泡对话请求加载动画捕获后端连接失败、接口报错、超时等全部异常并友好提示可折叠面板展示 Agent 完整思考决策链区分思考、工具调用、观测结果三种状态用不同颜色标识邮件发送结果实时提示发送成功弹出绿色提示未触发邮件给出灰色说明。前后端完全解耦前端仅通过 HTTP 请求调用后端接口后续可替换 Vue、React 网页端、小程序客户端无需修改智能体核心逻辑。四、项目部署完整操作流程4.1 环境依赖安装新建requirements.txt文件写入项目全部依赖包plaintextfastapi uvicorn langchain langgraph langchain-openai streamlit requests python-dotenv执行安装命令bash运行pip install -r requirements.txt4.2 私密配置文件.env 填写硅基流动官网注册账号创建 API Key 填入SILICONFLOW_API_KEYQQ 邮箱开启 POP3/SMTP 服务获取 16 位授权码填入SMTP_AUTH_CODE填写发件 QQ 邮箱MODEL_NAME 保持Qwen/Qwen2.5-32B-Instruct可自行替换硅基流动平台其他模型。4.3 启动服务顺序启动 FastAPI 后端服务端口 8000bash运行uvicorn app:app --host 0.0.0.0 --port 8000新开终端启动 Streamlit 前端页面bash运行streamlit run streamlit_app.py浏览器自动打开前端页面侧边栏填写邮箱即可开始对话测试。4.4 功能测试示例业务查询输入「查询订单 1001 物流状态」Agent 自主调用订单工具返回顺丰物流、预计送达时间库存咨询输入「电脑产品库存还有多少」自动调取 product_id1005 库存数据邮件归档输入「把我们刚刚的聊天记录发送到我的邮箱」Agent 调用 send_email 工具后端自动格式化对话历史发送至侧边栏填写的邮箱页面弹出发送成功提示。五、项目优势与落地价值5.1 技术层面优势自主推理脱离固定规则区别于传统关键词机器人Agent 自主判断工具调用支持混合多业务查询例如「订单 1002 到哪了同时看看耳机库存」可连续调用多个工具串联多业务数据记忆隔离多用户并发适配以邮箱为唯一标识区分对话记忆支持多人同时访问前端对话互不干扰低部署门槛无需本地部署大模型依托硅基流动云端 API低配电脑即可运行前端 Streamlit 零前端开发成本快速上线安全规范密钥、邮箱授权码统一存放.env 文件不硬编码在代码中降低信息泄露风险SMTP 使用 SSL 加密发送邮件传输过程安全全链路可视化调试决策链完整记录 Agent 每一步思考与工具调用快速定位模型幻觉、工具调用失败问题便于迭代优化 Prompt 与工具逻辑。5.2 业务落地价值降低客服人力成本自主 Agent 承接订单查询、库存咨询、对话归档标准化需求减少人工客服重复工作量对话自动归档合规可追溯用户一键获取全部对话记录无需人工导出整理解决售后纠纷无凭证的痛点轻量化中小企业适配整套项目无复杂中间件、数据库依赖内存对话记忆适配小体量业务后续可扩展 SQLite 数据库持久化记忆适配高并发场景高度可扩展新增业务仅需新增tool工具函数例如退款申请、优惠券查询、物流改地址等功能无需重构对话底层逻辑。六、项目优化扩展方向当前版本为基础可用版本可基于业务需求迭代升级提供四大扩展思路持久化对话记忆目前对话仅存储内存服务重启后记录丢失可接入 SQLite、Redis 数据库绑定用户邮箱永久存储对话多模态能力扩展接入硅基流动多模态模型支持用户上传订单截图、商品图片自动识别图片信息完成查询邮件模板美化当前邮件为纯文本格式修改 MIMEText 为 HTML 格式设计美观对话记录邮件模板区分用户、客服消息样式权限与限流机制增加 API 密钥鉴权、接口请求限流防止恶意调用大模型接口产生高额费用知识库 RAG 融合接入 LangChain 向量数据库上传商品说明书、售后政策文档实现专业知识库问答解决大模型专业知识幻觉问题多渠道接入前端接口标准化可对接企业微信、小程序、抖音商家后台一套客服 Agent 支撑全渠道咨询。七、总结本文搭建的自主智能客服 Agent打通了「大模型自主推理 - 业务工具调用 - 对话记录邮件归档」完整业务闭环依托 LangGraph ReAct 框架实现真正具备自主决策能力的客服机器人摒弃传统固定匹配规则的局限性。整套代码基于 Python 生态开发前后端分离架构清晰配置简单、部署轻量化中小企业无需投入大量算法、前端开发资源即可落地使用。硅基流动提供的开源大模型 API 大幅降低了大模型落地硬件门槛搭配 FastAPI 高性能接口、Streamlit 快速可视化界面兼顾开发效率与用户交互体验内置的邮件自动归档功能直击客服行业对话追溯、凭证留存的核心痛点具备极强的实用落地价值。同时项目模块化设计预留充足扩展空间可根据电商、线上服务、售后咨询等不同行业业务需求持续迭代工具集与交互能力是轻量化大模型 Agent 落地业务场景的优质实践方案。前端页面收到的邮件