第9章构建客户支持聊天机器人

本项目构建的是一个轻量级、可追踪、意图路由型客户支持工作流，完美展示了LangGraph的基础能力（节点+条件边+状态传递），并结合LangSmith实现了可观测性。若后续需要支持多轮对话或复杂协作，可在此基础上引入循环边（Conditional Loop）或多智能体协作模式（如Supervisor+Workers）。

1. 需求分析

（1）能够理解用户的问题意图（问题分类）。

（2）根据不同问题类型路由到不同的处理节点。

（3）能够回答常见问题（FAQ）。

（4）无法回答时能够转接人工客服。

（5）支持多轮对话，保持上下文。

（6）提供友好的Gradio交互界面。

（7）集成LangSmith进行追踪和调试。

2. 工作流设计

流程说明：

（1）流程以“用户输入消息”开始。

（2）进入“识别意图”节点，判断消息类型。

（3）根据意图将消息分派到4个不同的处理分支之一：answer_faq（常见问题）、handle_technical（技术问题）、handle_complaint（投诉）或transfer_to_human（转人工）。

（4）所有分支最终都汇聚到同一个结束节点。

图9.1清晰地展示了从消息输入、意图识别、分类处理到最终结束的完整工作流。所有处理节点（faq/technical/complaint/other）均为单向、无反馈、无循环，直接通向END，确保单次对话流程只执行一次完整推理。

图9.1 工作流设计图

9.2 项目架构设计

1. 整体架构

1）架构层次

├── 核心层

│ ├── 模型层：Qwen LLM 模型初始化与配置

│ ├── 图结构层：LangGraph节点与边的定义

│ └── 逻辑层：各节点的业务逻辑实现

├── 交互层

│ └── Gradio聊天界面：提供用户交互入口

└── 监控层

└── LangSmith：追踪对话流程和模型调用

2）架构流程（时序图）

时序图如图9.2所示。

图9.2 时序图

2. 技术栈明细

前端交互：Gradio，快速构建聊天界面，支持历史记录、示例、重试等。
工作流引擎：LangGraph (v1.0+)，构建状态驱动的有向图（DAG），支持条件路由、节点编排。
状态管理：TypedDict + StateGraph，明确定义状态结构，确保类型安全和可追踪性。
大语言模型：DashScope + qwen-turbo，阿里云通义千问API，低延迟、低成本，适合客服场景。
模型调用：dashscope.Generation.call，同步调用，指定temperature和result_format。
可观测性：LangChain Tracing V2 + LangSmith，自动记录完整执行链路，支持调试、监控、评估。
配置管理：python-dotenv，从.env安全加载API密钥。
运行环境：Python 3.9+，项目运行基础环境。
部署方式：Gradio.launch()，本地/内网Web服务（可扩展为FastAPI+Docker）。

3. 关键设计亮点

（1）解耦清晰：每个节点只负责一类任务，prompt模板与业务逻辑分离。

（2）安全可控：无递归、无循环，避免无限生成或token耗尽。

（3）可追踪、可审计：所有节点输入/输出、模型调用、意图分类结果均被LangSmith记录。

（4）低成本、高效：使用qwen-turbo平衡速度与成本，适合高频客服场景。

（5）快速原型能力：Gradio+LangGraph组合可在一天内完成MVP。

4. 后续可扩展方向

多轮对话支持：引入记忆机制（如LangGraph Checkpointing）。
人工接管接口：对接企业客服系统（如Zendesk）。
知识库增强：结合RAG（检索增强生成）提升FAQ准确率。
多语言支持：在prompt中选择指定语言。
部署优化：容器化（Docker）+反向智能体（Nginx）+HTTPS。

9.3 项目关键技术

9.3.1 初始化模型

该程序使用的是DashScope提供的云端API服务（通义千问Qwen-Turbo），属于无状态的远程模型调用。

1. 首先安装必要的依赖

# 安装基础依赖

pip install langgraph==1.0.5 langchain langchain-community gradio dashscope python-dotenv

# 安装LangSmith用于追踪

pip install langsmith

2. 模型初始化相关代码

