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

news2025/6/10 4:49:45

概述

本教程展示了如何使用 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: str
    category: str
    sentiment: str
    response: 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"]}).content
    return {"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"]}).content
    return {"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"]}).content
    return {"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"]}).content
    return {"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"]}).content
    return {"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))