ITS是多智能体协作最典型的落地场景之一——路口信号灯、自动驾驶车辆、路侧感知单元、交通指挥中心各自是独立Agent，需要去中心化协同决策。

一、ITS多智能体的核心架构认知

🧠 核心范式：交通信号灯Agent、自动驾驶车辆Agent、路侧单元Agent之间通过消息总线交换意图，形成”协商-承诺-执行-反思“循环。

二、技术选型（2026年7月最新）

三、零基础实战：构建”十字路口多智能体协作系统”

场景目标

两个相邻路口（路口A和路口B），各有一个信号灯Agent和一个路侧感知Agent。当路口A发生拥堵时，自动协商调整路口B的信号灯相位，避免溢出。

步骤1：定义每个Agent的基础结构（基于LangGraph）

每个路口Agent是一个独立的LangGraph实例，拥有自己的状态和决策逻辑：

python

from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI
import json

===== 路口Agent的专用状态 =====

class IntersectionState(MessagesState):
intersection_id: str # 路口ID
phase: str # 当前相位: “NS” / “EW” / “ALL_RED”
queue_length: int # 排队车辆数（感知获取）
pressure: float # 压力值（排队×等待时间）
neighbor_pressure: dict # 邻居路口的压力广播
action: str # 待执行动作

===== 创建单个路口Agent的工厂函数 =====

def create_intersection_agent(intersection_id: str):
model = ChatOpenAI(model=”gpt-4o-mini”)

1. 感知节点：读取本地传感器数据

def perceive(state: IntersectionState):

实际项目里这里调用SUMO API或摄像头数据

模拟：随机生成排队长度

import random
queue = random.randint(0, 30)
return {
“queue_length”: queue,
“pressure”: queue * (1 + state.get(“wait_time”, 0) / 60)
}

2. 协商节点：向邻居广播自己的压力，并接收邻居压力

def negotiate(state: IntersectionState):

广播自己的压力到Redis（模拟MQTT）

import redis
r = redis.Redis(host=’localhost’, decode_responses=True)
r.publish(f”intersection/{intersection_id}/pressure”,
json.dumps({“id”: intersection_id, “pressure”: state[“pressure”]}))

从Redis读取邻居压力（实际用订阅回调）

此处简化：直接硬编码邻居压力

return {“neighbor_pressure”: {“B”: 0.85}} # 0-1归一化压力值

3. 决策节点：根据自身+邻居压力，决定是否调整相位

def decide(state: IntersectionState):
my_pressure = state[“pressure”]
neighbor_p = state.get(“neighbor_pressure”, {}).get(“B”, 0)

决策规则（可让LLM生成更复杂策略）

if my_pressure > 0.7 and neighbor_p > 0.6:

双方向都堵 → 不调整（维持现状）

action = “HOLD”
elif my_pressure > 0.7 and neighbor_p < 0.4:

自己堵、邻居不堵 → 请求邻居让行

action = “REQUEST_GREEN_EXTENSION”
elif my_pressure < 0.3 and neighbor_p > 0.7:

自己不堵、邻居堵 → 给邻居让行

action = “YIELD_TO_NEIGHBOR”
else:
action = “HOLD”

return {“action”: action}

4. 执行节点：执行动作（调用SUMO API改信号灯）

def execute(state: IntersectionState):
action = state[“action”]
if action == “REQUEST_GREEN_EXTENSION”:
print(f”🚦 路口{intersection_id}: 请求延长绿灯”)

实际调用 SUMO 的 traci.trafficlight.setPhase()

elif action == “YIELD_TO_NEIGHBOR”:
print(f”🚦 路口{intersection_id}: 让行邻居，切换为黄灯”)
else:
print(f”🚦 路口{intersection_id}: 保持当前相位”)
return {“phase”: “ADJUSTED” if action != “HOLD” else “HOLD”}

5. 构建图

builder = StateGraph(IntersectionState)
builder.add_node(“perceive”, perceive)
builder.add_node(“negotiate”, negotiate)
builder.add_node(“decide”, decide)
builder.add_node(“execute”, execute)

