在构建复杂 AI 工作流时，静态流程往往无法满足实际业务需求。我在使用 LangGraph 开发多轮对话系统时，最大的挑战就是如何让 Agent 能够根据不同条件动态选择执行路径——这正是条件分支的核心价值所在。本文将深入讲解 LangGraph 中的条件路由（Conditional Edge）与循环控制（Loop），并通过 HolySheep AI 作为示例 API，演示完整的工程实践。

平台选型对比：HolySheep vs 官方 API vs 其他中转站

对比维度HolySheep AI官方 API其他中转站
美元汇率¥1 = $1（无损）¥7.3 = $1¥6.5-8.0 = $1
国内延迟<50ms（直连）200-500ms80-300ms
充值方式微信/支付宝/银行卡国际信用卡参差不齐
Claude Sonnet 4.5$15/MTok$15/MTok$12-18/MTok
GPT-4.1$8/MTok$8/MTok$7-12/MTok
DeepSeek V3.2$0.42/MTok$0.42/MTok$0.35-0.6/MTok
注册福利送免费额度无部分有

从我的实际项目经验来看，使用 HolySheep AI 后，同样的预算能多支撑 85% 以上的 API 调用量，特别适合需要大量测试条件分支逻辑的开发者。如果你是国内开发者，立即注册 HolySheep AI 可以避免繁琐的跨境支付问题。

LangGraph 条件分支核心概念

LangGraph 的控制流分为两种模式：

为什么需要条件分支？

在我的一个客服 Agent 项目中，需要根据用户意图分发到不同的处理模块：

这种 if-else 逻辑在 LangGraph 中通过 conditional_edges 实现，让工作流具备真正的决策能力。

实战代码：构建条件路由工作流

环境准备

pip install langgraph langchain-openai langchain-core


配置 HolySheep AI（国内直连，延迟 <50ms）

import os
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"

基础条件分支实现

from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal


定义状态类型

class RouterState(TypedDict):
    user_input: str
    intent: str
    response: str


意图识别函数（使用 HolySheep API）

def intent_classifier(state: RouterState) -> RouterState:
    """根据用户输入判断意图"""
    from langchain_openai import ChatOpenAI
    llm = ChatOpenAI(
        model="gpt-4.1",
        api_key="YOUR_HOLYSHEEP_API_KEY",
        base_url="https://api.holysheep.ai/v1"
    )
    
    prompt = f"判断用户意图，只能返回以下之一：order/refund/complaint/chat\n用户输入：{state['user_input']}"
    result = llm.invoke(prompt)
    state["intent"] = result.content.strip().lower()
    return state


各模块处理函数

def order_handler(state: RouterState) -> RouterState:
    state["response"] = "📦 您的订单编号：ORD-2024-00187，当前状态：已发货"
    return state

def refund_handler(state: RouterState) -> RouterState:
    state["response"] = "💰 退款申请已提交，预计3个工作日到账"
    return state

def complaint_handler(state: RouterState) -> RouterState:
    state["response"] = "📞 人工客服将在24小时内联系您"
    return state

def chat_handler(state: RouterState) -> RouterState:
    state["response"] = "😊 您好，有什么可以帮您的？"
    return state


路由函数 - 条件分支核心

def route_based_on_intent(state: RouterState) -> Literal["order_node", "refund_node", "complaint_node", "chat_node"]:
    """根据意图路由到对应节点"""
    intent = state.get("intent", "chat")
    route_map = {
        "order": "order_node",
        "refund": "refund_node",
        "complaint": "complaint_node"
    }
    return route_map.get(intent, "chat_node")


构建图

graph = StateGraph(RouterState)


添加节点

graph.add_node("classify", intent_classifier)
graph.add_node("order_node", order_handler)
graph.add_node("refund_node", refund_handler)
graph.add_node("complaint_node", complaint_handler)
graph.add_node("chat_node", chat_handler)


设置入口

graph.set_entry_point("classify")


添加条件分支（核心语法）

graph.add_conditional_edges(
    "classify",
    route_based_on_intent,
    {
        "order_node": "order_node",
        "refund_node": "refund_node",
        "complaint_node": "complaint_node",
        "chat_node": "chat_node"
    }
)


所有处理节点结束后结束

for node in ["order_node", "refund_node", "complaint_node", "chat_node"]:
    graph.add_edge(node, END)


编译并测试

app = graph.compile()


测试用例

test_inputs = [
    {"user_input": "我想查一下我的订单", "intent": "", "response": ""},
    {"user_input": "商品坏了要退款", "intent": "", "response": ""},
    {"user_input": "服务态度太差要投诉", "intent": "", "response": ""}
]

for input_data in test_inputs:
    result = app.invoke(input_data)
    print(f"输入: {input_data['user_input']}")
    print(f"意图: {result['intent']} → 响应: {result['response']}")
    print("-" * 50)

