项目背景与架构总览

智能客服系统是 LangChain 与 LangGraph 综合应用的一个经典场景。在电商、金融、SaaS 等业务中，客服系统需要同时处理 FAQ 类的知识检索问题、订单查询与退款等事务性操作，以及在遇到复杂情况时将对话无缝转接给人工坐席。传统的基于关键词匹配或固定决策树的客服系统难以应对用户多样化的表达方式和上下文依赖，而基于大语言模型的智能客服则可以利用语义理解能力和工具调用能力，在知识库检索与业务操作之间灵活切换。

我们的系统采用"意图识别 → RAG 检索 → Agent 决策"三级流水线架构。用户输入的任何消息首先经过一个轻量级的意图分类节点，这一节点的职责是快速判断用户当前消息的性质，是在咨询常见问题、需要查询订单、申请退款，还是明确要求转人工。意图分类的结果并不直接产生回复，而是作为后续节点的路由依据。如果意图被判定为 FAQ 类问题，系统进入 RAG 检索节点，从向量知识库中召回相关的产品说明、退换货政策或使用教程等文档片段，如果意图是订单查询或退款等事务性操作，RAG 节点将被跳过，消息直接传递到 Agent 决策节点，由 Agent 选择合适的业务工具来执行操作。Agent 决策节点是整个系统的核心，它持有一组业务工具，包括订单查询工具、退款审核工具和转人工工具，并基于系统提示词扮演电商客服的角色，综合上下文生成最终的用户回复。

这种三级路由设计的优势在于职责分离与可扩展性。意图识别专注于分类任务，可以采用更轻量或更便宜的模型以降低延迟和成本RAG 检索独立成节点，便于后续更换向量数据库或调整检索策略而不影响 Agent 的逻辑，Agent 则专注于工具调用与对话生成，不需要在每次推理时都判断"要不要检索"，从而减少了不必要的工具调用轮次，提升了响应速度。接下来我们将逐步实现这个系统的每一个组件。

开发环境与依赖

在开始编码之前，确保已安装以下核心依赖。LangGraph 是构建有状态工作流的基础框架，langchain.agents 提供了 create_agent 这一高级 API 用于快速创建具备工具调用能力的 Agent。向量存储方面，我们使用 Chroma 作为本地持久化方案，langchain-openai 或 langchain-siliconflow 提供模型与嵌入向量的统一接口。

# 核心依赖安装
pip install langgraph langchain langchain-openai langchain-community
pip install langchain-text-splitters chromadb python-dotenv

模型初始化统一使用 init_chat_model，这是 LangChain 推荐的模型工厂函数，可以通过 model_provider 参数指定不同的服务提供商，无需为每个平台写不同的初始化代码。嵌入模型用于将知识库文档转换为向量，这里选用与 Chat 模型同一提供商的嵌入接口。

from langchain.chat_models import init_chat_model
from langchain_openai import OpenAIEmbeddings
from dotenv import load_dotenv

load_dotenv()

llm = init_chat_model("deepseek-ai/DeepSeek-V4-Pro", model_provider="openai")

embeddings = OpenAIEmbeddings(model="Qwen/Qwen3-VL-Embedding-8B", check_embedding_ctx_length=False,)

这段代码是模型初始化的标准入口，它的核心任务是准备好后续整个智能客服系统所需的两个关键能力：对话推理与文本向量化。

首先导入了三个依赖。init_chat_model 是 LangChain 官方推荐的模型工厂函数，它的设计理念是屏蔽不同模型服务商之间的 API 差异，让开发者只需通过 model_provider 参数指定服务商，就能用同一套代码逻辑对接 OpenAI、DeepSeek、硅基流动等不同平台，无需为每个平台单独编写初始化逻辑。OpenAIEmbeddings 则专门负责将文本转换为向量，也就是把一段自然语言映射成一串高维数字，这是后续语义检索的基础。load_dotenv 的作用是把项目根目录下 .env 文件里的环境变量（例如 OPENAI_API_KEY、OPENAI_BASE_URL 等）注入到当前进程中，避免将密钥硬编码在代码里。

