企业客服每天面对海量咨询:产品问题、订单查询、技术支持、投诉处理……传统客服系统要么靠人工逐一处理(成本高、响应慢),要么用简单规则匹配(体验差、解决率低)。能否让AI像人类客服团队一样协作?——路由员分配任务、专家解决问题、经理处理升级,这正是Multi-Agent系统的用武之地。

一、概念

1.客服Multi-Agent架构模式

与上一篇的顺序协作不同,客服系统采用路由协作模式:


         用户输入 
            ↓ 
        路由Agent (Router) 
      ↙     ↓      ↘ 
   FAQ   知识库   人工转接 
  Agent   Agent    Agent 
 

角色分工:

Agent 职责 触发条件 输出
Router 意图识别、任务分发 所有请求 路由决策
FAQ Agent 处理常见问题 简单咨询 标准答案
Knowledge Agent 检索知识库 复杂问题 定制答案
Escalation Agent 人工转接 无法解决的问题 工单信息

2.实战

架构设计

技术栈:

  • LangGraph:编排Agent工作流(详细用法见系列一)
  • LangChain:LLM调用、RAG实现
  • Chainlit:快速搭建对话界面
  • Chroma:知识库向量存储

状态:


from typing import TypedDict, Literal 
 
class CustomerServiceState(TypedDict): 
    """客服系统的全局状态""" 
    user_query: str                    # 用户问题 
    intent: str                        # 意图类型:faq/knowledge/escalate 
    response: str                      # 最终回复 
    context: list[str]                 # 检索到的上下文 
    escalation_reason: str             # 转接原因 
 

工作流:


用户输入 → Router(意图分类) → 条件分支 → [FAQ/Knowledge/Escalation] → 输出结果 
 

3.实现路由Agent

路由Agent是整个系统的「调度中心」,负责快速判断用户意图。


from langchain_openai import ChatOpenAI 
from langchain.prompts import ChatPromptTemplate 
 
def router_agent(state: CustomerServiceState) -> CustomerServiceState: 
    """
    路由Agent:分析用户意图,决定分发给哪个专家Agent
    #...
    """ 
    query = state["user_query"] 
     
    # 构造意图分类Prompt 
    classification_prompt = ChatPromptTemplate.from_messages([ 
        ("system", """你是客服系统的路由助手。分析用户问题,判断意图类型。

意图类型定义:
- faq: 常见问题,如「怎么退货」「配送时间」
#...

"""), 
        ("user", "用户问题:{query}") 
    ]) 
     
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) 
    chain = classification_prompt | llm 
     
    # 调用LLM分类 
    result = chain.invoke({"query": query}) 
     
    # 解析结果 
    import json 
    intent_data = json.loads(result.content) 
     
    state["intent"] = intent_data["intent"] 
  
    return state 
 

关键设计:

  • 温度设为0:分类任务需要稳定性,避免随机性
  • 结构化输出:用JSON确保后续解析不出错
  • 置信度评估:可用于判断是否需要二次确认

4.实现FAQ Agent


def faq_agent(state: CustomerServiceState) -> CustomerServiceState: 
    """
    FAQ Agent:处理常见问题
    #...
    """ 
    query = state["user_query"] 
     
    # 简化示例:预定义FAQ库 
    faq_db = { 
        "怎么退货": "请在订单详情页点击「申请退货」,我们会在1-3个工作日内处理。", 
        #... 
    } 
     
    # 语义匹配(生产环境用向量检索) 
    from difflib import get_close_matches 
    matched = get_close_matches(query, faq_db.keys(), n=1, cutoff=0.6) 
     
    if matched: 
        state["response"] = faq_db[matched[0]] 
    else: 
        # 兜底:转到知识库Agent 
        state["intent"] = "knowledge" 
     
    return state 
 

优化方向:

  • 生产环境用向量检索替代字符串匹配,提升准确率
  • 引入点击率跟踪,自动优化FAQ库

5.实现Knowledge Agent

处理复杂问题,需要检索知识库生成答案。


