当前位置： 首页 > news >正文

智能客户服务支持智能体

news 2025/9/7 5:21:07

超越传统客服机器人。智能体可以深度查询知识库、调用订单系统API、甚至根据客户情绪灵活处理退货、退款、升级投诉等复杂流程。

案例：

客户说：“我上周买的鞋子尺码不对，想换货但是找不到订单页面了。”

智能体行动： ① 通过用户身份验证；② 调用订单查询API找到该订单；③ 检查换货政策；④ 生成换货链接并指导用户下一步操作；⑤ 若流程中断，后续可主动发送邮件提醒。

1. 核心功能

  • 多轮对话与上下文理解：能理解并记忆对话上下文。
  • 工具调用（Tool Use）：能调用外部API获取真实数据（如订单、用户信息）。
  • 知识库检索（RAG）：能从企业知识库（如手册、FAQ）中检索精准信息。
  • 安全性与权限控制：验证用户身份，并基于身份控制数据访问范围
  • 人工接管（Human-in-the-loop）：在无法处理或用户要求时，无缝转接人工客服

2. 技术栈选择（企业级）

  • LLM（大脑）： OpenAI GPT-4 Turbo (性价比和性能平衡)

  • 开发框架： LangChain / LangGraph (用于构建复杂、有状态的智能体工作流)

  • 工具调用： LangChain Tools & Custom Functions

  • 知识库： Chroma (向量数据库，轻量且高效) + OpenAI Embeddings

  • 后端API： FastAPI (高性能、异步支持好，适合生产环境)

  • 身份认证： JWT Tokens

  • 数据持久化： SQL Database (PostgreSQL) for logging

  • 部署： Docker & Kubernetes (容器化便于扩展和管理)

架构设计
在这里插入图片描述
在这里插入图片描述

一、环境依赖

# 创建项目目录并初始化虚拟环境
mkdir customer-support-agent
cd customer-support-agent
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate  # Windows# 安装核心依赖
pip install openai langchain langgraph chromadb langchain-openai fastapi uvicorn python-jose[cryptography] passlib sqlalchemy psycopg2-binary pydantic

二、数据模型

models.py

  • UserIdentity - 用户身份标识，跟踪用户身份和会话信息
  • AgentState - 智能体状态管理，管理对话智能体的完整状态，这是LangGraph框架中的核心概念
  • ConversationType - 对话类型枚举，定义对话的意图分类，限制可能的对话类型，便于处理后续的统计分析和逻辑分支
  • ConversationLog - 对话日志存储：用于持久化存储对话记录到数据库
from pydantic import BaseModel, Field #Pydantic: 提供数据验证和设置管理的现代Python数据验证库
from typing import Optional, List, Dict, Any #typing: 用于类型注解，提高代码可读性和类型安全
from enum import Enum #enum: 用于定义枚举类型，限制可能的取值class UserIdentity(BaseModel):"""用户身份标识"""user_id: Optional[str] = None #user_id: 可选字段，用于已登录用户session_id: str #session_id: 必需字段，唯一标识一个对话会话is_authenticated: bool = False #is_authenticated: 标识用户是否通过身份验证class AgentState(BaseModel):"""LangGraph智能体的状态"""#一个列表，列表中的每个元素都是字典，使用Field类来定义字段的详细配置，...表示该字段是必需的，不能为空messages: List[Dict[str, Any]] = Field(..., description="对话消息历史")user_identity: UserIdentity = Field(..., description="用户身份信息")current_step: str = Field("greeting", description="当前对话步骤")#"greeting": 默认值，表示如果没有提供值，就使用这个默认值#default_factory=dict: 使用工厂函数，每次创建新实例时都会调用dict()创建一个新的空字典extracted_info: Dict[str, Any] = Field(default_factory=dict, description="从对话中提取的结构化信息")class ConversationType(Enum):QUERY_ORDER = "query_order"  # 订单查询RETURN_EXCHANGE = "return_exchange"  # 退换货GENERAL_QUESTION = "general_question" # 一般问题class ConversationLog(BaseModel):"""存入数据库的对话日志"""session_id: str  # 会话标识user_id: Optional[str] # 可选用户IDmessage: str # 用户消息内容agent_response: str # 智能体回复内容intent: Optional[str] # 识别出的意图timestamp: str # 时间戳success: bool  # 处理是否成功