从功能角度看，模型调用所需的配置与封装可以视为“模型初始化”的逻辑部分。以下是程序中与模型初始化和调用准备相关的核心代码片段。

1）环境配置

# ====================== 1. 环境配置 ======================

import os

from dotenv import load_dotenv

import dashscope

# 加载环境变量

load_dotenv()

# 配置通义千问 API 密钥（这是使用 DashScope 模型的前提）

dashscope.api_key = os.getenv("DASHSCOPE_API_KEY") or ""

2）模型调用函数（封装模型接口）

# ====================== 2. 模型调用函数（封装模型接口） ======================

from dashscope import Generation

def call_qwen_model(prompt: str, temperature: float = 0.3) -> str:

"""调用通义千问模型（带LangSmith追踪）"""

try:

response = Generation.call(

model="qwen-turbo", # 指定模型

messages=[{"role": "user", "content": prompt}],

temperature=temperature,

result_format='text'

)

return response.output.text.strip()

except Exception as e:

return f"模型调用出错：{str(e)}"

dashscope.api_key=...是使用DashScope API的必要初始化步骤，相当于“认证初始化”。
call_qwen_model函数是对模型调用的统一封装，指定了：

模型名称：qwen-turbo。
输入格式：messages（符合聊天格式）。
输出格式：text。
控制参数：temperature。

9.3.2 构建图结构与定义节点的逻辑

构建图结构-定义节点的逻辑包含两个部分：

（1）节点函数的实现（即每个节点的具体逻辑）。

（2）在图中注册这些节点（通过graph.add_node()）。

定义节点的逻辑，我们聚焦于节点函数本身的实现逻辑，即每个节点做了什么。节点定义逻辑代码（不含图注册、路由或边连接代码）：

def identify_intent_node(state: SupportState) -> SupportState:

"""节点1：识别用户意图"""

prompt = """

你是一个客户支持意图识别助手，请分析用户的问题并归类到以下类别之一：

- faq: 常见问题（账户、产品使用、价格等）

- technical: 技术问题（故障、报错、功能异常等）

- complaint: 投诉（服务不满、产品质量、退款等）

- other: 无法分类的其他问题

只需要返回类别名称，不要返回其他内容。

用户问题：{user_input}

""".format(user_input=state["user_input"])

intent = call_qwen_model(prompt).lower()

valid_intents = ["faq", "technical", "complaint", "other"]

state["intent"] = intent if intent in valid_intents else "other"

return state

def answer_faq_node(state: SupportState) -> SupportState:

"""节点2：回答常见问题"""

prompt = """

你是一个专业的客户支持助手，请友好、简洁地回答用户的常见问题。

回答要准确、易懂，避免使用专业术语。

聊天历史：{chat_history}

用户问题：{user_input}

请给出你的回答：

""".format(

chat_history=state["chat_history"],

user_input=state["user_input"]

)

state["response"] = call_qwen_model(prompt)

return state

def handle_technical_node(state: SupportState) -> SupportState:

"""节点3：处理技术问题"""

prompt = """

你是一位技术支持专家，请详细解答用户的技术问题。

步骤要清晰，先给出解决方案概述，再分步骤说明。

聊天历史：{chat_history}

用户问题：{user_input}

请给出你的回答：

""".format(

chat_history=state["chat_history"],

user_input=state["user_input"]

)

state["response"] = call_qwen_model(prompt)

return state

def handle_complaint_node(state: SupportState) -> SupportState:

"""节点4：处理投诉问题"""

prompt = """

你是一名客户投诉处理专员，请先表达歉意，然后记录投诉内容，

并给出具体的解决措施和时间承诺。保持语气诚恳、专业。

聊天历史：{chat_history}

用户问题：{user_input}

请给出你的回复：

""".format(

chat_history=state["chat_history"],

user_input=state["user_input"]

)

state["response"] = call_qwen_model(prompt)

return state

def transfer_to_human_node(state: SupportState) -> SupportState:

"""节点5：转接人工客服"""

state["response"] = """

非常抱歉，我暂时无法解答你的问题。

正在为你转接人工客服，请稍候...

人工客服工作时间：9:00~18:00

你也可以通过以下方式联系我们：

- 客服热线：400-123-4567

- 邮箱：support@example.com

"""