from langchain_chroma import Chroma 
from langchain_openai import OpenAIEmbeddings 
from langchain.chains import RetrievalQA 
 
def knowledge_agent(state: CustomerServiceState) -> CustomerServiceState: 
    """
    Knowledge Agent:检索知识库,生成定制答案
# ....
    """ 
    query = state["user_query"] 
     
    # 初始化向量库(实际应用中应复用实例) 
    embeddings = OpenAIEmbeddings() 
    vectorstore = Chroma( 
        persist_directory="./customer_kb", 
        embedding_function=embeddings 
    ) 
     
    # 构建RAG链 
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3) 
    qa_chain = RetrievalQA.from_chain_type( 
        llm=llm, 
        retriever=vectorstore.as_retriever(search_kwargs={"k": 3}), 
        return_source_documents=True 
    ) 
     
    # 检索并生成答案 
    result = qa_chain.invoke({"query": query}) 
     
    state["response"] = result["result"] 
    state["context"] = [doc.page_content for doc in result["source_documents"]] 
    
    return state 
 

生产优化:

  • 缓存机制:相同问题直接返回缓存结果
  • 上下文压缩:对检索结果做重排序和去重
  • 答案验证:用另一个Agent检查答案的准确性

6.实现Escalation Agent

处理无法自动解决的问题,生成人工工单。


def escalation_agent(state: CustomerServiceState) -> CustomerServiceState: 
    """
    Escalation Agent:人工转接
    
    职责:
    1. 记录问题详情和处理历史
    2. 生成工单信息
    3. 告知用户预计响应时间
    """ 
    query = state["user_query"] 
     
    # 生成工单(实际应接工单系统API) 
    ticket_id = f"TK{hash(query) % 100000:05d}" 
     
    state["response"] = f"""感谢您的耐心,您的问题已转接人工客服。

工单号:{ticket_id}
#...
""" 
     
    state["escalation_reason"] = "超出自动处理范围" 
     
    print(f"🚨 [Escalation] 生成工单:{ticket_id}") 
     
    return state 
 

7.编排工作流


from langgraph.graph import StateGraph, END 
 
# 定义条件路由函数 
def route_by_intent(state: CustomerServiceState) -> str: 
    """根据意图决定下一个Agent""" 
    intent = state["intent"] 
    if intent == "faq": 
        return "faq" 
    elif intent == "knowledge": 
        return "knowledge" 
    else: 
        return "escalate" 
 
# 构建工作流 
workflow = StateGraph(CustomerServiceState) 
 
# 添加节点 
workflow.add_node("router", router_agent) 
workflow.add_node("faq", faq_agent) 
workflow.add_node("knowledge", knowledge_agent) 
workflow.add_node("escalate", escalation_agent) 
 
# 定义流程 
workflow.set_entry_point("router") 
workflow.add_conditional_edges( 
    "router", 
    route_by_intent, 
    { 
        "faq": "faq", 
        "knowledge": "knowledge", 
        "escalate": "escalate" 
    } 
) 
workflow.add_edge("faq", END) 
workflow.add_edge("knowledge", END) 
workflow.add_edge("escalate", END) 
 
# 编译应用 
app = workflow.compile() 
 

8.Chainlit界面集成


import chainlit as cl 
 
@cl.on_chat_start 
async def start(): 
    await cl.Message(content="您好!我是智能客服助手,有什么可以帮您?").send() 
 
@cl.on_message 
async def main(message: cl.Message): 
    # 调用Multi-Agent系统 
    result = app.invoke({ 
        "user_query": message.content, 
        "intent": "", 
        "response": "", 
        "context": [], 
        "escalation_reason": "" 
    }) 
     
    # 返回结果 
    response = result["response"] 
     
    # 如果是知识库回答,附加来源 
    if result["intent"] == "knowledge" and result["context"]: 
        response += "参考来源:" + "".join(result["context"][:2]) 
     
    await cl.Message(content=response).send() 
 

运行命令:


chainlit run customer_service.py -w 
 

