Back to Details

langgraph-routing

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

LangGraph Conditional Routing

LangGraph 条件路由

Route workflow execution dynamically based on state.
基于状态动态路由工作流执行。

Basic Conditional Edge

基础条件边

python
from langgraph.graph import StateGraph, END

def route_based_on_quality(state: WorkflowState) -> str:
    """Decide next step based on quality score."""
    if state["quality_score"] >= 0.8:
        return "publish"
    elif state["retry_count"] < 3:
        return "retry"
    else:
        return "manual_review"

workflow.add_conditional_edges(
    "quality_check",
    route_based_on_quality,
    {
        "publish": "publish_node",
        "retry": "generator",
        "manual_review": "review_queue"
    }
)
python
from langgraph.graph import StateGraph, END

def route_based_on_quality(state: WorkflowState) -> str:
    """根据质量分数决定下一步。"""
    if state["quality_score"] >= 0.8:
        return "publish"
    elif state["retry_count"] < 3:
        return "retry"
    else:
        return "manual_review"

workflow.add_conditional_edges(
    "quality_check",
    route_based_on_quality,
    {
        "publish": "publish_node",
        "retry": "generator",
        "manual_review": "review_queue"
    }
)

Quality Gate Pattern

质量门模式

python
def route_after_quality_gate(state: AnalysisState) -> str:
    """Route based on quality gate result."""
    if state["quality_passed"]:
        return "compress_findings"
    elif state["retry_count"] < 2:
        return "supervisor"  # Retry
    else:
        return END  # Return partial results

workflow.add_conditional_edges(
    "quality_gate",
    route_after_quality_gate,
    {
        "compress_findings": "compress_findings",
        "supervisor": "supervisor",
        END: END
    }
)
python
def route_after_quality_gate(state: AnalysisState) -> str:
    """根据质量门结果进行路由。"""
    if state["quality_passed"]:
        return "compress_findings"
    elif state["retry_count"] < 2:
        return "supervisor"  # 重试
    else:
        return END  # 返回部分结果

workflow.add_conditional_edges(
    "quality_gate",
    route_after_quality_gate,
    {
        "compress_findings": "compress_findings",
        "supervisor": "supervisor",
        END: END
    }
)

Retry Loop Pattern

重试循环模式

python
def llm_call_with_retry(state):
    """Retry failed LLM calls."""
    try:
        result = call_llm(state["input"])
        state["output"] = result
        state["retry_count"] = 0
        return state
    except Exception as e:
        state["retry_count"] += 1
        state["error"] = str(e)
        return state

def should_retry(state) -> str:
    if state.get("output"):
        return "success"
    elif state["retry_count"] < 3:
        return "retry"
    else:
        return "failed"

workflow.add_conditional_edges(
    "llm_call",
    should_retry,
    {
        "success": "next_step",
        "retry": "llm_call",  # Loop back
        "failed": "error_handler"
    }
)
python
def llm_call_with_retry(state):
    """重试失败的LLM调用。"""
    try:
        result = call_llm(state["input"])
        state["output"] = result
        state["retry_count"] = 0
        return state
    except Exception as e:
        state["retry_count"] += 1
        state["error"] = str(e)
        return state

def should_retry(state) -> str:
    if state.get("output"):
        return "success"
    elif state["retry_count"] < 3:
        return "retry"
    else:
        return "failed"

workflow.add_conditional_edges(
    "llm_call",
    should_retry,
    {
        "success": "next_step",
        "retry": "llm_call",  # 循环返回
        "failed": "error_handler"
    }
)

Routing Patterns

路由模式

Sequential:    A → B → C              (simple edges)
Branching:     A → (B or C)           (conditional edges)
Looping:       A → B → A              (retry logic)
Convergence:   (A or B) → C           (multiple inputs)
Diamond:       A → (B, C) → D         (parallel then merge)
顺序执行:    A → B → C              (简单边)
分支执行:     A → (B 或 C)           (条件边)
循环执行:       A → B → A              (重试逻辑)
汇聚执行:   (A 或 B) → C           (多输入)
菱形模式:       A → (B, C) → D         (并行后合并)