init_chat_model("deepseek-ai/DeepSeek-V4-Pro", model_provider="openai") 这一行值得展开说明。虽然模型名以 deepseek-ai 开头、提供商却填了 "openai"，这并不矛盾，因为 DeepSeek 的 API 在设计上兼容 OpenAI 的请求与响应格式，也就是说它用的是同一套 /v1/chat/completions 接口协议。当 model_provider 设为 "openai" 时，LangChain 底层会使用 ChatOpenAI 这个类来发起 HTTP 请求，但实际请求会被路由到 DeepSeek 的服务器（通过 OPENAI_BASE_URL 环境变量指定），而不是 OpenAI 的服务器。这样做的好处是开发者不需要等待 LangChain 为每个模型单独开发适配器，只要服务商遵循 OpenAI 兼容协议，就能立刻接入使用。

接下来 OpenAIEmbeddings(model="Qwen/Qwen3-VL-Embedding-8B") 创建了一个嵌入模型实例。Qwen3-VL-Embedding-8B 相比 DeepSeek-V4-Pro 是一个更轻量的模型，而且是专门用来嵌入的模型，把它用在向量化任务上是非常合理的选择，嵌入任务本质上只需要理解文本的语义并将其编码为向量，不需要复杂的推理链条或长篇生成能力，用更轻更快的模型可以在保证检索质量的前提下显著降低延迟和调用成本。相比之下，llm 使用的是 DeepSeek-V4-Pro，它负责对话生成、意图识别和工具调用等需要深度推理的任务，对模型的综合能力要求更高。这种"对话用大模型、嵌入用小模型"的分工策略，是在成本与性能之间取得平衡的典型做法。

构建向量知识库与 RAG 检索

知识库是智能客服系统的记忆中枢，存储着产品手册、退换货政策、常见问题解答等企业自有数据。构建知识库的第一步是加载文档，第二步是使用文本分割器将长文档切分成适合检索的语义块，第三步是将这些块通过嵌入模型向量化后存入向量数据库。

文档加载器支持多种格式，从本地的 TXT、Markdown、PDF 到网页内容均可处理。这里以简单的文本文件为例，实际项目中你可能需要对接 Confluence、Notion 或内部 Wiki 的内容管线。

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import Chroma

# 加载企业知识库文档
loader = TextLoader("./knowledge_base/faq.txt", encoding="utf-8")
documents = loader.load()

# 文本分割：按语义边界切分，保留重叠以维护上下文连贯性
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "，", "；", " ", ""]
)
chunks = splitter.split_documents(documents)

# 向量化并持久化到 Chroma
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./customer_service_db"
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

RecursiveCharacterTextSplitter 按照优先级依次尝试不同的分隔符：先按双换行（段落）切分，再按单换行，最后按中文标点和空格，确保每个文本块在语义上尽可能完整。chunk_overlap=50 让相邻块之间有 50 个字符的重叠，这在检索时能避免关键信息恰好落在块边界上而被截断。as_retriever 方法将向量存储包装成一个 Runnable 对象，可以直接在 LCEL 链或 LangGraph 节点中调用，k=4 表示每次检索返回相似度最高的 4 个文档片段。

在智能客服的场景中，RAG 检索并不是对所有请求都触发。我们将其封装为一个独立的节点函数，只有意图分类结果为 faq 时才执行检索。检索到的文档片段会被注入到状态中，供后续的 Agent 节点作为参考上下文使用。

def retrieve_knowledge(state: dict) -> dict:
    """仅在意图为 faq 时执行知识库检索"""
    intent = state.get("intent", "")
    if intent != "faq":
        return {"context": ""}
    user_msg = state["messages"][-1].content
    docs = retriever.invoke(user_msg)
    context = "\n\n".join([d.page_content for d in docs])
    return {"context": context}

这个 retrieve_knowledge 函数是 RAG 检索节点在 LangGraph 工作流中的具体实现，它的核心职责是充当一个智能的"开关+执行器"，只在必要的时候才触发向量知识库检索，而不是对所有请求都无差别地执行检索操作。