带循环的多轮对话系统

在实际项目中，我经常需要实现多轮对话循环。下面的代码展示了如何用 while 循环让 Agent 持续交互，直到用户明确结束：

from langgraph.graph import StateGraph, END, START
from typing import Annotated
import operator

class ConversationState(TypedDict):
    messages: Annotated[list, operator.add]
    should_continue: bool
    turn_count: int

def chat_node(state: ConversationState) -> ConversationState:
    """对话节点 - 每次交互增加轮次计数"""
    from langchain_openai import ChatOpenAI
    
    llm = ChatOpenAI(
        model="claude-sonnet-4.5",
        api_key="YOUR_HOLYSHEEP_API_KEY",
        base_url="https://api.holysheep.ai/v1"
    )
    
    # 获取历史消息
    history = state["messages"]
    response = llm.invoke(history)
    
    state["messages"].append({"role": "assistant", "content": response.content})
    state["turn_count"] += 1
    return state

def should_continue(state: ConversationState) -> Literal["continue", "end"]:
    """
    循环退出条件判断
    - 对话超过10轮自动结束
    - 用户输入'再见'结束
    """
    # 检查最后一条用户消息
    if state["turn_count"] >= 10:
        return "end"
    
    messages = state["messages"]
    if messages and messages[-1].get("role") == "user":
        user_msg = messages[-1].get("content", "").lower()
        if "再见" in user_msg or "结束" in user_msg or "退出" in user_msg:
            return "end"
    
    return "continue"


构建循环工作流

loop_graph = StateGraph(ConversationState)

loop_graph.add_node("chat", chat_node)
loop_graph.set_entry_point(START)


条件循环边 - 实现 while 循环

loop_graph.add_conditional_edges(
    "chat",
    should_continue,
    {
        "continue": "chat",  # 循环回自身
        "end": END          # 退出循环
    }
)

loop_app = loop_graph.compile()


运行多轮对话

initial_state = {
    "messages": [{"role": "user", "content": "你好，帮我推荐一本书"}],
    "should_continue": True,
    "turn_count": 0
}

print("=== 开始多轮对话（最多10轮）===\n")
for step in loop_app.stream(initial_state):
    node_name = list(step.keys())[0]
    node_state = step[node_name]
    print(f"[轮次 {node_state.get('turn_count', 0)}] {node_name}")
    if node_state.get("messages"):
        last_msg = node_state["messages"][-1]
        print(f"  {last_msg['role']}: {last_msg['content'][:100]}...")
    print()

高级技巧：条件边中使用 LLM 做决策

在我开发一个复杂的审批流程时，纯代码的条件判断不够灵活，于是我在条件边中直接调用 LLM 来决定路由方向：

from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.tools import tool

@tool
def approve_request(request_id: str) -> str:
    """批准请求"""
    return f"请求 {request_id} 已批准"

@tool
def reject_request(request_id: str) -> str:
    """拒绝请求"""
    return f"请求 {request_id} 已拒绝"

@tool
def escalate_to_manager(request_id: str) -> str:
    """升级给经理处理"""
    return f"请求 {request_id} 已升级到经理"

tools = [approve_request, reject_request, escalate_to_manager]

def llm_routing_decision(state: dict) -> str:
    """
    使用 LLM 做路由决策
    根据请求内容金额和类型决定处理路径
    """
    from langchain_openai import ChatOpenAI
    
    llm = ChatOpenAI(
        model="gpt-4.1",
        api_key="YOUR_HOLYSHEEP_API_KEY",
        base_url="https://api.holysheep.ai/v1"
    )
    
    request_text = state.get("request_text", "")
    prompt = f"""分析以下请求，决定处理方式：
    - 如果金额 <1000元且是常规采购：返回 "small_request"
    - 如果金额 1000-10000元：返回 "medium_request"  
    - 如果金额 >10000元：返回 "large_request"
    - 如果是敏感类型（人事、财务）：返回 "sensitive_request"
    
    请求内容：{request_text}
    只返回一个分类标签"""
    
    decision = llm.invoke(prompt)
    return decision.content.strip().lower()


路由映射

def route_decision(decision: str) -> str:
    route_map = {
        "small_request": "small_node",
        "medium_request": "medium_node",
        "large_request": "large_node",
        "sensitive_request": "sensitive_node"
    }
    return route_map.get(decision, "default_node")

实战经验分享

在我使用 LangGraph 构建企业级 AI 工作流的一年多时间里，有几点经验特别想分享给刚入门的开发者：

第一，选择稳定的 API 供应商至关重要。我之前使用官方 API 时，高峰期的延迟波动经常导致我的条件分支超时，特别是那些需要多轮 LLM 调用的复杂路由逻辑。切换到 HolySheep AI 后，国内直连的 <50ms 延迟让整个工作流流畅很多，而且汇率优势（¥1=$1）让我在调试阶段可以大胆测试各种分支路径，而不用担心费用。

