项目研究：使用 LangGraph 构建智能客服代理

概述

本教程展示了如何使用 LangGraph 构建一个智能客服代理。LangGraph 是一个强大的工具，可用于构建复杂的语言模型工作流。该代理可以自动分类用户问题、分析情绪，并根据需要生成回应或升级处理。

背景动机

在当今节奏飞快的商业环境中，高效、准确的客户支持至关重要。将用户交互的初始阶段自动化，能够显著缩短响应时间并提升客户满意度。本项目展示了如何将先进的语言模型与基于图的工作流结合，创建一个能够处理多种客户咨询的智能支持系统。

核心模块

• 状态管理：使用 TypedDict 管理每次用户交互的状态。
• 问题分类：将用户问题分类为技术类、账单类或通用类。
• 情绪分析：分析用户问题的情绪倾向（正面、中性或负面）。
• 响应生成：根据问题类别和情绪自动生成回应。
• 问题升级机制：对于负面情绪的问题自动转由人工处理。
• 工作流图构建：通过 LangGraph 构建灵活可扩展的工作流。

---

方法详解

1. 环境初始化与依赖安装

# 安装所需依赖
!pip install -q langgraph langchain-core langchain-openai python-dotenv ipython

2. 导入必要库

from typing import Dict, TypedDict
from langgraph.graph import StateGraph, END
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from IPython.display import display, Image
from langchain_core.runnables.graph import MermaidDrawMethod
from dotenv import load_dotenv
import os# 加载环境变量并设置 OpenAI API Key
load_dotenv()
os.environ["OPENAI_API_KEY"] = os.getenv('OPENAI_API_KEY')

---

3. 定义状态结构

class State(TypedDict):query: strcategory: strsentiment: strresponse: str

---

4. 定义各阶段处理函数（节点）

分类问题

def categorize(state: State) -> State:prompt = ChatPromptTemplate.from_template("Categorize the following customer query into one of these categories: Technical, Billing, General. Query: {query}")chain = prompt | ChatOpenAI(temperature=0)category = chain.invoke({"query": state["query"]}).contentreturn {"category": category}

情绪分析

def analyze_sentiment(state: State) -> State:prompt = ChatPromptTemplate.from_template("Analyze the sentiment of the following customer query. Respond with either 'Positive', 'Neutral', or 'Negative'. Query: {query}")chain = prompt | ChatOpenAI(temperature=0)sentiment = chain.invoke({"query": state["query"]}).contentreturn {"sentiment": sentiment}

响应生成（根据类别）

def handle_technical(state: State) -> State:prompt = ChatPromptTemplate.from_template("Provide a technical support response to the following query: {query}")chain = prompt | ChatOpenAI(temperature=0)response = chain.invoke({"query": state["query"]}).contentreturn {"response": response}def handle_billing(state: State) -> State:prompt = ChatPromptTemplate.from_template("Provide a billing support response to the following query: {query}")chain = prompt | ChatOpenAI(temperature=0)response = chain.invoke({"query": state["query"]}).contentreturn {"response": response}def handle_general(state: State) -> State:prompt = ChatPromptTemplate.from_template("Provide a general support response to the following query: {query}")chain = prompt | ChatOpenAI(temperature=0)response = chain.invoke({"query": state["query"]}).contentreturn {"response": response}

升级负面情绪的问题

def escalate(state: State) -> State:return {"response": "This query has been escalated to a human agent due to its negative sentiment."}

路由逻辑

def route_query(state: State) -> str:if state["sentiment"] == "Negative":return "escalate"elif state["category"] == "Technical":return "handle_technical"elif state["category"] == "Billing":return "handle_billing"else:return "handle_general"

---

5. 构建 LangGraph 图结构

workflow = StateGraph(State)# 添加节点
workflow.add_node("categorize", categorize)
workflow.add_node("analyze_sentiment", analyze_sentiment)
workflow.add_node("handle_technical", handle_technical)
workflow.add_node("handle_billing", handle_billing)
workflow.add_node("handle_general", handle_general)
workflow.add_node("escalate", escalate)# 添加边
workflow.add_edge("categorize", "analyze_sentiment")
workflow.add_conditional_edges("analyze_sentiment", route_query, {"handle_technical": "handle_technical","handle_billing": "handle_billing","handle_general": "handle_general","escalate": "escalate"
})
workflow.add_edge("handle_technical", END)
workflow.add_edge("handle_billing", END)
workflow.add_edge("handle_general", END)
workflow.add_edge("escalate", END)# 设置入口节点
workflow.set_entry_point("categorize")# 编译应用
app = workflow.compile()

---

6. 可视化工作流图

display(Image(app.get_graph().draw_mermaid_png(draw_method=MermaidDrawMethod.API,))
)

---

7. 执行客户查询处理函数

def run_customer_support(query: str) -> Dict[str, str]:results = app.invoke({"query": query})return {"category": results["category"],"sentiment": results["sentiment"],"response": results["response"]}

---

测试示例

# 示例1：技术类且情绪负面（应升级）
query = "My internet connection keeps dropping. Can you help?"
print(run_customer_support(query))# 示例2：技术类中性
query = "I need help talking to chatGPT"
print(run_customer_support(query))# 示例3：账单类中性
query = "where can i find my receipt?"
print(run_customer_support(query))# 示例4：通用类中性
query = "What are your business hours?"
print(run_customer_support(query))

---

总结

本教程展示了如何使用 LangGraph 构建一个智能客服代理系统。该系统结合了自然语言处理与结构化图工作流，可对客户问题进行分类、分析情绪并自动生成回应，甚至在必要时升级处理。该方法不仅可应用于客户支持领域，也适用于其他多步骤复杂流程的智能化处理场景。

参考文献

1. https://github.com/NirDiamant/GenAI_Agents/blob/main/all_agents_tutorials/customer_support_agent_langgraph.ipynb