函数签名 (state: dict) -> dict 遵循了 LangGraph 节点的标准约定：每个节点函数接收整个工作流的状态字典，然后返回一个仅包含自己想要更新的字段的字典，LangGraph 的归并器会自动将返回值合并回全局状态中。这种设计让节点之间保持松耦合，每个节点只关心自己负责的那一部分状态。

函数体的第一行 intent = state.get("intent", "") 从前一个节点（意图分类节点）写入的状态中读取意图标签。紧接着的 if intent != "faq": return {"context": ""} 就是那个关键的"开关"逻辑，如果用户当前的意图是订单查询、退款申请或转人工，那么检索知识库完全没有意义，因为这些问题需要的不是文档片段而是直接调用业务 API。返回空的 context 字符串既避免了浪费一次嵌入推理和向量检索的调用开销，也为后续的 Agent 节点提供了一个明确信号：这次没有知识库参考内容可用。

当意图确实是 faq 时，函数从状态中提取用户的最新消息。state["messages"][-1] 取的是消息列表的最后一个元素，因为 LangGraph 中消息是按时间顺序追加的，最后一个就是当前轮次的用户输入。.content 属性从中取出纯文本内容，这正是检索器所需要的查询字符串。然后 retriever.invoke(user_msg) 将用户的自然语言问题直接作为查询传入向量检索器，这里体现了语义检索的核心优势，用户不需要把问题拆成关键词，整句自然语言就能通过嵌入模型映射到向量空间，然后与知识库中的所有文档片段计算余弦相似度，召回语义上最相关的几个片段。k=4 表示每次最多返回 4 个最匹配的文档块，这个数字是在召回率和精确度之间取的平衡值，太少可能遗漏关键信息，太多则会引入噪音并膨胀上下文长度。

最后 "\n\n".join([d.page_content for d in docs]) 将检索到的多个文档片段用双换行拼接成一个完整的上下文字符串。page_content 是 LangChain Document 对象上存储文本内容的属性，它包含了当初被切分并向量化的原始文本块。用双换行分隔是为了在视觉上和语义上让不同来源的片段之间有一个清晰的边界，这样当大模型阅读这段拼接后的上下文时，能自然地感知到这些信息来自不同的文档段落。拼接完成的 context 被写入返回字典，随后会被注入到 Agent 决策节点的状态中，Agent 在生成回复时会将其作为"参考知识"来组织答案。

定义业务工具

工具是 Agent 与外部世界交互的桥梁。在智能客服系统中，我们至少需要三个业务工具：订单查询工具让 Agent 能够根据订单号或用户信息查找物流状态与订单详情、退款审核工具负责检查退款条件并执行或拒绝退款操作、转人工工具在 Agent 判断自己无法妥善处理时将对话连同上下文一起转交给人工坐席队列。

在 LangChain 中定义工具最简单的方式是使用 @tool 装饰器。每个工具需要一个清晰的函数名、一段描述用途的 docstring，以及类型注解，这些信息会被注入到模型的系统提示词中，帮助模型判断在什么情况下应该调用哪个工具。

from langchain.tools import tool

# 模拟订单数据库
ORDER_DB = {
    "ORD-2024-001": {"status": "已发货", "courier": "顺丰", "tracking": "SF1234567890"},
    "ORD-2024-002": {"status": "处理中", "courier": None, "tracking": None},
    "ORD-2024-003": {"status": "已签收", "courier": "中通", "tracking": "ZT9876543210"},
}


@tool
def order_query(order_id: str) -> str:
    """根据订单号查询订单状态和物流信息。参数 order_id 为订单编号，格式如 ORD-2024-XXX。"""
    order = ORDER_DB.get(order_id)
    if not order:
        return f"未找到订单 {order_id}，请确认订单号是否正确。"
    if order["status"] == "已发货":
        return f"订单 {order_id} 状态：{order['status']}，快递公司：{order['courier']}，运单号：{order['tracking']}。"
    return f"订单 {order_id} 状态：{order['status']}。"


