最近在做一个智能客服系统的升级项目，之前用的是规则引擎，后来也试过直接用大语言模型（LLM）接口，效果都不太理想。规则引擎太死板，稍微复杂点的对话就转不过弯；纯LLM方案呢，状态维护麻烦，响应速度也慢，成本还高。后来接触到了 LangGraph，感觉像是找到了“解题钥匙”。它用有向无环图（DAG）来编排对话流程，把LLM的智能和业务逻辑的确定性结合得很好。今天就来分享一下我们基于 LangGraph 搭建高可用智能客服的实战经验，希望能给有类似需求的同学一些参考。

智能客服系统架构示意图

一、为什么选择LangGraph？传统方案的痛点与对比

在深入技术细节前，我们先聊聊为什么需要 LangGraph。传统的智能客服方案主要有两类：

  1. 基于规则引擎的方案：比如早期的系统，我们定义了大量 if-else 规则。它的优点是确定性强、响应快。但缺点极其明显：对话逻辑僵化，无法处理规则外的情况；维护成本高，业务一变就要改一堆规则；扩展性差，多轮对话的状态管理简直就是噩梦。

  2. 基于纯LLM的方案：直接调用类似 GPT 的 API，把整个对话历史扔过去，让它生成回复。优点是灵活、智能，能处理开放性问题。但痛点也不少：响应延迟高，每次调用都涉及网络IO和模型推理；状态维护困难，需要自己管理冗长的对话历史上下文；成本高昂，每次交互都消耗 Token；可控性差，模型可能会“胡说八道”或者偏离预设的业务流程。

而 LangGraph 的核心思想是 “用图来编排智能体（Agent）”。它将一次对话会话视为在图上的一次遍历。节点（Node）可以是调用 LLM，也可以是执行一段确定性的业务代码（如查询数据库、调用 API），边（Edge）决定了流程的走向。这种架构带来了几个关键优势：

  • 状态集中管理：所有对话状态（用户信息、历史、当前意图等）都封装在一个 State 对象里，在图节点间流转，管理起来非常清晰。
  • 灵活性与可控性兼顾：既可以利用 LLM 处理自然语言理解、生成等不确定任务，又可以用确定性代码节点保障核心业务流程（如支付、查询）的稳定执行。
  • 易于实现复杂逻辑：循环、条件分支、并行处理等复杂对话逻辑，通过图的边（条件边）可以很直观地设计和实现。
  • 性能优化空间大：因为流程被拆解，可以对非LLM节点做缓存、异步化等优化，从而提升整体 QPS。

与 Rasa、Dialogflow 这类成熟的对话平台相比，LangGraph 更像一个底层框架或库，提供了极大的灵活性。Rasa 有自己的 NLU 和故事（Stories）概念，学习成本不低，且深度定制复杂。Dialogflow 是云服务，有厂商锁定风险。LangGraph 则允许你完全自主地构建和混合各种组件，特别适合需要深度集成到现有业务系统、对性能和可控性有极高要求的场景。

二、核心架构与实现：从StateGraph开始

下面我们用 Python 代码来一步步拆解如何构建一个简单的智能客服核心。

首先，定义整个对话的状态。这是 LangGraph 的基石，所有节点都读取和更新这个状态。

from typing import TypedDict, Annotated, List
from langgraph.graph.message import add_messages
import operator

class State(TypedDict):
    # 消息列表，记录对话历史
    messages: Annotated[List[str], add_messages]
    # 用户识别ID
    user_id: str
    # 当前识别出的用户意图（如：查询订单、投诉、转人工）
    current_intent: str
    # 从对话中提取的关键信息槽位（slots），例如订单号、日期
    extracted_slots: dict
    # 业务节点处理后的结果
    business_result: str
    # 标记对话是否需要结束或转人工
    should_end: bool
    should_transfer_to_human: bool

接下来，我们创建 StateGraph，并开始添加节点。

from langgraph.graph import StateGraph, END

# 初始化图
workflow = StateGraph(State)

1. 定义节点：路由、LLM处理与业务逻辑

一个典型的流程可能从“意图路由”节点开始。这个节点负责分析用户最新的输入，判断他想干什么。

def intent_router_node(state: State) -> dict:
    """根据用户最新消息，路由到不同处理节点"""
    latest_message = state[“messages”][-1].content if state[“messages”] else “”
    # 这里可以先用一个简单的规则或关键词匹配，也可以用一个小型/快速的LLM来判断
    # 为了示例，我们简单判断
    if “订单” in latest_message:
        intent = “query_order”
    elif “投诉” in latest_message:
        intent = “complaint”
    elif “人工” in latest_message:
        intent = “human”
    else:
        intent = “general_qa” # 通用问答

    # 更新状态中的意图
    return {“current_intent”: intent}

# 将函数添加为图节点
workflow.add_node(“intent_router”, intent_router_node)

然后是 LLM 处理节点。对于“通用问答”这类需要灵活生成回复的场景，我们调用 LLM。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# 初始化LLM，生产环境建议配置重试、超时等参数
llm = ChatOpenAI(model=“gpt-3.5-turbo”, temperature=0.1)