画外题：

#创建实例
state = AgentState(messages=[{"role": "user", "content": "你好"}],user_identity=UserIdentity(...),  # 需要提供UserIdentity实例# current_step 使用默认值 "greeting"# extracted_info 使用默认空字典
)# 访问属性
print(state.current_step)  # 输出: "greeting"
print(state.extracted_info)  # 输出: {}

在这里插入图片描述

三、核心工具

tools.py

  • 订单查询工具→ AI调用query_order_tool
  • 退换货政策查询工具→ AI调用query_return_policy_tool
  • 复杂问题无法解决 → AI调用create_support_ticket_tool
import requests
from langchain.tools import tool #LangChain的工具装饰器，将函数转换为AI可调用的工具
from typing import Type
from pydantic import BaseModel, Field
import os# 1. 订单查询工具
class OrderQueryInput(BaseModel):order_id: str = Field(..., description="The order ID to query")#装饰器是一种高级功能，它允许在不修改原函数代码的情况下，为函数添加额外的功能。
#将一个普通的 Python 函数"包装"成一个可以被 AI 大语言模型（LLM）识别和调用的工具（Tool）
#(args_schema=OrderQueryInput)：这是传递给 @tool 装饰器的参数，定义了工具输入参数的名称、类型、描述等，帮助 AI 模型理解如何正确地调用这个函数。
@tool(args_schema=OrderQueryInput)
def query_order_tool(order_id: str) -> str: # -> str:：返回类型提示，表明这个函数执行完毕后会返回一个字符串（String）类型的结果"""查询用户订单信息。需要先验证用户身份。"""# 模拟调用内部订单系统API# 真实环境中，这里会是 requests.get(f"{ORDER_API_URL}/{order_id}", headers=...)print(f"🔍 [Tool Call] Querying order: {order_id}")# 模拟响应 创建字典 构建了一个订单对象的详细信息，包括产品、尺寸、日期、状态mock_order_data = {"order_id": order_id,"product": "Running Shoes (Model X)","size": "42","order_date": "2024-09-15","status": "Delivered","customer_id": "cust_12345"}return f"Order Details: {str(mock_order_data)}"# 2. 退换货政策查询工具 (RAG)
@tool # @tool 装饰器告诉AI框架：“这个函数是我（AI）可以用的一个工具”。
def query_return_policy_tool(product_category: str) -> str:"""根据产品类别查询退换货政策。如果用户未指定类别，则查询通用政策。"""# 在真实环境中，这里会从向量数据库检索print(f"🔍 [Tool Call] Querying return policy for: {product_category}")policies = {"general": "You can return most items within 30 days of delivery. Items must be unworn and in original packaging.","shoes": "Shoes can be exchanged for a different size within 45 days. Must have original box and no signs of wear.","electronics": "Electronics can be returned within 14 days. Must be factory reset and all accessories included."}policy = policies.get(product_category.lower(), policies["general"])return f"Our return policy for {product_category}: {policy}"# 3. 创建客服工单工具
class CreateTicketInput(BaseModel):issue_summary: str = Field(..., description="A summary of the customer's issue")priority: str = Field("medium", description="Priority of the ticket: low, medium, high")@tool(args_schema=CreateTicketInput)
def create_support_ticket_tool(issue_summary: str, priority: str = "medium") -> str:"""在第三方系统（如Zendesk、Jira）中创建支持工单。用于当智能体无法解决问题时。"""print(f"🔍 [Tool Call] Creating Support Ticket. Priority: {priority}. Issue: {issue_summary}")# 模拟创建工单的API调用ticket_id = "TICKET-0987"return f"Successfully created a support ticket for you. Your ticket ID is {ticket_id}. A human agent will contact you shortly."

