在讨论“赵祺握住了豆包的方向盘”之前先说清楚这篇文章不是娱乐八卦也不是某款车型的新车评测。标题里出现的“赵祺”和“豆包”我更愿意理解为一个场景符号——赵祺代表开发者或用户豆包代表正在走向“智能座舱控制”的 AI 助手。真正值得拆解的是握住方向盘这件事背后的技术链路AI 助手怎样从“陪你聊天”变成“帮你开车”在它参与驾驶控制时权限边界、指令链路、安全兜底又该怎么设计如果你正准备做车载语音助手、智能座舱 Agent或者只是好奇“AI 驾驶到底会以什么方式落地”这篇文章会给你一套可运行的最小实验方案。我们会用一个模拟车控服务配合大模型 Agent 调度把“用户说一句话 - 识别意图 - 生成车控指令 - 控件执行 - 结果回执”的完整链路跑通。读完你至少能回答三个问题AI 助手控制汽车需要哪些核心模块开发一个最小可用的车载 Agent 要写多少代码上真车之前必须先守住哪些安全底线1. 这篇文章真正要解决的问题很多人看到“AI 握住方向盘”的第一反应是这难道就是自动驾驶其实不是。自动驾驶解决的是“车怎么自己开”而我们讨论的 AI 车载助手解决的是“人怎么用自然语言指挥车”。这两者有关联但工程边界完全不同。自动驾驶系统需要对感知、决策、规划、执行做实时闭环摄像头、激光雷达、毫米波雷达、高精地图缺一不可。而语音控制车机核心链路是语音识别 - 语义理解 - 工具调用 - 车控执行。它更像是把“大模型”接到“车内技能 API”上。但这里的坑也不少。最容易被低估的是两个第一大模型输出不稳定。你问它“今天天气怎么样”它胡说一点问题不大但如果让它发送“方向盘左转 10 度”输出一旦出现格式错误或者命令值超出车辆物理限制后果可能非常严重。所以车载 Agent 不能只有模型还要有指令校验、白名单和状态机。第二驾驶场景的时序要求很高。用户说“帮我右转”模型思考了 5 秒才返回指令这在真实道路上完全不可用。语音助手可以慢但车控指令必须快。这要求我们把很多能力前置意图分类提前做、常用指令模板预编译、低延迟链路单独优化。这篇文章要解决的是在一个完全可控的模拟环境里把 AI 车载 Agent 的骨架搭起来。你不需要真车也能验证整个工程链路。先跑通流程再谈真车接入这是比较稳妥的路径。2. AI 助手的“方向盘”是什么核心概念与技术边界2.1 从对话到车控指令中间发生了什么一句话要变成车能听懂的执行指令大概会经过这样几个阶段语音识别ASR把用户说的“帮我打开空调”转成文本。语义理解NLU/LLM理解这句话里的意图是“打开空调”参数是“空调状态开”。工具调用Tool CallingAgent 从预先注册的车控工具列表里选择合适的函数。协议转换把函数参数转换成车载总线或 MQTT 消息。执行与反馈车机执行命令后把结果返回给用户。在传统方案里语义理解和工具调用往往靠规则引擎或小模型分类实现。规则引擎的好处是响应快、可解释但坏处是泛化能力弱用户说“我有点冷”和“把空调温度调高”会被当成两个完全不同的意图。大模型 Agent 的出现改变了后两环。现在你可以让 LLM 根据工具描述自动选择合适的工具并生成结构化的调用参数。这大大提高了意图扩展的灵活性但它也把“不稳定”带进了控制链路。2.2 Agent、工具、权限一个容易混淆的三角关系很多初学者会把 Agent 本身当成“能执行任务的机器人”。更准确地说Agent 是一套决策循环用户输入 - 观察结合上下文 - 决策选择工具 - 行动调用工具 - 观察结果 - 下一步...在这个循环里“工具”是 Agent 的肢体“权限”是这些肢体的活动范围。在驾驶场景中权限尤其重要。工具可以注册很多查天气、讲故事、设置导航、调整座椅、打开车窗、控制方向盘。但每个工具的安全等级完全不同。查天气失败不会出事控制方向盘失败就是事故。所以你需要把工具分成“低风险工具”和“高风险工具”高风险工具必须加额外的校验和人工确认。2.3 方向盘控制不是“接口对接”而是状态机很多人以为车控就是“提供一个 APIAgent 调用就行”。真正落地时你还需要考虑车辆当前状态。比如用户说“向左打 10 度”这个指令本身没有意义。你必须知道车辆当前是否静止速度是多少方向盘当前角度是多少是不是已经在变道这是一组状态不是单一指令。所以车载 Agent 的“方向盘”从工程上讲不是一个 REST API而是一个状态机。Agent 只是状态机的输入源之一真正的执行器要依据状态判断“这条指令到底能不能执行”。这就是为什么在代码实现里我会把车控模拟器单独拆成一个服务而不是让 Agent 直接去操作硬件。3. 方案设计一个最小可行的 AI 车载实验 Agent我们不追求一步到位而是先做一个能在本地跑起来的 MVP。整个系统包含这几个部分FastAPI 入口服务对外提供 HTTP 接口接受文本指令。Agent 调度模块调用大模型让模型决定调用哪个工具、传入什么参数。车控模拟服务模拟车辆的部分状态比如空调、车窗、方向盘角度、车速。MQTT 消息总线负责在 Agent 调度模块和模拟车控服务之间传消息。安全校验模块对高风险工具做参数范围校验和二次确认。整体流程是这样的用户 POST /command {text: 左转一点} - FastAPI 接收 - Agent 调度器调用 LLM得到 tool_call - 安全校验模块检查工具白名单和参数范围 - 通过后发送 MQTT 消息到模拟车控服务 - 模拟车控服务执行并回发状态 - FastAPI 返回最终结果为什么要引入 MQTT而不直接在 FastAPI 里调用函数因为模拟实车场景时执行端通常不是一个 HTTP 服务而是靠近车机的中控模块。MQTT 更适合这类设备通信网络开销小而且天然支持发布订阅。先用 MQTT 把链路打通后面换真车协议时只需要替换执行端。3.1 工具注册表设计为了让 LLM 知道有哪些工具可用我们把工具描述组织成 JSON Schema。下面是我们需要实现的核心工具工具名功能参数风险等级set_temperature设置空调温度temperature: number低set_fan_speed设置空调风量speed: number低open_window打开车窗window_id: string, percent: number低set_steering_angle设置方向盘角度angle: number高set_speed_limit设置限速speed: number高emergency_stop紧急停车无最高在实际生产里高风险工具不会直接授权给模型自动执行。我们会让 Agent 生成执行意图但在真正执行前必须通过安全校验模块和人工确认。3.2 为什么选择“模拟器优先”你可能觉得模拟器不过瘾想一步到位接真车。但我的建议是先在模拟器里把所有边界情况测完再考虑硬件。模拟器的价值在于它能让你安全地测试各种异常场景。比如 Agent 生成了一个不合理的角度值模拟器可以直接拒绝并返回错误信息。这些测试做完之后你才会知道哪些规则需要写入 Agent 的系统提示词哪些规则需要在代码层硬编码。4. 环境准备与前置条件本文的示例代码基于 Python 3.10需要用到 FastAPI、paho-mqtt、uvicorn、openai SDK。如果你没有真实的 MQTT Broker可以用 Docker 快速启动一个 Mosquitto。依赖清单如下# requirements.txt fastapi0.104.1 uvicorn[standard]0.24.0 paho-mqtt1.6.1 openai1.6.1 pydantic2.5.2 python-dotenv1.0.0安装命令pip install -r requirements.txt启动 MQTT Broker使用 Dockerdocker run -d --name mqtt-broker -p 1883:1883 eclipse-mosquitto:2.0这只是一个测试用 Broker配置没有做任何权限控制。生产环境必须设置用户名密码并在内网隔离。5. 完整示例代码实现5.1 项目结构我们先规划目录结构doubao-driver/ ├── requirements.txt ├── .env ├── app/ │ ├── __init__.py │ ├── main.py │ ├── config.py │ ├── agent.py │ ├── safety.py │ └── vehicle_simulator/ │ ├── __init__.py │ ├── simulator.py │ └── mqtt_worker.py其中main.py是 FastAPI 入口agent.py负责大模型调用safety.py负责指令校验vehicle_simulator是模拟车控服务。5.2 车控模拟服务先用一个类模拟车辆状态。这里我们把方向盘角度、车速、空调状态放进一个状态字典。# 文件路径app/vehicle_simulator/simulator.py import threading import time class VehicleSimulator: 一个简单的车辆状态模拟器不连接真实硬件。 def __init__(self): self.lock threading.Lock() self.state { speed: 0.0, # 车速单位 km/h steering_angle: 0.0, # 方向盘角度单位 度负值为左转 temperature: 24.0, # 空调温度 fan_speed: 1, # 风量档位 0-7 window_state: { # 车窗位置0 表示关闭100 表示全开 driver: 0, passenger: 0, rear_left: 0, rear_right: 0, }, emergency_stop: False, } def get_status(self) - dict: with self.lock: return dict(self.state) def execute_control(self, control_type: str, params: dict) - dict: with self.lock: if control_type set_temperature: self.state[temperature] float(params.get(temperature, 24)) elif control_type set_fan_speed: self.state[fan_speed] int(params.get(speed, 1)) elif control_type open_window: wid params.get(window_id, driver) percent int(params.get(percent, 0)) self.state[window_state][wid] percent elif control_type set_steering_angle: self.state[steering_angle] float(params.get(angle, 0)) elif control_type set_speed_limit: # 这里做简化处理设定最高车速 self.state[speed] min(float(params.get(speed, 30)), 120) elif control_type emergency_stop: self.state[emergency_stop] True self.state[speed] 0.0 else: raise ValueError(funknown control_type: {control_type}) return dict(self.state)模拟器内部的锁是为了保证在多线程访问时状态一致。execute_control已经包含了一些基础限制比如最高限速不超过 120。但在生产项目中这些限制应该是车辆底层硬约束不能只靠模拟器实现。5.3 MQTT 通信模拟车控端需要一个独立进程监听 MQTT 消息。这里我们让mqtt_worker.py订阅车控主题vehicle/control/in执行完成后发布状态到vehicle/status/out。# 文件路径app/vehicle_simulator/mqtt_worker.py import json import threading import paho.mqtt.client as mqtt from app.vehicle_simulator.simulator import VehicleSimulator BROKER_HOST 127.0.0.1 BROKER_PORT 1883 CONTROL_TOPIC vehicle/control/in STATUS_TOPIC vehicle/status/out class MqttVehicleWorker: def __init__(self): self.simulator VehicleSimulator() self.client mqtt.Client() self.client.on_connect self._on_connect self.client.on_message self._on_message def _on_connect(self, client, userdata, flags, rc): print(MQTT connected with result code, rc) client.subscribe(CONTROL_TOPIC) def _on_message(self, client, userdata, msg): try: payload json.loads(msg.payload.decode(utf-8)) control_type payload[control_type] params payload.get(params, {}) print(f[MQTT Worker] receive control: {control_type} {params}) status self.simulator.execute_control(control_type, params) client.publish(STATUS_TOPIC, json.dumps(status)) except Exception as exc: print([MQTT Worker] error:, exc) client.publish(STATUS_TOPIC, json.dumps({error: str(exc)})) def start(self): self.client.connect(BROKER_HOST, BROKER_PORT, 60) self.client.loop_forever() def main(): worker MqttVehicleWorker() worker.start() if __name__ __main__: main()这里有一点值得注意在真实车辆控制中_on_message里不应该直接执行控制指令而应该先做身份认证、指令去重、时间戳校验和状态机校验。上面的代码只是演示消息链路不能直接搬到生产环境。5.4 Agent 调度模块Agent 调度模块的关键是让 LLM 输出结构化工具调用。我们使用 OpenAI 兼容的接口风格编写但模型名和接口地址都放在.env配置里避免写死。在.env中# .env MODEL_NAMEgpt-4o-mini LLM_BASE_URLhttps://api.openai.com/v1 LLM_API_KEYyour_api_key_here MQTT_BROKER_HOST127.0.0.1 MQTT_BROKER_PORT1883然后写agent.py。这段代码会维护一个工具描述列表并把用户文本、工具描述和系统提示词一起发给大模型。# 文件路径app/agent.py import os import json from openai import OpenAI from dotenv import load_dotenv load_dotenv() client OpenAI( base_urlos.getenv(LLM_BASE_URL), api_keyos.getenv(LLM_API_KEY), ) TOOLS [ { type: function, function: { name: set_temperature, description: 设置车内空调温度, parameters: { type: object, properties: { temperature: { type: number, description: 目标温度单位摄氏度建议范围 16 到 30 } }, required: [temperature] } } }, { type: function, function: { name: set_steering_angle, description: 设置方向盘转角。注意仅允许在车辆静止或极低速状态下调用, parameters: { type: object, properties: { angle: { type: number, description: 方向盘角度负值代表左转正值代表右转范围 -45 到 45 } }, required: [angle] } } }, { type: function, function: { name: open_window, description: 打开或关闭车窗, parameters: { type: object, properties: { window_id: { type: string, enum: [driver, passenger, rear_left, rear_right] }, percent: { type: integer, description: 车窗打开百分比0 表示关闭100 表示全开 } }, required: [window_id, percent] } } } ] SYSTEM_PROMPT 你是一个车载 AI 助手负责根据用户的自然语言指令输出车控工具调用。 规则 1. 如果用户表达模糊先向用户确认具体参数。 2. 对于涉及方向盘、速度、刹车的指令必须在结果里附带字段 safety_checktrue。 3. 如果用户指令超出工具能力范围不要强行调用工具直接返回拒绝原因。 class AgentDispatcher: def __init__(self): self.client client def plan(self, user_text: str) - dict: response self.client.chat.completions.create( modelos.getenv(MODEL_NAME), messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_text}, ], toolsTOOLS, tool_choiceauto, temperature0.1, ) message response.choices[0].message if not message.tool_calls: return { action: reply, content: message.content or 指令无法解析请重新描述 } tool_call message.tool_calls[0] return { action: tool, tool_name: tool_call.function.name, tool_args: json.loads(tool_call.function.arguments), raw_message: message.content, }这里有几处设计可以重点关注temperature0.1是为了让模型输出更稳定。在控制场景中创造性越少越好。系统提示词里明确要求高风险指令返回safety_checktrue但你要记住提示词约束只是软约束真正的安全不能靠提示词。工具描述里已经加入了参数范围帮助模型生成合法参数。5.5 安全校验模块安全校验模块是这套模拟系统里最重要的部分。它的职责是在 Agent 决定调用工具之后、真正发出 MQTT 消息之前再做一次指令合法性检查。# 文件路径app/safety.py class SafetyError(Exception): pass class SafetyGuard: def __init__(self): self.high_risk_tools { set_steering_angle, set_speed_limit, emergency_stop, } self.limits { set_temperature: {temperature: (16, 30)}, set_fan_speed: {speed: (0, 7)}, open_window: {percent: (0, 100)}, set_steering_angle: {angle: (-45, 45)}, set_speed_limit: {speed: (0, 120)}, } def check(self, tool_name: str, args: dict, vehicle_status: dict) - None: if tool_name not in self.limits: raise SafetyError(ftool {tool_name} is not in whitelist) for key, (low, high) in self.limits[tool_name].items(): if key not in args: raise SafetyError(fmissing param: {key}) value args[key] if not (low float(value) high): raise SafetyError(fparam {key}{value} out of range [{low}, {high}]) if tool_name set_steering_angle: speed vehicle_status.get(speed, 0) if speed 5: raise SafetyError(cannot change steering angle when speed 5 km/h) if tool_name in self.high_risk_tools: # 高风险工具需要人工确认标识这里简化成必须有 hard_confirmTrue if args.get(hard_confirm) is not True: raise SafetyError(high risk tool requires hard_confirmTrue) safety_guard SafetyGuard()这段代码的核心思想是防御式校验。模型说“到货了”并不算数校验器就是安全链路里的第三道闸门。5.6 FastAPI 入口与完整链路最后写main.py把 Agent、安全校验和 MQTT 发布串起来。# 文件路径app/main.py import json import os import time import uuid from fastapi import FastAPI from pydantic import BaseModel import paho.mqtt.publish as publish from app.agent import AgentDispatcher from app.safety import SafetyGuard, SafetyError from dotenv import load_dotenv load_dotenv() app FastAPI(titleDoubao Driver Agent) dispatcher AgentDispatcher() safety_guard SafetyGuard() MQTT_BROKER_HOST os.getenv(MQTT_BROKER_HOST, 127.0.0.1) MQTT_BROKER_PORT int(os.getenv(MQTT_BROKER_PORT, 1883)) CONTROL_TOPIC vehicle/control/in STATUS_TOPIC vehicle/status/out class CommandRequest(BaseModel): text: str class CommandResponse(BaseModel): request_id: str status: str message: str tool_name: str | None None tool_args: dict | None None vehicle_status: dict | None None def get_latest_vehicle_status() - dict: 简化实现实际项目中应通过 MQTT 订阅缓存状态而不是直接查询。 return { speed: 0.0, steering_angle: 0.0, temperature: 24.0, fan_speed: 1, window_state: { driver: 0, passenger: 0, rear_left: 0, rear_right: 0, }, } app.post(/command, response_modelCommandResponse) def handle_command(req: CommandRequest): request_id uuid.uuid4().hex[:8] try: plan dispatcher.plan(req.text) if plan[action] reply: return CommandResponse( request_idrequest_id, statusneed_more_info, messageplan[content], ) tool_name plan[tool_name] tool_args plan[tool_args] vehicle_status get_latest_vehicle_status() # 安全校验 safety_guard.check(tool_name, tool_args, vehicle_status) # 发送 MQTT 车控消息 payload { request_id: request_id, control_type: tool_name, params: tool_args, timestamp: int(time.time()), } publish.single( CONTROL_TOPIC, json.dumps(payload), hostnameMQTT_BROKER_HOST, portMQTT_BROKER_PORT, ) return CommandResponse( request_idrequest_id, statusaccepted, message指令已下发, tool_nametool_name, tool_argstool_args, vehicle_statusvehicle_status, ) except SafetyError as exc: return CommandResponse( request_idrequest_id, statusrejected, messagestr(exc), ) except Exception as exc: return CommandResponse( request_idrequest_id, statuserror, messagefsystem error: {exc}, )这个接口的逻辑非常直观接收文本Agent 规划安全校验MQTT 下发。但它也暴露出一个简化点get_latest_vehicle_status返回的是固定状态没有真正从 MQTT 订阅拿到实时状态。在完整项目里你应该有一个状态缓存服务持续订阅vehicle/status/out主题然后每个请求都从缓存读取最新状态。这样做的好处是即使一次指令已经执行系统也能掌握“当前车辆状态”而不是被模型“以为的状态”误导。6. 运行结果与效果验证先启动 MQTT Worker它会监听车控主题并更新模拟车状态。export PYTHONPATH. python -m app.vehicle_simulator.mqtt_worker然后启动 FastAPI 服务uvicorn app.main:app --reload --port 8000打开另一个终端调用接口curl -X POST http://127.0.0.1:8000/command \ -H Content-Type: application/json \ -d {text: 把空调调到24度}如果一切正常你会得到类似这样的响应{ request_id: a1b2c3d4, status: accepted, message: 指令已下发, tool_name: set_temperature, tool_args: { temperature: 24 }, vehicle_status: { speed: 0.0, steering_angle: 0.0, temperature: 24.0, fan_speed: 1, window_state: { driver: 0, passenger: 0, rear_left: 0, rear_right: 0 } } }MQTT Worker 那边应该打印[MQTT Worker] receive control: set_temperature {temperature: 24}再测试一个高风险指令curl -X POST http://127.0.0.1:8000/command \ -H Content-Type: application/json \ -d {text: 左转20度}由于我们没有在参数里传hard_confirmTrue安全校验会拒绝。响应大概是{ request_id: 9f8e7d6c, status: rejected, message: high risk tool requires hard_confirmTrue }这个失败其实是“预期内的成功”。它说明安全校验模块在正常工作。验证时建议按这个顺序排查先看 FastAPI 返回的status是accepted、rejected还是error。再去看 MQTT Worker 的日志确认指令是否真的到达模拟器。如果服务返回 500 或连接超时先确认 Broker 是否启动端口是否被占用。7. 常见问题与排查方法下面列几个我在把类似链路拼起来时最容易遇到的坑。问题现象可能原因排查方式解决方案curl 调用返回 500FastAPI 无法连接 MQTT Broker检查 Broker 是否启动docker ps再测试端口nc -vz 127.0.0.1 1883重新启动 Mosquitto 容器MQTT Worker 收不到消息发布了单条消息但 Worker 没有订阅成功在 Worker 启动日志中查看订阅信息用mosquitto_sub手动订阅主题检查主题名是否一致模型返回工具名不存在Tool 注册表和代码里的执行器不一致打印plan结果确认模型返回的tool_name确保agent.py里的TOOLS和safety.py里的白名单完全一致参数超出范围校验失败模型生成了不合理的参数比如温度 200查看 FastAPI 返回的message其中会写明哪个参数越界调整提示词中的参数描述或扩大小范围但扩范围要非常谨慎指令执行了但状态没有更新get_latest_vehicle_status是固定值没有真正读取实时状态查看vehicle/status/out主题的 MQTT 消息建立一个状态缓存模块订阅并保存最新状态8. 安全边界与最佳实践8.1 安全不能依赖大模型的“自觉”系统提示词里写得再清楚也只是软约束。大模型可能因为一次 prompt 注入或者一句模糊表达输出了预期之外的工具参数。所以凡是涉及安全风险的指令必须以白名单、参数校验、手动确认三重机制为准。在代码实现里安全校验模块是在模型输出之后、联网下发之前执行的。这个位置非常关键如果校验放在模型调用之前你校验的永远是“用户要什么”不是“模型要做什么”如果校验放在 MQTT 下发之后那只能给执行端多一些冗余不能阻止风险指令发出。8.2 高风险工具必须有人工确认环节我们示例里用hard_confirmTrue代替了人工确认在实际项目里这个字段应该来自一个独立的确认流程比如用户按下一个物理按钮或者中控屏弹窗点击确认。它不能由 Agent 自己生成。这一点值得强调在真车环境任何涉及转向、制动、速度变化的指令都需要在短时间内完成“二次确认”。你可以减少确认的交互成本但不能取消确认这个步骤。8.3 状态缓存和指令去重车载系统会有大量并发消息同一个指令可能因为网络重试被发送两次。所以你在生产环境必须加两个能力状态缓存Agent 服务本地保存车辆最新状态而不是每次重新查询。指令去重每个请求都有唯一request_id执行端对相同request_id的重复消息直接忽略。8.4 日志、追踪与审计在真实项目里每一句用户指令、每一条 Agent 规划、每一次安全校验结果、每一条 MQTT 消息都应该被记录下来。原因是一旦发生事故你要能完整复盘模型为什么这么理解校验为什么通过执行端为什么照做建议日志字段至少包含时间戳、request_id、用户文本、模型返回内容、工具名和参数、校验结果、执行结果、耗时。多人协作时使用统一的 trace_id 串起全链路。8.5 从小实验到真车之间的距离最后必须泼一盆冷水上面的代码只能在模拟器里跑通流程不能直接接入真实车辆。真车接入涉及总线协议、ISO 26262 功能安全标准、网络安全认证、整车厂权限管理等一系列问题。如果你真的要做车控项目务必在合规环境里由汽车电子工程师共同参与设计。9. 总结与后续学习方向我们做的事情本质上是把一个很“新闻标题”的概念拆成了一条可控的工程链路大模型 Agent 负责理解意图工具注册表负责描述能力安全校验负责守住边界MQTT 和模拟器负责低成本验证执行。你已经可以用这套骨架跑通“自然语言 - 车控指令 - 执行反馈”的最小闭环。如果你要继续深入我建议按这个顺序扩展把固定状态换成真正的 MQTT 状态缓存让模拟器状态实时回到 Agent。增加语音识别入口把音频转成文本后再调用/command。做一个简单的前端确认弹窗实现真正的人工确认流程。把工具注册表从代码里抽出来放到配置中心或数据库中方便动态调整。为每个工具写单元测试特别是针对参数越界、未知工具、模型幻觉场景的测试。“赵祺握住了豆包的方向盘”这句话听起来充满戏剧性但工程上真正重要的从来不是谁握住了方向盘而是当 AI 想要接管控制权时系统有没有足够多的安全护栏。在这个小实验里护栏就是白名单、参数范围、人工确认和状态回执。把这几个环节做扎实比让模型多学会几句聪明话重要得多。