@tool
def refund_check(order_id: str) -> str:
    """检查指定订单是否符合退款条件，并在符合条件时发起退款。参数 order_id 为订单编号。"""
    order = ORDER_DB.get(order_id)
    if not order:
        return f"未找到订单 {order_id}，无法处理退款。"
    if order["status"] == "已签收":
        return f"订单 {order_id} 已签收，不符合直接退款条件。建议引导用户先联系人工客服确认退款原因。"
    if order["status"] == "已发货":
        return f"订单 {order_id} 已在运输中。已为您拦截并提交退款申请，预计 3-5 个工作日内退款到账。"
    return f"订单 {order_id} 尚未发货，已直接取消并退款，预计 1-2 个工作日内到账。"


@tool
def transfer_human(reason: str) -> str:
    """将当前对话转接给人工客服。当用户明确要求转人工，或问题超出 AI 客服能力范围时调用。参数 reason 为转接原因。"""
    return f"已为您转接人工客服，转接原因：{reason}。请稍候，客服专员将很快接入。"

三个工具各有侧重，order_query 是典型的读操作，从数据库中查询信息并格式化返回，refund_check 包含业务规则判断，根据订单的不同状态做出不同的决策，这体现了工具可以封装任意复杂的业务逻辑，transfer_human 则是一个动作性工具，在实际生产环境中它会调用工单系统的 API 创建转接任务，而非简单地返回一句提示。

意图识别节点

意图识别是流水线的第一站。我们将其实现为一个独立的 LangGraph 节点，职责是从用户消息中提取意图标签。当前支持的意图包括 faq（常见问题咨询）、order_query（订单查询）、refund（退款申请）和 transfer_human（要求转人工）。为了保持响应速度，意图识别节点使用同一个 LLM 但配置了较低的温度参数，并给出了精简的分类指令。

from langchain_core.messages import HumanMessage, AIMessage

INTENT_PROMPT = """你是一个意图分类器。分析用户消息，仅返回以下意图之一（只返回标签，不要解释）：
- faq: 咨询产品信息、退换货政策、使用方法等常见问题
- order_query: 查询订单状态、物流信息
- refund: 申请退款、取消订单
- transfer_human: 明确要求人工客服

用户消息：{user_input}
意图："""


def classify_intent(state: dict) -> dict:
    """分析用户最新消息的意图并写入状态"""
    messages = state["messages"]
    if not messages:
        return {"intent": "faq"}
    user_msg = messages[-1].content if hasattr(messages[-1], "content") else messages[-1]["content"]
    response = llm.invoke(INTENT_PROMPT.format(user_input=user_msg))
    intent = response.content.strip().lower()
    # 标准化意图标签
    valid_intents = {"faq", "order_query", "refund", "transfer_human"}
    if intent not in valid_intents:
        intent = "faq"  # 兜底为 FAQ
    return {"intent": intent}

这段代码实现了智能客服系统三级流水线中的第一站，意图识别节点。它由两部分组成：一个精心设计的提示词模板 INTENT_PROMPT、一个在 LangGraph 工作流中作为节点运行的 classify_intent 函数。

先看 INTENT_PROMPT 这个提示词模板。它本质上是在用自然语言"编程"大模型的行为，开头"你是一个意图分类器"直接为模型设定了角色定位，让它明白自己此刻的任务不是闲聊或回答用户问题，而是执行一个狭窄的分类任务，紧接着列出四种意图标签及其含义，提示词末尾用 {user_input} 占位符预留了用户消息的插入位置，最后的"意图："是一个关键的输出引导，它让模型在读完指令后自然地在那个位置接着输出标签，而不容易啰嗦或发散。括号中的"只返回标签，不要解释"是一个防御性约束，防止模型在给出标签后又追加一段解释性文本，那样会增加后续解析的麻烦和推理开销。

再看 classify_intent 函数本身。它首先从 state["messages"] 中取出消息列表，如果列表为空（意味着这是对话的第一轮），就直接默认意图为 faq，这是一个合理的兜底策略，因为在没有任何用户消息的情况下，最安全的假设就是用户在咨询一般性问题。提取用户消息的那行 messages[-1].content if hasattr(messages[-1], "content") else messages[-1]["content"] 写得比较谨慎，它兼容了两种消息格式，一种是 LangChain 原生的 HumanMessage 对象，带有 .content 属性，另一种则是纯字典格式的消息，需要通过键 ["content"] 取值。这种兼容性在实际开发中很实用，因为 LangGraph 状态中的消息可能来自不同的上游来源，并非总是同一种数据结构。

