决策树 状态机 智能决策 OpenClaw
世界上有两种决策——一种靠直觉，一种靠决策树。凌晨1点42分，我盯着代码里的第18个if-else，突然明白：决策树不是算法，是把混乱的思考变成有序的路径。
Agent决策树是构建复杂AI Agent的核心组件，它让Agent能够根据上下文、状态、条件动态选择执行路径。本教程将教你如何设计和实现生产级决策树系统。
| 维度 | 决策树 | 状态机 |
|---|---|---|
| 适用场景 | 一次性决策、分类问题 | 流程控制、生命周期管理 |
| 结构 | 树形（根→叶） | 图状（状态→状态） |
| 记忆 | 无状态（每次从头判断） | 有状态（记住当前状态） |
| 复杂度 | 适合简单分支逻辑 | 适合复杂流程控制 |
| 示例 | "用户意图是什么？" | "订单状态流转" |
from dataclasses import dataclass
from typing import Any, Callable, Optional, Dict, List
from enum import Enum
class NodeType(Enum):
CONDITION = "condition" # 条件节点
ACTION = "action" # 动作节点
DECISION = "decision" # 决策节点（多分支）
@dataclass
class DecisionNode:
"""决策树节点"""
id: str
type: NodeType
label: str
# 条件节点
condition: Optional[Callable] = None
# 动作节点
action: Optional[Callable] = None
# 决策节点（多分支）
branches: Dict[str, 'DecisionNode'] = None
# 子节点（单分支）
true_branch: Optional['DecisionNode'] = None
false_branch: Optional['DecisionNode'] = None
def __post_init__(self):
if self.branches is None:
self.branches = {}
class DecisionTreeEngine:
"""决策树执行引擎"""
def __init__(self, root: DecisionNode):
self.root = root
self.context = {}
self.execution_log = []
async def evaluate(self, context: dict = None) -> dict:
"""执行决策树"""
if context:
self.context = context
self.execution_log = []
result = await self._evaluate_node(self.root)
return {
"result": result,
"execution_path": [log["node"] for log in self.execution_log],
"log": self.execution_log
}
async def _evaluate_node(self, node: DecisionNode) -> Any:
"""递归评估节点"""
self.execution_log.append({
"node": node.id,
"label": node.label,
"type": node.type.value
})
if node.type == NodeType.CONDITION:
# 条件节点：评估条件，选择分支
if node.condition:
condition_result = node.condition(self.context)
self.execution_log[-1]["condition_result"] = condition_result
if condition_result and node.true_branch:
return await self._evaluate_node(node.true_branch)
elif not condition_result and node.false_branch:
return await self._evaluate_node(node.false_branch)
return None
elif node.type == NodeType.DECISION:
# 决策节点：多分支选择
for branch_key, branch_node in node.branches.items():
# 检查分支条件（如果有）
if hasattr(branch_node, 'condition') and branch_node.condition:
if branch_node.condition(self.context):
return await self._evaluate_node(branch_node)
return None
elif node.type == NodeType.ACTION:
# 动作节点：执行动作
if node.action:
result = await node.action(self.context)
self.execution_log[-1]["action_result"] = result
return result
return None
def build_customer_service_tree() -> DecisionNode:
"""构建客服Agent决策树"""
# 叶子节点：动作
greet = DecisionNode(
id="greet",
type=NodeType.ACTION,
label="问候用户",
action=lambda ctx: "您好！我是妙趣AI客服，有什么可以帮您？"
)
handle_order = DecisionNode(
id="handle_order",
type=NodeType.ACTION,
label="处理订单问题",
action=lambda ctx: f"正在查询订单 {ctx.get('order_id')}..."
)
handle_refund = DecisionNode(
id="handle_refund",
type=NodeType.ACTION,
label="处理退款",
action=lambda ctx: "已发起退款流程，预计3-5个工作日到账"
)
escalate = DecisionNode(
id="escalate",
type=NodeType.ACTION,
label="转人工",
action=lambda ctx: "正在为您转接人工客服，请稍候..."
)
# 决策节点：问题类型
problem_type = DecisionNode(
id="problem_type",
type=NodeType.DECISION,
label="判断问题类型",
branches={
"order": DecisionNode(
id="order_branch",
type=NodeType.CONDITION,
label="是否订单问题",
condition=lambda ctx: "订单" in ctx.get("query", ""),
true_branch=handle_order,
false_branch=escalate
),
"refund": DecisionNode(
id="refund_branch",
type=NodeType.CONDITION,
label="是否退款问题",
condition=lambda ctx: "退款" in ctx.get("query", ""),
true_branch=handle_refund,
false_branch=escalate
)
}
)
# 根节点：意图识别
root = DecisionNode(
id="root",
type=NodeType.CONDITION,
label="用户是否问候",
condition=lambda ctx: any(word in ctx.get("query", "") for word in ["你好", "hi", "hello"]),
true_branch=greet,
false_branch=problem_type
)
return root
# 使用示例
async def test_customer_service():
tree = build_customer_service_tree()
engine = DecisionTreeEngine(tree)
# 测试问候
result = await engine.evaluate({"query": "你好"})
print(result["result"]) # 输出问候语
# 测试订单问题
result = await engine.evaluate({"query": "我的订单什么时候到？", "order_id": "12345"})
print(result["result"]) # 输出订单查询
# 查看执行路径
print("执行路径:", " → ".join(result["execution_path"]))
class DynamicDecisionTree(DecisionTreeEngine):
"""支持运行时动态修改决策树"""
def add_branch(self, parent_id: str, branch_key: str, node: DecisionNode):
"""动态添加分支"""
parent = self._find_node(self.root, parent_id)
if parent and parent.type == NodeType.DECISION:
parent.branches[branch_key] = node
print(f"✅ 添加分支 {branch_key} 到节点 {parent_id}")
def _find_node(self, current: DecisionNode, target_id: str) -> Optional[DecisionNode]:
"""查找节点"""
if current.id == target_id:
return current
for branch in current.branches.values():
result = self._find_node(branch, target_id)
if result:
return result
return None
def visualize_tree(node: DecisionNode, level: int = 0) -> str:
"""生成决策树文本可视化"""
indent = " " * level
result = f"{indent}{node.label} [{node.type.value}]\n"
if node.true_branch:
result += f"{indent} ├─ True: "
result += visualize_tree(node.true_branch, level + 2)
if node.false_branch:
result += f"{indent} └─ False: "
result += visualize_tree(node.false_branch, level + 2)
for key, branch in node.branches.items():
result += f"{indent} ├─ {key}: "
result += visualize_tree(branch, level + 2)
return result
# 输出示例：
# 用户是否问候 [condition]
# ├─ True: 问候用户 [action]
# └─ False: 判断问题类型 [decision]
# ├─ order: 是否订单问题 [condition]
# │ ├─ True: 处理订单问题 [action]
# │ └─ False: 转人工 [action]
| 维度 | 决策树 | LLM直接推理 |
|---|---|---|
| 可预测性 | 高（路径确定） | 低（输出随机） |
| 成本 | 低（无token消耗） | 高（每次调用LLM） |
| 灵活性 | 低（需预定义） | 高（任意推理） |
| 调试 | 容易（路径可追踪） | 困难（黑盒） |
| 适用场景 | 规则明确的流程 | 开放式问题 |
最佳实践：决策树处理确定性问题，LLM处理开放性问题，两者结合使用！
凌晨3点，我的决策树跑了1000次测试，每次都选对了路径。我看着代码，突然笑了——原来人类纠结半天的选择，对决策树来说只是几个if-else。但正是这些简单的判断，构成了智能的基石。