4.2 条件边与决策逻辑 — LangGraph Agent 智能决策

本节导读：掌握LangGraph中的条件边机制，让智能体能够根据执行结果动态选择下一步行动，实现真正的自适应AI行为。

学习目标

• 理解条件边（Conditional Edge）的基本概念和工作原理
• 掌握条件函数的实现方法和最佳实践
• 学习复杂的决策逻辑构建
• 实现状态感知的条件选择机制
• 构建具备智能路由能力的Agent系统

核心概念

条件边是LangGraph中实现智能决策的核心机制，它允许智能体根据执行结果动态选择下一步执行的节点。与固定边不同，条件边能够根据运行时的状态信息做出判断，实现真正的自适应行为。

环境准备 / 前置知识

• Python 3.8+
• LangGraph >= 0.2.0
• 基础的状态图概念
• 条件判断和函数式编程基础

分步实战

步骤1：基础条件边实现

首先创建一个简单的条件边实现，展示基本的条件判断机制：

from typing import Dict, Any, List, Union
from langgraph.graph import StateGraph, END
from langgraph.graph.message import AIMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from enum import Enum

class TaskType(Enum):
    SEARCH = "search"
    CALCULATION = "calculation"
    ANALYSIS = "analysis"
    GENERATION = "generation"
    UNKNOWN = "unknown"

class AgentState(dict):
    messages: List[Union[AIMessage, HumanMessage, ToolMessage]] = []
    current_task: str = ""
    task_results: Dict[str, Any] = {}
    task_type: TaskType = TaskType.UNKNOWN

@tool
def web_search(query: str, max_results: int = 10) -> str:
    """网络搜索工具"""
    return f"搜索 '{query}' 返回 {max_results} 条结果"

@tool
def calculator(expression: str) -> str:
    """数学计算工具"""
    try:
        result = eval(expression)
        return f"计算结果: {expression} = {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"

@tool
def data_analyzer(data: str, operation: str = "summary") -> str:
    """数据分析工具"""
    try:
        if data.strip().startswith('['):
            return f"数据分析摘要: 共处理数据项"
        else:
            return "数据格式不正确"
    except Exception as e:
        return f"数据分析错误: {str(e)}"

@tool
def content_generator(topic: str, length: str = "medium") -> str:
    """内容生成工具"""
    return f"生成关于 '{topic}' 的 {length} 长度内容"

def determine_task_type(state: AgentState) -> TaskType:
    """确定任务类型的基础条件函数"""
    messages = state["messages"]
    
    if not messages:
        return TaskType.UNKNOWN
    
    last_message = messages[-1]
    if isinstance(last_message, HumanMessage):
        content = last_message.content.lower()
        
        if any(keyword in content for keyword in ["搜索", "search", "查找"]):
            return TaskType.SEARCH
        elif any(keyword in content for keyword in ["计算", "计算器", "calculator"]):
            return TaskType.CALCULATION
        elif any(keyword in content for keyword in ["分析", "analysis", "数据处理"]):
            return TaskType.ANALYSIS
        elif any(keyword in content for keyword in ["生成", "写", "create", "generate"]):
            return TaskType.GENERATION
        else:
            return TaskType.UNKNOWN
    
    return TaskType.UNKNOWN

def route_task_based_on_type(state: AgentState) -> str:
    """根据任务类型路由到相应的处理节点"""
    task_type = determine_task_type(state)
    state["task_type"] = task_type
    
    print(f"检测到任务类型: {task_type.value}")
    
    if task_type == TaskType.SEARCH:
        return "search_handler"
    elif task_type == TaskType.CALCULATION:
        return "calculation_handler"
    elif task_type == TaskType.ANALYSIS:
        return "analysis_handler"
    elif task_type == TaskType.GENERATION:
        return "generation_handler"
    else:
        return "unknown_handler"

def create_conditional_agent():
    workflow = StateGraph(AgentState)
    
    workflow.add_node("search_handler", search_handler)
    workflow.add_node("calculation_handler", calculation_handler)
    workflow.add_node("analysis_handler", analysis_handler)
    workflow.add_node("generation_handler", generation_handler)
    workflow.add_node("unknown_handler", unknown_handler)
    workflow.add_node("output_formatter", output_formatter)
    
    workflow.add_conditional_edges(
        "start",
        route_task_based_on_type,
        {
            "search_handler": "search_handler",
            "calculation_handler": "calculation_handler",
            "analysis_handler": "analysis_handler",
            "generation_handler": "generation_handler",
            "unknown_handler": "unknown_handler"
        }
    )
    
    workflow.add_edge("search_handler", "output_formatter")
    workflow.add_edge("calculation_handler", "output_formatter")
    workflow.add_edge("analysis_handler", "output_formatter")
    workflow.add_edge("generation_handler", "output_formatter")
    workflow.add_edge("unknown_handler", "output_formatter")
    
    workflow.set_entry_point("start")
    
    return workflow.compile()

