智能客户服务协助智能体

卓恒前方汽车销售服务有限公司

超越传统客服机器人。智能体可以深度查询知识库、调用订单系统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] # 可选用户ID
message: 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_id
user_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 uvicorn
uvicorn.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-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["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?"}'

来源：https://www.cnblogs.com/yjbjingcha/p/19082153