基于规则的单区域共享出行Agent实现

场景

工作日下午五点，你站在体育场外，刚刚结束的一场活动散场了。你打开打车应用，预计等待时间：12分钟。与此同时，三个街区之外，司机们正在空荡荡的街道上转圈，没有人在叫车。

这种情况在任何一个打车平台上都在不断发生。这不是 bug——只是供需自行失衡了。

某个角色——人或系统——必须监控城市的每个区域。它需要察觉某个区域何时失衡，然后决定该怎么做：

涨价，以吸引更多司机

给司机发奖金，让他们重新布局

引导乘客去一个相对空闲的区域

或者干脆不管，如果它自己能恢复平衡的话

这个决策者，就是我用 LangGraph 在这个系列中构建的东西——一次一个能力。

这是五部分系列的第一部分。每一部分在前一部分的基础上精确地增加一个真正有意义的新能力，让这个 agent 越来越接近生产环境实际需要的形态：

第一部分（也就是这篇）：针对单一区域的基于规则的 agent——没有 LLM，一切都是结构化的数字

第二部分：引入 LLM，但只用于查表做不到的两件事——读取人类的自由文本备注，以及用自然语言解释决策结果

第三部分：加入记忆，让 agent 能记住它在一个区域上已经尝试过什么，而不是决策一次就忘掉

第四部分：加入人工介入的暂停点，让有风险或不寻常的决策不会无人看管地直接执行

第五部分：同时协调两个区域，因为从一个区域抽调司机是一个会影响两个区域的决策

第一部分里，上述所有能力都还没有构建。这是刻意为之的。上述每一项新增都是真实的复杂性，而真实的复杂性只应该在问题真正需要它时才出现——不是因为听起来很酷。

所以第一部分保持在这个问题的复杂程度允许的最低限度：一个区域，一次决策循环，以及一个运营人员可以在白板上写出来的规则。

数据：城市的一个快照

在有 agent 之前，先有一个城市需要观察。我生成了 8 个区域的城市快照——市中心、机场、几个郊区、一个大学区，等等。

包括：

一个乘客请求数量

几个上下文标志——是否下雨、附近是否有活动

zones = generate_all_zones(hour=17, rain=False, event=True, seed=42)

df = pd.DataFrame(zones)[[
    'zone_name', 'zone_type', 'driver_count',
    'rider_request_count', 'avg_wait_time', 'rain_flag', 'event_flag', 'traffic'
]]
df['ratio'] = (df['rider_request_count'] / df['driver_count']).round(2)
df

ratio 列是乘客请求数除以司机数。这就是 agent 需要的全部信号：

高于 1.3 → 供给不足（司机太少）

低于 0.5 → 供给过剩（闲置司机太多）

介于两者之间 → 平衡

市中心核心区的 ratio 是 1.93——真实的供给不足，加上附近有活动导致情况更糟。这就是我接下来要逐步演示的区域。

决策的完整流程

以下是决策的端到端形态：

detect_imbalance
      ↓
classify_severity
      ↓ (conditional edge)
 ┌────┴────────────┬──────────────┐
 ↓                 ↓              ↓
demand_action   supply_action  do_nothing
 └────────┬────────┴──────────────┘
          ↓
   resolved_imbalance
          ↓
   choose_best_policy
          ↓
   simulate_and_report

detect_imbalance 计算 ratio 值。

classify_severity 将其标记为轻度、中度或严重。这个标记仅用于报告——不影响决策。

然后是条件边，真实的分支在这里发生。它读取 imbalance_type 并决定哪些策略值得尝试：

供给不足 → surge_pricing、driver_bonus、demand_redirect 或 do_nothing

供给过剩 → reallocation_nudge 或 do_nothing（一个闲置司机太多的区域不需要需求侧的激励）

平衡 → do_nothing

之后还有两个节点运行：

resolved_imbalance 对每个候选策略评估一次。它记录每个策略的利润，以及它是否真正解决了失衡。

choose_best_policy 在所有能解决失衡的策略中选择利润最高的。如果没有一个能解决失衡，就退而求其次，选择总体利润最高的。

还有一个值得专门指出的点：这里的每个策略都是区域本地的。没有任何策略会查看所选区域自身数据之外的信息。

从邻近区域调入司机是一个真实可行的想法。但这需要先有两个区域在视野中——那是第五部分的内容，不是第一部分。

LangGraph 的 StateGraph

LangGraph 的 StateGraph 运行在单一共享状态上。每个节点都从状态中读取并写入状态。在 Python 中，这只是一个 TypedDict：