return state

说明：

每个节点函数接收SupportState类型的状态对象，处理后原地更新并返回。
所有节点均为纯函数式设计：无副作用，仅依赖输入状态，输出更新后的状态。
除transfer_to_human_node外，其余节点均通过call_qwen_model调用大模型生成响应。
节点职责明确：

意图识别→分类。
FAQ/技术/投诉→领域化生成。
转人工→静态兜底回复。

注意：call_qwen_model和 SupportState类型定义虽未在此段落内，但属于节点逻辑的必要依赖，在实际使用中需一并保留。

9.3.3 连接节点与转向边缘

“连接节点-转向边缘”（即定义节点之间如何连接，特别是基于条件的动态路由）主要体现在以下两个部分：

（1）使用条件路由函数（route_by_intent）决定从意图识别节点转向哪个处理节点。

（2）使用add_conditional_edges添加条件边。

（3）使用add_edge添加静态终止边（所有处理节点→END）。

以下是完整抽取的“连接节点-转向边缘”逻辑代码：

# 定义条件路由函数：根据意图状态动态选择下一个节点

def route_by_intent(state: SupportState) -> str:

"""根据意图转向不同节点"""

intent = state["intent"]

routing_map = {

"faq": "answer_faq",

"technical": "handle_technical",

"complaint": "handle_complaint",

"other": "transfer_to_human"

}

return routing_map.get(intent, "transfer_to_human")

# 添加从意图识别节点出发的条件边（动态路由）

graph.add_conditional_edges(

source="identify_intent",

path=route_by_intent,

path_map={

"answer_faq": "answer_faq",

"handle_technical": "handle_technical",

"handle_complaint": "handle_complaint",

"transfer_to_human": "transfer_to_human"

}

)

# 所有处理节点执行完毕后直接结束（静态边，无循环）

graph.add_edge("answer_faq", END)

graph.add_edge("handle_technical", END)

graph.add_edge("handle_complaint", END)

graph.add_edge("transfer_to_human", END)

说明：

add_conditional_edges：实现动态分支，根据state["intent"]的值在运行时决定流向哪个节点。
path_map：显式声明所有可能的返回值与目标节点的映射关系（LangGraph 1.x 推荐写法，增强可读性和校验）。
add_edge(..., END)：确保每个处理路径单向终止，符合“无递归、单次执行”的设计目标。
该逻辑位于build_support_graph()函数内部，依赖已添加的节点（如"identify_intent""answer
_faq" 等）。

9.3.4 LangSmith追踪

LangSmith 是一个用于调试、测试、评估和监控大型语言模型应用的开发平台，特别适合与 LangChain 等框架配合使用。本节将讲解LangSmith API密钥的申请和使用方法。

1. LangSmith API密钥的申请

首先到LangSmith网站https://smith.langchain.com申请LangSmith API密钥，步骤说明如下：

（1）访问LangSmith官网：打开浏览器，访问网址https://smith.langchain.com。

（2）登录或注册账号：如果你已经有LangChain账号（例如通过GitHub或邮箱注册过），直接登录。如果没有，请点击Sign Up注册一个新账号（支持GitHub、Google或邮箱注册）。

（3）进入API密钥管理页面：点击右上角你的头像或用户名，在下拉菜单中选择“API Keys”（或访问：https://smith.langchain.com/settings/api_keys）。

（4）创建新API密钥：点击“Create API Key”按钮，为密钥添加一个描述（如My LangGraph Project），再点击Create按钮，系统会生成一个形如lsk_abc123def456...的密钥。

（5）使用API密钥，在你的代码或环境变量中设置：

export LANGCHAIN_API_KEY=lsk_abc123def456...
export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_PROJECT=YourProjectName # 可选，指定项目名

启用LANGCHAIN_TRACING_V2=true后，LangChain会自动将trace数据发送到LangSmith网站，方便读者调试和分析。简单说，使用LangSmith可以：

查看每个节点的执行过程。
监控模型调用的耗时和成本。
调试对话流程中的问题。
记录用户对话历史。
分析意图识别的准确性。

