我第一次尝试用 LangChain 构建客服机器人时，遇到最大的坑就是"对话状态丢失"——用户问完地址，问优惠码，再问退货政策，机器人每次都像失忆了一样从头开始。这就是为什么我后来转向了 LangGraph，它用状态机的方式管理对话流程，让多轮对话变得清晰可控。今天这篇文章，我会从零开始，手把手教你用 LangGraph + HolySheheep AI 构建一个真正能用的客服 Agent。

一、先搞懂什么是"状态管理"，为什么要用它

普通聊天机器人的逻辑是：用户发一条 → AI 回复一条 → 结束。没有记忆。而真实客服场景是这样的：

用户: 我想买个手机
客服: 好的，我们有这几款热门机型...
用户: 256G的有哪些？
客服: 这三款都是256G的...
用户: 那256G的最便宜的多少钱？
客服: 是XXX品牌的，售价XXX元...
用户: 可以用优惠码吗？
客服: 可以的，我们目前有以下优惠...

看，这种连续追问就叫多轮对话。LangGraph 的核心思想是：把对话拆成一个个"节点"（Node），用"状态"（State）在节点间传递信息。就像工厂流水线，每个工位都能看到前面的加工结果。

二、环境准备：安装依赖

我建议用 Python 3.10+，先建个虚拟环境（这步很重要，避免依赖冲突）：

mkdir customer-service-bot
cd customer-service-bot
python -m venv venv

Windows

venv\Scripts\activate

Mac/Linux

source venv/bin/activate


安装核心依赖

pip install langgraph langchain-core langchain-community
pip install openai  # HolySheep API 兼容 OpenAI SDK

注册 HolySheep AI 后，在控制台获取 API Key（格式像 sk-holysheep-xxxxxx）。HolySheheep 的优势是国内直连延迟<50ms，比调 OpenAI 官方快 5-10 倍，而且汇率是 ¥1=$1，比官方的 ¥7.3=$1 便宜 85% 以上。

三、用 LangGraph 构建客服 Agent（完整代码）

3.1 定义对话状态

首先定义一个状态类，包含对话历史和上下文信息：

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
import operator

class CustomerServiceState(TypedDict):
    """客服 Agent 的状态结构"""
    messages: Annotated[Sequence[BaseMessage], operator.add]  # 对话历史
    current_intent: str | None  # 当前意图识别结果
    user_info: dict  # 用户信息（姓名、订单号等）
    context: dict  # 额外上下文（当前商品、会话阶段等）

def create_initial_state():
    """创建初始状态"""
    return {
        "messages": [],
        "current_intent": None,
        "user_info": {},
        "context": {
            "stage": "greeting",  # greeting -> inquiry -> resolution -> closing
            "current_product": None,
            "pending_action": None
        }
    }

3.2 接入 HolySheep API 实现 LLM 调用

这里我用 HolySheep 的 GPT-4.1 模型，价格是 $8/MTok 输出，汇率 ¥1=$1，换算下来非常划算。接入方式完全兼容 OpenAI SDK，只需改 base_url 和 api_key：

from openai import OpenAI

class HolySheepLLM:
    """HolySheep AI LLM 封装"""
    
    def __init__(self, api_key: str):
        self.client = OpenAI(
            api_key=api_key,
            base_url="https://api.holysheep.ai/v1"  # 必须是这个地址
        )
    
    def invoke(self, prompt: str, system_prompt: str = "") -> str:
        """
        调用 LLM 生成回复
        
        参数:
            prompt: 用户输入
            system_prompt: 系统提示词
        
        返回:
            LLM 生成的文本回复
        """
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": prompt})
        
        response = self.client.chat.completions.create(
            model="gpt-4.1",  # HolySheep 支持的模型
            messages=messages,
            temperature=0.7,
            max_tokens=500
        )
        
        return response.choices[0].message.content


使用示例

llm = HolySheepLLM(api_key="YOUR_HOLYSHEEP_API_KEY")  # 换成你的真实 Key
response = llm.invoke("你好，我想咨询退货流程")
print(response)

3.3 定义意图识别节点

客服 Agent 的核心是判断用户想干什么。我定义了几种常见意图：

INTENT_PROMPT = """你是一个电商客服机器人，你需要判断用户消息的意图。

可能的意图类型：
- greeting: 问候/打招呼
- product_inquiry: 产品咨询
- order_status: 订单查询
- refund: 退货/退款
- discount: 优惠/折扣
- complaint: 投诉
- farewell: 道别/结束
- other: 其他问题

请只输出意图类型，不要输出其他内容。

用户消息：{user_message}"""

INTENT_KEYWORDS = {
    "greeting": ["你好", "hi", "hello", "在吗", "在不在"],
    "product_inquiry": ["这个", "那个", "多少钱", "规格", "参数", "有货吗"],
    "order_status": ["订单", "发货", "物流", "到了吗", "什么时候到"],
    "refund": ["退货", "退款", "换货", "不满意", "不要了"],
    "discount": ["优惠", "打折", "降价", "便宜", "优惠码", "券"],
    "complaint": ["投诉", "差评", "质量问题", "坏了", "投诉"],
    "farewell": ["谢谢", "拜拜", "再见", "好的", "知道了"]
}