builder.add_edge(START, “perceive”)
builder.add_edge(“perceive”, “negotiate”)
builder.add_edge(“negotiate”, “decide”)
builder.add_edge(“decide”, “execute”)
builder.add_edge(“execute”, END)

return builder.compile()

步骤2：多Agent协作的”共识仲裁层”

当多个路口Agent的决策冲突时（例如两个路口都想抢绿灯），需要一个轻量级共识协议：

python

from typing import List, Dict
import random

class ConsensusCoordinator:
“””分布式共识协调器（去中心化投票）”””

def init(self, agent_ids: List[str]):
self.agent_ids = agent_ids
self.proposals = {} # {agent_id: proposal}

def collect_proposals(self, agent_id: str, proposal: str, priority: float):
“””收集各Agent的提案，priority越高越优先”””
self.proposals[agent_id] = {“proposal”: proposal, “priority”: priority}

def reach_consensus(self) -> Dict[str, str]:
“””基于优先级加权投票，返回每个Agent最终执行的动作”””

如果所有提案都是HOLD，保持现状

if all(p[“proposal”] == “HOLD” for p in self.proposals.values()):
return {aid: “HOLD” for aid in self.agent_ids}

选出优先级最高的非HOLD提案

best = max(
[(aid, p) for aid, p in self.proposals.items() if p[“proposal”] != “HOLD”],
key=lambda x: x[1][“priority”]
)

最佳提案的Agent执行，其他Agent配合（让行或等待）

result = {}
for aid in self.agent_ids:
if aid == best[0]:
result[aid] = best[1][“proposal”]
else:
result[aid] = “YIELD” # 配合模式
return result

使用示例

coordinator = ConsensusCoordinator([“A”, “B”])

路口A提案：请求绿灯延长（优先级0.9）

coordinator.collect_proposals(“A”, “REQUEST_GREEN_EXTENSION”, 0.9)

路口B提案：保持（优先级0.3）

coordinator.collect_proposals(“B”, “HOLD”, 0.3)

final_actions = coordinator.reach_consensus()
print(final_actions) # {‘A’: ‘REQUEST_GREEN_EXTENSION’, ‘B’: ‘YIELD’}

步骤3：集成SUMO仿真（让Agent”活”在真实交通流中）

python

import traci # SUMO的Python接口
from sumo_rl import SumoEnvironment # 强化学习环境封装

1. 启动SUMO仿真

sumo_cmd = [“sumo-gui”, “-c”, “intersection.sumocfg”]
traci.start(sumo_cmd)

2. 主循环：每个仿真步长执行多智能体决策

step = 0
while step < 10000:

每个路口Agent独立感知、决策

for intersection_id in [“A”, “B”]:

获取排队长度（从SUMO读取）

queue = traci.edge.getLastStepVehicleNumber(f”edge_{intersection_id}”)

调用Agent决策（上面定义的LangGraph实例）

agent = create_intersection_agent(intersection_id)
result = agent.invoke({
“intersection_id”: intersection_id,
“queue_length”: queue,
“neighbor_pressure”: get_neighbor_pressure(intersection_id) # 从Redis读
})

执行动作（通过SUMO设置信号灯）

if result[“action”] == “REQUEST_GREEN_EXTENSION”:
traci.trafficlight.setPhase(intersection_id, 2) # 延长绿灯
elif result[“action”] == “YIELD”:
traci.trafficlight.setPhase(intersection_id, 1) # 黄灯

… 其他动作

traci.simulationStep()
step += 1

traci.close()

四、2026年ITS多智能体的前沿特性

五、零基础学习路线（ITS方向）

六、避坑指南（ITS多智能体特有）

如果你今天就想动手，建议先跑通步骤1的单路口Agent代码（不依赖SUMO，纯逻辑模拟），感受Agent”感知-协商-决策-执行”的循环。完成后，我可以为你提供SUMO环境配置文件，让代码接入真实交通仿真。😊

本作品采用《CC 协议》，转载必须注明作者和本文链接