生产环境改进建议

def query_order_tool(order_id:str) -> str:try:response = requests.get(r"{os.getenv('ORDER_API_URL')}/{order_id}",headers = {"Authorization": f"Bearer {os.getenv('API_TOKEN')}"},timeout = 10)return f"Order Details:{response.json()}"except requests.RequestException as e:return f"Sorry, I couldn't retrieve your order details. Error: {str(e)}"

四、构建智能体工作流

        +-------------+|   agent     |  # 决策节点+-------------+|v (条件路由)+---------+---------+|         |         |v         v         v
+-------+   END   +-------------+
| tools |         | human_agent |
+-------+         +-------------+|                   |+--------+          |v          v+-------------+|    agent    |  # 循环回到agent+-------------+

agent.py

  • 智能体决策节点，接收当前对话状态，调用LLM会自动决定是否需要调用工具，返回AI消息到状态中
  • 路由函数 - 核心决策逻辑
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_openai import ChatOpenAI
from models import AgentState
from tools import query_order_tool, query_return_policy_tool, create_support_ticket_tool
from typing import Literal, Dict, Any# 初始化LLM
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0) #模型选择: gpt-4-turbo 提供强大的推理能力 温度设置: temperature=0 确保确定性输出，适合客服场景
# 绑定工具 工具绑定: bind_tools() 让LLM知道可用的工具及其功能
llm_with_tools = llm.bind_tools([query_order_tool, query_return_policy_tool, create_support_ticket_tool])# 定义工具节点 ToolNode 专门处理工具调用的执行
tool_node = ToolNode(tools=[query_order_tool, query_return_policy_tool, create_support_ticket_tool])def agent_step(state: AgentState):#接收当前对话状态"""智能体决策节点"""print(f"🤖 [Agent Step] Current Step: {state['current_step']}")messages = state["messages"]response = llm_with_tools.invoke(messages) #调用LLM生成响应，LLM会自动决定是否需要调用工具return {"messages": [response]}# 返回AI消息到状态中def route_to_tools(state: AgentState) -> Literal["tools", "end", "human_agent"]:"""路由函数，决定下一步是调用工具、结束还是转人工"""ai_msg = state["messages"][-1]#检查工具调用: 如果AI消息包含tool_calls，转到"tools"节点if not hasattr(ai_msg, 'tool_calls') or len(ai_msg.tool_calls) == 0:# 如果没有工具调用，检查是否需要结束或转人工if "thank you" in state["messages"][-2].content.lower():#结束条件: 用户说"thank you"时结束对话return "end"if "human" in state["messages"][-2].content.lower():return "human_agent"return "end"return "tools"def call_human_agent(state: AgentState):"""调用创建工单工具并结束对话"""issue_summary = f"Customer requested human agent. Conversation history: {state['messages']}"tool_input = {"issue_summary": issue_summary, "priority": "medium"}result = create_support_ticket_tool.invoke(tool_input)#返回友好的转接消息return {"messages": [AIMessage(content=f"I've escalated your issue to our human team. {result}")]}# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_step)
workflow.add_node("tools", tool_node)
workflow.add_node("human_agent", call_human_agent)workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent",route_to_tools,{"tools": "tools","end": END,"human_agent": "human_agent"}
)
workflow.add_edge("tools", "agent")
workflow.add_edge("human_agent", END)# 编译图
app = workflow.compile()

实际对话示例：

用户: "我想查询订单12345"
AI: (决定调用query_order_tool) → 工具节点 → 返回结果
用户: "谢谢你的帮助"
AI: (检测到感谢) → 结束对话用户: "我要找人工客服"  
AI: (检测到"human") → 转人工节点 → 创建工单 → 结束

五、创建FastAPI后端与安全中间件