State-Based Router

基于状态的路由器

python
def dynamic_router(state: WorkflowState) -> str:
    """Route based on multiple state conditions."""
    if state.get("error"):
        return "error_handler"
    if not state.get("validated"):
        return "validator"
    if state["confidence"] < 0.5:
        return "enhance"
    return "finalize"
python
def dynamic_router(state: WorkflowState) -> str:
    """基于多状态条件进行路由。"""
    if state.get("error"):
        return "error_handler"
    if not state.get("validated"):
        return "validator"
    if state["confidence"] < 0.5:
        return "enhance"
    return "finalize"

Command vs Conditional Edges (2026 Best Practice)

Command 与条件边（2026最佳实践）

python
from langgraph.types import Command
from typing import Literal
python
from langgraph.types import Command
from typing import Literal

Use CONDITIONAL EDGES when: Pure routing, no state updates

当仅需纯路由、无需更新状态时，使用 CONDITIONAL EDGES

def simple_router(state: WorkflowState) -> str: if state["score"] > 0.8: return "approve" return "reject"
workflow.add_conditional_edges("evaluate", simple_router)
def simple_router(state: WorkflowState) -> str: if state["score"] > 0.8: return "approve" return "reject"
workflow.add_conditional_edges("evaluate", simple_router)

Use COMMAND when: Updating state AND routing together

当需要同时更新状态和路由时，使用 COMMAND

def router_with_state(state: WorkflowState) -> Command[Literal["approve", "reject"]]: if state["score"] > 0.8: return Command( update={"route_reason": "high score", "routed_at": time.time()}, goto="approve" ) return Command( update={"route_reason": "low score", "routed_at": time.time()}, goto="reject" )
workflow.add_node("evaluate", router_with_state)
def router_with_state(state: WorkflowState) -> Command[Literal["approve", "reject"]]: if state["score"] > 0.8: return Command( update={"route_reason": "高分", "routed_at": time.time()}, goto="approve" ) return Command( update={"route_reason": "低分", "routed_at": time.time()}, goto="reject" )
workflow.add_node("evaluate", router_with_state)

No conditional edges needed - Command handles routing

无需条件边 - Command 会处理路由

undefined
undefined

Semantic Routing Implementation

语义路由实现

python
from sentence_transformers import SentenceTransformer
import numpy as np

embedder = SentenceTransformer("all-MiniLM-L6-v2")
python
from sentence_transformers import SentenceTransformer
import numpy as np

embedder = SentenceTransformer("all-MiniLM-L6-v2")

Pre-compute route embeddings

预计算路由嵌入向量

ROUTE_EMBEDDINGS = { "technical": embedder.encode("technical implementation code programming engineering"), "business": embedder.encode("business strategy revenue customers sales marketing"), "support": embedder.encode("help troubleshoot error problem fix support issue"), "creative": embedder.encode("design creative writing content marketing copy"), }
def semantic_router(state: WorkflowState) -> str: """Route based on semantic similarity.""" query = state["query"] query_embedding = embedder.encode(query)
# Calculate cosine similarities
similarities = {}
for route, route_embedding in ROUTE_EMBEDDINGS.items():
    similarity = np.dot(query_embedding, route_embedding) / (
        np.linalg.norm(query_embedding) * np.linalg.norm(route_embedding)
    )
    similarities[route] = similarity

# Return highest similarity route
best_route = max(similarities, key=similarities.get)

# Optional: threshold check
if similarities[best_route] < 0.3:
    return "general"  # Fallback