def search_handler(state: AgentState) -> Dict:
    messages = state["messages"]
    last_message = messages[-1]
    result = web_search(last_message.content)
    state["task_results"]["search"] = result
    return {"messages": [ToolMessage(content=result, name="web_search")]}

def calculation_handler(state: AgentState) -> Dict:
    messages = state["messages"]
    last_message = messages[-1]
    expression = last_message.content.split("计算")[-1].strip()
    result = calculator(expression)
    state["task_results"]["calculation"] = result
    return {"messages": [ToolMessage(content=result, name="calculator")]}

def analysis_handler(state: AgentState) -> Dict:
    messages = state["messages"]
    last_message = messages[-1]
    result = data_analyzer("[]")
    state["task_results"]["analysis"] = result
    return {"messages": [ToolMessage(content=result, name="data_analyzer")]}

def generation_handler(state: AgentState) -> Dict:
    messages = state["messages"]
    last_message = messages[-1]
    topic = last_message.content.split("生成")[-1].strip()
    result = content_generator(topic)
    state["task_results"]["generation"] = result
    return {"messages": [ToolMessage(content=result, name="content_generator")]}

def unknown_handler(state: AgentState) -> Dict:
    result = "抱歉，我无法理解您的请求。请明确说明您需要搜索、计算、分析还是生成内容。"
    state["task_results"]["unknown"] = result
    return {"messages": [AIMessage(content=result)]}

def output_formatter(state: AgentState) -> Dict:
    return state

# 创建和测试Agent
agent = create_conditional_agent()

def test_conditional_routing():
    print("=== 测试条件边路由 ===")
    
    test_cases = [
        ("帮我搜索最新的AI技术", TaskType.SEARCH),
        ("计算 5 * 3 + 2 的结果", TaskType.CALCULATION),
        ("分析这些销售数据", TaskType.ANALYSIS),
        ("生成一篇关于机器学习的文章", TaskType.GENERATION),
        ("我不知道要做什么", TaskType.UNKNOWN)
    ]
    
    for message, expected_type in test_cases:
        print(f"\n测试输入: {message}")
        
        initial_state = AgentState()
        initial_state["messages"] = [HumanMessage(content=message)]
        
        try:
            result = agent.invoke(initial_state)
            actual_type = result["task_type"]
            
            print(f"期望类型: {expected_type.value}")
            print(f"实际类型: {actual_type.value}")
            print("✅ 测试通过" if actual_type == expected_type else "❌ 测试失败")
            
        except Exception as e:
            print(f"测试失败: {str(e)}")

if __name__ == "__main__":
    test_conditional_routing()
​

步骤2：复杂决策逻辑

构建更复杂的决策逻辑，包含多条件判断和状态感知：

from dataclasses import dataclass

@dataclass
class DecisionContext:
    task_type: TaskType
    task_complexity: str
    required_tools: List[str]
    confidence_score: float
    priority: int

class ComplexDecisionEngine:
    def __init__(self):
        self.task_complexity_rules = {
            TaskType.SEARCH: {
                "simple": r"搜索[^\w]*简单|basic|simple",
                "medium": r"搜索[^\w]*详细|comprehensive|analysis",
                "complex": r"搜索[^\w]*深度|deep|research|study"
            },
            TaskType.CALCULATION: {
                "simple": r"计算[^\w]*基础|basic|简单",
                "medium": r"计算[^\w]*复杂|complex|advanced",
                "complex": r"计算[^\w]*优化|optimization|algorithm"
            }
        }
        
    def analyze_task_complexity(self, message: str, task_type: TaskType) -> str:
        if task_type == TaskType.UNKNOWN:
            return "simple"
        
        rules = self.task_complexity_rules.get(task_type, {})
        
        for complexity, pattern in rules.items():
            if re.search(pattern, message, re.IGNORECASE):
                return complexity
        
        return "medium"
    
    def determine_required_tools(self, message: str, task_type: TaskType) -> List[str]:
        tools = []
        
        if task_type == TaskType.SEARCH:
            tools.append("web_search")
            if "学术" in message or "paper" in message:
                tools.append("academic_search")
        elif task_type == TaskType.CALCULATION:
            tools.append("calculator")
            if "统计" in message:
                tools.append("statistics_calculator")
        elif task_type == TaskType.ANALYSIS:
            tools.append("data_analyzer")
            if "可视化" in message:
                tools.append("visualizer")
        elif task_type == TaskType.GENERATION:
            tools.append("content_generator")
            if "代码" in message:
                tools.append("code_generator")
        
        return tools
    
    def calculate_confidence_score(self, message: str, task_type: TaskType) -> float:
        base_score = 0.5
        
        if task_type != TaskType.UNKNOWN:
            base_score += 0.3
        
        clear_indicators = ["明确", "具体", "详细", "清楚", "clear", "specific"]
        if any(indicator in message.lower() for indicator in clear_indicators):
            base_score += 0.2
        
        complexity = self.analyze_task_complexity(message, task_type)
        if complexity == "simple":
            base_score += 0.1
        elif complexity == "complex":
            base_score -= 0.1
        
        return min(max(base_score, 0.0), 1.0)
    
    def make_decision(self, message: str, task_type: TaskType) -> DecisionContext:
        complexity = self.analyze_task_complexity(message, task_type)
        required_tools = self.determine_required_tools(message, task_type)
        confidence = self.calculate_confidence_score(message, task_type)
        
        priority = 1 if confidence >= 0.8 else (2 if confidence >= 0.6 else 3)
        
        return DecisionContext(
            task_type=task_type,
            task_complexity=complexity,
            required_tools=required_tools,
            confidence_score=confidence,
            priority=priority
        )