from fastapi import FastAPI, HTTPException, Depends, Header
from fastapi.middleware.cors import CORSMiddleware
from models import UserIdentity, AgentState, ConversationLog
from agent import app as agent_app
from typing import Annotated, Optional
import uuid
import datetime
import json# 初始化FastAPI应用
app = FastAPI(title="Customer Support Agent API")# 中间件：CORS（允许前端访问）
app.add_middleware(CORSMiddleware,allow_origins=["*"],  # 生产环境应指定具体域名allow_credentials=True,allow_methods=["*"],allow_headers=["*"],
)# 模拟用户身份验证（生产环境应使用JWT）
async def verify_token(authorization: Annotated[Optional[str], Header()] = None) -> UserIdentity:session_id = str(uuid.uuid4())if authorization and authorization.startswith("Bearer "):token = authorization[7:]# 这里应验证JWT token并提取user_id# 为简单起见，我们模拟一个已验证用户user_identity = UserIdentity(user_id="cust_12345", session_id=session_id, is_authenticated=True)else:# 未认证用户只有session_iduser_identity = UserIdentity(session_id=session_id, is_authenticated=False)return user_identity# API路由
@app.post("/chat")
async def chat_endpoint(message: str,user_identity: UserIdentity = Depends(verify_token)
):"""主聊天端点"""try:# 1. 初始化或获取对话状态（这里应使用Redis等持久化状态，为简单起见我们每次新建）initial_state = AgentState(messages=[HumanMessage(content=message)],user_identity=user_identity,current_step="greeting")# 2. 执行智能体图final_state = agent_app.invoke(initial_state)# 3. 获取最终响应agent_response = final_state["messages"][-1].content# 4. 记录日志（应异步写入数据库）log_entry = ConversationLog(session_id=user_identity.session_id,user_id=user_identity.user_id,message=message,agent_response=agent_response,intent=final_state.get("current_step"),timestamp=datetime.datetime.utcnow().isoformat(),success=True)print(f"📝 Logging conversation: {log_entry.json()}")# 5. 返回响应return {"response": agent_response,"session_id": user_identity.session_id,"user_id": user_identity.user_id}except Exception as e:print(f"❌ Error in chat endpoint: {e}")raise HTTPException(status_code=500, detail=str(e))@app.get("/health")
async def health_check():return {"status": "OK"}if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)

六、测试智能体

import asyncio
from models import AgentState, UserIdentity
from agent import app# 模拟一个测试用户
test_user = UserIdentity(user_id="cust_12345", session_id="test_session_001", is_authenticated=True)def test_conversation():# 模拟用户消息：询问订单状态test_messages = ["Hi, I want to check the status of my order ORD-67890"]state = AgentState(messages=[], user_identity=test_user, current_step="start")for msg in test_messages:state["messages"].append(HumanMessage(content=msg))print(f"User: {msg}")# 调用智能体state = app.invoke(state)agent_msg = state["messages"][-1]print(f"Agent: {agent_msg.content}")if hasattr(agent_msg, 'tool_calls') and agent_msg.tool_calls:print(f"Agent called tools: {agent_msg.tool_calls}")print("---")if __name__ == "__main__":test_conversation()

七、使用Docker容器化部署

FROM python:3.11-slimWORKDIR /appCOPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txtCOPY . .EXPOSE 8000CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]

docker-compose.yml

version: '3.8'
services:support-agent:build: .ports:- "8000:8000"environment:- OPENAI_API_KEY=${OPENAI_API_KEY}# 生产环境应添加以下依赖# depends_on:#   - postgres#   - redis# postgres:#   image: postgres:13#   environment:#     POSTGRES_DB: agent_db#     POSTGRES_USER: agent#     POSTGRES_PASSWORD: password# redis:#   image: redis:7-alpine

如何运行

设置环境变量：

export OPENAI_API_KEY='your-openai-api-key'

启动服务

uvicorn main:app --reload

测试API