return best_route
workflow.add_conditional_edges( "classifier", semantic_router, { "technical": "tech_agent", "business": "business_agent", "support": "support_agent", "creative": "creative_agent", "general": "general_agent" } )
undefined
ROUTE_EMBEDDINGS = { "technical": embedder.encode("technical implementation code programming engineering"), "business": embedder.encode("business strategy revenue customers sales marketing"), "support": embedder.encode("help troubleshoot error problem fix support issue"), "creative": embedder.encode("design creative writing content marketing copy"), }
def semantic_router(state: WorkflowState) -> str: """基于语义相似度进行路由。""" query = state["query"] query_embedding = embedder.encode(query)
# 计算余弦相似度
similarities = {}
for route, route_embedding in ROUTE_EMBEDDINGS.items():
    similarity = np.dot(query_embedding, route_embedding) / (
        np.linalg.norm(query_embedding) * np.linalg.norm(route_embedding)
    )
    similarities[route] = similarity

# 返回相似度最高的路由
best_route = max(similarities, key=similarities.get)

# 可选：阈值检查
if similarities[best_route] < 0.3:
    return "general"  # 兜底路由

return best_route
workflow.add_conditional_edges( "classifier", semantic_router, { "technical": "tech_agent", "business": "business_agent", "support": "support_agent", "creative": "creative_agent", "general": "general_agent" } )
undefined

Key Decisions

关键决策建议

DecisionRecommendation
Max retries2-3 for LLM calls
FallbackAlways have END fallback
Routing functionKeep pure (no side effects)
Edge mappingExplicit mapping for clarity
Command vs ConditionalCommand when updating state + routing
Semantic routingPre-compute embeddings, use cosine similarity
决策项建议方案
最大重试次数LLM调用建议设置2-3次
兜底路由始终设置END作为兜底
路由函数保持纯函数（无副作用）
边映射使用显式映射以提升可读性
Command vs 条件边同时更新状态+路由时使用Command
语义路由预计算嵌入向量，使用余弦相似度

Common Mistakes

常见错误

  • No END fallback (workflow hangs)
  • Infinite loops (no max retry)
  • Side effects in router (hard to debug)
  • Missing edge mappings (runtime error)
  • 未设置END兜底（导致工作流挂起）
  • 无限循环（未设置最大重试次数）
  • 路由函数中存在副作用（难以调试）
  • 缺失边映射（运行时错误）

Evaluations

评估用例

See references/evaluations.md for test cases.
查看 references/evaluations.md 获取测试用例。

Related Skills

相关技能

  • langgraph-state
    - State design for routing decisions
  • langgraph-supervisor
    - Supervisor pattern with dynamic routing
  • langgraph-parallel
    - Route to parallel branches
  • langgraph-human-in-loop
    - Route based on human decisions
  • langgraph-tools
    - Route after tool execution results
  • agent-loops
    - ReAct loop patterns with conditional routing
  • langgraph-state
    - 为路由决策设计状态
  • langgraph-supervisor
    - 带动态路由的监督者模式
  • langgraph-parallel
    - 路由至并行分支
  • langgraph-human-in-loop
    - 基于人工决策进行路由
  • langgraph-tools
    - 根据工具执行结果路由
  • agent-loops
    - 带条件路由的ReAct循环模式

Capability Details

能力详情

conditional-routing

conditional-routing

Keywords: conditional, branch, decision, if-else Solves:
  • Route based on conditions
  • Implement branching logic
  • Create decision nodes
关键词: conditional, branch, decision, if-else 解决场景:
  • 基于条件路由
  • 实现分支逻辑
  • 创建决策节点

semantic-routing

semantic-routing

Keywords: semantic, embedding, similarity, intent Solves:
  • Route by semantic similarity
  • Intent-based routing
  • Embedding-based decisions
关键词: semantic, embedding, similarity, intent 解决场景:
  • 基于语义相似度路由
  • 基于意图的路由
  • 基于嵌入向量的决策

router-template

router-template

Keywords: template, router, semantic, implementation Solves:
  • Semantic router template
  • Production router code
  • Copy-paste implementation
关键词: template, router, semantic, implementation 解决场景:
  • 语义路由器模板
  • 生产级路由器代码
  • 可直接复用的实现方案