大语言模型（LLM）的崛起为机器人领域带来了前所未有的变革。通过将 GPT-4、PaLM、LLaMA 等模型与 ROS2 系统深度集成，机器人获得了理解自然语言指令、进行常识推理、生成执行策略的能力。本章将深入探讨如何在 ROS2 中构建 LLM 驱动的智能机器人系统，实现从语言理解到物理执行的完整闭环。
LLM 任务规划器作为机器人系统的”大脑”，负责将高层次的自然语言指令转换为可执行的机器人动作序列。在 ROS2 中，我们通过专门的规划节点来封装 LLM 功能：
┌─────────────────────────────────────────────────┐
│ 用户语音/文本输入 │
└────────────────────┬────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────┐
│ LLM Task Planner Node │
│ ┌───────────────────────────────────────────┐ │
│ │ Prompt Engineering Module │ │
│ │ - 环境状态编码 │ │
│ │ - 任务历史管理 │ │
│ │ - 约束条件注入 │ │
│ └───────────────────┬───────────────────────┘ │
│ │ │
│ ┌───────────────────▼───────────────────────┐ │
│ │ LLM Inference Engine │ │
│ │ - API 调用管理 │ │
│ │ - 流式响应处理 │ │
│ │ - 缓存与优化 │ │
│ └───────────────────┬───────────────────────┘ │
│ │ │
│ ┌───────────────────▼───────────────────────┐ │
│ │ Plan Validation Module │ │
│ │ - 语法检查 │ │
│ │ - 可行性验证 │ │
│ │ - 安全性审核 │ │
│ └───────────────────┬───────────────────────┘ │
└──────────────────────┬──────────────────────────┘
│
▼
┌─────────────────────────────────────────────────┐
│ Action Execution Layer │
│ (Nav2 / MoveIt2 / ros2_control) │
└─────────────────────────────────────────────────┘
有效的 Prompt 设计是 LLM 任务规划的核心。我们采用结构化的 Prompt 模板，包含以下关键要素：
典型的 Prompt 结构：
System: You are a robot controller for a mobile manipulator. Your available actions are:
- navigate_to(location): Move to specified location
- pick(object): Grasp the specified object
- place(object, location): Place object at location
- scan(): Perform 360-degree environment scan
Current state:
- Robot position: (x=2.5, y=1.0, theta=0.0)
- Gripper: empty
- Detected objects: [red_cube at table_1, blue_sphere at floor]
- Battery: 85%
Task: {user_instruction}
Generate a step-by-step plan in JSON format with safety checks.
为了处理复杂的多轮对话和长期任务，我们实现了分层记忆系统：
class LLMContextManager:
def __init__(self, max_short_term: int = 10, max_long_term: int = 100):
self.short_term_memory = deque(maxlen=max_short_term) # 最近交互
self.long_term_memory = [] # 重要事件与学习经验
self.episodic_memory = {} # 任务执行历史
self.semantic_memory = {} # 概念与知识图谱
def add_interaction(self, instruction: str, plan: str, result: str):
"""记录完整的交互循环"""
interaction = {
'timestamp': time.time(),
'instruction': instruction,
'plan': plan,
'execution_result': result,
'context_embedding': self._compute_embedding(instruction)
}
self.short_term_memory.append(interaction)
# 评估是否转入长期记忆
if self._is_important(interaction):
self.long_term_memory.append(interaction)
def retrieve_relevant_context(self, query: str, k: int = 5):
"""基于相似度检索相关历史"""
query_embedding = self._compute_embedding(query)
# 结合短期和长期记忆进行检索
all_memories = list(self.short_term_memory) + self.long_term_memory
similarities = [
cosine_similarity(query_embedding, m['context_embedding'])
for m in all_memories
]
top_k_indices = np.argsort(similarities)[-k:]
return [all_memories[i] for i in top_k_indices]
将 LLM 生成的高层计划转换为可执行的行为树是关键步骤：
<!-- LLM 生成的任务转换为行为树 -->
<root main_tree_to_execute="MainTree">
<BehaviorTree ID="MainTree">
<Sequence>
<!-- LLM 规划验证 -->
<Action ID="ValidateLLMPlan" plan="{llm_plan}"/>
<!-- 动态生成的子任务 -->
<Fallback>
<Sequence>
<Action ID="NavigateToTable" goal="{table_position}"/>
<Action ID="IdentifyObject" target="{object_name}"/>
<Action ID="GraspObject" approach="{grasp_strategy}"/>
</Sequence>
<!-- 失败恢复：请求 LLM 重新规划 -->
<Action ID="RequestLLMReplan"
context="{failure_reason}"
max_attempts="3"/>
</Fallback>
<!-- 执行确认 -->
<Action ID="ReportToLLM" status="{execution_status}"/>
</Sequence>
</BehaviorTree>
</root>
视觉-语言模型使机器人能够理解视觉场景并用自然语言描述，实现”看图说话”和”听话看图”的能力：
class VLMPerceptionNode(Node):
def __init__(self):
super().__init__('vlm_perception_node')
# 订阅多模态输入
self.image_sub = self.create_subscription(
Image, '/camera/image_raw', self.image_callback, 10)
self.pointcloud_sub = self.create_subscription(
PointCloud2, '/velodyne/points', self.pointcloud_callback, 10)
# VLM 服务
self.vlm_service = self.create_service(
VLMQuery, 'vlm_query', self.handle_vlm_query)
# 场景描述发布
self.scene_desc_pub = self.create_publisher(
String, '/scene_description', 10)
# 初始化 VLM 模型（如 CLIP, BLIP-2, LLaVA）
self.vlm_model = self._load_vlm_model()
def image_callback(self, msg: Image):
"""处理图像输入并生成场景描述"""
# 图像预处理
cv_image = self.bridge.imgmsg_to_cv2(msg, "rgb8")
# VLM 推理
scene_features = self.vlm_model.encode_image(cv_image)
description = self.vlm_model.generate_description(scene_features)
# 发布场景描述
desc_msg = String()
desc_msg.data = description
self.scene_desc_pub.publish(desc_msg)
def handle_vlm_query(self, request, response):
"""处理视觉问答请求"""
image = self.bridge.imgmsg_to_cv2(request.image, "rgb8")
question = request.question
# 多模态推理
answer = self.vlm_model.answer_question(image, question)
response.answer = answer
response.confidence = self.vlm_model.get_confidence()
return response
VLM 不仅识别物体，还理解它们之间的空间关系和语义联系：
class SceneGraphGenerator:
def __init__(self, vlm_model):
self.vlm = vlm_model
self.scene_graph = nx.DiGraph()
def build_scene_graph(self, image, depth_map):
"""构建场景图谱"""
# 1. 物体检测与分割
objects = self.detect_objects(image)
# 2. 空间关系提取
for obj1, obj2 in itertools.combinations(objects, 2):
relation = self.vlm.infer_relation(image, obj1, obj2)
if relation.confidence > 0.7:
self.scene_graph.add_edge(
obj1.id, obj2.id,
relation=relation.type,
confidence=relation.confidence
)
# 3. 属性标注
for obj in objects:
attributes = self.vlm.extract_attributes(image, obj.bbox)
self.scene_graph.nodes[obj.id].update(attributes)
# 4. 场景标题生成
scene_caption = self.vlm.generate_caption(
image,
context=self.scene_graph
)
return self.scene_graph, scene_caption
将自然语言描述定位到图像中的具体区域：
class VisualGrounding:
def __init__(self):
self.grounding_model = load_model('grounding_dino')
self.sam_model = load_model('segment_anything')
def ground_referring_expression(self, image, expression):
"""
定位指称表达式对应的图像区域
如："红色杯子右边的蓝色笔"
"""
# 1. 提取关键实体
entities = self.parse_entities(expression)
# 2. 目标检测
detections = self.grounding_model.detect(image, entities)
# 3. 关系推理
target = self.resolve_spatial_relations(detections, expression)
# 4. 精确分割
mask = self.sam_model.segment(image, target.bbox)
return {
'bbox': target.bbox,
'mask': mask,
'confidence': target.confidence,
'attributes': target.attributes
}
## 21.3 自然语言导航（VLN）
### 21.3.1 语言指令到路径规划
自然语言导航（Vision-and-Language Navigation）使机器人能够理解和执行复杂的导航指令，如"走到厨房，绕过餐桌，在冰箱前停下"：
```python
class VLNNavigator(Node):
def __init__(self):
super().__init__('vln_navigator')
# 语言编码器
self.language_encoder = BertModel.from_pretrained('bert-base-chinese')
# 视觉编码器
self.vision_encoder = ResNetEncoder()
# 跨模态融合网络
self.cross_modal_attention = CrossModalTransformer(
d_model=768,
n_heads=12,
n_layers=6
)
# 动作预测头
self.action_predictor = nn.Sequential(
nn.Linear(768, 512),
nn.ReLU(),
nn.Linear(512, 6) # 前进/后退/左转/右转/停止/扫描
)
# ROS2 接口
self.nav_goal_pub = self.create_publisher(
NavigateToPose, '/navigate_to_pose', 10)
self.instruction_sub = self.create_subscription(
String, '/navigation_instruction',
self.process_instruction, 10)
def process_instruction(self, msg):
"""处理导航指令"""
instruction = msg.data
# 1. 指令解析与分解
sub_goals = self.decompose_instruction(instruction)
# 2. 执行导航序列
for sub_goal in sub_goals:
# 获取当前观测
current_view = self.get_current_observation()
# 预测下一步动作
action = self.predict_action(sub_goal, current_view)
# 执行动作
self.execute_navigation_action(action)
# 检查是否到达子目标
if self.check_subgoal_reached(sub_goal, current_view):
self.get_logger().info(f"Reached subgoal: {sub_goal}")
def decompose_instruction(self, instruction: str) -> List[str]:
"""将复杂指令分解为子目标"""
# 使用 LLM 或规则分解
prompt = f"""
将以下导航指令分解为简单的子步骤：
指令：{instruction}
输出格式：
1. 第一步动作
2. 第二步动作
...
"""
sub_goals = self.llm_decompose(prompt)
return sub_goals
def predict_action(self, instruction, observation):
"""基于指令和观测预测动作"""
# 编码语言指令
inst_tokens = self.tokenizer(instruction, return_tensors='pt')
inst_features = self.language_encoder(**inst_tokens).last_hidden_state
# 编码视觉观测
vis_features = self.vision_encoder(observation)
# 跨模态注意力
fused_features = self.cross_modal_attention(
inst_features, vis_features
)
# 预测动作
action_logits = self.action_predictor(fused_features.mean(dim=1))
action = torch.argmax(action_logits, dim=-1)
return action
理解空间关系词汇并映射到机器人坐标系：
class SpatialLanguageGrounder:
def __init__(self):
self.spatial_relations = {
'前面': (1.0, 0.0),
'后面': (-1.0, 0.0),
'左边': (0.0, 1.0),
'右边': (0.0, -1.0),
'旁边': 'proximity',
'之间': 'between',
'附近': 'near',
'远离': 'far_from'
}
def ground_spatial_expression(self, expression, scene_objects):
"""
将空间表达接地到具体位置
例如："桌子和椅子之间的位置"
"""
# 解析空间关系
relation, anchors = self.parse_spatial_relation(expression)
# 找到锚点物体
anchor_objects = []
for anchor in anchors:
obj = self.find_object_by_description(anchor, scene_objects)
if obj:
anchor_objects.append(obj)
# 计算目标位置
if relation == 'between':
target_pos = self.compute_between_position(anchor_objects)
elif relation == 'near':
target_pos = self.compute_near_position(anchor_objects[0])
else:
# 相对方向
direction = self.spatial_relations.get(relation)
target_pos = self.compute_relative_position(
anchor_objects[0], direction
)
return target_pos
def compute_between_position(self, objects):
"""计算两个物体之间的位置"""
if len(objects) < 2:
return None
pos1 = np.array([objects[0].x, objects[0].y])
pos2 = np.array([objects[1].x, objects[1].y])
# 中点位置
midpoint = (pos1 + pos2) / 2
# 考虑碰撞，稍作偏移
offset = self.compute_collision_free_offset(midpoint)
return midpoint + offset
处理指令中的歧义并通过对话澄清：
class AmbiguityResolver:
def __init__(self):
self.dialogue_manager = DialogueManager()
self.context_tracker = ContextTracker()
def resolve_ambiguous_reference(self, instruction, scene):
"""解决指称歧义"""
# 提取指称词
references = self.extract_references(instruction)
ambiguous_refs = []
for ref in references:
candidates = self.find_candidates(ref, scene)
if len(candidates) == 0:
# 未找到匹配
question = f"我没有看到{ref}，您能详细描述一下吗？"
clarification = self.ask_for_clarification(question)
# 重新搜索
candidates = self.find_candidates(clarification, scene)
elif len(candidates) > 1:
# 多个候选
question = self.generate_disambiguation_question(
ref, candidates
)
answer = self.ask_for_clarification(question)
candidates = self.filter_by_answer(candidates, answer)
if len(candidates) == 1:
self.context_tracker.add_resolution(ref, candidates[0])
return self.context_tracker.get_resolved_instruction(instruction)
def generate_disambiguation_question(self, reference, candidates):
"""生成消歧问题"""
# 找出区分特征
distinguishing_features = self.find_distinguishing_features(candidates)
if 'color' in distinguishing_features:
colors = [c.color for c in candidates]
return f"您指的是哪个颜色的{reference}？可选：{', '.join(colors)}"
elif 'location' in distinguishing_features:
locations = [self.describe_location(c) for c in candidates]
return f"您指的是哪个{reference}？{', '.join(locations)}"
else:
# 使用索引
return f"我看到{len(candidates)}个{reference}，您指的是第几个？"
Code as Policies 方法让 LLM 直接生成可执行的机器人控制代码：
class CodeAsPolicies:
def __init__(self):
self.llm = GPT4Interface()
self.code_validator = CodeValidator()
self.sandbox_executor = SandboxExecutor()
# 定义机器人 API
self.robot_api = """
# 可用的机器人控制函数
def move_to(x: float, y: float, theta: float = 0.0):
'''移动到指定位置'''
def pick_object(obj_name: str, grasp_type: str = 'top'):
'''抓取指定物体'''
def place_object(location: str):
'''放置物体到指定位置'''
def get_object_position(obj_name: str) -> Tuple[float, float, float]:
'''获取物体位置'''
def detect_objects() -> List[str]:
'''检测当前可见物体'''
"""
def generate_policy(self, task_description: str, scene_context: dict):
"""生成任务执行策略代码"""
prompt = f"""
任务：{task_description}
场景信息：
{json.dumps(scene_context, indent=2)}
使用以下 API 生成 Python 代码完成任务：
{self.robot_api}
要求：
1. 添加必要的错误处理
2. 在关键步骤添加日志
3. 考虑执行失败的恢复策略
生成的代码：
```python
"""
# LLM 生成代码
generated_code = self.llm.complete(prompt)
# 代码验证
validation_result = self.code_validator.validate(generated_code)
if not validation_result.is_valid:
# 基于错误信息重新生成
retry_prompt = f"""
上次生成的代码有以下问题：
{validation_result.errors}
请修正并重新生成：
"""
generated_code = self.llm.complete(retry_prompt)
return generated_code
def execute_policy(self, code: str, timeout: float = 30.0):
"""在沙盒环境中执行生成的代码"""
# 注入运行时环境
runtime_env = {
'move_to': self.robot_move_to,
'pick_object': self.robot_pick,
'place_object': self.robot_place,
'get_object_position': self.get_object_pos,
'detect_objects': self.detect_objects,
'np': np,
'math': math
}
# 沙盒执行
result = self.sandbox_executor.run(
code,
globals=runtime_env,
timeout=timeout
)
return result
确保生成的代码安全可靠：
class SafetyValidator:
def __init__(self):
self.forbidden_modules = ['os', 'subprocess', 'eval', 'exec', '__import__']
self.workspace_limits = {
'x': (-5.0, 5.0),
'y': (-5.0, 5.0),
'z': (0.0, 2.0)
}
self.velocity_limits = {
'linear': 1.0, # m/s
'angular': 1.0 # rad/s
}
def validate_code(self, code: str) -> ValidationResult:
"""静态代码分析"""
result = ValidationResult()
# AST 分析
try:
tree = ast.parse(code)
except SyntaxError as e:
result.add_error(f"语法错误: {e}")
return result
# 检查危险操作
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for alias in node.names:
if alias.name in self.forbidden_modules:
result.add_error(f"禁止导入模块: {alias.name}")
elif isinstance(node, ast.Call):
if isinstance(node.func, ast.Name):
if node.func.id in ['eval', 'exec']:
result.add_error(f"禁止使用函数: {node.func.id}")
# 模拟执行验证
if result.is_valid:
sim_result = self.simulate_execution(code)
if not sim_result.is_safe:
result.add_error(f"模拟执行失败: {sim_result.reason}")
return result
def validate_runtime_action(self, action_type: str, params: dict) -> bool:
"""运行时动作验证"""
if action_type == 'move_to':
x, y = params.get('x'), params.get('y')
# 检查工作空间限制
if not (self.workspace_limits['x'][0] <= x <= self.workspace_limits['x'][1]):
return False
if not (self.workspace_limits['y'][0] <= y <= self.workspace_limits['y'][1]):
return False
# 检查速度限制
if params.get('velocity', 0) > self.velocity_limits['linear']:
return False
elif action_type == 'pick_object':
# 检查抓取力度
if params.get('force', 0) > 50.0: # N
return False
return True
支持基于执行反馈的动态代码调整：
class AdaptiveCodeGenerator:
def __init__(self):
self.execution_history = []
self.learned_patterns = {}
def generate_with_feedback(self, task, max_attempts=3):
"""带反馈循环的代码生成"""
for attempt in range(max_attempts):
# 生成初始代码
if attempt == 0:
code = self.generate_initial_code(task)
else:
# 基于失败原因重新生成
code = self.regenerate_code(
task,
self.execution_history[-1]
)
# 执行代码
result = self.execute_with_monitoring(code)
self.execution_history.append({
'code': code,
'result': result,
'task': task
})
if result.success:
# 成功，学习模式
self.learn_from_success(task, code)
return code
else:
# 失败，分析原因
self.analyze_failure(result)
raise Exception(f"Failed to complete task after {max_attempts} attempts")
def regenerate_code(self, task, last_execution):
"""基于执行反馈重新生成代码"""
error_analysis = self.analyze_error(last_execution['result'].error)
prompt = f"""
任务：{task}
上次执行失败：
代码：{last_execution['code']}
错误：{last_execution['result'].error}
错误分析：{error_analysis}
请生成修正后的代码，避免相同错误：
"""
return self.llm.generate(prompt)
def learn_from_success(self, task, code):
"""从成功执行中学习模式"""
# 提取代码模式
pattern = self.extract_pattern(code)
# 关联任务类型
task_type = self.classify_task(task)
if task_type not in self.learned_patterns:
self.learned_patterns[task_type] = []
self.learned_patterns[task_type].append({
'pattern': pattern,
'code': code,
'success_rate': 1.0
})
Google 的 PaLM-E（Pathways Language Model Embodied）是一个 5620 亿参数的多模态模型，专门为机器人控制设计：
┌────────────────────────────────────────────────┐
│ PaLM-E Architecture │
├────────────────────────────────────────────────┤
│ │
│ ┌──────────────────────────────────────────┐ │
│ │ Vision Encoder (ViT-22B) │ │
│ │ - 4K resolution support │ │
│ │ - Multi-view fusion │ │
│ │ - Temporal encoding │ │
│ └─────────────────┬────────────────────────┘ │
│ │ │
│ ┌─────────────────▼────────────────────────┐ │
│ │ Language Model (PaLM-540B) │ │
│ │ - Instruction following │ │
│ │ - Chain-of-thought reasoning │ │
│ │ - Multi-task learning │ │
│ └─────────────────┬────────────────────────┘ │
│ │ │
│ ┌─────────────────▼────────────────────────┐ │
│ │ Robot Control Decoder │ │
│ │ - Action tokenization │ │
│ │ - Continuous control │ │
│ │ - Safety constraints │ │
│ └──────────────────────────────────────────┘ │
│ │
└────────────────────────────────────────────────┘
class PaLMERobotController:
def __init__(self):
self.model = PaLMEModel.from_pretrained('palm-e-562b')
self.robot_interface = RobotInterface()
# 性能优化配置
self.config = {
'inference_batch_size': 1,
'max_sequence_length': 2048,
'vision_resolution': (1024, 1024),
'control_frequency': 10, # Hz
'planning_horizon': 50, # steps
}
def process_multimodal_input(self, instruction, images, sensor_data):
"""处理多模态输入"""
# 图像编码
vision_tokens = self.model.encode_images(images)
# 传感器数据编码
sensor_tokens = self.encode_sensors(sensor_data)
# 指令编码
text_tokens = self.model.tokenize(instruction)
# 多模态融合
input_sequence = torch.cat([
text_tokens,
vision_tokens,
sensor_tokens
], dim=1)
return input_sequence
def generate_robot_plan(self, instruction, scene):
"""生成机器人执行计划"""
# 多模态输入处理
input_seq = self.process_multimodal_input(
instruction,
scene['images'],
scene['sensors']
)
# 模型推理
with torch.no_grad():
output = self.model.generate(
input_seq,
max_length=self.config['max_sequence_length'],
temperature=0.7,
top_p=0.9
)
# 解码动作序列
action_sequence = self.decode_actions(output)
return action_sequence
Robotics Transformer 2 (RT-2) 将视觉-语言模型直接用于机器人控制：
class RT2Model:
"""
RT-2: Vision-Language-Action Model
将 VLM 的输出空间扩展到包含机器人动作
"""
def __init__(self):
# 基础 VLM (PaLI-X 或 PaLM-E)
self.vlm_backbone = load_pretrained_vlm('pali-x-55b')
# 动作 tokenizer
self.action_tokenizer = ActionTokenizer(
num_bins=256, # 离散化精度
action_dim=7 # 机器人自由度
)
# 扩展词表以包含动作 token
self.extended_vocab = self.vlm_backbone.vocab + self.action_tokenizer.vocab
def forward(self, images, instruction):
"""
输入：图像序列 + 语言指令
输出：机器人动作序列
"""
# 视觉编码
visual_features = self.vlm_backbone.encode_images(images)
# 语言编码
text_features = self.vlm_backbone.encode_text(instruction)
# 跨模态注意力
fused_features = self.vlm_backbone.cross_attention(
visual_features, text_features
)
# 生成动作 token
action_tokens = self.vlm_backbone.generate(
fused_features,
max_new_tokens=10, # 最多10个动作
vocab=self.extended_vocab
)
# 解码为连续动作
actions = self.action_tokenizer.decode(action_tokens)
return actions
RT-2 采用多阶段训练策略：
class RT2Training:
def __init__(self):
self.stages = [
'vision_language_pretraining',
'robot_data_finetuning',
'co_finetuning'
]
def stage1_vl_pretraining(self, model, dataset):
"""第一阶段：视觉-语言预训练"""
# 使用大规模互联网数据
for batch in dataset:
images, captions = batch
# 标准 VLM 训练目标
loss = model.compute_captioning_loss(images, captions)
loss.backward()
# 梯度累积
if step % accumulation_steps == 0:
optimizer.step()
optimizer.zero_grad()
def stage2_robot_finetuning(self, model, robot_dataset):
"""第二阶段：机器人数据微调"""
# 冻结部分层
model.freeze_backbone_layers(keep_top_n=12)
for episode in robot_dataset:
observations = episode['observations']
instructions = episode['instructions']
actions = episode['actions']
# 动作预测损失
pred_actions = model(observations, instructions)
action_loss = F.mse_loss(pred_actions, actions)
# 辅助任务：场景理解
scene_desc = model.describe_scene(observations)
desc_loss = compute_description_loss(scene_desc, episode['scene_labels'])
total_loss = action_loss + 0.1 * desc_loss
total_loss.backward()
def stage3_co_finetuning(self, model, mixed_dataset):
"""第三阶段：混合微调"""
# 同时使用互联网数据和机器人数据
for batch in mixed_dataset:
if batch['type'] == 'internet':
loss = self.compute_vl_loss(model, batch)
else: # robot data
loss = self.compute_robot_loss(model, batch)
# 动态权重平衡
loss = self.balance_losses(loss, batch['type'])
loss.backward()
class RT2Deployment:
def __init__(self):
self.quantization_config = {
'weight_bits': 8,
'activation_bits': 8,
'kv_cache_bits': 4
}
def optimize_for_edge(self, model):
"""边缘设备优化"""
# 1. 模型量化
quantized_model = torch.quantization.quantize_dynamic(
model,
{nn.Linear, nn.Conv2d},
dtype=torch.qint8
)
# 2. 知识蒸馏到小模型
student_model = RT2StudentModel(hidden_size=768) # 原模型的 1/10
student_model = self.distill_knowledge(model, student_model)
# 3. 模型剪枝
pruned_model = self.structured_pruning(
student_model,
pruning_ratio=0.3
)
# 4. 编译优化
compiled_model = torch.compile(
pruned_model,
mode='reduce-overhead',
backend='inductor'
)
return compiled_model
def streaming_inference(self, model, instruction, image_stream):
"""流式推理优化"""
action_buffer = []
for image in image_stream:
# 增量处理
features = model.encode_image_incremental(image)
# 预测下一个动作
action = model.predict_next_action(
instruction,
features,
history=action_buffer[-10:] # 使用最近10个动作作为上下文
)
action_buffer.append(action)
# 并行执行和推理
yield action
关键论文推荐：
class HallucinationMitigation:
"""LLM 幻觉缓解策略"""
def __init__(self):
self.fact_checker = FactChecker()
self.confidence_threshold = 0.8
def validate_llm_output(self, llm_response, context):
"""验证 LLM 输出的真实性"""
# 1. 检查物理可行性
if not self.check_physical_feasibility(llm_response):
return False, "违反物理约束"
# 2. 检查环境一致性
if not self.check_environmental_consistency(llm_response, context):
return False, "与环境状态不一致"
# 3. 检查动作序列合理性
if not self.check_action_sequence_validity(llm_response):
return False, "动作序列不合理"
return True, "验证通过"
class LatencyOptimization:
"""延迟优化策略"""
def __init__(self):
self.cache = LRUCache(maxsize=1000)
self.predictor = ActionPredictor()
def reduce_inference_latency(self, instruction):
# 1. 查询缓存
if instruction in self.cache:
return self.cache[instruction]
# 2. 预测性执行
predicted_action = self.predictor.predict(instruction)
# 3. 异步 LLM 调用
future = self.async_llm_call(instruction)
# 4. 使用预测结果，同时等待 LLM
return predicted_action, future
class SafetyMonitor:
"""安全监控系统"""
def __init__(self):
self.safety_rules = self.load_safety_rules()
self.emergency_stop = EmergencyStop()
def monitor_execution(self, action_stream):
for action in action_stream:
# 1. 预检查
if not self.pre_check_action(action):
self.emergency_stop.trigger()
raise SafetyViolation(f"Unsafe action: {action}")
# 2. 执行监控
self.monitor_action_execution(action)
# 3. 后验证
if not self.post_validate_state():
self.rollback_action(action)
本章深入探讨了大语言模型与 ROS2 机器人系统的深度集成。我们学习了：
LLM 任务规划：如何构建基于 LLM 的高层任务规划器，包括 Prompt 工程、上下文管理和行为树集成
多模态感知：VLM 模型在场景理解、视觉问答和 Grounding 任务中的应用
自然语言导航：实现从语言指令到机器人导航的完整管道，包括空间语言理解和歧义消解
Code as Policies：让 LLM 直接生成可执行代码，实现灵活的任务执行策略
产业实践：通过 Google PaLM-E 案例了解大规模部署经验
前沿技术：RT-2 等最新模型架构和训练策略
关键要点：
练习 21.1：设计一个简单的 LLM 任务规划器，能够将”清理桌子”这样的高层指令分解为具体的机器人动作序列。
练习 21.2：实现一个基础的视觉 Grounding 系统，能够定位”红色杯子左边的书”这样的指称表达。
练习 21.3：设计一个 Code as Policies 的安全验证器，防止生成的代码执行危险操作。
练习 21.4：设计一个多轮对话系统，能够通过询问用户来消除指令中的歧义。
练习 21.5：实现一个 LLM 驱动的错误恢复系统，当任务执行失败时能自动分析原因并尝试新策略。
练习 21.6：设计一个分布式 LLM 机器人系统，多个机器人可以协作完成复杂任务。
练习 21.7：实现一个自适应 Prompt 优化系统，根据执行效果自动改进 Prompt 模板。
练习 21.8：设计一个边缘-云协同的 LLM 机器人架构，平衡实时性和计算能力。