LangSmith追踪（LangSmith Tracing）的配置与使用主要体现在两个部分：

在系统环境变量LANGSMITH_API_KEY中配置（启用追踪并设置项目信息）。
在Gradio接口中使用tracing_v2_enabled上下文管理器包裹图的执行。

2. LangSmith追踪相关代码

以下是完整抽取的LangSmith追踪相关代码。

（1）环境配置（启用LangSmith追踪）：

# ====================== 1. 环境配置（启用 LangSmith 追踪） ======================

import os

from dotenv import load_dotenv

from langchain_core.tracers.context import tracing_v2_enabled

load_dotenv()

# 启用 LangSmith 追踪

os.environ["LANGCHAIN_TRACING_V2"] = "true"

os.environ["LANGCHAIN_API_KEY"] = os.getenv("LANGCHAIN_API_KEY") or ""

os.environ["LANGCHAIN_PROJECT"] = "客户支持聊天机器人-无递归版"

os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"

（2）在图执行时启用追踪上下文：

# ====================== 2. 在图执行时启用追踪上下文 ======================

def chat_interface(message, history):

# ...（构建 initial_state）

# 使用 tracing_v2_enabled 上下文管理器包裹图的invoke调用

with tracing_v2_enabled(project_name="客户支持聊天机器人"):

result = support_graph.invoke(initial_state)

return result["response"]

说明：

LANGCHAIN_TRACING_V2="true"：全局开启LangChain v2追踪功能。
LANGCHAIN_API_KEY：用于认证LangSmith服务（需在LangSmith获取）。
LANGCHAIN_PROJECT：指定追踪数据归属的项目名称（用于LangSmith控制台分组）。
LANGCHAIN_ENDPOINT：LangSmith API地址（默认官方地址）。
tracing_v2_enabled(project_name=...)：上下文级追踪控制，确保本次invoke的完整执行链（包括所有节点）被记录到指定项目。

该代码通过“环境变量全局启用+上下文管理器精细控制”的方式，实现了对LangGraph工作流执行过程的端到端可观测性，所有节点输入/输出、模型调用、状态变更均可在LangSmith控制台中查看、调试和评估。

9.4 项目代码实现

【示例9.1】LangGraph构建使用小型模型的客户支持聊天机器人。

（1）依赖安装

# 卸载现有冲突包

pip uninstall -y gradio pydantic langgraph langchain langchain-core

# 安装兼容版本组合（经过验证的稳定版本）

pip install gradio==4.24.0 pydantic==2.5.2 langgraph==1.0.5 langchain==0.2.14 langchain-core==0.2.35 dashscope==1.14.0 python-dotenv

（2）.env配置

DASHSCOPE_API_KEY=你的通义千问API密钥

LANGCHAIN_API_KEY=你的LangSmith API密钥（可选）

（3）访问系统界面

打开浏览器访问：http://localhost:7860，即可看到交互界面，开始对话。

（4）项目代码实现（customer_support_bot.py）

# -*- coding: utf-8 -*-

import os

import gradio as gr

from dotenv import load_dotenv

import dashscope

from dashscope import Generation

from typing import Dict, TypedDict, Optional

from langgraph.graph import StateGraph, END

from langchain_core.tracers.context import tracing_v2_enabled

# ===================== 1. 环境配置（含LangSmith） ======================

load_dotenv()

# 配置通义千问API密钥

dashscope.api_key = os.getenv("DASHSCOPE_API_KEY") or ""

# 配置LangSmith追踪（核心：启用追踪）

os.environ["LANGCHAIN_TRACING_V2"] = "true"

os.environ["LANGCHAIN_API_KEY"] = os.getenv("LANGCHAIN_API_KEY") or ""

os.environ["LANGCHAIN_PROJECT"] = "客户支持聊天机器人-无递归版"

os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"

# =========================== 2. 模型调用函数 ============================

def call_qwen_model(prompt: str, temperature: float = 0.3) -> str:

"""调用通义千问模型（带LangSmith追踪）"""

try:

response = Generation.call(

model="qwen-turbo",

messages=[{"role": "user", "content": prompt}],

temperature=temperature,

result_format='text'

)

return response.output.text.strip()