第二，条件分支的路由函数必须返回值在映射表中存在。这是我踩过最多次的坑——当 LLM 返回的意图不在预设的 route_map 中时，整个图会抛出 KeyError。解决方案是在路由函数中用 .get(intent, "default_node") 兜底。

第三，循环退出条件要设上界。我第一次实现多轮对话时忘记设置最大轮次，导致用户触发了一个 bug 后对话陷入了无限循环。使用 HolySheep AI 的低价优势，我建议在调试阶段就设置合理的轮次上限（比如 10-20 轮），避免意外开销。

常见报错排查

错误 1：KeyError in conditional_edges 路由映射

# ❌ 错误写法
def route_intent(state):
    return state["intent"]  # 可能返回不在映射表中的值

graph.add_conditional_edges(
    "classify",
    route_intent,
    {"order": "order_node", "refund": "refund_node"}  # 如果返回 "complaint" 会报错
)


✅ 正确写法 - 添加兜底默认值

def route_intent(state):
    route_map = {
        "order": "order_node",
        "refund": "refund_node",
        "complaint": "complaint_node"
    }
    return route_map.get(state["intent"], "default_node")  # 兜底

graph.add_conditional_edges(
    "classify",
    route_intent,
    {
        "order_node": "order_node",
        "refund_node": "refund_node",
        "complaint_node": "complaint_node",
        "default_node": "default_node"  # 需要包含兜底节点
    }
)

错误 2：循环节点未设置退出条件导致超时

# ❌ 危险写法 - 可能无限循环
def should_loop(state):
    # 只检查是否继续，没有上限
    return "continue" if not state["done"] else "end"


✅ 正确写法 - 强制退出机制

def should_loop(state):
    MAX_TURNS = 20
    
    # 检查轮次上限
    if state.get("turn_count", 0) >= MAX_TURNS:
        print(f"⚠️ 达到最大轮次 {MAX_TURNS}，强制退出")
        return "end"
    
    # 检查用户是否主动结束
    messages = state.get("messages", [])
    if messages and "再见" in messages[-1].get("content", ""):
        return "end"
    
    return "continue"

错误 3：API 认证失败或 base_url 配置错误

# ❌ 常见错误配置
os.environ["OPENAI_API_BASE"] = "https://api.openai.com/v1"  # 错误！
os.environ["OPENAI_API_KEY"] = "sk-xxxx"  # 使用了错误的 key 格式


✅ HolySheep AI 正确配置

import os
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"


如果是 Claude 模型，直接用相同 base_url

llm = ChatOpenAI(
    model="claude-sonnet-4.5",
    api_key="YOUR_HOLYSHEEP_API_KEY",  # 与 OpenAI 使用相同的 key
    base_url="https://api.holysheep.ai/v1"
)

错误 4：状态传递丢失问题

# ❌ 错误：直接覆盖状态而不是更新
def bad_node(state):
    return {"user_input": "new input"}  # 丢失了其他字段！


✅ 正确：保持状态连续性

def good_node(state):
    new_state = state.copy()  # 或 state | {"new_field": "value"}
    new_state["processed"] = True
    return new_state


或者在节点内直接修改

def best_node(state: RouterState) -> RouterState:
    state["processed"] = True
    state["response"] = "处理完成"
    return state

错误 5：conditional_edges 路由函数类型错误

# ❌ 错误：返回了状态对象而不是字符串
def wrong_route(state):
    return state  # 返回了整个状态，类型不对


✅ 正确：返回字符串路由键

def correct_route(state) -> str:
    return state.get("next_action", "default")


或者使用 Literal 类型提示

from typing import Literal

def typed_route(state) -> Literal["node_a", "node_b", "node_c"]:
    return state["route_key"]

性能与成本优化建议

在我优化 LangGraph 工作流时，发现几个显著影响成本和响应速度的点：

使用 HolySheep AI 的汇率优势（¥1=$1），我在生产环境中日均调用量约 5000 次，月度费用控制在 ¥800 以内，相比官方 API 节省超过 85% 成本。

总结

LangGraph 的条件分支和循环控制为构建智能 AI 工作流提供了强大的表达能力。通过本文的讲解，你应该已经掌握了：

如果你还没体验过 HolySheheep AI 的优势，强烈建议立即注册，利用首月赠送的免费额度开始你的 LangGraph 项目。国内直连 + 微信支付宝充值 + 无损汇率，让 AI 工作流开发不再受制于网络和成本问题。

完整的示例代码已在上文提供，你可以直接复制运行。如果有任何问题，欢迎在评论区交流！

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

相关资源

相关文章