拿到用户文本后，函数用 INTENT_PROMPT.format(user_input=user_msg) 将消息填入模板，然后调用 llm.invoke(...) 发送给大模型进行推理。注意这里调用的是同一个 llm 实例（DeepSeek-V4-Pro），但通过 prompt 中的严格约束将它的行为从"对话生成"切换到了"分类判断"。

接下来是一个防御性的"标准化"步骤：将模型的输出与预定义的合法意图集合 {"faq", "order_query", "refund", "transfer_human"} 做比对，如果模型返回了任何不在这个集合内的文本（比如它不小心多写了几个字，或者返回了一个近义词），就会被强制归入 "faq" 这个兜底类别。这个防护层保证了无论模型输出什么奇怪的结果，下游的 RAG 检索节点和 Agent 节点永远只会看到一个规范化的意图值，整个工作流不会因为一个分类异常而崩溃。

最后函数返回 {"intent": intent}，只包含一个键值对。意图识别节点只修改全局状态中的 intent 字段，完全不动 messages 列表。LangGraph 的状态归并机制会自动将这个返回的字典合并到全局 CustomerState 中，intent 字段被更新为新值，而其他字段（如 messages 和 context）保持不变。正是这种"只返回自己关心的字段"的设计，让每个节点可以独立演化而不相互干扰，后续的 retrieve_knowledge 节点和 customer_agent 节点都可以通过 state["intent"] 读取这个分类结果来做各自的路由和决策。

Agent 决策节点与工作流组装

Agent 决策节点是系统的核心。它使用 create_agent 创建，这是一个封装了 ReAct 循环（推理-行动-观察）的高级 API，内部实际上编译成了一个 LangGraph 子图。我们将其作为一个节点嵌入到外层的工作流图中。

create_agent 接收三个关键参数，model 指定使用的语言模型、tools 是工具列表，模型会在推理过程中决定调用哪些工具、system_prompt 则定义了 Agent 的角色、行为规范和输出要求。在客服场景中，系统提示词需要明确告知 Agent 它代表一家电商公司，应该用友好、专业的中文回复用户，并在必要时主动使用工具。

from langchain.agents import create_agent

customer_agent = create_agent(
    model=llm,
    tools=[order_query, refund_check, transfer_human],
    system_prompt=(
        "你是「极客商城」的智能客服助手小 G。你的职责是帮助用户解决问题。"
        "当用户咨询产品信息或使用问题时，请参考提供的知识库内容进行回答。"
        "当用户询问订单状态时，使用 order_query 工具查询。"
        "当用户要求退款时，先使用 refund_check 工具检查退款条件。"
        "当你无法解决用户问题或用户明确要求转人工时，使用 transfer_human 工具。"
        "始终保持热情、耐心的语气，用中文回复。"
    ),
)

接下来是最关键的步骤，用 LangGraph 的 StateGraph 将意图识别、RAG 检索和 Agent 决策三个节点串联成一个完整的工作流。我们使用 TypedDict 定义状态结构，包含消息列表、意图标签和检索到的上下文。

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver


class CustomerState(TypedDict):
    messages: Annotated[list, add_messages]
    intent: str
    context: str


# 构建工作流图
graph = StateGraph(CustomerState)

# 添加三个节点
graph.add_node("classify", classify_intent)
graph.add_node("retrieve", retrieve_knowledge)
graph.add_node("agent", customer_agent)

# 连接边：classify → retrieve → agent → END
graph.add_edge(START, "classify")
graph.add_edge("classify", "retrieve")
graph.add_edge("retrieve", "agent")
graph.add_edge("agent", END)

# 编译图（注入 checkpointer 以启用记忆）
memory = InMemorySaver()
app = graph.compile(checkpointer=memory)