二、总结

  1. 路由准确性 - 错误的路由会导致用户体验下降,需要持续优化意图识别模型
  2. 分层处理 - 80%问题用FAQ解决,只有20%调用大模型,大幅降低成本
  3. 知识库 - Knowledge Agent的效果取决于知识库的质量和覆盖度
  4. 兜底机制 - 无法处理的问题要有明确的转接流程,避免用户困在系统中
  5. 持续优化 - 记录所有对话,分析FAQ未匹配、Knowledge失败的case,不断优化

如何学习大模型 AI ?

由于新岗位的生产效率,要优于被取代岗位的生产效率,所以实际上整个社会的生产效率是提升的。

但是具体到个人,只能说是:

“最先掌握AI的人,将会比较晚掌握AI的人有竞争优势”。

这句话,放在计算机、互联网、移动互联网的开局时期,都是一样的道理。

我在一线互联网企业工作十余年里,指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家,也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑,所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限,很多互联网行业朋友无法获得正确的资料得到学习提升,故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

在这里插入图片描述

第一阶段(10天):初阶应用

该阶段让大家对大模型 AI有一个最前沿的认识,对大模型 AI 的理解超过 95% 的人,可以在相关讨论时发表高级、不跟风、又接地气的见解,别人只会和 AI 聊天,而你能调教 AI,并能用代码将大模型和业务衔接。

  • 大模型 AI 能干什么?
  • 大模型是怎样获得「智能」的?
  • 用好 AI 的核心心法
  • 大模型应用业务架构
  • 大模型应用技术架构
  • 代码示例:向 GPT-3.5 灌入新知识
  • 提示工程的意义和核心思想
  • Prompt 典型构成
  • 指令调优方法论
  • 思维链和思维树
  • Prompt 攻击和防范

第二阶段(30天):高阶应用

该阶段我们正式进入大模型 AI 进阶实战学习,学会构造私有知识库,扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架,抓住最新的技术进展,适合 Python 和 JavaScript 程序员。

  • 为什么要做 RAG
  • 搭建一个简单的 ChatPDF
  • 检索的基础概念
  • 什么是向量表示(Embeddings)
  • 向量数据库与向量检索
  • 基于向量检索的 RAG
  • 搭建 RAG 系统的扩展知识
  • 混合检索与 RAG-Fusion 简介
  • 向量模型本地部署

第三阶段(30天):模型训练

恭喜你,如果学到这里,你基本可以找到一份大模型 AI相关的工作,自己也能训练 GPT 了!通过微调,训练自己的垂直大模型,能独立训练开源多模态大模型,掌握更多技术方案。

到此为止,大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗?

  • 为什么要做 RAG
  • 什么是模型
  • 什么是模型训练
  • 求解器 & 损失函数简介
  • 小实验2:手写一个简单的神经网络并训练它
  • 什么是训练/预训练/微调/轻量化微调
  • Transformer结构简介
  • 轻量化微调
  • 实验数据集的构建

第四阶段(20天):商业闭环

对全球大模型从性能、吞吐量、成本等方面有一定的认知,可以在云端和本地等多种环境下部署大模型,找到适合自己的项目/创业方向,做一名被 AI 武装的产品经理。

  • 硬件选型
  • 带你了解全球大模型
  • 使用国产大模型服务
  • 搭建 OpenAI 代理
  • 热身:基于阿里云 PAI 部署 Stable Diffusion
  • 在本地计算机运行大模型
  • 大模型的私有化部署
  • 基于 vLLM 部署大模型
  • 案例:如何优雅地在阿里云私有部署开源大模型
  • 部署一套开源 LLM 项目
  • 内容安全
  • 互联网信息服务算法备案

学习是一个过程,只要学习就会有挑战。天道酬勤,你越努力,就会成为越优秀的自己。

如果你能在15天内完成所有的任务,那你堪称天才。然而,如果你能完成 60-70% 的内容,你就已经开始具备成为一名大模型 AI 的正确特征了。

这份完整版的大模型 AI 学习资料已经上传CSDN,朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费

在这里插入图片描述

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