def general_qa_llm_node(state: State) -> dict:
    """处理通用问答的LLM节点"""
    prompt = ChatPromptTemplate.from_messages([
        (“system”, “你是一个专业的客服助手，请根据对话历史友好、准确地回答用户问题。”),
        (“placeholder”, “{messages}”) # LangGraph会自动注入完整的messages历史
    ])
    chain = prompt | llm
    # 调用LLM获取回复
    ai_message = chain.invoke({“messages”: state[“messages”]})
    # 将AI回复作为一条新消息添加到状态中
    new_messages = state[“messages”] + [ai_message]
    return {“messages”: new_messages}

workflow.add_node(“general_qa_processor”, general_qa_llm_node)

对于确定的业务，如 “查询订单”，我们使用业务逻辑节点。这里不调用LLM，而是查询数据库或内部API。

def query_order_business_node(state: State) -> dict:
    """业务节点：查询订单状态"""
    # 1. 从提取的槽位中获取订单号（假设之前有节点填充了）
    order_no = state[“extracted_slots”].get(“order_number”)
    if not order_no:
        # 如果槽位信息不足，则准备一个追问消息，并更新状态要求返回“追问节点”
       追问消息 = “请问您的订单号是多少？”
        new_messages = state[“messages”] + [HumanMessage(content=追问消息)]
        return {“messages”: new_messages, “business_result”: “need_more_info”}

    # 2. 模拟查询数据库或服务
    # db_query = f“SELECT status FROM orders WHERE order_no='{order_no}' AND user_id='{state[‘user_id’]}'”
    # order_status = execute_query(db_query) # 伪代码
    order_status = “已发货” # 模拟结果

    # 3. 构造业务回复
    business_reply = f“您的订单 {order_no} 当前状态为：{order_status}。”
    new_messages = state[“messages”] + [AIMessage(content=business_reply)]

    # 4. 更新状态
    return {
        “messages”: new_messages,
        “business_result”: “query_success”,
        “should_end”: True # 这个业务完成后，可以结束本轮对话
    }

workflow.add_node(“query_order_processor”, query_order_business_node)

2. 编排流程：设置边与条件路由

节点定义好后，我们需要把它们连接起来，定义执行流程。

# 设置入口点
workflow.set_entry_point(“intent_router”)

# 从路由节点出来后，根据 `current_intent` 的值决定下一步去哪
def route_after_intent(state: State) -> str:
    intent = state[“current_intent”]
    if intent == “query_order”:
        return “query_order_processor”
    elif intent == “general_qa”:
        return “general_qa_processor”
    elif intent == “human”:
        return “transfer_to_human”
    else:
        # 默认路由到通用处理
        return “general_qa_processor”

# 添加条件边
workflow.add_conditional_edges(
    “intent_router”, # 源节点
    route_after_intent, # 路由判断函数
    {
        “query_order_processor”: “query_order_processor”,
        “general_qa_processor”: “general_qa_processor”,
        “transfer_to_human”: “transfer_to_human” # 需要提前定义此节点
    }
)

# 对于业务节点，执行完后通常直接结束或返回路由节点进行下一轮
workflow.add_edge(“query_order_processor”, END) # 本例中查询完直接结束
workflow.add_edge(“general_qa_processor”, END) # 通用问答后也结束，实际中可能返回路由节点

# 编译图，得到可执行对象
app = workflow.compile()

现在，一个最简单的可运行图就构建好了。你可以通过 app.invoke(initial_state) 来运行一次对话。

对话流程图解

三、向生产级迈进：性能优化与避坑指南

上面的基础框架跑通后，要应对生产环境的高并发和稳定性要求，还需要做大量优化。

1. 性能优化：异步、批处理与缓存

异步执行：很多节点，尤其是LLM调用和外部API调用，是IO密集型的。使用异步可以极大提升吞吐量。LangGraph 支持异步节点。

import asyncio
from langchain_openai import AsyncChatOpenAI

async_llm = AsyncChatOpenAI(model=“gpt-3.5-turbo”)

async def async_general_qa_node(state: State) -> dict:
    prompt = ChatPromptTemplate.from_messages([...])
    chain = prompt | async_llm
    ai_message = await chain.ainvoke({“messages”: state[“messages”]})
    return {“messages”: state[“messages”] + [ai_message]}

workflow.add_node(“async_qa_processor”, async_general_qa_node)

对话状态缓存：每次对话都从零开始构建完整的消息历史调用LLM，成本高且慢。我们可以缓存“对话摘要”或“向量化表示”。这里用 Redis 示例缓存经过压缩的上文。

import redis
import json
from hashlib import md5

redis_client = redis.Redis(host=‘localhost’, port=6379, db=0)

def get_cached_history(user_id: str, message_hash: str) -> Optional[List]:
    """根据用户ID和最新消息的哈希获取缓存的压缩历史"""
    cache_key = f“chat_history:{user_id}:{message_hash}”
    cached = redis_client.get(cache_key)
    return json.loads(cached) if cached else None

