多模态机器人对话导航系统:从感知到执行的 Agent 链路
基于 Python + Pydantic + FastAPI 风格 HTTP 服务搭建的多模态对话 Agent 原型,覆盖 VLM 图像理解、多轮状态管理、结构化输出校验、端云路由、工具调用、主动交互与 Case 分析,面向机器人交互场景的完整技术实践。
多模态机器人对话导航系统:从感知到执行的 Agent 链路
这个项目是我围绕**多模态对话系统(机器人交互方向)**做的一个端到端原型。核心目标不是做一个通用聊天框架,而是把机器人导航场景下的对话链路跑通:从用户的多模态输入(语音/文本 + 摄像头图像 + 传感器),到 Agent 的理解、规划、工具调用,再到最终的结构化回复与执行计划。
整个系统用 Python 重写自一个 Go 版本的机器人导航 demo(eino-vlm-agent-demo),在保留原有主链路(Observer → Planner → Skill → Safety → ExecutionPlan)的同时,把对话状态层、结构化输出校验、端云路由、工具调用和 case 分析做得更系统化。这篇文章会逐项对应 JD 中的知识点,把技术实现摊开讲清楚。
Agent-Native 架构总览
技术栈全景
| 能力 | 具体实现 | 对应 JD |
|---|---|---|
| 多模态输入 | text + image_b64 + sensors | 多模态对话 |
| VLM 视觉理解 | Qwen3.5-0.8B + OpenAI-compatible API | VLM、多模态模型 |
| 文本推理 | Qwen3.5-0.8B | LLM 应用 |
| 模型部署 | vLLM(原 Go 版支持,Python 版兼容 same API) | vLLM |
| 结构化输出 | Pydantic v2 + JSON Schema + 自定义 JSON 提取器 | Pydantic、JSON Schema |
| HTTP 服务 | 类 FastAPI 契约(RESTful JSON + SSE) | FastAPI、API 调用 |
| 多轮状态 | SessionManager + RLock + 自动裁剪 | 多轮状态管理 |
| 端云路由 | RouteStrategy 四态 + shadow 对比 | 端云协同 |
| 工具调用 | RuleBasedToolPlanner + LLM Skill Agent | Tool Use、Function Calling |
| 主动交互 | 澄清 / 恢复 / 安全兜底 | 主动交互 |
| Case 分析 | JSONL 追加日志 + 字段级诊断 | 问题拆解 |
| 图像处理 | PIL + base64 + JPEG 压缩 | 图像理解 |
1. 多模态对话链路搭建
对话链路的入口是一个 HTTP API 服务,暴露 /api/chat 端点,接受多模态输入:
class UserCommand(SchemaModel):
text: str = ""
history: list[str] = Field(default_factory=list)
sensors: dict[str, Any] = Field(default_factory=dict)
image_b64: str = ""
链路内部采用异步消息总线 + 请求/回复模式,每个 Agent 通过 correlation_id 追踪一次完整的对话回合:
class Message(SchemaModel):
id: str
correlation_id: str
from_: str = Field(alias="from")
to: str
role: AgentRole
type: MessageType
payload: Any
timestamp: datetime
流程上,Coordinator 作为主控节点,依次调度 Observer(图像+传感器感知)、Planner(意图理解)、Skill(导航动作选择)、Safety(安全检查),最终产出 ExecutionPlan。整个流程支持dry-run 模式,在未经显式审批前不会触发真实运动,保证调试安全。
服务端同时支持 SSE(Server-Sent Events)向客户端推送 mission 状态变更,前端可以实时看到机器人的执行进度。
2. VLM 图像理解与多模态感知
系统中的 Observer Agent 负责处理图像输入。它接收用户上传的 image_b64,经过预处理后送给 VLM 做场景理解。
2.1 图像预处理
class ImageProcessor:
def __init__(self, max_side: int, jpeg_quality: int, output_dir: str) -> None:
self.max_side = max_side
self.jpeg_quality = jpeg_quality
self.output_dir = output_dir
def process(self, raw_base64: str, correlation_id: str) -> ImageResult | None:
image_bytes = _decode_image_data(raw_base64)
image = Image.open(io.BytesIO(image_bytes))
image = image.convert("RGB")
image = _resize_to_max_side(image, self.max_side)
image.save(filename, format="JPEG", quality=self.jpeg_quality)
这里做了几件事:
- base64 解码:支持标准 base64 和 data URL 格式(
data:image/jpeg;base64,...) - PIL 图像处理:统一转 RGB,按最长边等比例缩放(默认 640px)
- JPEG 压缩:质量 85,平衡清晰度和传输体积
- 落盘存档:每张图按
correlation_id命名,便于事后 case 回溯
2.2 VLM 调用
图像通过 OpenAI-compatible 的 chat_completion_with_images 接口送入 VLM:
class Client(LLMClientProtocol):
def chat_completion_with_images(
self,
endpoint_name: str,
system_prompt: str,
user_prompt: str,
image_paths: list[str],
temperature: float,
max_tokens: int,
) -> str:
content_parts = [{"type": "text", "text": user_prompt}]
for image_path in image_paths:
image_bytes = Path(image_path).read_bytes()
mime, _ = mimetypes.guess_type(image_path)
content_parts.append({
"type": "image_url",
"image_url": {
"url": f"data:{mime};base64,{base64.b64encode(image_bytes).decode('ascii')}"
},
})
body = {
"model": endpoint.model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": content_parts},
],
"temperature": temperature,
"max_tokens": max_tokens,
}
这里用的是 Qwen3.5-0.8B 作为视觉感知和文本推理模型(OBSERVER_MODEL / PLANNER_MODEL / SKILL_MODEL),通过 vLLM 本地部署在 :8003 端口,以 OpenAI-compatible API 接入。同一个客户端也可以无缝切换到云端 DashScope(只需要改 base_url 和 model),实现端云两用。
2.3 结构化视觉输出
VLM 的 system prompt 被约束为严格 JSON 输出:
VISUAL_DETECT_SYSTEM_PROMPT = """You are a visual object detector...
Output strict JSON only, no markdown fences:
{"objects": [{"name": "red_ball", "distance_m": 1.5, "direction": "front"}],
"scene_summary": "前方1.5米有红色球"}"""
Observer 会把 VLM 返回的 JSON 解析成 ObservationPayload,包含 visible_objects、navigable_space、scene_description 等字段,供后续 Planner 使用。
3. 多轮对话状态管理
对话系统要支持任务续接和失败恢复,状态层必须足够清晰。我设计了两层状态:
- Session 级别:维护对话生命周期、当前阶段、活跃任务、用户反馈
- Task 级别:存储当前目标、任务状态字典、工具调用历史
class Session(SchemaModel):
id: str
turns: list[Turn] = Field(default_factory=list)
turns_dropped: int = 0
current_phase: ScenePhase = ScenePhase.IDLE
current_target: str = ""
active_mission_id: str = ""
task_state: dict[str, Any] = Field(default_factory=dict)
user_feedback: str = ""
last_route: str = ""
last_tool_name: str = ""
SessionManager 用 threading.RLock 保证并发安全,提供 get_or_create、set_mission、clear_mission 等原子操作。add_turn 会自动裁剪历史(保留最近 20 轮),防止上下文膨胀。
当用户说”继续刚才的任务”时,系统会检查 session.current_target 和 session.current_phase,构造恢复指令,实现任务续接。
4. 结构化输出设计与校验
LLM 的输出不可控,但我要求系统关键节点必须产出可校验的结构化数据。全项目使用 Pydantic v2 定义 schema,并在 LLM 返回后做 model_validate 校验。
4.1 核心对话状态 Schema
核心的对话状态 schema 覆盖了 JD 中提到的几乎所有结构化字段:
class DialogState(SchemaModel):
speech_act: SpeechAct = SpeechAct.COMMAND
route: RouteStrategy = RouteStrategy.LOCAL_PROVISIONAL
route_reason: str = ""
confidence: float = 0.0
need_clarification: bool = False
clarification_question: str = ""
emotion_style: str = "neutral"
tool_call: ToolCall | None = None
direct_response: str = ""
direct_handled: bool = False
task_state: dict[str, Any] = Field(default_factory=dict)
其中 SpeechAct 枚举区分了 command / question / confirm / clarify / cancel / report 六种言语行为,RouteStrategy 枚举实现了端云路由策略。
4.2 JSON 提取与容错
对于 LLM 返回的 JSON,我实现了一个多层提取器 JSONExtractor,处理各种 LLM 的”坏毛病”:
class JSONExtractor:
def __init__(self, allow_markdown: bool = True, allow_partial: bool = False) -> None:
self.allow_markdown = allow_markdown
self.allow_partial = allow_partial
def extract(self, text: str) -> str:
# 1. 尝试从 markdown code block 中提取
# 2. 尝试从纯文本中扫描 JSON object(处理嵌套括号)
# 3. 直接验证整个字符串是否是合法 JSON
提取器会处理:
```json ... ```包裹的 JSON- 无 fence 的纯 JSON 字符串
- JSON 后面跟了解释文本的情况(通过括号深度匹配只取 object)
- 部分截断 JSON(
allow_partial=True时兜底)
这对应了 JD 中要求的JSON 校验能力。
4.3 Function Calling vs Tool Calling
我在系统中有意区分了这两种模式:
- Function Calling:Planner / Skill Agent 调用 LLM 时,通过 system prompt 约束模型输出结构化字段(如
intent、goal_pose、action_type),然后服务端用 Pydantic 解析。这是模型侧的结构化生成。 - Tool Calling:
tool_planner.py中的RuleBasedToolPlanner,根据规则直接命中工具(如INTRO_TOOL、CANCEL_TOOL),不经过 LLM。这是系统侧的确定性调用。
两者互补:高频、意图明确的指令走 Tool Calling(延迟低、确定性强),复杂语义理解走 Function Calling(灵活度高)。
5. Agent / Tool Use / Planning / Router 原型
5.1 规则驱动的直接工具规划
对于高频、意图明确的用户指令,不需要经过完整的 LLM planning,而是走规则引擎快速响应:
@dataclass(frozen=True)
class ToolPlanningRule:
name: str
predicate: Callable[[str, Session | None], bool]
apply: Callable[[DialogState, str, Session | None], None]
class RuleBasedToolPlanner:
def plan(self, state: DialogState, text: str, session: Session | None = None) -> bool:
lowered = text.lower()
for rule in self._rules:
if rule.predicate(lowered, session):
rule.apply(state, lowered, session)
return True
return False
规则库覆盖的场景:
| 工具 | 触发词 | 行为 |
|---|---|---|
INTRO_TOOL | ”你是谁” / “介绍一下” | 自我介绍,直接回复 |
STATUS_TOOL | ”当前状态” / “在做什么” | 报告 session 和 mission 状态 |
CANCEL_TOOL | ”取消任务” / “停止搜索” | 安全取消活跃任务 |
CONFIRM_TOOL | ”确认一下” / “当前目标” | 确认 current_target |
RESUME_TOOL | ”继续” / “resume” | 任务续接 |
5.2 LLM 驱动的导航动作选择
对于”导航到某地""搜索目标”等复杂意图,由 Skill Agent 调用 LLM 选择 Nav2 动作类型:
class Nav2ActionPayload(SchemaModel):
action_type: str = "" # navigate_to_pose / spin / backup / search_pattern
goal_pose: Pose | None = None
waypoints: list[Pose] = Field(default_factory=list)
spin_angle_deg: float = 0.0
backup_dist_m: float = 0.0
behavior_tree_id: str = ""
reason: str = ""
5.3 意图归一化与地点匹配
Planner 返回的意图会经过 normalize_intent 处理,做校验和降级:
ALLOWED_INTENTS = {"NAVIGATE_TO", "SEARCH_FOR", "AVOID_LEFT", "AVOID_RIGHT", "STOP"}
def normalize_intent(intent: IntentPayload) -> None:
if intent.intent not in ALLOWED_INTENTS:
raise ValueError(f"unsupported intent {intent.intent!r}")
if intent.intent == "STOP":
intent.goal_pose = None
elif not has_reliable_goal_pose(intent.goal_pose):
intent.intent = "SEARCH_FOR"
intent.context = append_fallback_note(intent.context, "Exact goal pose unavailable; downgraded to SEARCH_FOR.")
地点匹配通过 match_known_place 实现,支持 label 和 aliases 的多别名匹配:
def match_known_place(target_label: str, user_command: str, places: list[KnownPlace]) -> tuple[KnownPlace | None, bool]:
for place in places:
for candidate in [place.label, *place.aliases]:
if target == normalized or normalized in command:
return place, True
return None, False
6. 端侧 / 云端路由实验
这是我最贴合 JD 要求的部分之一。我设计了一个四态路由策略,让系统可以根据对话场景、网络状况和安全等级,灵活选择本地模型或云端模型:
class RouteStrategy(str, Enum):
LOCAL_FINAL = "LOCAL_FINAL" # 本地规则或轻量模型直接决策
LOCAL_PROVISIONAL = "LOCAL_PROVISIONAL" # 本地模型出草案,可选 shadow 对比
CLOUD_FINAL = "CLOUD_FINAL" # 云端大模型最终决策
SAFE_FALLBACK = "SAFE_FALLBACK" # 安全兜底,不走 LLM
路由规则写在 dialogue.py 中,根据 speech_act、用户输入特征、session 状态综合判断:
- 疑问句 →
LOCAL_PROVISIONAL(需要本地推理,但不确定时可选 shadow 对比) - 介绍/描述性请求 →
CLOUD_FINAL(云端大模型生成能力更强) - 停止/取消 →
SAFE_FALLBACK(安全相关,不走 LLM,直接规则命中) - 任务续接 →
LOCAL_FINAL(有明确 session 状态支撑,本地直接决策)
def planner_endpoint_for_route(route: RouteStrategy | str) -> str:
route = RouteStrategy(route)
if route == RouteStrategy.CLOUD_FINAL:
return "planner_cloud"
return "planner_local"
6.1 Shadow 对比
同时支持 EINO_ROUTER_ENABLE_SHADOW_COMPARE 环境变量开启 shadow 对比:在 LOCAL_PROVISIONAL 下,本地 planner 和云端 planner 并行执行,记录两者结果差异,用于后续路由策略优化。
class PipelineDiagnostics(SchemaModel):
shadow_planner_endpoint: str = ""
shadow_intent: str = ""
shadow_target_label: str = ""
shadow_confidence: float = 0.0
shadow_matches_primary: bool | None = None
shadow_error: str = ""
这直接对应了 JD 中”构造 local / cloud 对比样本,支持策略验证”的要求。
7. 主动交互与澄清逻辑
系统支持多种主动交互能力,通过 build_dialog_state 统一判断:
| 场景 | 触发条件 | 行为 |
|---|---|---|
| 主动澄清 | 输入为空 / 意图模糊 / 无 session 的代词引用 | 设置 need_clarification=True,返回澄清问题 |
| 任务恢复 | 用户说”继续”且 session 有 current_target | 构造 RESUME_TOOL,恢复之前任务 |
| 失败重试 | mission loop 中某 step 失败 | 自动 replan,进入 REPLANNING 状态 |
| 安全拦截 | 检测到 stop 关键词或前方障碍 | SAFE_FALLBACK,直接生成停止指令 |
| 主动问候 | 系统启动 / 长时间无交互后 | 前端可配置主动问候语 |
if not text:
state.need_clarification = True
state.confidence = 0.0
state.route_reason = "empty_input"
state.clarification_question = "请再说一下你的目标,或者告诉我是导航、搜索还是停止。"
return state
当用户输入模糊(如”那个""继续”)且没有活跃 session 时,系统不会乱猜,而是明确请求澄清:“你希望我做什么?例如导航到某地、搜索目标,或者停止。“
8. Demo 场景工程化
Demo 场景是机器人在室内环境中接收语音/文本指令,完成导航、搜索、介绍等任务。我把这个场景的完整交互链路落成了可运行的原型:
8.1 对话脚本示例
User: "去会议室"
Robot: "收到,正在规划去会议室的路线。"
[导航中...]
User: "停一下"
Robot: "已暂停导航。"
User: "继续"
Robot: "继续刚才的导航任务,目标是会议室。"
8.2 场景阶段管理
class ScenePhase(str, Enum):
IDLE = "idle"
OBSERVATION = "observation"
NAVIGATION = "navigation"
SEARCHING = "searching"
WAITING_USER = "waiting_user"
RECOVERY = "recovery"
coordinator.py 根据当前阶段决定下一步行为。例如 SEARCHING 阶段会进入 mission loop,按预设的搜索步序(spin → move → observe)循环,直到找到目标或超时。
8.3 执行边界
- Chat 默认只返回结构化 plan(dry-run)
- 真实执行需要显式
execute=true或走/api/execution/approve - Mission loop 也受同样的 dry-run 边界约束
9. Case 分析与问题定位
我实现了一个只追加的 JSONL case logger,每次对话结束后自动记录完整上下文,用于后续的问题定位:
class CaseRecord(SchemaModel):
timestamp: str
event_type: str = "chat"
session_id: str = ""
correlation_id: str = ""
user_text: str = ""
route: str = ""
route_reason: str = ""
speech_act: str = ""
need_clarification: bool = False
shadow_matches_primary: bool | None = None
intent: str = ""
target_label: str = ""
action_type: str = ""
safety_status: str = ""
errors: list[str]
diagnostics_notes: list[str]
通过分析 case log,我可以快速定位对话失败的根因:
| 问题类型 | 诊断方法 |
|---|---|
| 模型理解错误 | 对比 user_text → intent → action_type 的映射是否合理 |
| 状态丢失 | 检查 session_id 的 turns 是否连续,turns_dropped 是否非零 |
| 工具调用失败 | 查看 tool_name 和 errors 字段 |
| 结构化输出错误 | 看 diagnostics_notes 中是否有 JSON parse error |
| 路由错误 | 对比 route 与 route_reason,检查 shadow 对比结果 |
这对应了 JD 中”具备较好的问题拆解能力,能将对话体验问题拆成状态、模型、工具、路由或数据问题”的要求。
验证脚本 ./scripts/validate.sh 会跑 backend 单测、路由语义检查、case logging 验证、mission loop 验证和 server 契约测试,最终输出聚合 JSON:
{"result":"PASS","backend":"PASS","smoke":"PASS","frontend":"PASS"}
10. 模型应用与端侧部署
项目中实际接入和验证过的模型:
| 角色 | 模型 | 部署方式 |
|---|---|---|
| Planner(意图+视觉) | Qwen3.5-0.8B | vLLM 本地 :8003 |
| Skill(动作选择) | Qwen3.5-0.8B | vLLM 本地 :8003 |
| Observer(场景描述) | Qwen3.5-0.8B | vLLM 本地 :8003 |
所有模型都通过 OpenAI-compatible API 接入,这意味着:
- 可以一键切换到 GPT-4o、Claude、InternVL、MiniCPM、LLaVA 等任意兼容模型
- 本地部署时可以用 vLLM 做高吞吐推理(原 Go 版本已在 Jetson Orin 上验证)
- 端云路由逻辑与模型具体实现解耦,只做 endpoint 选择
11. 未来演进方向
当前原型是一个最小可运行对话 Agent,下一步可以按以下方向扩展,进一步贴合 JD 的加分项:
11.1 RAG 向量检索
目前的地点匹配是简单的字符串比对(match_known_place)。未来可以引入:
- FAISS 向量索引:把地点描述、场景图片 embedding 存入向量库
- BM25:支持关键词模糊匹配
- Query Rewrite:把用户的口语化表达(“那个开会的地方”)改写为标准查询
- Reranker:对多路召回结果做精排
11.2 更丰富的工具生态
当前工具库只有 5 个基础工具。可以扩展:
- LangChain / LlamaIndex:接入外部知识库、文档问答
- Redis:缓存 session 状态和频繁查询结果,降低 LLM 调用成本
- Gradio / Streamlit:快速搭建可视化调试界面,方便 case 分析
11.3 场景迁移
当前场景是室内机器人导航,但对话系统的核心链路(多轮状态 + 结构化输出 + 工具调用 + 端云路由)可以迁移到:
- 智能座舱:语音控制空调、导航、音乐
- 语音助手:智能家居控制
- 客服助手:多轮问答 + 工单创建
- 数字人:多模态交互 + 情感化回复(
emotion_style字段已预留)
面试真题映射(2026 Agent / 多模态方向面经对照)
以下问题来自 2026 年公开面经(阿里/字节/美团/京东/蚂蚁/联想/淘天等 Agent / AI后端 / 多模态岗),我把它们与本文项目中的具体实现做了一一对应。
Agent 架构与模式
| 面经真题 | 项目中的对应实现 |
|---|---|
| Agent 和 RAG 的区别是什么? | §1 多模态链路(Agent = 感知→规划→工具→执行闭环) vs §11.1 RAG(检索增强生成,未来扩展方向) |
| 为什么要用 Agent,而不是直接单轮问答? | §2 多轮状态管理 + §6 主动交互(任务续接、失败恢复、澄清) |
| ReAct / Workflow / Multi-Agent 的差别? | 当前是 Workflow 风格(Coordinator 固定调度 Observer→Planner→Skill→Safety),未来可扩展为 ReAct(Planner 动态决定下一步调用谁) |
| 单 Agent 和 Multi-Agent 分别适合什么场景? | 当前单 Agent(Coordinator 统一编排),复杂场景可拆分为 Multi-Agent(每个 Agent 独立进程 + Bus 通信) |
| 主 Agent 和子 Agent 如何协作? | Coordinator(Master)通过 Message Bus 向各 Agent 发请求/收回复,correlation_id 追踪全链路 |
上下文工程与记忆
| 面经真题 | 项目中的对应实现 |
|---|---|
| 上下文窗口溢出怎么处理? | §2 add_turn 自动裁剪,保留最近 20 轮,防止 token 膨胀 |
| 长短期记忆怎么分层? | Session(长期:turns / current_target / active_mission_id)→ task_state(中期:user_feedback / resume_target)→ DialogState(短期:单轮对话状态) |
| 记忆压缩怎么做? | §2 summarize_history 提取最近 8 轮,用 “User: … Robot: …” 格式压缩注入 prompt |
| Lost in the Middle 怎么处理? | 历史按时间序排列,最近轮次在 prompt 尾部,降低被中间信息淹没的概率 |
| token 消耗怎么控制? | image 预处理(resize 640px + JPEG 85 质量)+ history 裁剪 + max_tokens 限制(默认 256) |
工具调用协议
| 面经真题 | 项目中的对应实现 |
|---|---|
| MCP 和 Function Calling 的差别? | §4.3 明确区分:Function Calling = LLM 生成结构化参数(模型侧);Tool Calling = 规则引擎直接命中工具(系统侧) |
| Skill 和 MCP 的区别? | 当前 Skill = 规则化工具(INTRO_TOOL / CANCEL_TOOL 等),未来可接入 MCP 标准协议做动态发现 |
| 模型如何选择调用哪个 Skill? | RuleBasedToolPlanner 通过 predicate 匹配输入文本特征 + confidence 阈值 + 白名单校验 |
| Skill 能否保证百分百被调起? | 不能。规则未命中时 fallback 到 LLM Function Calling;need_clarification 兜底 |
| 工具调用传错参怎么兜底? | §5 Safety Agent + ALLOWED_INTENTS / ALLOWED_ACTIONS 白名单 + normalize_intent 校验降级 |
RAG 与知识库
| 面经真题 | 项目中的对应实现 |
|---|---|
| RAG 解决什么问题? | §11.1 未来规划:当前 match_known_place 是关键词检索,未来将引入 FAISS 向量检索 + BM25 混合召回 |
| 向量检索和关键词检索区别? | 关键词:精确匹配 label/aliases(match_known_place);向量:语义相似度(未来 FAISS embedding) |
| 是否做混合检索、Rerank? | 当前未实现,§11.1 已规划混合检索 + Reranker 精排 |
| 知识库怎么持续保鲜? | known_places 从运行时 JSON 配置加载,支持热更新;case log 持续追加用于模型迭代 |
质量、安全、评测
| 面经真题 | 项目中的对应实现 |
|---|---|
| 幻觉怎么处理? | Pydantic Schema 硬约束 + normalize_intent 校验 + ALLOWED_INTENTS 白名单 + has_reliable_goal_pose 降级 |
| Prompt 注入怎么防? | looks_like_stop_command 安全关键词检测 + SAFE_FALLBACK 路由直接走规则,绕过 LLM |
| Agent 死循环怎么避免? | timeout_sec 默认 30s + mission loop 最大时长 300s + max_duration_ms 硬限制 |
| 失败重试怎么做? | §7 mission loop 的 REPLANNING 状态,单 step 失败后自动 replan |
| 如何评估任务完成质量? | §8 CaseLogger 字段级诊断(intent / action_type / safety_status / errors)+ validate.sh JSON 基线对比 |
| 准确率、召回率如何衡量? | shadow_matches_primary 记录端云决策一致性;diagnostics_notes 记录每轮结构化输出错误率 |
大模型推理优化(华为 AI 岗笔试同源)
| 面经真题 | 项目中的对应实现 |
|---|---|
| 为什么用 vLLM 而不是直接调 API? | vLLM 本地部署 Qwen3.5-0.8B,PageAttention + Continuous Batching,降低 TTFT 和 token 成本 |
| KV Cache 是什么? | vLLM 默认启用,缓存历史 Key/Value 避免重复计算;PagedAttention 分页管理减少显存碎片 |
| 量化怎么做? | vLLM 支持 AWQ / GPTQ / FP8 量化;当前 0.8B 模型显存占用极低,Jetson Orin 可直接跑 |
| Prefill vs Decode 瓶颈差异? | Prefill(首 token):处理多模态输入(image + text)计算密集;Decode(生成):结构化 JSON 输出短,带宽压力小 |
| Temperature 的数学作用? | §2 vlm_temperature = 0.1(Observer)和 default_temperature = 0.1(Planner),低温度保证结构化输出确定性 |
| 生成策略怎么选? | 结构化输出场景用 Greedy(temperature→0);开放性描述用 Top-p 采样 |
跨语言后端基础(Java/Go/C++ 面经交叉)
| 面经真题 | 项目中的对应实现 |
|---|---|
| 项目深挖:最复杂的模块是什么? | Coordinator(Master):异步消息总线 + 多 Agent 并发调度 + 状态机切换 + 错误降级 |
| 高并发怎么设计? | SessionManager 用 threading.RLock;NavServer 用 ThreadingHTTPServer;消息总线非阻塞 |
| 数据一致性怎么保证? | Session 状态原子操作(get_or_create / set_mission / clear_mission),case log 只追加不写回 |
| 性能瓶颈在哪?怎么排查? | LLM 调用延迟(~10s)是瓶颈,已通过 shadow 对比 + case log 量化诊断;image 预处理(PIL resize)是次瓶颈 |
| 如果落地失败怎么复盘? | §8 case log 分析矩阵:模型理解错误 / 状态丢失 / 工具调用失败 / 结构化输出错误 / 路由错误 |
总结
这个项目让我对以下能力有了系统性的工程实践:
- ✅ 多模态对话链路:text + image + sensors → VLM/LLM → structured output → tool call → reply
- ✅ 多轮状态管理:Session / task state / user feedback / current target / scene phase
- ✅ 结构化输出校验:Pydantic v2 + JSON Schema + 多层 JSON 提取器
- ✅ Agent / Tool Use:RuleBasedToolPlanner + LLM Function Calling + 意图归一化
- ✅ 端云路由:LOCAL_FINAL / LOCAL_PROVISIONAL / CLOUD_FINAL / SAFE_FALLBACK + shadow 对比
- ✅ 主动交互:澄清、恢复、重试、安全兜底
- ✅ 场景工程化:Demo 对话脚本 + 阶段状态机 + dry-run 边界
- ✅ Case 分析:字段级诊断 + 问题拆解(状态/模型/工具/路由/数据)
- ✅ 模型应用:Qwen3.5-0.8B / vLLM / OpenAI-compatible API
- ✅ 工程能力:Python + Pydantic + 清晰可维护的脚本和服务逻辑