except Exception as e:

return f"模型调用出错：{str(e)}"

# ====================== 3. 状态定义（LangGraph核心） ======================

class SupportState(TypedDict):

"""定义图的状态结构"""

user_input: str

chat_history: list

intent: Optional[str]

response: Optional[str]

# ================== 4. 节点定义（核心：每个节点只做单一任务） ==================

def identify_intent_node(state: SupportState) -> SupportState:

"""节点1：识别用户意图"""

prompt = """

你是一个客户支持意图识别助手，请分析用户的问题并归类到以下类别之一：

- faq: 常见问题（账户、产品使用、价格等）

- technical: 技术问题（故障、报错、功能异常等）

- complaint: 投诉（服务不满、产品质量、退款等）

- other: 无法分类的其他问题

只需要返回类别名称，不要返回其他内容。

用户问题：{user_input}

""".format(user_input=state["user_input"])

intent = call_qwen_model(prompt).lower()

valid_intents = ["faq", "technical", "complaint", "other"]

state["intent"] = intent if intent in valid_intents else "other"

return state

def answer_faq_node(state: SupportState) -> SupportState:

"""节点2：回答常见问题"""

prompt = """

你是一个专业的客户支持助手，请友好、简洁地回答用户的常见问题。

回答要准确、易懂，避免使用专业术语。

聊天历史：{chat_history}

用户问题：{user_input}

请给出你的答案：

""".format(

chat_history=state["chat_history"],

user_input=state["user_input"]

)

state["response"] = call_qwen_model(prompt)

return state

def handle_technical_node(state: SupportState) -> SupportState:

"""节点3：处理技术问题"""

prompt = """

你是一个技术支持专家，请详细解答用户的技术问题。

步骤要清晰，先给出解决方案概述，再分步骤说明。

聊天历史：{chat_history}

用户问题：{user_input}

请给出你的回答：

""".format(

chat_history=state["chat_history"],

user_input=state["user_input"]

)

state["response"] = call_qwen_model(prompt)

return state

def handle_complaint_node(state: SupportState) -> SupportState:

"""节点4：处理投诉问题"""

prompt = """

你是一个客户投诉处理专员，请先表达歉意，然后记录投诉内容，

并给出具体的解决措施和时间承诺。保持语气诚恳、专业。

聊天历史：{chat_history}

用户问题：{user_input}

请给出你的回复：

""".format(

chat_history=state["chat_history"],

user_input=state["user_input"]

)

state["response"] = call_qwen_model(prompt)

return state

def transfer_to_human_node(state: SupportState) -> SupportState:

"""节点5：转接人工客服"""

state["response"] = """

非常抱歉，我暂时无法解答你的问题。

正在为你转接人工客服，请稍候...

人工客服工作时间：9:00~18:00

你也可以通过以下方式联系我们：

- 客服热线：400-123-4567

- 邮箱：support@example.com

"""

return state

# ================== 5. 构建图结构（核心：无循环，单次流程） ==================

def build_support_graph():

"""

构建LangGraph图结构：

1. 定义节点

2. 定义条件边（意图路由）

3. 无循环，流程到END终止

"""

# （1）创建状态图

graph = StateGraph(SupportState)

# （2）添加所有节点

graph.add_node("identify_intent", identify_intent_node)

graph.add_node("answer_faq", answer_faq_node)

graph.add_node("handle_technical", handle_technical_node)

graph.add_node("handle_complaint", handle_complaint_node)

graph.add_node("transfer_to_human", transfer_to_human_node)

# （3）设置入口节点

graph.set_entry_point("identify_intent")

# （4）定义条件边（意图路由逻辑）

def route_by_intent(state: SupportState) -> str:

"""根据意图转向不同节点"""

intent = state["intent"]

routing_map = {

"faq": "answer_faq",

"technical": "handle_technical",

"complaint": "handle_complaint",

"other": "transfer_to_human"

}

return routing_map.get(intent, "transfer_to_human")

# 添加意图路由的条件边

graph.add_conditional_edges(

source="identify_intent",

path=route_by_intent,

path_map={

"answer_faq": "answer_faq",

"handle_technical": "handle_technical",

"handle_complaint": "handle_complaint",

"transfer_to_human": "transfer_to_human"

}

)

