在讨论“赵祺握住了豆包的方向盘”之前,先说清楚:这篇文章不是娱乐八卦,也不是某款车型的新车评测。标题里出现的“赵祺”和“豆包”,我更愿意理解为一个场景符号——赵祺代表开发者或用户,豆包代表正在走向“智能座舱控制”的 AI 助手。真正值得拆解的,是握住方向盘这件事背后的技术链路:AI 助手怎样从“陪你聊天”变成“帮你开车”?在它参与驾驶控制时,权限边界、指令链路、安全兜底又该怎么设计?
如果你正准备做车载语音助手、智能座舱 Agent,或者只是好奇“AI 驾驶到底会以什么方式落地”,这篇文章会给你一套可运行的最小实验方案。我们会用一个模拟车控服务,配合大模型 Agent 调度,把“用户说一句话 -> 识别意图 -> 生成车控指令 -> 控件执行 -> 结果回执”的完整链路跑通。读完你至少能回答三个问题:AI 助手控制汽车需要哪些核心模块?开发一个最小可用的车载 Agent 要写多少代码?上真车之前必须先守住哪些安全底线?
1. 这篇文章真正要解决的问题
很多人看到“AI 握住方向盘”的第一反应是:这难道就是自动驾驶?其实不是。自动驾驶解决的是“车怎么自己开”,而我们讨论的 AI 车载助手解决的是“人怎么用自然语言指挥车”。这两者有关联,但工程边界完全不同。
自动驾驶系统需要对感知、决策、规划、执行做实时闭环,摄像头、激光雷达、毫米波雷达、高精地图缺一不可。而语音控制车机,核心链路是:语音识别 -> 语义理解 -> 工具调用 -> 车控执行。它更像是把“大模型”接到“车内技能 API”上。
但这里的坑也不少。最容易被低估的是两个:
第一,大模型输出不稳定。你问它“今天天气怎么样”,它胡说一点问题不大;但如果让它发送“方向盘左转 10 度”,输出一旦出现格式错误,或者命令值超出车辆物理限制,后果可能非常严重。所以车载 Agent 不能只有模型,还要有指令校验、白名单和状态机。
第二,驾驶场景的时序要求很高。用户说“帮我右转”,模型思考了 5 秒才返回指令,这在真实道路上完全不可用。语音助手可以慢,但车控指令必须快。这要求我们把很多能力前置:意图分类提前做、常用指令模板预编译、低延迟链路单独优化。
这篇文章要解决的是:在一个完全可控的模拟环境里,把 AI 车载 Agent 的骨架搭起来。你不需要真车,也能验证整个工程链路。先跑通流程,再谈真车接入,这是比较稳妥的路径。
2. AI 助手的“方向盘”是什么:核心概念与技术边界
2.1 从对话到车控指令,中间发生了什么
一句话要变成车能听懂的执行指令,大概会经过这样几个阶段:
- 语音识别(ASR):把用户说的“帮我打开空调”转成文本。
- 语义理解(NLU/LLM):理解这句话里的意图是“打开空调”,参数是“空调状态=开”。
- 工具调用(Tool Calling):Agent 从预先注册的车控工具列表里选择合适的函数。
- 协议转换:把函数参数转换成车载总线或 MQTT 消息。
- 执行与反馈:车机执行命令后,把结果返回给用户。
在传统方案里,语义理解和工具调用往往靠规则引擎或小模型分类实现。规则引擎的好处是响应快、可解释,但坏处是泛化能力弱,用户说“我有点冷”和“把空调温度调高”会被当成两个完全不同的意图。
大模型 Agent 的出现改变了后两环。现在你可以让 LLM 根据工具描述自动选择合适的工具,并生成结构化的调用参数。这大大提高了意图扩展的灵活性,但它也把“不稳定”带进了控制链路。
2.2 Agent、工具、权限:一个容易混淆的三角关系
很多初学者会把 Agent 本身当成“能执行任务的机器人”。更准确地说,Agent 是一套决策循环:
用户输入 -> 观察(结合上下文) -> 决策(选择工具) -> 行动(调用工具) -> 观察结果 -> 下一步...在这个循环里,“工具”是 Agent 的肢体,“权限”是这些肢体的活动范围。
在驾驶场景中,权限尤其重要。工具可以注册很多:查天气、讲故事、设置导航、调整座椅、打开车窗、控制方向盘。但每个工具的安全等级完全不同。查天气失败不会出事,控制方向盘失败就是事故。所以你需要把工具分成“低风险工具”和“高风险工具”,高风险工具必须加额外的校验和人工确认。
2.3 方向盘控制不是“接口对接”,而是状态机
很多人以为车控就是“提供一个 API,Agent 调用就行”。真正落地时,你还需要考虑车辆当前状态。
比如用户说“向左打 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 fastapi==0.104.1 uvicorn[standard]==0.24.0 paho-mqtt==1.6.1 openai==1.6.1 pydantic==2.5.2 python-dotenv==1.0.0安装命令:
pip install -r requirements.txt启动 MQTT Broker(使用 Docker):
docker 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(f"unknown 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_NAME=gpt-4o-mini LLM_BASE_URL=https://api.openai.com/v1 LLM_API_KEY=your_api_key_here MQTT_BROKER_HOST=127.0.0.1 MQTT_BROKER_PORT=1883然后写agent.py。这段代码会维护一个工具描述列表,并把用户文本、工具描述和系统提示词一起发给大模型。
# 文件路径:app/agent.py import os import json from openai import OpenAI from dotenv import load_dotenv load_dotenv() client = OpenAI( base_url=os.getenv("LLM_BASE_URL"), api_key=os.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_check=true。 3. 如果用户指令超出工具能力范围,不要强行调用工具,直接返回拒绝原因。 """ class AgentDispatcher: def __init__(self): self.client = client def plan(self, user_text: str) -> dict: response = self.client.chat.completions.create( model=os.getenv("MODEL_NAME"), messages=[ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_text}, ], tools=TOOLS, tool_choice="auto", temperature=0.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, }这里有几处设计可以重点关注:
temperature=0.1是为了让模型输出更稳定。在控制场景中,创造性越少越好。- 系统提示词里明确要求高风险指令返回
safety_check=true,但你要记住,提示词约束只是软约束,真正的安全不能靠提示词。 - 工具描述里已经加入了参数范围,帮助模型生成合法参数。
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(f"tool {tool_name} is not in whitelist") for key, (low, high) in self.limits[tool_name].items(): if key not in args: raise SafetyError(f"missing param: {key}") value = args[key] if not (low <= float(value) <= high): raise SafetyError(f"param {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_confirm=True if args.get("hard_confirm") is not True: raise SafetyError("high risk tool requires hard_confirm=True") 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(title="Doubao 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_model=CommandResponse) def handle_command(req: CommandRequest): request_id = uuid.uuid4().hex[:8] try: plan = dispatcher.plan(req.text) if plan["action"] == "reply": return CommandResponse( request_id=request_id, status="need_more_info", message=plan["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), hostname=MQTT_BROKER_HOST, port=MQTT_BROKER_PORT, ) return CommandResponse( request_id=request_id, status="accepted", message="指令已下发", tool_name=tool_name, tool_args=tool_args, vehicle_status=vehicle_status, ) except SafetyError as exc: return CommandResponse( request_id=request_id, status="rejected", message=str(exc), ) except Exception as exc: return CommandResponse( request_id=request_id, status="error", message=f"system 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_confirm=True,安全校验会拒绝。响应大概是:
{ "request_id": "9f8e7d6c", "status": "rejected", "message": "high risk tool requires hard_confirm=True" }这个失败其实是“预期内的成功”。它说明安全校验模块在正常工作。
验证时建议按这个顺序排查:
- 先看 FastAPI 返回的
status,是accepted、rejected还是error。 - 再去看 MQTT Worker 的日志,确认指令是否真的到达模拟器。
- 如果服务返回 500 或连接超时,先确认 Broker 是否启动,端口是否被占用。
7. 常见问题与排查方法
下面列几个我在把类似链路拼起来时最容易遇到的坑。
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| curl 调用返回 500 | FastAPI 无法连接 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_confirm=True代替了人工确认,在实际项目里,这个字段应该来自一个独立的确认流程,比如用户按下一个物理按钮,或者中控屏弹窗点击确认。它不能由 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 想要接管控制权时,系统有没有足够多的安全护栏。在这个小实验里,护栏就是白名单、参数范围、人工确认和状态回执。把这几个环节做扎实,比让模型多学会几句聪明话重要得多。