超越传统客服机器人。智能体可以深度查询知识库、调用订单系统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?"}'