# （5）所有处理节点都指向END（核心：无循环，避免递归）

graph.add_edge("answer_faq", END)

graph.add_edge("handle_technical", END)

graph.add_edge("handle_complaint", END)

graph.add_edge("transfer_to_human", END)

# （6）编译图（适配LangGraph 1.0.5，无config参数）

return graph.compile()

# 初始化图（全局实例）

support_graph = build_support_graph()

# ================== 6. Gradio聊天接口（带LangSmith追踪） ==================

def chat_interface(message, history):

"""

Gradio聊天接口：

1. 构建初始状态

2. 运行LangGraph图（单次流程）

3. 返回结果（带LangSmith追踪）

"""

try:

# （1）构建初始状态

initial_state = {

"user_input": message,

"chat_history": history,

"intent": None,

"response": None

}

# （2）运行图（带LangSmith追踪）

with tracing_v2_enabled(project_name="客户支持聊天机器人"):

result = support_graph.invoke(initial_state)

# （3）返回回复

return result["response"]

except Exception as e:

return f"处理出错：{str(e)}\n请检查LangSmith密钥或重试。"

# ====================== 7. 创建Gradio界面 ======================

iface = gr.ChatInterface(

fn=chat_interface,

title="客户支持聊天机器人（LangGraph+LangSmith版）",

description="""

基于LangGraph + 通义千问(Qwen)构建的智能客户支持系统

✅ 完整LangGraph图结构 | ✅ 节点连接与条件边 | ✅ LangSmith追踪

""",

examples=[

["如何修改我的账户密码？"],

["我的软件安装后无法启动怎么办？"],

["我对你们的服务非常不满意！"],

["你们的产品价格是多少？"]

retry_btn="重新回答",

undo_btn="撤销上一步",

clear_btn="清空对话"

)

# ====================== 8. 运行应用 ======================

if __name__ == "__main__":

print("=== 客户支持聊天机器人启动中 ===")

print("✅ LangGraph图结构已构建")

print("✅ LangSmith追踪已启用")

print("✅ 访问地址：http://0.0.0.0:7860")

iface.launch(

server_name="0.0.0.0",

server_port=7860,

share=False,

show_error=True

)

运行输出（如图9.3所示）：

=== 客户支持聊天机器人启动中 ===

✅ LangGraph图结构已构建

✅ LangSmith追踪已启用

✅ 访问地址：http://0.0.0.0:7860

Running on local URL: http://0.0.0.0:7860

IMPORTANT: You are using gradio version 3.50.2, however version 4.44.1 is available, please upgrade.

--------

To create a public link, set `share=True` in `launch()`.

<IPython.core.display.HTML object>

图9.3 项目运行界面

这里可以输入任意问题，比如：

“如何修改密码？”→得到FAQ类回复。

“软件闪退怎么办？”→得到技术类回复。

“服务太差了！”→得到投诉类回复。

“你好”→转接人工回复。

9.5 本章小结

本章实现了一个基于LangGraph与通义千问（Qwen-Turbo）的无递归客户支持聊天机器人系统。系统采用状态驱动的有向图架构，将对话流程清晰划分为意图识别、分类处理与响应生成三个阶段。通过定义SupportState统一管理对话状态，并封装5个职责单一的节点函数（意图识别、FAQ回答、技术问题处理、投诉响应、转人工），确保逻辑解耦与可维护性。节点间通过条件边实现动态路由：意图识别结果决定后续处理路径，所有路径最终单向终止于END节点，避免循环，保障单次推理的安全性与确定性。系统深度集成LangSmith追踪能力，通过环境变量全局启用追踪，并在图执行时使用tracing_v2_enabled上下文管理器，实现对完整执行链路（包括输入、状态变更、模型调用）的自动记录，为调试、监控与效果评估提供可视化支持。前端采用Gradio快速构建交互界面，结合DashScope云端大模型API，实现低延迟、低成本的智能客服响应。整体设计体现了模块化、可观测、可扩展的工程思想，为多智能体协作与复杂工作流系统奠定了坚实基础。