curl -X 'POST' \'http://localhost:8000/chat' \-H 'Authorization: Bearer fake_jwt_token_for_testing' \-H 'Content-Type: application/json' \-d '{"message": "I need to return my shoes, what is your policy?"}'

在这里插入图片描述


文章转载自：

http://Bf48LeGi.xjpnq.cn
http://3z26FILh.xjpnq.cn
http://Pkc9MuKo.xjpnq.cn
http://ZzFElLas.xjpnq.cn
http://DDoVt5fu.xjpnq.cn
http://uOUOY6hp.xjpnq.cn
http://uaF5H0be.xjpnq.cn
http://2zgGFWCH.xjpnq.cn
http://WogP61j3.xjpnq.cn
http://piFf7yxY.xjpnq.cn
http://5x88nxdB.xjpnq.cn
http://6tVYc45h.xjpnq.cn
http://THIbODk1.xjpnq.cn
http://GFLJEIUA.xjpnq.cn
http://l64U0ZF3.xjpnq.cn
http://Md2WQ1SL.xjpnq.cn
http://ebtjnliq.xjpnq.cn
http://YyScmNhu.xjpnq.cn
http://hr2O0cpk.xjpnq.cn
http://wRZIxd8R.xjpnq.cn
http://U9DJDHb5.xjpnq.cn
http://Zq6j1yOE.xjpnq.cn
http://Wmo0nLxo.xjpnq.cn
http://F8eY3Q1f.xjpnq.cn
http://LdEP3DUN.xjpnq.cn
http://L8wjRHLF.xjpnq.cn
http://Fm3bcKcB.xjpnq.cn
http://0OQyxyGD.xjpnq.cn
http://acRNPrsB.xjpnq.cn
http://cfxgFDcR.xjpnq.cn
http://www.dtcms.com/a/369829.html

相关文章：

  • Gutenberg块编辑器：WordPress 2025高效内容开发指南
  • JUC、JVM八股补充
  • windows找不到gpedit.msc（本地组策略编辑器）
  • 【洛谷】队列相关经典算法题详解：模板队列、机器翻译、海港
  • 激光频率梳 3D 轮廓测量 - 油路板的凹槽深度和平面度测量
  • 24.线程概念和控制（一）
  • Altium Designer（AD24）切换工作界面为浅灰色的方法
  • 让字符串变成回文串的最少插入次数-二维dp
  • 零基础入门深度学习：从理论到实战，GitHub+开源资源全指南（2025最新版）
  • 从文本到知识：使用LLM图转换器构建知识图谱的详细指南
  • 【开题答辩全过程】以 停车场管理系统的设计与实现为例，包含答辩的问题和答案
  • 带fat32文件系统的bin二进制文件制作教程
  • 【Redis】缓存的穿透、击穿和雪崩
  • C++经典的数据结构与算法之经典算法思想：分治法（Divide and Conquer）
  • PDF教程｜如何把想要的网页保存下来？
  • DevOps实战(2) - 使用Arbess+GitPuk+Docker实现Java项目自动化部署
  • Git reset 回退版本
  • PostgreSQL与Greenplum数据库的编程语言连接
  • git在Linux中的使用
  • 全面剖析TENGJUN防水TYPE-C板上双排贴（L7.55/舌片外露1.1/双耳带螺孔）：认证、防水与结构设计的三重突破
  • fastapi通过sqlmodel连接Mysql实现crud功能
  • 百度竞价推广：搜索竞价信息流推广代运营
  • Go基础（④指针）
  • 【开题答辩全过程】以 基于JSP的高校健康体育活动管理系统的设计与实现为例，包含答辩的问题和答案
  • 贪心算法应用：基因编辑靶点选择问题详解
  • webrtc弱网-LossBasedBandwidthEstimation类源码分析与算法原理
  • 01-线上问题处理-树形结构拼接
  • uniapp | 解决组件样式不生效问题
  • 尚硅谷宋红康JVM全套教程（详解java虚拟机）
  • uniapp基础组件概述