def classify_intent(state: CustomerServiceState) -> CustomerServiceState:
    """意图识别节点 - 判断用户想干什么"""
    user_message = state["messages"][-1].content
    
    # 先用关键词快速匹配
    for intent, keywords in INTENT_KEYWORDS.items():
        if any(kw in user_message for kw in keywords):
            state["current_intent"] = intent
            state["context"]["stage"] = "inquiry" if intent != "greeting" else "greeting"
            return state
    
    # 关键词匹配失败，用 LLM 判断
    llm = HolySheepLLM(api_key="YOUR_HOLYSHEEP_API_KEY")
    prompt = INTENT_PROMPT.format(user_message=user_message)
    intent = llm.invoke(prompt).strip().lower()
    
    state["current_intent"] = intent if intent in INTENT_KEYWORDS else "other"
    return state

3.4 定义回复生成节点

根据意图生成专业回复，这是最核心的节点：

SYSTEM_PROMPT = """你是一个专业的电商客服，代表店铺与用户沟通。

回复要求：
1. 语气友好、专业、有耐心
2. 简洁明了，不要啰嗦
3. 如果需要用户提供信息，主动询问
4. 涉及到订单号、金额等重要信息时，提醒用户核对

对话历史：
{history}

当前意图：{intent}
用户信息：{user_info}
上下文：{context}

用户最新消息：{user_message}

请生成回复："""

def generate_response(state: CustomerServiceState) -> CustomerServiceState:
    """生成回复节点 - 根据意图生成专业客服回复"""
    user_message = state["messages"][-1].content
    intent = state.get("current_intent", "other")
    user_info = state.get("user_info", {})
    context = state.get("context", {})
    
    # 构造提示词
    history_text = "\n".join([
        f"{'用户' if isinstance(m, HumanMessage) else '客服'}: {m.content}"
        for m in state["messages"][:-1]
    ])
    
    prompt = SYSTEM_PROMPT.format(
        history=history_text or "（无历史对话）",
        intent=intent,
        user_info=user_info,
        context=context,
        user_message=user_message
    )
    
    # 调用 HolySheep API
    llm = HolySheepLLM(api_key="YOUR_HOLYSHEEP_API_KEY")
    response_text = llm.invoke(prompt, system_prompt="你是一个专业的电商客服。")
    
    # 将 AI 回复加入消息历史
    ai_message = AIMessage(content=response_text)
    state["messages"] = state["messages"] + [ai_message]
    
    return state

3.5 搭建状态图（Graph）

现在把节点串起来，形成完整的状态图：

from langgraph.graph import StateGraph, END

def should_continue(state: CustomerServiceState) -> str:
    """判断是否继续对话"""
    last_message = state["messages"][-1].content if state["messages"] else ""
    
    # 检测结束意图
    farewell_keywords = ["谢谢", "好的", "拜拜", "再见", "没问题了", "知道了"]
    if any(kw in last_message for kw in farewell_keywords):
        return "end"
    
    # 最多对话 20 轮
    if len(state["messages"]) >= 40:  # 20 轮 = 40 条消息
        return "end"
    
    return "continue"

def route_after_intent(state: CustomerServiceState) -> str:
    """意图识别后的路由"""
    return "generate_response"


创建状态图

workflow = StateGraph(CustomerServiceState)


添加节点

workflow.add_node("classify_intent", classify_intent)
workflow.add_node("generate_response", generate_response)


设置入口点和边

workflow.set_entry_point("classify_intent")
workflow.add_edge("classify_intent", "generate_response")


添加条件边：generate_response 之后判断是否继续

workflow.add_conditional_edges(
    "generate_response",
    should_continue,
    {
        "continue": "classify_intent",  # 继续下一轮
        "end": END
    }
)


编译图

customer_service_graph = workflow.compile()

def chat(message: str, state: dict = None):
    """对话入口函数"""
    if state is None:
        state = create_initial_state()
    
    # 添加用户消息
    human_message = HumanMessage(content=message)
    state["messages"] = state["messages"] + [human_message]
    
    # 运行状态图
    result = customer_service_graph.invoke(state)
    
    return result

3.6 运行测试

现在来测试一下完整的对话流程：

# 初始化对话状态
session_state = None


第一轮：打招呼

print("=" * 50)
print("用户: 你好，我想问一下手机")
session_state = chat("你好，我想问一下手机", session_state)
print("客服:", session_state["messages"][-1].content)
print(f"当前意图: {session_state['current_intent']}")


第二轮：追问价格

print("\n" + "=" * 50)
print("用户: 256G的多少钱")
session_state = chat("256G的多少钱", session_state)
print("客服:", session_state["messages"][-1].content)


第三轮：问优惠

print("\n" + "=" * 50)
print("用户: 可以用优惠码吗")
session_state = chat("可以用优惠码吗", session_state)
print("客服:", session_state["messages"][-1].content)


第四轮：结束对话

print("\n" + "=" * 50)
print("用户: 谢谢，知道了")
session_state = chat("谢谢，知道了", session_state)
print("客服:", session_state["messages"][-1].content)