这里有几个值得注意的细节。add_messages 是 LangGraph 内置的消息列表归并器，它确保每次节点返回的新消息被追加到已有消息列表末尾，而不是覆盖。当 create_agent 返回一个 AIMessage 时，add_messages 会正确地将其合并。InMemorySaver 是 LangGraph 提供的内存级检查点存储，用于保存每个线程的对话状态，这是实现多轮对话记忆的关键基础设施。在生产环境中，应替换为 PostgresSaver 或 SqliteSaver 等持久化方案。

运行与测试

现在我们可以用各种真实场景来测试系统。每次调用 app.invoke 时传入一个 config 字典，其中的 thread_id 标识了一个独立的对话会话。相同的 thread_id 意味着共享同一段对话历史。

# 场景一：FAQ 咨询
config = {"configurable": {"thread_id": "user-001"}}
result = app.invoke(
    {"messages": [HumanMessage(content="你们的退货政策是什么样的？")]},
    config
)
print(result["messages"][-1].content)

# 场景二：订单查询（同一用户，多轮对话）
result = app.invoke(
    {"messages": [HumanMessage(content="帮我查一下订单 ORD-2024-001 到哪了")]},
    config
)
print(result["messages"][-1].content)

# 场景三：利用记忆引用前文
result = app.invoke(
    {"messages": [HumanMessage(content="刚才那个订单能退款吗？")]},
    config
)
print(result["messages"][-1].content)

在场景三中，用户说"刚才那个订单"而没有重复订单号，Agent 能否正确理解取决于两件事，一是消息历史通过 add_messages 被完整保留在状态中，模型可以看到之前关于 ORD-2024-001 的对话，二是 InMemorySaver 确保即使 invoke 调用是分开进行的，相同 thread_id 下的状态也能被恢复。这就是对话记忆的工作原理——不需要额外开发任何存储逻辑，LangGraph 的检查点机制自动处理了状态的持久化与恢复。

添加对话记忆的深入理解

在上面的代码中，我们已经在编译图时注入了 InMemorySaver 作为 checkpointer，这是 LangGraph 实现短期记忆的标准方式。每当图执行到一个节点结束时，checkpointer 会自动保存一份状态快照，关联到当前的 thread_id。下一次使用相同的 thread_id 调用 invoke 时，框架会从最近的检查点恢复状态，消息历史、意图标签和检索上下文都会被完整还原。

对于需要跨会话持久化的场景，可以替换为 Postgres 检查点存储。langgraph-checkpoint-postgres 包提供了 PostgresSaver，它将检查点写入 PostgreSQL 数据库，适合生产环境部署。使用方式几乎与 InMemorySaver 完全相同，只需在编译图时传入不同的 checkpointer 实例即可。

如果你希望客服系统具备更长期的记忆能力，例如记住用户是 VIP 会员、偏好某种沟通风格、或者有过投诉记录，可以在状态中增加额外的字段并在节点中读写这些字段。LangGraph 状态的灵活性意味着你可以自由扩展状态结构，而检查点机制会一视同仁地将所有状态字段一并持久化。

工作流可视化与调试

LangGraph 提供了内置的图可视化功能，可以将编译后的工作流渲染为 Mermaid 图表。在开发阶段，调用 app.get_graph().draw_mermaid_png() 可以生成一张直观的流程图，帮助你验证节点和边的连接是否符合预期。

from IPython.display import Image, display

display(Image(app.get_graph().draw_mermaid_png()))

此外，建议配置 LangSmith 追踪来观察每次调用的完整执行轨迹，包括意图分类的结果、检索到的文档片段、Agent 的推理链以及每个工具的调用参数与返回值。这对于调试 Agent 的错误行为（例如调用了错误的工具或检索到了无关文档）至关重要。

练习任务

• 实现完整的智能客服工作流
• 集成至少 3 个业务工具
• 添加对话记忆功能

考核点 ✅

1. 工作流运行：提交完整可运行的智能客服 LangGraph 代码，能处理真实用户问题
2. 工具验收：展示 3 个业务工具各自被 Agent 调用成功的日志输出
3. 记忆功能：演示多轮对话中 Agent 能引用前文信息（提交对话截图）
4. 架构评审：口头解释意图识别→RAG→Agent 三级路由的设计思路