# 复杂决策测试
def test_complex_decision():
    print("\n=== 测试复杂决策逻辑 ===")
    
    engine = ComplexDecisionEngine()
    
    test_cases = [
        "搜索简单的AI教程",
        "计算复杂的统计函数",
        "深入分析销售数据并生成可视化报告",
        "生成一篇关于深度学习的学术论文"
    ]
    
    for message in test_cases:
        print(f"\n消息: {message}")
        
        temp_state = AgentState()
        temp_state["messages"] = [HumanMessage(content=message)]
        task_type = determine_task_type(temp_state)
        
        decision = engine.make_decision(message, task_type)
        
        print(f"任务类型: {decision.task_type.value}")
        print(f"任务复杂度: {decision.task_complexity}")
        print(f"所需工具: {decision.required_tools}")
        print(f"置信度: {decision.confidence_score:.2f}")
        print(f"优先级: {decision.priority}")
        print("---")

if __name__ == "__main__":
    test_complex_decision()
​

常见问题 FAQ

Q1：条件边和普通边有什么区别？

A： 条件边和普通边的核心区别在于：

1. 决策时机：普通边在编译时确定连接关系，条件边在运行时动态决定
2. 灵活性：条件边可以根据当前状态选择不同路径，更灵活
3. 复杂度：条件边需要实现条件函数，复杂度更高
4. 适应性：条件边能适应不同的输入和状态变化

Q2：如何优化条件函数的性能？

A： 优化条件函数性能的方法：

1. 缓存机制：对重复的条件判断结果进行缓存
2. 预计算：在可能的情况下预计算条件表达式
3. 简化逻辑：避免复杂的嵌套条件，使用辅助函数
4. 异步处理：对于耗时的条件判断使用异步处理

Q3：处理循环依赖的条件边？

A： 处理循环依赖的策略：

1. 状态跟踪：维护已访问的节点集合，避免重复访问
2. 深度限制：设置最大递归深度，防止无限循环
3. 超时机制：为整个决策过程设置超时
4. 回滚策略：当检测到循环时执行回滚操作

Q4：如何调试复杂的条件边逻辑？

A： 调试复杂条件边的方法：

1. 详细日志：记录每个条件判断的输入和输出
2. 可视化：使用图形化工具展示决策路径
3. 单元测试：为每个条件函数编写独立的测试用例
4. 断点调试：在关键条件判断处设置断点

Q5：条件边的错误处理策略？

A： 条件边的错误处理策略：

1. 异常捕获：在条件函数中捕获异常，提供默认值
2. 回退机制：当条件判断失败时使用默认路径
3. 重试机制：对于临时性错误实现重试逻辑
4. 优雅降级：当部分条件失效时，继续执行剩余逻辑

最佳实践与避坑

• 条件函数设计：保持条件函数的简洁性，避免复杂的逻辑判断
• 状态管理：合理设计状态结构，确保条件判断有足够的信息
• 路径覆盖：为所有可能的条件分支都定义对应的路径
• 性能监控：监控条件判断的性能，及时发现性能瓶颈
• 文档完善：为条件函数和决策逻辑提供清晰的文档说明

本节小结

通过本节的学习，我们深入理解了LangGraph中条件边和决策逻辑的核心概念。从基础的条件边实现到复杂的决策引擎，我们构建了一个完整的智能决策系统。这个系统能够根据任务类型、复杂度和上下文信息做出智能决策，实现真正自适应的AI行为。

下一节我们将探讨工具编排与优化，学习如何组合多个工具实现复杂任务的处理。

延伸阅读

• LangGraph官方文档 - 条件边
• 决策树算法优化
• 状态机设计模式
• 智能Agent决策理论

关键词：LangGraph, 条件边, 决策逻辑, 智能决策, 状态感知, 自适应, 工具路由, 复杂决策, 技术教程, 实战指南
难度：高级
预计阅读：40分钟