四、HolySheep API 价格对比（帮你们省钱）

我用 HolySheep AI 做过详细对比，2026 年主流模型价格：

我自己的经验是：客服 Agent 用 Gemini 2.5 Flash 完全够用，响应快、成本低、效果好。一条典型客服对话（输入+输出约 2000 tokens），成本不到 ¥0.005，换算成人民币几乎可以忽略不计。

五、部署到 Web 服务

最后一步，把 Agent 部署成 Web API，方便前端调用。我用 FastAPI：

# pip install fastapi uvicorn
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uuid

app = FastAPI(title="客服 Agent API")


存储会话状态（生产环境建议用 Redis）

sessions = {}

class ChatRequest(BaseModel):
    session_id: str | None = None
    message: str

class ChatResponse(BaseModel):
    session_id: str
    response: str
    history: list

@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
    """对话接口"""
    # 获取或创建会话
    if request.session_id and request.session_id in sessions:
        session_state = sessions[request.session_id]
    else:
        session_id = request.session_id or str(uuid.uuid4())
        session_state = None
    
    # 调用 Agent
    try:
        result = chat(request.message, session_state)
        sessions[session_id] = result
        
        return ChatResponse(
            session_id=session_id,
            response=result["messages"][-1].content,
            history=[
                {"role": "user" if isinstance(m, HumanMessage) else "assistant", 
                 "content": m.content}
                for m in result["messages"]
            ]
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health():
    return {"status": "ok"}


启动服务: uvicorn main:app --host 0.0.0.0 --port 8000

常见报错排查

报错1: "AuthenticationError: Incorrect API key provided"

这是最常见的错误，意味着 API Key 格式不对或已过期。

# ❌ 错误写法
client = OpenAI(api_key="sk-holysheep-xxx", base_url="https://api.holysheep.ai/v1")


✅ 正确写法

client = OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",  # 替换成真实 Key，从控制台复制
    base_url="https://api.holysheep.ai/v1"
)


检查 Key 是否正确

print(client.api_key)  # 确认输出的是你的真实 Key，不是占位符

报错2: "RateLimitError: Rate limit reached"

请求太快被限流了。解决方案是加延迟和重试机制：

import time
from openai import RateLimitError

def call_with_retry(client, model, messages, max_retries=3):
    """带重试的 API 调用"""
    for i in range(max_retries):
        try:
            response = client.chat.completions.create(
                model=model,
                messages=messages
            )
            return response
        except RateLimitError:
            if i < max_retries - 1:
                wait_time = 2 ** i  # 指数退避: 1s, 2s, 4s
                print(f"触发限流，等待 {wait_time}s 后重试...")
                time.sleep(wait_time)
            else:
                raise Exception("API 调用失败，已达到最大重试次数")


使用示例

result = call_with_retry(client, "gpt-4.1", messages)

报错3: "State schema missing required key: 'messages'"

状态字典缺少必要字段。确保状态初始化时包含所有必需字段：

# ❌ 错误：状态不完整
bad_state = {
    "messages": [HumanMessage(content="你好")]
    # 缺少 current_intent, user_info, context
}


✅ 正确：使用初始化工厂函数

good_state = create_initial_state()
good_state["messages"] = [HumanMessage(content="你好")]


或者手动补全所有字段

full_state = {
    "messages": [HumanMessage(content="你好")],
    "current_intent": None,
    "user_info": {},
    "context": {
        "stage": "greeting",
        "current_product": None,
        "pending_action": None
    }
}

报错4: "Invalid URL: Cannot connect to endpoint"

base_url 配置错误，常见于复制代码时遗漏或拼写错误。

# ❌ 常见错误写法
base_url = "https://api.holysheep.ai/v1/"  # 多了尾部斜杠（部分版本有问题）
base_url = "https://api.holysheep.ai/"      # 缺少 /v1 路径
base_url = "https://api.holysheep.com/v1"    # 拼写错误


✅ 正确写法

base_url = "https://api.holysheep.ai/v1"    # 无尾部斜杠

报错5: "Context window exceeded"

对话历史太长，超出了模型上下文限制。

def trim_messages(messages: list, max_messages: int = 20) -> list:
    """裁剪消息历史，保留最近 N 条"""
    if len(messages) <= max_messages:
        return messages
    
    # 保留系统消息（如果有）和最近的 max_messages 条
    system_messages = [m for m in messages if hasattr(m, 'content') and 
                       getattr(m, 'content', '').startswith('你是一个')]
    
    recent_messages = messages[-max_messages:]
    
    if system_messages:
        return system_messages + recent_messages
    return recent_messages


在生成回复前调用

state["messages"] = trim_messages(state["messages"], max_messages=20)

六、我的实战经验总结

我在生产环境跑这个客服 Agent 大半年，有几点心得：

调试的时候，我习惯在每轮对话后打印 state，清楚看到意图识别对不对、上下文传没传对。LangGraph 的优势就是这种透明性，不像传统 LangChain 那样黑盒。

👉 免费注册 HolySheep AI，获取首月赠额度

相关资源

相关文章