1. 项目概述:一个能“思考”和“纠错”的本地语音助手
最近在折腾一个挺有意思的项目,我把它叫做 AnovaX。这名字听着有点学术,但它的核心想法其实很直接:做一个完全运行在你本地电脑上的、能像人一样“思考”和“纠错”的语音助手。市面上很多语音助手,比如手机内置的,或者一些智能音箱,它们要么是“一锤子买卖”——你说一句,它执行一个固定动作,要么就是所有“思考”过程都在云端,你的隐私和数据安全总让人有点不放心。
AnovaX 想解决的就是这两个痛点。首先,它完全本地运行,从语音识别到任务规划再到执行,所有数据都在你自己的机器上处理,这对于处理一些敏感信息或者追求极致隐私的用户来说,是刚需。其次,它不是一个简单的“命令-响应”系统,而是一个由大语言模型驱动的“多智能体”系统。你可以把它想象成一个小型团队:有一个“大脑”负责理解你的意图并制定分步计划,还有几个各司其职的“专家”负责执行具体任务,比如查天气、写邮件、控制智能家居。更关键的是,这个“大脑”还具备“自适应恢复”能力——当某个“专家”执行任务出错时,“大脑”能察觉并尝试换种方法或者给出补救方案,而不是直接摆烂告诉你“出错了”。
这个项目融合了几个当下很热的技术点:LLM 规划、类型化执行器、自适应恢复机制,并用 Python 的 Flask 框架作为粘合剂,通过 JSON 进行标准化的数据交换。接下来,我就详细拆解一下我是如何从零开始构建这个系统的,包括核心设计思路、每一步的实操细节、以及过程中踩过的那些坑。
2. 核心架构与设计思路拆解
在动手写代码之前,花时间把架构想清楚至关重要。AnovaX 不是一个简单的脚本,而是一个微服务化的智能体系统,设计上的考量直接决定了后续开发的复杂度和系统的健壮性。
2.1 为什么选择“多智能体”与“规划-执行”范式?
传统的语音助手架构通常是“意图识别 -> 槽位填充 -> 调用单一API”。这种模式对于“播放周杰伦的歌”这类简单指令很有效,但一旦遇到复杂任务就捉襟见肘。比如用户说:“我有点冷,而且房间太亮了,顺便提醒我明天上午十点开会。”这个指令包含了三个子任务:调节温度、调节灯光、设置提醒。一个单体模型很难同时处理好这种多意图的、有逻辑关联的复合指令。
因此,我采用了“规划-执行”范式。其核心思想是引入一个“规划器”(Planner),通常由 LLM 担任,它的职责不是直接执行,而是将用户的自然语言指令解析成一个结构化的任务计划。这个计划就像一份项目甘特图,明确了要做什么、按什么顺序做、以及每个步骤需要什么参数。然后,不同的“执行器”(Executor)来认领并完成计划中的具体任务。这样做的好处非常明显:
- 解耦与复用:规划逻辑和执行逻辑分离。增加新功能(如“发送邮件”)只需增加一个新的执行器,无需改动规划器。执行器可以被不同的计划重复调用。
- 处理复杂指令:LLM 擅长理解和分解复杂、模糊的人类指令,将其转化为明确的步骤序列。
- 易于调试:整个任务的执行过程被清晰地记录为“计划”,哪里出了问题一目了然,方便回溯和优化。
“多智能体”在这里体现为多个独立的、功能专一的执行器。每个执行器都是一个独立的智能体,封装了特定领域的知识和能力(如日历管理、智能家居控制)。
2.2 技术栈选型背后的逻辑
确定了架构范式,接下来就是技术选型。每一个选择都有其具体的考量:
- 核心推理引擎(LLM):这是系统的大脑。我选择了能在本地运行的Llama 2/3 或 Mistral系列的 7B/13B 参数模型。为什么不选 ChatGPT 的 API?核心原因就是隐私和成本。所有语音数据、日程信息都在本地处理,零数据泄露风险。虽然本地模型能力稍弱,但对于规划、文本生成这类任务,经过微调的 7B 模型已经足够可用。使用
llama.cpp或Text Generation Inference库可以高效地在消费级 GPU 甚至 CPU 上运行。 - 服务框架与通信:Flask是一个轻量级、灵活的 Python Web 框架,非常适合快速构建 RESTful API。每个执行器都可以封装成一个独立的 Flask 服务,通过 HTTP 接口提供服务。规划器与执行器之间,以及内部各模块之间,使用JSON作为数据交换格式。JSON 结构清晰、语言无关、易于调试,非常适合用来定义“任务计划”这种结构化的数据。一个计划可能长这样:
{ “plan_id”: “plan_001”, “user_query”: “房间太热且太亮”, “steps”: [ { “step_id”: 1, “action”: “adjust_thermostat”, “parameters”: {“device”: “living_room_ac”, “temperature”: 22}, “depends_on”: [] }, { “step_id”: 2, “action”: “adjust_light”, “parameters”: {“device”: “main_light”, “brightness”: 30}, “depends_on”: [1] } ] } - 语音接口:采用Vosk或Whisper的本地部署版本进行语音识别(STT),使用Coqui TTS或Edge TTS进行语音合成(TTS)。同样,坚持所有组件本地化。
- 执行器类型化:这是提升系统可靠性的关键设计。每个执行器的输入输出都通过Pydantic模型进行严格定义。例如,天气查询执行器,其输入模型会强制要求必须有
location字段,且必须是字符串;输出模型则定义必须包含temperature、condition等字段。这能在运行时尽早发现参数错误,避免把错误的数据传给执行器导致不可预知的崩溃。
2.3 自适应恢复机制的设计
这是 AnovaX 区别于普通系统的亮点。自适应恢复意味着系统不能一错就停。我的设计是让规划器(LLM)也承担一部分“监控”和“恢复”的职责。
- 执行状态反馈:每个执行器完成任务后,不仅要返回业务结果(如“温度已设定”),还必须返回一个标准化的执行状态,包括:
SUCCESS、FAILED、PARTIAL_SUCCESS。如果失败,还需提供错误码和描述。 - 规划器作为协调者:主控服务(或规划器本身)会监控每个步骤的执行状态。当收到
FAILED状态时,它不会直接向用户报错,而是会将原始计划、已执行的步骤、当前失败步骤的详细信息,再次提交给 LLM 规划器。 - LLM 驱动的重规划:规划器基于新的上下文(“我们在做A计划,执行到第二步‘开灯’时失败了,原因是灯泡未连接”),生成一个恢复计划。这个新计划可能包括:重试、跳过该步骤、尝试替代方案(如“打开台灯”)、或者调整后续步骤的参数。
- 循环与终止:系统尝试执行恢复计划。这个过程可以设置最大重试次数。如果最终无法解决,再向用户反馈一个清晰的、包含上下文信息的错误,比如:“抱歉,无法调节主灯(设备离线),但我已经为您降低了空调温度。”
这个机制极大地增强了系统的鲁棒性,让它更像一个能应对意外情况的智能助手,而不是一个脆弱的自动化脚本。
3. 核心模块实现细节
理论说完了,我们来看看代码层面具体怎么实现。我会以最核心的规划器和执行器为例,展示关键代码和配置。
3.1 LLM 规划器的提示工程与输出结构化
规划器的核心是构造一个能引导 LLM 输出结构化计划的提示词。这里面的技巧很多。
基础提示词结构:
planning_prompt_template = “”” 你是一个任务规划助手。请将用户的请求分解为一个逐步执行的任务计划。 你可以调用的执行器动作包括: - get_weather: 获取天气。参数:location (字符串) - control_light: 控制灯光。参数:device_name (字符串), action (枚举:”on”, “off”, “dim”), brightness (可选,整数 0-100) - create_calendar_event: 创建日历事件。参数:title (字符串), start_time (ISO 8601 字符串), end_time (ISO 8601 字符串) - send_email: 发送邮件。参数:to (字符串列表), subject (字符串), body (字符串) 请严格按照以下 JSON 格式输出计划,不要有任何其他解释: {{ “plan_id”: “生成一个唯一UUID”, “user_query”: “用户原始请求”, “steps”: [ {{ “step_id”: 1, “action”: “动作名称”, “parameters”: {{动作所需参数}}, “depends_on”: [] // 依赖的前置步骤ID列表,没有则为空 }} ] }} 用户请求:{user_input} “””关键技巧与注意事项:
- 明确边界:在提示词开头就清晰定义角色和任务,让 LLM 进入状态。
- 枚举能力:把系统能做的所有事(执行器列表)明确告诉 LLM,这是它的“工具库”。LLM 不会使用你没告诉它的工具。
- 参数示例化:每个动作的参数不仅要有名字,最好给出类型和示例(如
brightness (0-100)),这能极大提高 LLM 填充参数的准确性。 - 强制结构化输出:使用“请严格按照以下 JSON 格式输出”这样的强指令,并提供一个几乎完整的 JSON 模板。这对于引导本地小模型输出稳定格式至关重要。更高级的做法可以使用
json mode或function calling,但对于纯本地部署,清晰的模板提示是最实用的。 - 依赖关系:
depends_on字段用于定义步骤间的依赖。比如“关空调”必须在“开窗”之后,这需要 LLM 理解常识。在提示词中可以通过例子来说明。
处理 LLM 输出:LLM 的回复需要被解析和验证。这里一定要做防御性编程。
import json import re def parse_llm_planning_response(llm_output: str): “””解析LLM输出的计划,并做基本验证””” # 1. 尝试从输出中提取JSON块,LLM有时会附带一些说明文字 json_match = re.search(r‘\{.*\}’, llm_output, re.DOTALL) if not json_match: raise ValueError(“无法从LLM响应中找到JSON结构”) json_str = json_match.group() try: plan = json.loads(json_str) except json.JSONDecodeError as e: raise ValueError(f“LLM返回的JSON格式无效: {e}”) # 2. 基础结构验证 required_keys = {“plan_id”, “user_query”, “steps”} if not all(k in plan for k in required_keys): raise ValueError(f“计划缺少必要字段,需要: {required_keys}”) # 3. 步骤验证 for step in plan[“steps”]: if “action” not in step or “parameters” not in step: raise ValueError(“计划中的步骤缺少 ‘action‘ 或 ‘parameters‘ 字段”) # 这里可以进一步验证 action 是否在允许的列表中,参数类型是否匹配 # ... # 4. 生成唯一 plan_id (如果LLM没生成或生成的不合规) if not plan[“plan_id”] or not isinstance(plan[“plan_id”], str): import uuid plan[“plan_id”] = f“plan_{uuid.uuid4().hex[:8]}” return plan3.2 类型化执行器的 Flask 服务实现
执行器是干实事的。我们以实现一个“控制灯光”的执行器为例,展示如何用 Flask 和 Pydantic 构建一个健壮的服务。
首先,定义严格的输入输出模型:
from pydantic import BaseModel, Field, validator from typing import Literal, Optional class LightControlRequest(BaseModel): “””控制灯光请求体””” device_name: str = Field(…, description=“设备名称,如 ‘客厅主灯‘”) action: Literal[“on”, “off”, “dim”] = Field(…, description=“执行的动作”) brightness: Optional[int] = Field(None, ge=0, le=100, description=“亮度百分比,仅当 action=‘dim‘ 时有效”) @validator(‘brightness’) def validate_brightness(cls, v, values): if values.get(‘action’) == ‘dim’ and v is None: raise ValueError(‘“dim”动作必须提供 brightness 参数’) if values.get(‘action’) != ‘dim’ and v is not None: # 非调光动作,忽略 brightness 参数 return None return v class LightControlResponse(BaseModel): “””控制灯光响应体””” success: bool message: str previous_state: Optional[dict] = None # 可选的,返回操作前的状态 new_state: Optional[dict] = None # 可选的,返回操作后的状态使用 Pydantic 的好处是,它能自动进行数据验证和类型转换。如果请求中brightness传了字符串”50″,Pydantic 会尝试将其转为整数50。如果转换失败或不符合ge=0, le=100的约束,在进入业务逻辑前就会抛出清晰的验证错误。
然后,实现 Flask 服务端点:
from flask import Flask, request, jsonify import logging # 假设有一个虚拟的智能家居客户端 from smart_home_client import HomeAssistantClient app = Flask(__name__) client = HomeAssistantClient() # 初始化客户端 logging.basicConfig(level=logging.INFO) @app.route(‘/api/execute/control_light’, methods=[‘POST’]) def control_light(): “””控制灯光执行器端点””” try: # 1. 用Pydantic解析并验证请求 req_data = request.get_json() if not req_data: return jsonify({“success”: False, “message”: “请求体必须为JSON”}), 400 light_req = LightControlRequest(**req_data) # 2. 记录日志 app.logger.info(f“执行控制灯光: device={light_req.device_name}, action={light_req.action}, brightness={light_req.brightness}”) # 3. 调用实际硬件或模拟接口 # 这里根据你的智能家居平台(如Home Assistant, MQTT)进行调用 result = client.control_light( device_name=light_req.device_name, action=light_req.action, brightness=light_req.brightness ) # 4. 构造标准化响应 if result[“success”]: resp = LightControlResponse( success=True, message=f“成功将 {light_req.device_name} 设置为 {light_req.action}”, new_state=result.get(“state”) ) return jsonify(resp.dict()), 200 else: resp = LightControlResponse( success=False, message=f“操作失败: {result[‘error’]}”, ) return jsonify(resp.dict()), 500 except Exception as e: app.logger.error(f“控制灯光执行器内部错误: {e}”, exc_info=True) # 返回一个兜底的错误响应,但依然符合我们的响应模型 resp = LightControlResponse( success=False, message=f“服务器内部错误: {str(e)}” ) return jsonify(resp.dict()), 500 if __name__ == ‘__main__’: # 在生产环境中,应使用 Gunicorn 或 uWSGI app.run(host=‘0.0.0.0’, port=5001, debug=False) # 每个执行器使用不同端口关键点:
- 清晰的路由:
/api/execute/<action_name>的命名规则让主控服务易于动态发现和调用。 - 统一的响应格式:所有执行器都返回包含
success和message的标准响应,主控服务可以统一处理。 - 全面的错误处理:使用 try-except 包裹核心逻辑,确保任何异常都不会导致服务崩溃,而是返回一个友好的错误响应。详细的日志对于后期排查问题不可或缺。
- 端口管理:每个执行器作为一个独立服务运行在不同端口(如5001, 5002),方便独立部署和扩展。
3.3 主控服务与自适应恢复流程
主控服务是系统的调度中心。它接收语音识别后的文本,调用规划器生成计划,然后按顺序调度执行器,并管理整个恢复流程。
主控流程伪代码:
class AnovaXController: def __init__(self, planner_url, executor_urls): self.planner_url = planner_url # 规划器服务地址 self.executor_map = executor_urls # 动作到执行器URL的映射 def process_query(self, user_query: str, max_retries=2): “””处理用户查询的核心流程””” # 1. 生成初始计划 initial_plan = self._call_planner(user_query) execution_plan = initial_plan executed_steps = [] # 记录成功步骤 failed_attempts = 0 # 2. 执行循环 while execution_plan[“steps”] and failed_attempts <= max_retries: current_step = self._get_next_step(execution_plan, executed_steps) if not current_step: break # 所有步骤完成 # 3. 执行当前步骤 step_result = self._execute_single_step(current_step) # 4. 处理结果 if step_result[“success”]: executed_steps.append({ “step_id”: current_step[“step_id”], “result”: step_result }) # 重置失败计数,因为成功执行了一步 failed_attempts = 0 else: # 步骤执行失败 failed_attempts += 1 app.logger.warning(f“步骤 {current_step[‘step_id’]} 执行失败: {step_result[‘message’]}”) # 5. 触发自适应恢复:请求新的恢复计划 recovery_plan = self._request_recovery_plan( original_plan=initial_plan, executed_steps=executed_steps, failed_step=current_step, failure_reason=step_result[‘message’] ) if recovery_plan: execution_plan = recovery_plan # 用新计划替换旧计划 app.logger.info(f“已生成并切换到恢复计划”) else: # 如果规划器也无法生成恢复计划,则彻底失败 return { “overall_success”: False, “message”: f“步骤‘{current_step[‘action’]}’执行失败且无法恢复: {step_result[‘message’]}”, “partial_results”: executed_steps } # 6. 最终结果汇总 if not execution_plan[“steps”]: return {“overall_success”: True, “message”: “所有任务已完成”, “details”: executed_steps} else: return {“overall_success”: False, “message”: “达到最大重试次数,任务未完成”, “details”: executed_steps} def _request_recovery_plan(self, original_plan, executed_steps, failed_step, failure_reason): “””请求LLM生成恢复计划””” recovery_prompt = f“”” 原始计划:{json.dumps(original_plan)} 已成功完成的步骤:{json.dumps(executed_steps)} 当前失败的步骤:{json.dumps(failed_step)} 失败原因:{failure_reason} 请基于以上情况,生成一个恢复计划。你可以: 1. 跳过当前失败步骤(如果它不重要)。 2. 调整参数后重试该步骤。 3. 用一个替代动作替换该步骤(例如,无法打开主灯,改为打开台灯)。 4. 调整后续步骤以适应现状。 请输出新的完整计划JSON。 “”” # 调用规划器服务,传入 recovery_prompt # … 调用逻辑与生成初始计划类似 # 返回新的计划,或 None(如果规划器认为无法恢复)这个流程实现了带状态的重试和重规划。_get_next_step函数需要根据步骤间的depends_on依赖关系来决定哪个步骤是当前可执行的。
4. 部署、调试与性能优化
将各个模块开发完成后,如何把它们有机地组合起来并稳定运行,是另一个挑战。
4.1 本地多服务部署与管理
你会在本地启动多个进程:语音识别服务、TTS服务、LLM服务(或连接本地Ollama)、规划器服务、多个执行器服务、以及主控服务。手动管理这些进程是噩梦。
推荐使用 Docker Compose:
version: ‘3.8’ services: llm-service: image: ollama/ollama:latest # 或你的自定义LLM服务镜像 ports: - “11434:11434” volumes: - ./ollama_data:/root/.ollama command: serve planner: build: ./planner ports: - “5000:5000” environment: - LLM_API_URL=http://llm-service:11434/api/generate depends_on: - llm-service executor-light: build: ./executors/light ports: - “5001:5001” # 可以挂载本地设备配置文件 volumes: - ./config/lights.yaml:/app/config.yaml executor-weather: build: ./executors/weather ports: - “5002:5002” main-controller: build: ./main_controller ports: - “8080:8080” # 对外提供主API environment: - PLANNER_URL=http://planner:5000 - EXECUTOR_LIGHT_URL=http://executor-light:5001 - EXECUTOR_WEATHER_URL=http://executor-weather:5002 depends_on: - planner - executor-light - executor-weather # 语音服务可以单独部署,或集成到main-controller中使用docker-compose up就能一键拉起所有服务。每个服务在独立的容器中运行,互不干扰,端口通过内部网络通信,非常清晰。
如果没有Docker,可以用进程管理工具如 PM2:
# 为每个Python服务创建一个ecosystem.config.js module.exports = { apps: [ { name: “planner”, script: “./planner/app.py”, interpreter: “python3” }, { name: “executor-light”, script: “./executors/light/app.py”, interpreter: “python3” }, { name: “main-controller”, script: “./main_controller/app.py”, interpreter: “python3” }, ] } # 然后 pm2 start ecosystem.config.js4.2 问题排查与日志追踪
当系统行为异常时,清晰的日志链路是救命稻草。你需要为整个系统建立一个统一的请求标识。
- 生成唯一请求ID:在主控服务收到用户查询时,立即生成一个
request_id(如UUID),并在所有后续的日志、HTTP请求头中传递这个ID。# 在主控服务中 request_id = str(uuid.uuid4()) logger.info(f“[{request_id}] 开始处理查询: {user_query}”) # 调用规划器时,在HTTP头中传递 headers = {‘X-Request-ID’: request_id} requests.post(planner_url, json={…}, headers=headers) # 在每个服务(规划器、执行器)中,从请求头获取并记录 request_id = request.headers.get(‘X-Request-ID’, ‘unknown’) app.logger.info(f“[{request_id}] 收到规划请求”) - 结构化日志:使用
structlog或json-logging库输出 JSON 格式的日志,方便用 ELK 或 Loki 等工具收集和检索。每条日志都应包含request_id,service_name,level,timestamp,message等关键字段。 - 监控执行状态:主控服务应维护一个内存或 Redis 中的状态表,记录每个
request_id对应的计划、当前步骤、执行结果等。可以提供一个简单的管理端点来查询这些状态,对于调试非常有用。
4.3 性能优化与缓存策略
本地 LLM 推理是性能瓶颈。以下是一些优化方向:
- 规划结果缓存:对于相同或相似的用户查询,直接返回缓存的结果,避免重复调用 LLM。可以使用语义相似度(如 Sentence-BERT 生成向量,计算余弦相似度)来判断查询是否相似。缓存可以放在 Redis 中。
import redis from sentence_transformers import SentenceTransformer encoder = SentenceTransformer(‘all-MiniLM-L6-v2’) # 轻量级模型 r = redis.Redis() def get_cached_plan(user_query, threshold=0.9): query_vec = encoder.encode(user_query).tobytes() # 简单策略:遍历已有缓存的向量(生产环境应用近似最近邻搜索,如FAISS) for key in r.scan_iter(“plan_cache:*”): cached_vec = r.get(key) similarity = cosine_similarity(query_vec, cached_vec) if similarity > threshold: return json.loads(r.get(key.replace(‘_vec’, ‘’))) # 取回对应的计划 return None - LLM 服务优化:使用
vLLM或llama.cpp的-ngl参数进行 GPU 层卸载,能显著提升推理速度。对于规划任务,可以适当降低生成参数(如temperature=0.1,top_p=0.9)来获得更确定、更快的输出。 - 异步执行:如果计划中的步骤没有依赖关系,可以使用
asyncio或Celery并发执行,缩短整体响应时间。主控服务需要处理好步骤间的依赖图调度。 - 执行器超时与重试:网络调用可能失败。在执行器调用时,必须设置合理的超时时间,并实现简单的重试机制(如最多3次,使用指数退避)。
import requests from tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10)) def call_executor_with_retry(url, payload, request_id): headers = {‘X-Request-ID’: request_id} # 设置5秒连接超时,30秒读取超时 response = requests.post(url, json=payload, headers=headers, timeout=(5, 30)) response.raise_for_status() return response.json()
5. 常见问题与实战避坑指南
在开发和测试 AnovaX 的过程中,我遇到了不少典型问题,这里总结一下,希望能帮你绕过这些坑。
5.1 LLM 规划不稳定的问题与对策
问题:本地小模型(如 7B)生成的计划格式可能飘忽不定,有时会多输出一些解释文字,有时 JSON 格式错误,有时“脑补”出不存在的执行器动作。
解决方案:
- 后处理与重试:像前面
parse_llm_planning_response函数那样,实现一个健壮的解析器。如果解析失败,可以尝试用简单的正则或字符串操作修复常见的格式错误(如多余的逗号、未闭合的引号)。如果修复失败,则将错误信息和原始提示重新发给 LLM,要求它“修正 JSON 格式”。通常第二次它会做得更好。 - 更严格的提示词:在提示词中强调“只输出 JSON,不要有任何其他文字”。使用
json …这样的 Markdown 代码块包裹示例,有时能提高模型输出纯 JSON 的概率。 - 输出引导:如果使用
llama.cpp的grammar功能,可以强制模型输出符合特定 JSON 模式的文本,这是终极解决方案。或者,使用 OpenAI 格式的 API,开启json_mode。 - 动作白名单验证:在解析计划后,务必检查每个
step[“action”]是否在你预先定义好的执行器白名单中。如果不在,要么直接将该步骤标记为失败,要么在生成恢复计划时让 LLM 替换成一个有效动作。
5.2 执行器服务通信与错误处理
问题:网络是不可靠的,执行器服务可能崩溃、重启慢、或者返回非预期的数据。
解决方案:
- 服务发现与健康检查:主控服务不应硬编码执行器地址。可以引入一个简单的服务注册中心(甚至用一个共享的 JSON 文件或 Redis),执行器启动时注册自己(名称、地址、健康状态)。主控服务定期对执行器进行健康检查(
/health端点),只将请求发给健康的服务。 - 熔断与降级:对于频繁失败的执行器,可以使用熔断器模式(如
pybreaker)。短时间内失败次数超过阈值,则暂时“熔断”对该服务的调用,直接返回一个预定义的降级响应(如“天气服务暂时不可用”),并定期尝试恢复。 - 响应格式契约:严格执行 Pydantic 响应模型。即使执行器内部出错,也要捕获异常并返回格式正确的
{“success”: false, “message”: “…”}。这能防止主控服务因为解析响应而崩溃。
5.3 自适应恢复中的逻辑死循环
问题:恢复机制可能陷入死循环。例如,步骤A失败 -> 生成恢复计划B -> B又失败 -> 生成恢复计划C(可能又绕回A)-> 无限循环。
解决方案:
- 设置全局重试上限:如主流程中的
max_retries,对整个恢复过程设置一个上限(如3次)。 - 记录恢复历史:在请求恢复计划时,不仅传递当前失败信息,也传递已经尝试过的恢复方案。提示词中可以加入:“已经尝试过方案X和Y,但都失败了,请提供新的方案。” 这能引导 LLM 避免重复。
- 引入人工干预或默认降级:当达到重试上限时,停止自动化恢复,转而通过 TTS 向用户询问该怎么办(“无法打开客厅灯,您是希望我跳过这一步,还是继续尝试?”),或者执行一个安全的默认操作。
5.4 语音交互的延迟与体验
问题:完整的“语音输入 -> LLM规划 -> 多步执行 -> 语音输出”链路可能很长,用户会感到明显的延迟和“卡顿”。
解决方案:
- 流式响应与进度反馈:在语音识别结束后,可以立即用 TTS 给出一个中间反馈,如“好的,我来处理”。在执行过程中,对于耗时较长的步骤,可以通过声音提示(如一个简短的提示音)或灯光变化来表明系统正在工作。
- 异步执行与后续通知:对于非常耗时的任务(如“帮我整理上个月的所有文档”),系统可以在接受指令后立即响应“好的,这可能需要几分钟,完成后我通知您”,然后在后台异步执行,完成后通过通知音或闪烁灯光提示用户。
- 优化流水线:语音识别、LLM推理、执行器调用,这三者尽可能并行。例如,在 LLM 生成计划的同时,可以提前预加载一些可能用到的执行器客户端连接。
构建 AnovaX 这样的系统是一个持续迭代的过程。从最简单的“开灯关灯”开始,逐步增加执行器,优化规划提示词,完善错误处理机制。最重要的不是一步到位实现所有功能,而是建立一个灵活、可扩展、健壮的框架。这个框架本身,就是应对未来各种复杂语音交互需求的最强武器。