LangGraph教程（四）：智能客服项目示例（流程+完整代码）

一、项目背景与目标

在传统客服场景中，人工客服需要处理大量重复性问题，效率低下且成本高昂。基于LangGraph的智能客服系统通过状态机管理对话流程，结合大语言模型（LLM）实现意图识别、信息检索和响应生成，能够显著提升服务效率。本示例将实现一个支持多轮对话的智能客服，具备以下功能：

意图分类（如查询订单、投诉建议）
实体抽取（订单号、用户ID等）
工具调用（数据库查询、API调用）
上下文管理（多轮对话状态跟踪）

二、核心流程设计

1. 状态机架构

采用LangGraph的StateGraph构建有限状态机，定义5个核心状态：

from langgraph.prebuilt import StateGraph
class CustomerServiceStates:
    INIT = "init"
    INTENT_RECOGNITION = "intent_recognition"
    ENTITY_EXTRACTION = "entity_extraction"
    TOOL_INVOCATION = "tool_invocation"
    RESPONSE_GENERATION = "response_generation"

2. 状态转移逻辑

当前状态	触发条件	下一状态	数据流
INIT	用户输入	INTENT_RECOGNITION	原始文本
INTENT_RECOGNITION	意图分类完成	ENTITY_EXTRACTION	意图标签
ENTITY_EXTRACTION	实体抽取完成	TOOL_INVOCATION	结构化参数
TOOL_INVOCATION	工具返回结果	RESPONSE_GENERATION	工具输出
RESPONSE_GENERATION	响应生成完成	INIT	客服回复

3. 工具集成设计

定义3类工具接口：

class CustomerServiceTools:
    def query_order(self, order_id: str) -> dict:
        """模拟订单查询"""
        return {"status": "shipped", "tracking": "123456"}
    def log_complaint(self, user_id: str, content: str) -> str:
        """记录投诉"""
        return f"Complaint from {user_id} logged"
    def fetch_faq(self, query: str) -> str:
        """知识库检索"""
        return "Standard FAQ response"

三、完整代码实现

1. 环境准备

pip install langgraph openai python-dotenv

2. 主程序实现

import os
from dotenv import load_dotenv
from langgraph.prebuilt import StateGraph
from langchain_community.llms import OpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StructuredOutputParser
load_dotenv()
class CustomerServiceAgent:
    def __init__(self):
        self.llm = OpenAI(temperature=0.3)
        self.tools = CustomerServiceTools()
        self.graph = self._build_graph()
    def _build_graph(self):
        graph = StateGraph(CustomerServiceStates.INIT)
        # 意图识别节点
        graph.add_node(
            CustomerServiceStates.INTENT_RECOGNITION,
            self._recognize_intent
        )
        # 实体抽取节点
        graph.add_node(
            CustomerServiceStates.ENTITY_EXTRACTION,
            self._extract_entities
        )
        # 工具调用节点
        graph.add_node(
            CustomerServiceStates.TOOL_INVOCATION,
            self._invoke_tool
        )
        # 响应生成节点
        graph.add_node(
            CustomerServiceStates.RESPONSE_GENERATION,
            self._generate_response
        )
        # 定义状态转移
        graph.set_edge(
            CustomerServiceStates.INIT,
            CustomerServiceStates.INTENT_RECOGNITION,
            condition=lambda x: True
        )
        graph.set_edge(
            CustomerServiceStates.INTENT_RECOGNITION,
            CustomerServiceStates.ENTITY_EXTRACTION,
            condition=lambda x: x.get("intent") in ["query_order", "complaint"]
        )
        # ...其他边定义
        return graph
    def _recognize_intent(self, input_text: str):
        prompt = ChatPromptTemplate.from_template("""
        用户说: {input_text}
        请分类意图，选项: 查询订单/投诉建议/一般咨询
        """)
        message = prompt.format_message(input_text=input_text)
        response = self.llm.invoke([message])
        return {"intent": response.content.strip()}
    def _extract_entities(self, state: dict):
        intent = state["intent"]
        input_text = state["input_text"]
        if intent == "query_order":
            parser = StructuredOutputParser.from_response_schema(
                {"type": "object", "properties": {"order_id": {"type": "string"}}}
            )
            prompt = ChatPromptTemplate.from_template("""
            从以下文本提取订单号: {input_text}
            输出格式: {{"order_id": "XXX"}}
            """)
            # ...实现实体抽取逻辑
            return {"entities": {"order_id": "123"}}  # 简化示例
        # ...其他意图处理
    def run(self, input_text: str):
        state = {"input_text": input_text}
        while True:
            current_state = state.get("current_state", CustomerServiceStates.INIT)
            if current_state == CustomerServiceStates.RESPONSE_GENERATION:
                response = self._generate_response(state)
                return response
            # ...状态机执行逻辑

3. 工具类实现

class CustomerServiceTools:
    def query_order(self, order_id: str) -> dict:
        # 实际应连接数据库
        return {
            "status": "delivered",
            "items": [{"name": "Product A", "qty": 1}],
            "tracking": f"TRACK{order_id}"
        }
    def log_complaint(self, user_id: str, content: str) -> str:
        # 实际应写入投诉系统
        return f"Complaint #{len(self.complaints)+1} logged: {content[:50]}..."
    def fetch_faq(self, query: str) -> str:
        # 实际应查询知识库
        faq_db = {
            "return policy": "30天无理由退换",
            "shipping fee": "满99元免运费"
        }
        return faq_db.get(query.lower(), "未找到相关答案")

四、关键技术点解析

1. 上下文管理

使用状态字典维护对话上下文：

class DialogState:
    def __init__(self):
        self.history = []
        self.current_intent = None
        self.entities = {}
    def update(self, new_state: dict):
        self.history.append(new_state)
        if "intent" in new_state:
            self.current_intent = new_state["intent"]
        if "entities" in new_state:
            self.entities.update(new_state["entities"])

2. 错误处理机制

实现三级容错：

输入校验层：过滤无效字符
模型输出校验：验证JSON结构
工具调用重试：3次重试机制

3. 性能优化

缓存常用FAQ响应
异步工具调用
意图分类模型微调

五、部署与扩展建议

1. 部署方案

graph LR
    A[用户请求] --> B[API网关]
    B --> C[负载均衡器]
    C --> D[客服实例集群]
    D --> E[Redis缓存]
    D --> F[PostgreSQL数据库]

2. 扩展方向

多语言支持：添加语言检测模块
情感分析：集成情绪识别模型
主动学习：收集难例优化模型

六、完整运行示例

if __name__ == "__main__":
    agent = CustomerServiceAgent()
    # 测试用例1：订单查询
    print("测试订单查询:")
    response = agent.run("我的订单12345到哪里了？")
    print(response)
    # 测试用例2：投诉处理
    print("\n测试投诉处理:")
    response = agent.run("我要投诉，物流太慢了！")
    print(response)

七、最佳实践总结

状态设计原则：每个状态应对应明确的业务动作
工具调用规范：
- 工具接口应保持无状态
- 输入参数严格校验
- 输出结果标准化
模型优化方向：
- 微调专用意图分类模型
- 构建领域特定的实体识别模型
- 使用RAG增强知识库查询

本实现通过LangGraph的状态机架构，将复杂的对话流程分解为可管理的模块，结合LLM的语义理解能力和工具集成，构建出可扩展的智能客服系统。实际部署时建议增加监控模块，跟踪意图识别准确率、工具调用成功率等关键指标。

LangGraph教程（四）：智能客服实战指南