def update_history_cache(user_id: str, message_hash: str, compressed_history: List):
    cache_key = f“chat_history:{user_id}:{message_hash}”
    redis_client.setex(cache_key, timeout=300, value=json.dumps(compressed_history)) # 缓存5分钟

然后在LLM节点前，可以插入一个“历史压缩/缓存查询”节点来优化。

2. 避坑指南：循环、冷启动与安全

循环依赖检测：LangGraph 虽然基于 DAG，但支持通过条件边实现“循环”（比如多轮澄清）。但设计不当会导致无限循环。务必确保每个循环路径都有明确的退出条件（如最大轮数、用户明确指令、should_end 标志）。可以在状态中增加 turn_count 并在路由节点中进行判断。

冷启动性能调优：图在第一次编译和节点初始化时可能有开销。对于常驻服务，建议在服务启动时完成图的编译和预热（如用空状态或模拟状态 invoke 一次）。另外，将LLM客户端等重型对象设为全局单例，避免每次调用都创建。

敏感词过滤集成：绝对不能完全信任LLM的输出。必须在最终回复发送给用户前，经过一个安全过滤节点。这个节点应该是确定性的，可以集成本地敏感词库或调用风控API。

def safety_filter_node(state: State) -> dict:
    """安全过滤节点，检查AI生成的消息"""
    latest_ai_message = state[“messages”][-1].content
    # 调用过滤函数或服务
    if contains_sensitive_content(latest_ai_message): # 伪代码
        filtered_reply = “您的问题涉及敏感内容，我无法回答。请问还有其他可以帮您的吗？”
        new_messages = state[“messages”][:-1] + [AIMessage(content=filtered_reply)] # 替换最后一条
        return {“messages”: new_messages}
    # 如果安全，则原样返回状态
    return state

workflow.add_node(“safety_filter”, safety_filter_node)
# 确保所有AI生成回复的流程最后都经过这个节点
workflow.add_edge(“general_qa_processor”, “safety_filter”)
workflow.add_edge(“safety_filter”, END)

四、延伸思考：动态调整的DAG与未来展望

在实战中，我们还在思考一个更进阶的问题：能否根据业务指标动态调整DAG结构？比如：

  • 当发现用户对“查询订单”的满意度下降时，是否可以在流程中自动插入一个“确认订单号”的澄清节点，以减少错误？
  • 当“转人工”的比率在某个环节异常高时，是否能够动态简化该环节前的流程，让用户更快接触到人工客服？

理论上，这是可行的。我们可以：

  1. 埋点与监控：在每个图节点和边上埋点，收集耗时、成功率、用户后续行为（如满意度评分、是否转人工）等数据。
  2. 策略引擎：建立一个外部的策略引擎，定期分析这些指标。当触发某些规则（如“节点A的转人工率 > 阈值”），引擎生成一个“图调整指令”。
  3. 热更新：LangGraph 的 StateGraph 对象在编译后虽然不可变，但我们可以设计一个图版本管理机制。当收到调整指令时，根据模板和规则，动态生成一个新的图定义并编译，然后用新的 app 替换旧的。对于正在进行的会话，可以等其自然结束或迁移到新图版本。

这听起来有点像“在线学习”或“A/B测试”的架构。实现起来复杂度很高，涉及到状态迁移、版本兼容性等问题，但无疑是提升系统长期自进化能力的一个激动人心的方向。

结语

从规则引擎到纯LLM，再到 LangGraph，我们走了一条追求“智能”与“可控”平衡的路径。LangGraph 提供的图编排范式，让构建复杂、稳定的对话系统变得模块化和清晰。它不是一个开箱即用的产品，而是一套强大的“乐高”组件，需要你根据业务场景去设计和搭建。

这次实战让我们团队的智能客服系统在核心指标上有了显著提升，尤其是复杂业务场景的对话成功率和系统整体QPS。当然，挑战依然存在，比如更精细的状态设计、更优的缓存策略、以及上面提到的动态化探索。希望这篇分享能为你带来一些启发，也欢迎一起交流在智能体（Agent）应用开发中的更多心得。

Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

Agent 工程中的模型缓存优化经验分享

avatar AI Agent技术社区

CC-Switch不只是切换API：从GitHub更新日志看懂它的功能和底层原理

CC Switch：从配置切换器到AI编程统一管理平台 摘要： CC Switch已从最初的Claude Code/Codex供应商切换工具，发展为功能全面的AI编程管理平台。它通过统一界面管理多个AI编程工具（Claude Code、Codex、Gemini CLI等）的配置，支持供应商切换、本地代理路由、跨工具能力同步等功能。核心演进包括：采用SSOT架构集中管理供应商数据、扩展支持6+工具、

avatar AI Agent技术社区

Harness Engineering在传统软件工程的应用

能力定义关键问题可读性 (Readability)AI Agent 能理解项目的规则、边界和约束Agent 在编码前是否知道"这里不能改"、“这个模块只能做什么”？防御性 (Defense)项目的规则可以被强制执行，违规会被阻断Agent 违反边界时，系统是否能自动阻止而非事后发现？反馈性 (Feedback)项目的健康状态可以被自动化度量Agent 完成任务后，系统能否自动判定"完成度"和"健康

avatar AI Agent技术社区