class ZoneState(TypedDict):
    zone:                dict  # the single zone this graph run studies
    imbalance_ratio:     float
    imbalance_type:      str   # "deficit" | "surplus" | "balanced"
    severity:            str   # "none" | "mild" | "moderate" | "critical"
    candidate_policies:  list  # which policies are worth evaluating, set by the routed branch
    policy_evaluations:  dict  # {policy_name: profit} for every candidate considered
    policy_resolutions:  dict  # {policy_name: bool} — did this candidate resolve the imbalance
    recommended_policy:  str
    outcome:             dict
    outcome_delta:       dict
    report:              str

这里没有任何 LangGraph 特有的魔法。只是一个普通字典的结构。框架唯一的工作是确保每个节点都认同这个结构。

每个节点都是一个普通的 Python 函数。它接收当前状态作为输入，然后返回它所修改的字段。节点之间不直接互相调用——由图的边来决定下一步运行什么。

将它们连接在一起需要三个步骤：

1. 为每个节点注册一个名称
2. 用 add_edge 连接那些直通的节点
3. 在唯一需要根据状态决定下一步的地方使用 add_conditional_edges

g = StateGraph(ZoneState)

g.add_node("detect_imbalance",    detect_imbalance)
g.add_node("classify_severity",   classify_severity)
g.add_node("demand_action",       demand_action)
g.add_node("supply_action",       supply_action)
g.add_node("do_nothing",          do_nothing)
g.add_node("resolved_imbalance",  resolved_imbalance)
g.add_node("choose_best_policy",  choose_best_policy)
g.add_node("simulate_and_report", simulate_and_report)

g.add_edge(START, "detect_imbalance")
g.add_edge("detect_imbalance", "classify_severity")

g.add_conditional_edges("classify_severity", route_action, {
    "demand_action": "demand_action",
    "supply_action": "supply_action",
    "do_nothing":    "do_nothing",
})

g.add_edge("demand_action",       "resolved_imbalance")
g.add_edge("supply_action",       "resolved_imbalance")
g.add_edge("do_nothing",          "resolved_imbalance")
g.add_edge("resolved_imbalance",  "choose_best_policy")
g.add_edge("choose_best_policy",  "simulate_and_report")
g.add_edge("simulate_and_report", END)

app = g.compile()

route_action 是那个条件边背后的函数。它从状态中读取 imbalance_type，然后返回要走分支的名称。

这就是这个阶段全部的路由逻辑——一个 if 形状的决策，但表达为数据而不是一串 if 语句。

在市中心核心区上运行

图编译完成后，运行它只需一次调用：invoke()，以市中心核心区的快照作为起始状态。

result = app.invoke(initial_state)

以下是返回结果：

surge_pricing 在利润上胜出——这与没有 resolution 检查时的结果相同。但注意，没有任何策略在一个周期内解决了失衡。

这不是 bug。resolved_imbalance 只衡量一个策略在 15 分钟窗口内的效果。如此规模的供给不足——29 个乘客对 15 个司机——确实需要不止一次干预才能关闭缺口。

Agent 仍然做出了它能做的最佳选择。它只是无法承诺在单次尝试中完全解决问题，而且它诚实地说出了这一点（resolved=NO），而不是假装没事。赋予它再次尝试的能力，并记住它已经尝试过什么——这正是第三部分要添加的内容。

快进结果

最后一个节点是 simulate_and_report。它将选中的策略向前推演 15 分钟。

模拟是随机的。司机对激励的响应是概率性的，就像真实的人一样——而不是按固定时间表行动。

方向正确：更多司机、更少待处理请求、更短的等待时间。只是不足以在一个周期内将市中心核心区从供给不足扭转为平衡状态。

这个 agent 在这里查看的一切都是已经结构化的——数字和布尔标志。这正是普通规则能够处理它的原因。没有任何歧义需要语言模型来化解。

但运营人员不会写结构化字段。他们写的是类似这样的话："体育场活动刚刚取消了，而且附近发生了一场事故导致交通堵塞。"第一部分的 agent 没有办法读取这样的句子。没有任何查表能解决这个问题——它需要一个 LLM，用于恰好一个狭义的任务。这就是第二部分要切入的地方。

本系列代码：github.com/ebiarian/zone-balancing-ridesharing-langgraph-agent

下一篇——第二部分：教 agent 读取人类运营备注，并用自然语言解释它自己的决策——以及对五款本地 LLM 的正面比较，看哪一款在这个任务上真正站得住。