夜雨聆风凌晨两点，工厂A线三号机台发出异响。值班巡检员按计划要到凌晨四点才会走到这台设备前——传统定时巡检的时间表不会因为一台设备的异常而提前。等巡检员发现时，轴承已经过热变形，停机维修损失按小时计算。如果有人能在异响出现的第一时间说一句"去看看A线三号机台"，机器人立刻出发、拍照、分析、查手册、语音汇报，整个响应链路压缩到几分钟。这个实战项目把前面积累的能力串成一个完整的智能巡检机器人系统：用自然语言下达巡检指令，LLM 规划路线，Nav2 自主导航，VLM 视觉判读设备状态，RAG 检索维修手册，语音播报巡检结果，全程有安全防护兜底。
一、场景引入：工厂巡检的困境
工厂车间的设备巡检是一项高频但低灵活度的工作。传统方案依赖人工定时巡检或固定路线自动巡检，两者各有短板：
传统定时巡检：
04:00 巡检员到达A线 → 记录正常
04:15 巡检员到达B线 → 记录正常
04:30 巡检员到达C线 → 发现C线泵机漏液（但漏液02:00就开始了）
固定路线自动巡检：
机器人按预设路径逐点拍照 → 照片存档 → 人工事后查看
问题：发现异常后无法自主查询处理方案，仍需人工介入核心矛盾在于：巡检的"触发时机"和"异常处置"都缺乏弹性。设备不会按照巡检时间表出故障，而固定路线机器人只会拍照存档，不会判断照片里是否有问题，更不会查阅维修手册给出建议。
用大模型重新设计这条链路，巡检可以从"按时间表走流程"变成"按需响应"：
用户："去检查一下A线的设备状态"
↓
LLM 解析指令 → 规划A线巡检路线（3个机台点位）
↓
机器人自主导航至每个点位 → 拍照 → VLM 分析是否有异常
↓
发现异常 → RAG 检索设备手册和维修指南
↓
巡检完成 → 语音汇报结果 + 生成结构化报告这条链路把前十四篇的技术串在一起：LLM 桥接负责理解指令和规划路线（第01、02篇），VLM 负责视觉判读（第03篇），语音交互负责输入指令和输出汇报（第04篇），异常恢复负责处理导航失败（第05篇），RAG 负责知识检索（第06篇），安全防护负责兜底（第13篇）。
二、项目架构：整合前14篇技术
2.1 技术栈全景
┌─────────────────────────────────────────────────────────────┐
│ 用户交互层 │
│ Whisper ASR（语音→文字） edge-tts TTS（文字→语音） │
└──────────────────────────┬──────────────────────────────────┘
│
┌──────────────────────────▼──────────────────────────────────┐
│ AI 决策层 │
│ LLM 路线规划（GPT-4o） VLM 异常检测（GPT-4o Vision） │
│ 异常恢复策略选择 RAG 知识检索（ChromaDB） │
└──────────────────────────┬──────────────────────────────────┘
│
┌──────────────────────────▼──────────────────────────────────┐
│ 编排执行层 │
│ InspectionOrchestrator（状态机编排各模块） │
└──────────┬───────────────┬───────────────┬──────────────────┘
│ │ │
┌──────────▼─────┐ ┌───────▼───────┐ ┌────▼──────────┐
│ Nav2 导航 │ │ 摄像头采集 │ │ 安全过滤层 │
│ 自主移动 │ │ 图像获取 │ │ 规则引擎+急停 │
└────────────────┘ └───────────────┘ └───────────────┘2.2 模块与前序教程的对应关系
2.3 巡检流程状态机
整个巡检过程由一个状态机驱动，每个状态对应一个明确的动作：
IDLE（待命）
│ 收到巡检指令
▼
PLANNING（路线规划）
│ LLM 返回巡检点位列表
▼
NAVIGATING（导航中）──── 导航失败 ────► RECOVERING（异常恢复）
│ 到达点位 │ 选择策略：重试/跳过/上报
│ ◄────────────────────────────────────┘
▼
INSPECTING（拍照检测）
│ 拍照 → VLM 分析
├── 正常 ──► 记录结果
└── 异常 ──► QUERYING（知识检索）
│ RAG 查询维修指南
▼
记录结果
│
▼
NEXT_POINT（下一个点位）
│ 还有未巡检点位 ──► NAVIGATING
│ 全部巡检完毕
▼
REPORTING（生成报告）
│ 汇总结果 → 语音播报
▼
IDLE（待命）三、系统设计
3.1 巡检任务定义：自然语言到巡检路线
用户用一句话下达巡检指令，LLM 负责把这句话翻译成一组有序的巡检点位。关键在于：LLM 不直接生成坐标，而是从预定义的点位库中选择。
用户："检查A线设备"
↓
LLM 理解：目标区域 = A线，任务 = 巡检
↓
从点位库筛选A线相关点位 → 按物理顺序排列
↓
返回巡检路线：[A线1号机台, A线2号机台, A线3号机台]LLM 的 System Prompt 中注入点位库的描述信息，让它知道有哪些点位、各点位属于哪条产线、检测什么设备。这样即使用户说"看看A线那几台机器"，LLM 也能匹配到正确的点位。
3.2 VLM 异常检测：拍照到判读
机器人到达巡检点位后，调用摄像头拍照，把照片交给 GPT-4o 的视觉接口分析。VLM 的输入是图像加上该设备的检查要点，输出是结构化的判读结果：
{
"point_id":"A_line_01",
"equipment":"A线1号注塑机",
"has_anomaly":true,
"anomaly_type":"泄漏",
"description":"设备底部液压管路接口处有液体渗出，疑似液压油泄漏",
"severity":"warning",
"confidence":0.85
}检查要点随设备类型变化：注塑机看液压管路和温度，电机看是否有异响和振动痕迹，配电柜看指示灯状态和是否有焦痕。这些检查要点写在配置文件中，巡检时注入 VLM 的 prompt。
3.3 RAG 查询：发现异常时检索手册
VLM 判读出异常后，系统自动用异常描述作为查询条件，在 ChromaDB 知识库中检索相关文档片段。知识库中预先导入了设备操作手册、维修指南、故障代码表等文档。
VLM 判读："液压管路接口处液体渗出，疑似液压油泄漏"
↓
RAG 检索查询："液压油泄漏 管路接口"
↓
检索结果：匹配到《A线注塑机维修手册》第3.2节"液压系统泄漏处理"
↓
LLM 综合异常描述和检索内容 → 生成处置建议3.4 语音报告：巡检完成后汇报
巡检全部点位走完后，系统汇总所有检测结果，生成一份结构化报告，同时用 TTS 语音播报摘要。报告分三部分：
• 巡检概览：巡检了多少个点位，耗时多久 • 异常清单：哪些点位发现异常，异常类型和严重程度 • 处置建议：基于 RAG 检索结果给出的维修建议
语音播报只读摘要，完整报告写入日志文件供后续查阅。
四、配置文件设计
4.1 巡检点位配置
# config/inspection_points.yaml
# 巡检点位定义：每个点位包含坐标、设备信息和检查要点
inspection_points:
A_line_01:
name:"A线1号注塑机"
line:"A"
pose:
x:8.5
y:3.2
yaw:1.57
equipment:
type:"injection_molding_machine"
model:"海天MA2500"
id:"IM-A01"
check_items:
-"液压管路是否有渗漏"
-"料筒温度指示是否在正常范围（180-220℃）"
-"设备外壳是否有异常变形或焦痕"
photo_orientation:"正面操作面板方向"
A_line_02:
name:"A线2号注塑机"
line:"A"
pose:
x:8.5
y:5.8
yaw:1.57
equipment:
type:"injection_molding_machine"
model:"海天MA2500"
id:"IM-A02"
check_items:
-"液压管路是否有渗漏"
-"料筒温度指示是否在正常范围（180-220℃）"
-"合模机构运行声音是否正常"
photo_orientation:"正面操作面板方向"
A_line_03:
name:"A线3号电机组"
line:"A"
pose:
x:8.5
y:8.4
yaw:1.57
equipment:
type:"motor_unit"
model:"ABB M3BP 160L"
id:"MO-A03"
check_items:
-"电机外壳温度是否异常"
-"是否有异响或异常振动痕迹"
-"接线盒是否有烧焦痕迹"
photo_orientation:"侧面电机主体方向"
B_line_01:
name:"B线1号配电柜"
line:"B"
pose:
x:12.0
y:3.0
yaw:0.0
equipment:
type:"power_cabinet"
model:"施耐德Prisma P"
id:"PC-B01"
check_items:
-"各回路指示灯状态是否正常（绿灯=运行，红灯=故障）"
-"柜内是否有焦糊气味或烧痕"
-"电流表读数是否在额定范围内"
photo_orientation:"正面柜门仪表盘方向"
B_line_02:
name:"B线2号水泵"
line:"B"
pose:
x:12.0
y:6.0
yaw:0.0
equipment:
type:"water_pump"
model:"格兰富CR45"
id:"WP-B02"
check_items:
-"泵体密封处是否有水渍渗漏"
-"压力表读数是否在正常范围（0.3-0.5MPa）"
-"运行声音是否平稳无杂音"
photo_orientation:"正面泵体压力表方向"
# 产线到点位的映射，供LLM快速检索
line_mapping:
A: ["A_line_01", "A_line_02", "A_line_03"]
B: ["B_line_01", "B_line_02"]
all: ["A_line_01", "A_line_02", "A_line_03", "B_line_01", "B_line_02"]4.2 安全规则配置
# config/inspection_safety_rules.yaml
# 巡检专用安全规则，在第13篇基础上针对工厂环境调整
safety_rules:
# 巡检速度限制（工厂地面可能有油渍，速度要慢）
-name:inspection_speed
type:range
param:linear_velocity
min:-0.3
max:0.3
action:clamp
-name:inspection_angular_speed
type:range
param:angular_velocity
min:-0.6
max:0.6
action:clamp
# 生产区域禁入（运行中的设备内部区域）
-name:production_forbidden_zones
type:zone
zones:
-name:"A线设备内部"
polygon: [[8.0, 3.0], [9.0, 3.0], [9.0, 3.8], [8.0, 3.8]]
-name:"B线配电柜内部"
polygon: [[11.5, 2.5], [12.5, 2.5], [12.5, 3.5], [11.5, 3.5]]
action:reject
# 单次巡检最大点位数（防止LLM规划出过长的路线）
-name:max_points_per_inspection
type:threshold
param:point_count
max:10
action:reject
# 导航超时（单个点位导航超过120秒视为失败）
-name:navigation_timeout
type:threshold
param:nav_duration
max:120.0
action:reject
# 巡检时间窗口（仅允许在非生产高峰期巡检）
-name:inspection_time_window
type:time
start:"22:00"
end:"06:00"
action:warn
message:"当前为生产高峰时段，巡检可能影响产线运行"4.3 知识库文档结构
knowledge_base/
├── equipment_manuals/
│ ├── 海天MA2500注塑机操作手册.md
│ ├── 海天MA2500注塑机维修手册.md
│ ├── ABB_M3BP电机维护指南.md
│ ├── 施耐德Prisma配电柜手册.md
│ └── 格兰富CR45水泵手册.md
├── fault_codes/
│ ├── 注塑机故障代码表.md
│ └── 配电柜报警代码表.md
├── repair_procedures/
│ ├── 液压系统泄漏处理流程.md
│ ├── 电机过热排查步骤.md
│ └── 配电柜跳闸恢复流程.md
└── emergency/
├── 火灾应急流程.md
└── 化学品泄漏处置.md五、核心模块实现
5.1 巡检路线规划器
InspectionPlanner 调用 LLM，把用户的自然语言指令转化为有序的巡检点位列表。它把点位库的描述信息注入 prompt，让 LLM 在已知点位中选择，而非凭空生成坐标。
# inspection_robot/inspection_planner.py
"""巡检路线规划器：自然语言 → 巡检点位列表"""
import json
import yaml
from openai import OpenAI
classInspectionPlanner:
"""使用 LLM 将自然语言巡检指令转化为有序点位列表"""
def__init__(self, api_key: str, model: str = "gpt-4o",
points_file: str = "config/inspection_points.yaml"):
self.client = OpenAI(api_key=api_key)
self.model = model
withopen(points_file, "r", encoding="utf-8") as f:
config = yaml.safe_load(f)
self.points = config.get("inspection_points", {})
self.line_mapping = config.get("line_mapping", {})
defplan(self, user_command: str) -> dict:
"""
规划巡检路线
参数:
user_command: 用户自然语言指令，如"检查A线设备"
返回:
{
"understood": True,
"target_line": "A",
"route": ["A_line_01", "A_line_02", "A_line_03"],
"reason": "用户要求检查A线，共3个点位"
}
"""
# 构建点位描述供 LLM 参考
points_desc = self._build_points_description()
system_prompt = (
"你是一个工厂巡检机器人助手。用户会用自然语言下达巡检指令，"
"你需要根据指令从以下巡检点位中选择合适的点位，"
"并按物理顺序（从近到远）排列成巡检路线。\n\n"
f"可用巡检点位：\n{points_desc}\n\n"
"返回 JSON 格式：\n"
'{"understood": bool, "target_line": str, '
'"route": [点位ID列表], "reason": str}\n'
"如果无法理解指令，understood 设为 false，route 为空列表。"
)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_command},
],
temperature=0.1,
response_format={"type": "json_object"},
)
result = json.loads(response.choices[0].message.content)
# 校验 LLM 返回的点位是否真实存在
valid_route = []
for point_id in result.get("route", []):
if point_id inself.points:
valid_route.append(point_id)
result["route"] = valid_route
return result
def_build_points_description(self) -> str:
"""构建点位描述文本，注入 LLM prompt"""
lines = []
for pid, info inself.points.items():
lines.append(
f"- {pid}: {info['name']}，产线={info['line']}，"
f"设备={info['equipment']['model']}({info['equipment']['id']})"
)
return"\n".join(lines)5.2 VLM 异常检测器
AnomalyDetector 调用 GPT-4o 的视觉接口，把设备照片和检查要点一起送入模型，获取结构化的异常判读结果。
# inspection_robot/anomaly_detector.py
"""VLM 异常检测器：设备照片 → 异常判读结果"""
import base64
import json
from openai import OpenAI
classAnomalyDetector:
"""使用 GPT-4o Vision 分析设备照片是否有异常"""
def__init__(self, api_key: str, model: str = "gpt-4o"):
self.client = OpenAI(api_key=api_key)
self.model = model
defanalyze(self, image_path: str, point_info: dict) -> dict:
"""
分析设备照片
参数:
image_path: 照片文件路径
point_info: 点位信息，包含设备名称和检查要点
返回:
{
"has_anomaly": bool,
"anomaly_type": str,
"description": str,
"severity": "info" | "warning" | "critical",
"confidence": float
}
"""
# 读取并编码图片
withopen(image_path, "rb") as f:
image_base64 = base64.b64encode(f.read()).decode("utf-8")
check_items = "\n".join(
f" - {item}"for item in point_info.get("check_items", [])
)
prompt = (
f"你是工厂设备巡检专家。请仔细分析这张照片，"
f"判断设备是否存在异常。\n\n"
f"设备名称：{point_info['name']}\n"
f"设备型号：{point_info['equipment']['model']}\n"
f"检查要点：\n{check_items}\n\n"
f"请逐项检查上述要点，并给出整体判断。\n"
f"以 JSON 格式返回结果：\n"
'{"has_anomaly": bool, "anomaly_type": "异常类型或无", '
'"description": "详细描述", '
'"severity": "info/warning/critical", '
'"confidence": 0.0-1.0}'
)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_base64}"
},
},
],
}
],
temperature=0.1,
response_format={"type": "json_object"},
max_tokens=500,
)
result = json.loads(response.choices[0].message.content)
result["point_id"] = point_info.get("equipment", {}).get("id", "")
return result5.3 RAG 知识库查询
KnowledgeQuery 使用 ChromaDB 存储设备文档的向量索引，发现异常时检索相关维修指南。
# inspection_robot/knowledge_query.py
"""RAG 知识检索：异常描述 → 维修指南"""
import os
import chromadb
from openai import OpenAI
classKnowledgeQuery:
"""基于 ChromaDB 的设备知识库检索"""
def__init__(self, api_key: str, db_path: str = "data/chroma_db",
collection_name: str = "equipment_manuals"):
self.embed_client = OpenAI(api_key=api_key)
self.chroma_client = chromadb.PersistentClient(path=db_path)
self.collection = self.chroma_client.get_or_create_collection(
name=collection_name
)
defquery(self, anomaly_description: str, top_k: int = 3) -> list:
"""
根据异常描述检索相关文档
参数:
anomaly_description: VLM 给出的异常描述
top_k: 返回的文档片段数量
返回:
[{"source": "文档名", "content": "片段内容", "score": 相似度}]
"""
# 生成查询向量
embedding_response = self.embed_client.embeddings.create(
model="text-embedding-3-small",
input=anomaly_description,
)
query_embedding = embedding_response.data[0].embedding
# ChromaDB 检索
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=top_k,
)
documents = []
for i inrange(len(results["ids"][0])):
documents.append({
"source": results["metadatas"][0][i].get("source", "未知"),
"content": results["documents"][0][i],
"score": 1.0 - results["distances"][0][i],
})
return documents
defbuild_advice(self, anomaly: dict, retrieved_docs: list,
api_key: str) -> str:
"""用 LLM 综合异常信息和检索内容生成处置建议"""
client = OpenAI(api_key=api_key)
docs_text = "\n\n".join(
f"【来源：{d['source']}】\n{d['content']}"for d in retrieved_docs
)
prompt = (
f"设备巡检发现异常，请根据检索到的维修手册内容给出处置建议。\n\n"
f"异常信息：\n"
f" 设备：{anomaly.get('point_id', '未知')}\n"
f" 异常类型：{anomaly.get('anomaly_type', '未知')}\n"
f" 描述：{anomaly.get('description', '未知')}\n"
f" 严重程度：{anomaly.get('severity', '未知')}\n\n"
f"检索到的手册内容：\n{docs_text}\n\n"
f"请给出具体的处置步骤，如果手册中没有相关内容请说明。"
)
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0.2,
max_tokens=400,
)
return response.choices[0].message.content知识库的构建脚本，把设备手册文档导入 ChromaDB：
# inspection_robot/build_knowledge_base.py
"""构建设备知识库：把文档导入 ChromaDB"""
import os
import chromadb
from openai import OpenAI
defbuild_knowledge_base(kb_dir: str, db_path: str, api_key: str):
"""
遍历知识库目录，把所有 .md 文档分块后导入 ChromaDB
参数:
kb_dir: 知识库文档目录
db_path: ChromaDB 存储路径
api_key: OpenAI API Key
"""
client = OpenAI(api_key=api_key)
chroma_client = chromadb.PersistentClient(path=db_path)
collection = chroma_client.get_or_create_collection(
name="equipment_manuals"
)
doc_id = 0
for root, dirs, files in os.walk(kb_dir):
for fname in files:
ifnot fname.endswith(".md"):
continue
fpath = os.path.join(root, fname)
withopen(fpath, "r", encoding="utf-8") as f:
content = f.read()
# 按段落分块（以 ## 标题分割）
chunks = split_by_section(content)
for chunk in chunks:
iflen(chunk.strip()) < 20:
continue
# 生成向量
emb_resp = client.embeddings.create(
model="text-embedding-3-small",
input=chunk,
)
embedding = emb_resp.data[0].embedding
collection.add(
ids=[f"doc_{doc_id}"],
embeddings=[embedding],
documents=[chunk],
metadatas=[{"source": fname}],
)
doc_id += 1
print(f"知识库构建完成，共导入 {doc_id} 个文档片段")
defsplit_by_section(text: str, max_length: int = 500) -> list:
"""按章节标题分割文档，超长段落进一步切分"""
chunks = []
current = []
for line in text.split("\n"):
if line.startswith("## ") and current:
chunk = "\n".join(current)
iflen(chunk) > max_length:
chunks.extend(split_long_text(chunk, max_length))
else:
chunks.append(chunk)
current = [line]
else:
current.append(line)
if current:
chunk = "\n".join(current)
iflen(chunk) > max_length:
chunks.extend(split_long_text(chunk, max_length))
else:
chunks.append(chunk)
return chunks
defsplit_long_text(text: str, max_length: int) -> list:
"""按句号切分超长文本"""
sentences = text.replace("。", "。\n").split("\n")
chunks = []
current = ""
for s in sentences:
iflen(current) + len(s) > max_length and current:
chunks.append(current)
current = s
else:
current += s
if current:
chunks.append(current)
return chunks
if __name__ == "__main__":
build_knowledge_base(
kb_dir="knowledge_base",
db_path="data/chroma_db",
api_key=os.environ.get("OPENAI_API_KEY", ""),
)5.4 语音报告模块
VoiceReporter 使用 edge-tts 把巡检结果合成语音播报，同时生成文本报告。
# inspection_robot/voice_reporter.py
"""语音报告模块：巡检结果 → 语音播报 + 文本报告"""
import os
import json
import asyncio
from datetime import datetime
import edge_tts
classVoiceReporter:
"""巡检结果语音播报和报告生成"""
def__init__(self, voice: str = "zh-CN-YunxiNeural"):
self.voice = voice
defgenerate_report(self, inspection_results: list,
route_info: dict) -> dict:
"""
生成结构化巡检报告
参数:
inspection_results: 各点位的检测结果列表
route_info: 路线信息（目标产线、点位数等）
返回:
{"summary": str, "anomalies": list, "report_path": str}
"""
total = len(inspection_results)
anomalies = [r for r in inspection_results if r.get("has_anomaly")]
normal = total - len(anomalies)
# 生成语音播报文本
summary = (
f"巡检完成。本次巡检目标为{route_info.get('target_line', '全部')}线，"
f"共检查{total}个点位。"
f"其中{normal}个点位正常，发现{len(anomalies)}处异常。"
)
if anomalies:
summary += "异常详情："
for a in anomalies:
summary += (
f"{a.get('point_name', '未知设备')}发现"
f"{a.get('anomaly_type', '未知异常')}，"
f"严重程度{a.get('severity', '未知')}。"
)
if a.get("advice"):
summary += f"处置建议：{a['advice']}。"
else:
summary += "所有设备运行正常。"
# 写入报告文件
report_path = f"reports/inspection_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
os.makedirs("reports", exist_ok=True)
full_report = {
"timestamp": datetime.now().isoformat(),
"route_info": route_info,
"total_points": total,
"anomaly_count": len(anomalies),
"results": inspection_results,
"summary": summary,
}
withopen(report_path, "w", encoding="utf-8") as f:
json.dump(full_report, f, ensure_ascii=False, indent=2)
return {
"summary": summary,
"anomalies": anomalies,
"report_path": report_path,
}
asyncdefspeak(self, text: str, output_path: str = "reports/tts_output.mp3"):
"""使用 edge-tts 合成语音"""
communicate = edge_tts.Communicate(text, self.voice)
await communicate.save(output_path)
return output_path
defspeak_sync(self, text: str, output_path: str = "reports/tts_output.mp3"):
"""同步封装语音合成"""
return asyncio.run(self.speak(text, output_path))六、InspectionOrchestrator：巡检编排节点
InspectionOrchestrator 是整个系统的核心，它是一个 ROS2 节点，内部用状态机编排各模块。它接收巡检指令，依次执行路线规划、导航、拍照检测、知识检索、报告生成。
#!/usr/bin/env python3
"""
InspectionOrchestrator - 智能巡检编排节点
整合 LLM 路线规划、Nav2 导航、VLM 异常检测、RAG 知识检索、语音报告
"""
import os
import json
import math
import yaml
import rclpy
from rclpy.node import Node
from rclpy.action import ActionClient
from rclpy.executors import MultiThreadedExecutor
from rclpy.callback_groups import ReentrantCallbackGroup
from nav2_msgs.action import NavigateToPose
from action_msgs.msg import GoalStatus
from geometry_msgs.msg import PoseStamped
from sensor_msgs.msg import Image
from std_msgs.msg import String
from cv_bridge import CvBridge
import cv2
from inspection_robot.inspection_planner import InspectionPlanner
from inspection_robot.anomaly_detector import AnomalyDetector
from inspection_robot.knowledge_query import KnowledgeQuery
from inspection_robot.voice_reporter import VoiceReporter
# 巡检状态机枚举
STATE_IDLE = "IDLE"
STATE_PLANNING = "PLANNING"
STATE_NAVIGATING = "NAVIGATING"
STATE_INSPECTING = "INSPECTING"
STATE_QUERYING = "QUERYING"
STATE_RECOVERING = "RECOVERING"
STATE_REPORTING = "REPORTING"
classInspectionOrchestrator(Node):
"""巡检编排主节点：状态机驱动整个巡检流程"""
def__init__(self):
super().__init__("inspection_orchestrator")
# 声明参数
self.declare_parameter("openai_api_key", "")
self.declare_parameter("model", "gpt-4o")
self.declare_parameter("points_file", "")
self.declare_parameter("rules_file", "")
self.declare_parameter("photo_dir", "photos")
api_key = self.get_parameter("openai_api_key").value
model = self.get_parameter("model").value
points_file = self.get_parameter("points_file").value
# 初始化各模块
self.planner = InspectionPlanner(
api_key=api_key, model=model, points_file=points_file
)
self.detector = AnomalyDetector(api_key=api_key, model=model)
self.knowledge = KnowledgeQuery(api_key=api_key)
self.reporter = VoiceReporter()
self.bridge = CvBridge()
# 加载点位配置
withopen(points_file, "r", encoding="utf-8") as f:
self.points_config = yaml.safe_load(f).get("inspection_points", {})
# Nav2 导航客户端
self.nav_client = ActionClient(
self, NavigateToPose, "navigate_to_pose"
)
# 订阅摄像头图像
self.latest_image = None
self.image_sub = self.create_subscription(
Image, "/camera/image_raw", self.image_callback, 10
)
# 订阅巡检指令（语音节点解析后发布到这里）
self.cmd_sub = self.create_subscription(
String, "/inspection_command", self.command_callback, 10
)
# 发布巡检状态和结果
self.status_pub = self.create_publisher(String, "/inspection_status", 10)
self.result_pub = self.create_publisher(String, "/inspection_result", 10)
# 巡检状态
self.state = STATE_IDLE
self.current_route = []
self.current_point_index = 0
self.inspection_results = []
self.route_info = {}
self.nav_retry_count = 0
self.max_nav_retries = 2
# 确保照片目录存在
os.makedirs(self.get_parameter("photo_dir").value, exist_ok=True)
# 等待导航服务
self.get_logger().info("等待 Nav2 导航服务...")
self.nav_client.wait_for_server()
self.get_logger().info("导航服务已连接，巡检编排节点就绪")
defimage_callback(self, msg: Image):
"""缓存最新一帧摄像头图像"""
self.latest_image = msg
defcommand_callback(self, msg: String):
"""接收巡检指令，启动巡检流程"""
ifself.state != STATE_IDLE:
self.get_logger().warn(
f"当前状态为 {self.state}，忽略新指令：{msg.data}"
)
self._publish_status("BUSY", "巡检进行中，请稍后再试")
return
command = msg.data
self.get_logger().info(f"收到巡检指令：{command}")
self._start_inspection(command)
def_start_inspection(self, command: str):
"""启动巡检流程"""
self.state = STATE_PLANNING
self._publish_status("PLANNING", f"正在规划巡检路线：{command}")
self.inspection_results = []
self.current_point_index = 0
self.nav_retry_count = 0
# 调用 LLM 规划路线
plan_result = self.planner.plan(command)
ifnot plan_result.get("understood") ornot plan_result.get("route"):
self.get_logger().warn(f"无法理解巡检指令：{plan_result}")
self._publish_status("ERROR", "无法理解巡检指令")
self.state = STATE_IDLE
return
self.current_route = plan_result["route"]
self.route_info = {
"target_line": plan_result.get("target_line", "all"),
"total_points": len(self.current_route),
"reason": plan_result.get("reason", ""),
}
self.get_logger().info(
f"巡检路线规划完成：{self.current_route}，"
f"共 {len(self.current_route)} 个点位"
)
self._publish_status(
"ROUTE_PLANNED",
f"路线：{self.current_route}，共{len(self.current_route)}个点位"
)
# 开始导航至第一个点位
self._navigate_to_current_point()
def_navigate_to_current_point(self):
"""导航至当前巡检点位"""
ifself.current_point_index >= len(self.current_route):
# 所有点位巡检完毕，生成报告
self._generate_report()
return
self.state = STATE_NAVIGATING
point_id = self.current_route[self.current_point_index]
point_info = self.points_config[point_id]
self.get_logger().info(
f"导航至第 {self.current_point_index + 1}/{len(self.current_route)} "
f"个点位：{point_info['name']} ({point_id})"
)
self._publish_status(
"NAVIGATING",
f"导航至 {point_info['name']}（{self.current_point_index + 1}/"
f"{len(self.current_route)}）"
)
# 构建 Nav2 目标
goal_msg = NavigateToPose.Goal()
goal_msg.pose = PoseStamped()
goal_msg.pose.header.frame_id = "map"
goal_msg.pose.header.stamp = self.get_clock().now().to_msg()
goal_msg.pose.pose.position.x = float(point_info["pose"]["x"])
goal_msg.pose.pose.position.y = float(point_info["pose"]["y"])
goal_msg.pose.pose.position.z = 0.0
yaw = float(point_info["pose"]["yaw"])
goal_msg.pose.pose.orientation.z = math.sin(yaw / 2.0)
goal_msg.pose.pose.orientation.w = math.cos(yaw / 2.0)
# 发送导航目标
send_goal_future = self.nav_client.send_goal_async(
goal_msg,
feedback_callback=self._nav_feedback_callback,
)
send_goal_future.add_done_callback(self._nav_goal_response_callback)
def_nav_feedback_callback(self, feedback_msg):
"""导航反馈回调"""
# 可在此处理导航进度反馈
pass
def_nav_goal_response_callback(self, future):
"""导航目标响应回调"""
goal_handle = future.result()
ifnot goal_handle.accepted:
self.get_logger().warn("导航目标被拒绝")
self._handle_nav_failure("导航目标被Nav2拒绝")
return
result_future = goal_handle.get_result_async()
result_future.add_done_callback(self._nav_result_callback)
def_nav_result_callback(self, future):
"""导航结果回调"""
if future.result().status == GoalStatus.STATUS_SUCCEEDED:
# 导航成功，开始拍照检测
self.nav_retry_count = 0
self._inspect_current_point()
else:
self._handle_nav_failure("导航未到达目标点")
def_handle_nav_failure(self, reason: str):
"""处理导航失败：重试或跳过"""
self.state = STATE_RECOVERING
point_id = self.current_route[self.current_point_index]
self.get_logger().warn(
f"导航失败：{reason}，点位={point_id}，"
f"重试次数={self.nav_retry_count}/{self.max_nav_retries}"
)
self._publish_status("RECOVERING", f"导航失败：{reason}，尝试恢复")
ifself.nav_retry_count < self.max_nav_retries:
# 重试导航
self.nav_retry_count += 1
self.get_logger().info(f"重试导航（第{self.nav_retry_count}次）...")
self._navigate_to_current_point()
else:
# 超过重试次数，记录失败并跳过
self.get_logger().warn(
f"点位 {point_id} 导航失败，已达最大重试次数，跳过该点位"
)
self.inspection_results.append({
"point_id": point_id,
"point_name": self.points_config[point_id]["name"],
"has_anomaly": None,
"anomaly_type": "巡检失败",
"description": f"导航失败，已跳过：{reason}",
"severity": "warning",
"confidence": 0.0,
})
self.nav_retry_count = 0
self.current_point_index += 1
self._navigate_to_current_point()
def_inspect_current_point(self):
"""在当前点位拍照并进行 VLM 异常检测"""
self.state = STATE_INSPECTING
point_id = self.current_route[self.current_point_index]
point_info = self.points_config[point_id]
self.get_logger().info(f"开始检测：{point_info['name']}")
self._publish_status("INSPECTING", f"正在检测 {point_info['name']}")
# 等待获取一帧图像
ifself.latest_image isNone:
self.get_logger().warn("未收到摄像头图像，等待中...")
self.create_timer(2.0, self._retry_capture)
return
# 保存照片
photo_dir = self.get_parameter("photo_dir").value
photo_path = os.path.join(
photo_dir, f"{point_id}_{self.get_clock().now().to_msg().sec}.jpg"
)
cv_image = self.bridge.imgmsg_to_cv2(self.latest_image, "bgr8")
cv2.imwrite(photo_path, cv_image)
self.get_logger().info(f"照片已保存：{photo_path}")
# 调用 VLM 分析
try:
anomaly_result = self.detector.analyze(photo_path, point_info)
except Exception as e:
self.get_logger().error(f"VLM 分析失败：{e}")
anomaly_result = {
"has_anomaly": None,
"anomaly_type": "检测失败",
"description": f"VLM 分析异常：{e}",
"severity": "warning",
"confidence": 0.0,
}
anomaly_result["point_id"] = point_id
anomaly_result["point_name"] = point_info["name"]
anomaly_result["photo_path"] = photo_path
self.get_logger().info(
f"检测结果：{point_info['name']} - "
f"异常={anomaly_result.get('has_anomaly')}, "
f"类型={anomaly_result.get('anomaly_type')}, "
f"严重程度={anomaly_result.get('severity')}"
)
# 如果发现异常，查询知识库
if anomaly_result.get("has_anomaly"):
self._query_knowledge(anomaly_result)
else:
self.inspection_results.append(anomaly_result)
self._next_point()
def_retry_capture(self):
"""重试拍照"""
ifself.latest_image isnotNone:
self._inspect_current_point()
def_query_knowledge(self, anomaly_result: dict):
"""发现异常时查询 RAG 知识库"""
self.state = STATE_QUERYING
point_id = anomaly_result["point_id"]
self.get_logger().info(
f"发现异常，查询知识库：{anomaly_result['anomaly_type']} - "
f"{anomaly_result['description']}"
)
self._publish_status(
"QUERYING",
f"查询维修指南：{anomaly_result['anomaly_type']}"
)
try:
# RAG 检索
retrieved_docs = self.knowledge.query(anomaly_result["description"])
self.get_logger().info(
f"检索到 {len(retrieved_docs)} 条相关文档"
)
# LLM 生成处置建议
api_key = self.get_parameter("openai_api_key").value
advice = self.knowledge.build_advice(
anomaly_result, retrieved_docs, api_key
)
anomaly_result["advice"] = advice
anomaly_result["references"] = [
d["source"] for d in retrieved_docs
]
self.get_logger().info(f"处置建议：{advice[:100]}...")
except Exception as e:
self.get_logger().error(f"知识库查询失败：{e}")
anomaly_result["advice"] = f"知识库查询失败：{e}"
self.inspection_results.append(anomaly_result)
self._next_point()
def_next_point(self):
"""前往下一个巡检点位"""
self.state = STATE_NAVIGATING
self.current_point_index += 1
ifself.current_point_index < len(self.current_route):
self.get_logger().info(
f"前往下一个点位（{self.current_point_index + 1}/"
f"{len(self.current_route)}）"
)
self._navigate_to_current_point()
else:
self._generate_report()
def_generate_report(self):
"""生成巡检报告并语音播报"""
self.state = STATE_REPORTING
self.get_logger().info("巡检完成，生成报告...")
self._publish_status("REPORTING", "正在生成巡检报告")
report = self.reporter.generate_report(
self.inspection_results, self.route_info
)
self.get_logger().info(f"巡检摘要：{report['summary']}")
self.get_logger().info(f"报告已保存：{report['report_path']}")
# 语音播报
try:
audio_path = self.reporter.speak_sync(report["summary"])
self.get_logger().info(f"语音播报已生成：{audio_path}")
except Exception as e:
self.get_logger().error(f"语音合成失败：{e}")
# 发布巡检结果
result_msg = String()
result_msg.data = json.dumps({
"summary": report["summary"],
"report_path": report["report_path"],
"anomaly_count": len(report["anomalies"]),
"results": self.inspection_results,
}, ensure_ascii=False)
self.result_pub.publish(result_msg)
self._publish_status("COMPLETED", report["summary"])
self.state = STATE_IDLE
self.get_logger().info("巡检流程结束，回到待命状态")
def_publish_status(self, status: str, message: str):
"""发布巡检状态"""
msg = String()
msg.data = json.dumps(
{"state": status, "message": message}, ensure_ascii=False
)
self.status_pub.publish(msg)
defmain(args=None):
rclpy.init(args=args)
node = InspectionOrchestrator()
# 使用多线程执行器，避免回调阻塞
executor = MultiThreadedExecutor()
executor.add_node(node)
try:
executor.spin()
except KeyboardInterrupt:
pass
finally:
node.destroy_node()
rclpy.shutdown()
if __name__ == "__main__":
main()七、Launch 文件
巡检流程涉及多个节点协同工作，用 Launch 文件统一启动。
# launch/inspection_robot.launch.py
"""智能巡检机器人系统启动文件"""
import os
from ament_index_python.packages import get_package_share_directory
from launch import LaunchDescription
from launch.actions import DeclareLaunchArgument, IncludeLaunchDescription
from launch.launch_description_sources import PythonLaunchDescriptionSource
from launch.substitutions import LaunchConfiguration
from launch_ros.actions import Node
defgenerate_launch_description():
pkg_share = get_package_share_directory("inspection_robot")
nav2_bringup_share = get_package_share_directory("nav2_bringup")
# 启动参数
openai_api_key = LaunchConfiguration("openai_api_key")
model = LaunchConfiguration("model")
points_file = os.path.join(pkg_share, "config", "inspection_points.yaml")
rules_file = os.path.join(pkg_share, "config", "inspection_safety_rules.yaml")
declare_args = [
DeclareLaunchArgument(
"openai_api_key",
default_value=os.environ.get("OPENAI_API_KEY", ""),
description="OpenAI API Key",
),
DeclareLaunchArgument(
"model",
default_value="gpt-4o",
description="使用的 LLM/VLM 模型",
),
DeclareLaunchArgument(
"map_file",
default_value=os.path.join(pkg_share, "maps", "factory.yaml"),
description="工厂地图文件",
),
]
# Nav2 导航栈
nav2_launch = IncludeLaunchDescription(
PythonLaunchDescriptionSource(
os.path.join(nav2_bringup_share, "launch", "bringup_launch.py")
),
launch_arguments={
"map": LaunchConfiguration("map_file"),
"use_sim_time": "false",
"params_file": os.path.join(
pkg_share, "config", "nav2_params.yaml"
),
}.items(),
)
# 巡检编排节点
orchestrator_node = Node(
package="inspection_robot",
executable="inspection_orchestrator",
name="inspection_orchestrator",
output="screen",
parameters=[
{
"openai_api_key": openai_api_key,
"model": model,
"points_file": points_file,
"rules_file": rules_file,
"photo_dir": os.path.expanduser("~/inspection_photos"),
}
],
)
# 语音交互节点（第04篇实现）
voice_node = Node(
package="inspection_robot",
executable="voice_interaction",
name="voice_interaction_node",
output="screen",
parameters=[
{
"openai_api_key": openai_api_key,
"whisper_model": "base",
"tts_voice": "zh-CN-YunxiNeural",
"command_topic": "/inspection_command",
}
],
)
# 安全过滤节点（第13篇实现）
safety_node = Node(
package="robot_safety",
executable="safety_filter",
name="safety_filter_node",
output="screen",
parameters=[
{
"openai_api_key": openai_api_key,
"model": model,
"rules_path": rules_file,
}
],
)
# 急停节点（第13篇实现）
estop_node = Node(
package="robot_safety",
executable="emergency_stop",
name="emergency_stop_node",
output="screen",
)
return LaunchDescription(
declare_args
+ [nav2_launch, orchestrator_node, voice_node, safety_node, estop_node]
)setup.py 入口点配置
# setup.py
import os
from glob import glob
from setuptools import setup
package_name = "inspection_robot"
setup(
name=package_name,
version="0.1.0",
packages=[package_name],
data_files=[
("share/ament_index/resource_index/packages",
["resource/" + package_name]),
("share/" + package_name, ["package.xml"]),
("share/" + package_name + "/config",
glob("config/*.yaml")),
("share/" + package_name + "/launch",
glob("launch/*.launch.py")),
("share/" + package_name + "/maps",
glob("maps/*.yaml")),
],
install_requires=["setuptools"],
entry_points={
"console_scripts": [
"inspection_orchestrator = "
"inspection_robot.inspection_orchestrator:main",
"voice_interaction = "
"inspection_robot.voice_interaction_node:main",
"build_knowledge_base = "
"inspection_robot.build_knowledge_base:main",
],
},
)package.xml 依赖声明
<?xml version="1.0"?>
<packageformat="3">
<name>inspection_robot</name>
<version>0.1.0</version>
<description>AI驱动的智能巡检机器人</description>
<maintaineremail="user@example.com">Inspection Robot</maintainer>
<license>MIT</license>
<buildtool_depend>ament_python</buildtool_depend>
<depend>rclpy</depend>
<depend>std_msgs</depend>
<depend>geometry_msgs</depend>
<depend>sensor_msgs</depend>
<depend>nav2_msgs</depend>
<depend>nav2_bringup</depend>
<depend>cv_bridge</depend>
<depend>robot_safety</depend>
<test_depend>ament_copyright</test_depend>
<test_depend>ament_flake8</test_depend>
<test_depend>ament_pep257</test_depend>
</package>八、实战演示
8.1 环境准备
# 编译工作空间
cd ~/ai_robot_ws
colcon build --packages-select inspection_robot robot_safety
# source 环境
source install/setup.bash
# 设置 API Key
export OPENAI_API_KEY="your-api-key-here"
# 构建知识库（首次运行）
ros2 run inspection_robot build_knowledge_base8.2 启动系统
# 启动整个巡检系统
ros2 launch inspection_robot inspection_robot.launch.py \
openai_api_key:=$OPENAI_API_KEY \
model:=gpt-4o8.3 发起巡检
通过语音或话题发布巡检指令：
# 方式1：通过话题发布文字指令
ros2 topic pub --once /inspection_command std_msgs/String \
"data: '检查A线设备'"
# 方式2：通过语音节点说话（语音节点自动识别后发布到 /inspection_command）
# 对着麦克风说："检查A线设备"8.4 巡检流程日志
以下是一次完整巡检的运行日志：
[inspection_orchestrator]: 等待 Nav2 导航服务...
[inspection_orchestrator]: 导航服务已连接，巡检编排节点就绪
[inspection_orchestrator]: 收到巡检指令：检查A线设备
[inspection_orchestrator]: 正在规划巡检路线：检查A线设备
[inspection_orchestrator]: 巡检路线规划完成：['A_line_01', 'A_line_02', 'A_line_03']，共 3 个点位
[inspection_orchestrator]: 导航至第 1/3 个点位：A线1号注塑机 (A_line_01)
[inspection_orchestrator]: 导航至 A线1号注塑机（1/3）
[inspection_orchestrator]: 开始检测：A线1号注塑机
[inspection_orchestrator]: 照片已保存：~/inspection_photos/A_line_01_1782263259.jpg
[inspection_orchestrator]: 检测结果：A线1号注塑机 - 异常=False, 类型=无, 严重程度=info
[inspection_orchestrator]: 前往下一个点位（2/3）
[inspection_orchestrator]: 导航至第 2/3 个点位：A线2号注塑机 (A_line_02)
[inspection_orchestrator]: 导航至 A线2号注塑机（2/3）
[inspection_orchestrator]: 开始检测：A线2号注塑机
[inspection_orchestrator]: 照片已保存：~/inspection_photos/A_line_02_1782263285.jpg
[inspection_orchestrator]: 检测结果：A线2号注塑机 - 异常=True, 类型=泄漏, 严重程度=warning
[inspection_orchestrator]: 发现异常，查询知识库：泄漏 - 设备底部液压管路接口处有液体渗出，疑似液压油泄漏
[inspection_orchestrator]: 检索到 3 条相关文档
[inspection_orchestrator]: 处置建议：根据《海天MA2500注塑机维修手册》第3.2节，液压系统泄漏处理步骤如下：1. 立即停机并切断电源；2. 在泄漏点下方放置接油盘；3. 使用扳手紧固管路接口螺母，扭矩...
[inspection_orchestrator]: 前往下一个点位（3/3）
[inspection_orchestrator]: 导航至第 3/3 个点位：A线3号电机组 (A_line_03)
[inspection_orchestrator]: 导航失败：导航未到达目标点，点位=A_line_03，重试次数=0/2
[inspection_orchestrator]: 重试导航（第1次）...
[inspection_orchestrator]: 导航至第 3/3 个点位：A线3号电机组 (A_line_03)
[inspection_orchestrator]: 开始检测：A线3号电机组
[inspection_orchestrator]: 照片已保存：~/inspection_photos/A_line_03_1782263320.jpg
[inspection_orchestrator]: 检测结果：A线3号电机组 - 异常=False, 类型=无, 严重程度=info
[inspection_orchestrator]: 巡检完成，生成报告...
[inspection_orchestrator]: 巡检摘要：巡检完成。本次巡检目标为A线，共检查3个点位。其中2个点位正常，发现1处异常。异常详情：A线2号注塑机发现泄漏，严重程度warning。处置建议：根据《海天MA2500注塑机维修手册》第3.2节...
[inspection_orchestrator]: 报告已保存：reports/inspection_20260624_022015.json
[inspection_orchestrator]: 语音播报已生成：reports/tts_output.mp3
[inspection_orchestrator]: 巡检流程结束，回到待命状态8.5 查看巡检结果
# 查看巡检状态话题
ros2 topic echo /inspection_status
# 查看巡检结果
ros2 topic echo /inspection_result
# 查看生成的报告文件
cat reports/inspection_20260624_022015.json | python3 -m json.tool报告文件内容示例：
{
"timestamp":"2026-06-24T02:20:15",
"route_info":{
"target_line":"A",
"total_points":3,
"reason":"用户要求检查A线，共3个点位"
},
"total_points":3,
"anomaly_count":1,
"results":[
{
"point_id":"A_line_01",
"point_name":"A线1号注塑机",
"has_anomaly":false,
"anomaly_type":"无",
"severity":"info",
"confidence":0.92,
"photo_path":"~/inspection_photos/A_line_01_1782263259.jpg"
},
{
"point_id":"A_line_02",
"point_name":"A线2号注塑机",
"has_anomaly":true,
"anomaly_type":"泄漏",
"description":"设备底部液压管路接口处有液体渗出，疑似液压油泄漏",
"severity":"warning",
"confidence":0.85,
"advice":"根据《海天MA2500注塑机维修手册》第3.2节，液压系统泄漏处理步骤如下：1. 立即停机并切断电源；2. 在泄漏点下方放置接油盘；3. 使用扳手紧固管路接口螺母...",
"references":["海天MA2500注塑机维修手册.md","液压系统泄漏处理流程.md"]
},
{
"point_id":"A_line_03",
"point_name":"A线3号电机组",
"has_anomaly":false,
"anomaly_type":"无",
"severity":"info",
"confidence":0.90,
"photo_path":"~/inspection_photos/A_line_03_1782263320.jpg"
}
],
"summary":"巡检完成。本次巡检目标为A线，共检查3个点位。其中2个点位正常，发现1处异常。异常详情：A线2号注塑机发现泄漏，严重程度warning。"
}九、部署注意事项
9.1 网络与 API 依赖
本系统依赖 OpenAI API 进行 LLM 路线规划、VLM 视觉分析和 RAG 向量生成。工厂网络环境通常有防火墙限制，部署前需确认：
• 机器人主控计算机能访问 api.openai.com（或配置代理）• API 调用延迟在可接受范围（单次巡检约调用 3-10 次 API） • 准备 API 费用预算：单次巡检约 0.05-0.15 美元（GPT-4o）
如果网络不可靠，可参考第12篇部署本地大模型（如 Qwen-VL），把 LLM 和 VLM 替换为本地推理，RAG 的向量生成也可换用本地嵌入模型（如 bge-small-zh）。
9.2 摄像头与光照
工厂车间光照条件复杂，VLM 的判读准确率受图像质量直接影响：
• 巡检点位的光照强度应保证摄像头能拍到清晰设备面板 • 夜间巡检需开启机器人补光灯 • 摄像头分辨率建议 1080p 以上，VLM 对细节的识别依赖图像清晰度 • 拍照前等待 1-2 秒让机器人完全停止，避免运动模糊
9.3 地图与导航
巡检点位坐标基于工厂地图，部署时需要：
• 用 SLAM 建图确保地图覆盖所有巡检区域 • 点位坐标在建图后实地标定，不能凭图纸估算 • 生产期间设备位置可能变动（叉车搬运物料），定期更新地图 • Nav2 的代价地图参数需针对工厂地面调整（油渍、减速带等）
9.4 知识库维护
RAG 知识库的效果取决于文档质量：
• 设备手册更新后需重新运行 build_knowledge_base导入• 文档分块策略影响检索精度，按章节标题分块优于固定长度切分 • 检索结果中包含来源信息，便于操作人员核验建议的依据 • 敏感设备参数不应放入云端 LLM 上下文，本地 RAG + 本地模型更安全
十、常见问题排查
问题1：LLM 规划的巡检路线包含不存在的点位
原因：LLM 在理解指令时生成了点位库中没有的点位 ID。
解决方案：InspectionPlanner 中已做校验，过滤掉不存在的点位。如果过滤后路线为空，检查点位配置文件中的 ID 是否与 LLM 返回的一致。可以在 System Prompt 中更明确地列出所有可用点位 ID，减少 LLM 的猜测。
# 在 _build_points_description 中强调 ID 格式
lines.append(
f"- 点位ID: {pid}（必须使用此ID）, "
f"名称: {info['name']}, 产线: {info['line']}"
)问题2：VLM 分析结果不稳定，同一设备多次检测结果不同
原因：GPT-4o 的视觉判断存在随机性，温度参数和光照变化都会影响结果。
解决方案：
• 将 VLM 调用的 temperature设为 0.1（已在代码中设置），降低随机性• 对关键设备进行多次拍照分析，取多数结果 • 在检查要点中提供更具体的判断标准，如"料筒温度指示灯绿色为正常，红色为异常"
# 多次拍照取多数结果
def_inspect_with_consensus(self, point_info, num_shots=3):
results = []
for i inrange(num_shots):
# 间隔1秒拍一张
time.sleep(1.0)
photo_path = self._capture_photo(point_info)
result = self.detector.analyze(photo_path, point_info)
results.append(result)
# 取多数投票
anomaly_votes = sum(1for r in results if r.get("has_anomaly"))
consensus = anomaly_votes > num_shots / 2
return consensus, results[0] # 返回共识结果和第一张详情问题3：导航频繁失败，巡检效率低
原因：工厂环境动态障碍物多（物料、人员、叉车），Nav2 路径规划受阻。
解决方案：
• 调整 Nav2 的代价地图膨胀半径，给障碍物留更多余量 • 在巡检时间窗口配置中限制为非生产高峰时段 • 增大导航超时时间，允许 Nav2 等待障碍物移开 • 对长期被阻挡的点位，在配置中标注备用观测位置
# nav2_params.yaml 关键调整
local_costmap:
inflation_layer:
inflation_radius:0.5# 增大膨胀半径
obstacle_layer:
observation_persistence:5.0# 障碍物保持时间问题4：RAG 检索结果与异常不相关
原因：异常描述过于简短或用词与手册中的术语不一致，导致向量相似度低。
解决方案：
• 在 VLM 的 prompt 中要求使用标准术语描述异常（如"液压油泄漏"而非"漏油"） • 扩大检索的 top_k 值，从 3 增加到 5，提高召回率 • 在知识库文档中补充同义词索引（如"漏油"→"液压油泄漏"） • 对检索结果做二次过滤，用 LLM 判断相关性
问题5：语音播报在工厂噪音环境下听不清
原因：工厂车间背景噪音大，TTS 生成的语音音量不足。
解决方案：
• 使用外接扬声器而非机器人内置喇叭，提高音量 • 选择音色浑厚的 TTS 语音（如 zh-CN-YunxiNeural）• 同时生成文本报告，通过工厂 MES 系统推送到管理人员手机 • 考虑在异常严重程度为 critical 时触发声光报警，而非仅语音播报
问题6：知识库构建时部分文档导入失败
原因：文档编码不是 UTF-8，或文档内容包含特殊字符导致分块异常。
解决方案：
# 检查文档编码
file -i knowledge_base/equipment_manuals/海天MA2500注塑机维修手册.md
# 转换编码为 UTF-8
iconv -f GBK -t UTF-8 input.md > output.md在 build_knowledge_base.py 中添加编码容错：
defread_file_safe(fpath: str) -> str:
"""尝试多种编码读取文件"""
for encoding in ["utf-8", "gbk", "gb2312", "latin-1"]:
try:
withopen(fpath, "r", encoding=encoding) as f:
return f.read()
except UnicodeDecodeError:
continue
return""总结
从用户说一句"检查A线设备"开始，到语音播报巡检结果结束，整条链路串联了六项核心技术。InspectionPlanner 把自然语言指令转化为有序巡检点位列表，LLM 在预定义点位库中选择而非凭空生成坐标；机器人沿规划路线逐点移动，导航失败时触发异常恢复机制重试或跳过；到达点位后 AnomalyDetector 调用 GPT-4o Vision 分析设备照片，结合检查要点输出结构化判读结果；发现异常时 KnowledgeQuery 在 ChromaDB 中检索设备手册，LLM 综合检索内容生成处置建议；Whisper 接收语音指令，edge-tts 播报巡检摘要；安全规则引擎校验运动参数，急停节点在物理异常时切断动力。
InspectionOrchestrator 作为编排核心，用状态机管理 IDLE→PLANNING→NAVIGATING→INSPECTING→QUERYING→REPORTING 的完整流程，每个状态对应明确的模块调用和状态转移条件。整套系统不是各模块的简单堆砌，而是通过状态机把异构的异步流程（网络请求、导航动作、图像采集、语音合成）编排成一条可靠的巡检链路。工程上的核心取舍在于：LLM 和 VLM 负责决策层的理解和判断，Nav2 和安全规则负责执行层的确定性和可靠性。决策层允许不确定性（LLM 可能误判），但执行层必须可控（规则引擎兜底）。这种分层设计让系统既具备自然语言交互的灵活性，又保留了工业场景所需的安全边界。
巡检场景跑通之后，下一篇换一个实战方向——语音控制分拣，看怎么用语音指令驱动机械臂完成物品分拣任务。