1. 项目概述为什么我们需要一个“本地化、多智能体”的语音助手如果你和我一样对市面上的语音助手总感觉“差那么点意思”那这个项目可能就是你一直在找的答案。我们习惯了那些需要联网、反应迟缓、功能固定且无法深度定制的语音助手。它们要么听不懂复杂的指令要么听懂了也做不了什么更别提在断网环境下完全瘫痪了。今天要聊的AnovaX就是一次彻底打破这些限制的尝试。它不是一个简单的“语音转文本再查天气”的工具而是一个运行在你本地电脑上的、由多个“专家”智能体Agent协同工作的、具备高级规划和自我纠错能力的语音指挥中心。简单来说AnovaX 的核心价值在于三个关键词本地Local、多智能体Multi-Agent、规划与执行Planning Execution。本地化意味着你的所有语音数据、对话历史和任务执行过程都留在你自己的机器上隐私和安全得到最大程度的保障。多智能体架构则像组建了一个“专家顾问团”——当你下达一个模糊指令如“帮我安排下周的会议并写个总结”系统会先让一个“规划师”智能体基于LLM拆解任务第一步访问日历第二步起草邮件第三步生成会议纪要。然后它会调用对应的“执行器”智能体Typed Executor去完成每一步的具体操作。而“自适应恢复”能力则让这个系统具备了“韧性”——当某个步骤出错比如日历API暂时不可用它不会直接崩溃而是尝试分析错误原因调整执行路径或者向你请求更多信息。从技术栈来看它巧妙地融合了当前最热门的技术方向用Python作为粘合剂Flask提供轻量级的Web服务接口JSON作为智能体间通信和数据交换的标准格式而这一切的大脑则是一个本地部署或通过API调用的大语言模型LLM。这不仅仅是又一个“玩具项目”它为我们展示了如何将前沿的AI能力LLM的规划与推理与传统的、可靠的软件工程实践类型化接口、错误处理相结合构建出真正实用、可控的自动化工具。接下来我将带你深入AnovaX的内部看看它是如何工作的以及我们如何能亲手搭建并定制一个属于自己的智能语音助手。2. 架构核心拆解LLM规划器、类型化执行器与状态机要理解AnovaX我们必须先抛开“语音助手”这个表象把它看作一个任务驱动的自动化系统。它的核心工作流可以抽象为感知语音输入→ 理解与规划LLM→ 分发与执行多智能体→ 反馈与调整自适应恢复。下面我们来逐一拆解这三个核心组件。2.1 LLM作为“总指挥”从模糊指令到可执行计划LLM在这里扮演的角色是“首席规划官”。它的输入是你的语音转文字后的自然语言指令输出是一个结构化的、可执行的任务计划Plan。这个计划通常是一个JSON数组其中每个元素代表一个子任务。为什么是LLM传统的规则引擎或意图识别模型在处理“帮我查一下明天天气如果下雨就提醒我带伞并取消下午的户外跑步计划”这类嵌套、多步骤、带条件的指令时会变得极其复杂和脆弱。LLM凭借其强大的语义理解和上下文推理能力可以自然地理解这种复杂指令并将其分解为逻辑步骤。一个典型的工作流程如下语音识别STT将用户的音频流实时转换为文本。这一步可以使用本地模型如Vosk、Whisper.cpp或云服务API。指令理解与规划转换后的文本被送入LLM。我们通过精心设计的系统提示词System Prompt来引导LLM的行为。这个提示词会定义系统的角色“你是一个任务规划助手”、可用的工具集“你可以使用以下功能查询天气、管理日历、发送邮件…”以及输出格式规范“请以如下JSON格式输出你的计划…”。计划生成LLM根据指令和可用工具生成一个计划。例如对于指令“安排一个明天下午3点与团队的会议主题是项目评审”LLM可能输出[ { “step_id”: 1, “action”: “check_calendar_availability”, “parameters”: { “time”: “tomorrow 15:00”, “duration”: “1h”, “attendees”: [“team”] } }, { “step_id”: 2, “action”: “create_calendar_event”, “parameters”: { “title”: “项目评审会议”, “start_time”: “2023-10-27T15:00:00”, “end_time”: “2023-10-27T16:00:00”, “attendees”: [“aliceexample.com”, “bobexample.com”] }, “depends_on”: [1] } ]这个JSON计划清晰地定义了动作序列、参数和依赖关系第二步依赖于第一步的成功。提示设计一个稳定可靠的系统提示词是成功的关键。它需要明确边界LLM不能做什么、格式化输出要求并包含足够的示例Few-shot Learning。一个常见的技巧是让LLM在无法理解或缺少信息时输出一个特定的“请求澄清request_clarification”动作而不是胡乱猜测。2.2 类型化执行器让每个“专家”可靠地干活计划中的每一个action如check_calendar_availability都需要一个对应的执行器Executor来具体实现。AnovaX强调“类型化Typed”这是其工程健壮性的基石。什么是类型化执行器它意味着每个执行器都有严格定义的输入和输出接口Schema。这通常通过Python的Pydantic库来实现。例如创建日历事件执行器的输入模型可能定义为from pydantic import BaseModel from datetime import datetime from typing import List class CreateEventInput(BaseModel): title: str start_time: datetime end_time: datetime attendees: List[str] description: str “”而它的输出模型可能包含一个event_id字段。这样做的好处是自动验证在代码执行前系统就能确保传入的参数类型和结构是正确的避免了运行时因数据格式错误导致的崩溃。清晰契约每个执行器做什么、需要什么、返回什么一目了然便于开发和维护。安全过滤可以防止LLM生成的计划中包含恶意或意外的参数。执行器如何工作每个执行器被注册到一个中央执行器注册表中。当调度器需要执行action: “create_calendar_event”时它会从注册表中找到对应的执行器函数将parameters字典反序列化成定义好的Pydantic模型然后调用该函数。函数执行完毕后再将结果序列化传递给下一个步骤或作为最终结果返回。注意执行器的实现应尽量保持“无状态”和“幂等”。即多次执行相同参数应产生相同的结果且执行器内部不依赖全局可变状态。这有助于错误恢复和重试机制的实现。2.3 自适应恢复状态机系统的“免疫系统”这是AnovaX区别于普通脚本最酷的部分。多步骤任务难免会出错网络波动、API限流、资源不存在、权限不足……一个脆弱的系统会直接报错退出。而AnovaX的“自适应恢复”机制试图让系统像人一样遇到挫折后尝试其他办法。其核心是一个状态机State Machine它管理着每个任务的生命周期PENDING-PLANNING-EXECUTING-SUCCEEDED/FAILED/NEEDS_RECOVERY。当某个步骤执行失败时系统不会立即将整个任务标记为FAILED而是进入NEEDS_RECOVERY状态。此时恢复处理器Recovery Handler被触发。恢复策略可以有多层简单重试对于瞬时错误如网络超时自动重试几次。参数调整将错误信息如“未找到文件A”反馈给LLM规划器请求其重新规划或调整参数“请使用文件B代替”生成一个新的子计划。备选路径如果某个执行器彻底失败如服务不可用LLM可能会被询问是否有替代方案“无法通过Google日历创建事件是否改为发送一封会议邀请邮件”。人工介入当自动恢复尝试多次后仍失败系统可以生成一个清晰的摘要并通过TTS语音或界面提示向用户请求帮助。这个循环执行→失败→分析→重规划/调整→再执行构成了“自适应”的核心。它要求系统不仅能执行计划还能理解执行结果包括错误并具备一定的元认知能力来调整自己的行为。3. 从零开始搭建技术栈选型与核心模块实现理解了架构我们就可以动手搭建了。这里我会基于常见的开源工具和库给出一个可运行的实现方案。我们的技术栈如下Python主语言、FlaskWeb服务框架、LangChain/LlamaIndexLLM应用框架可选但推荐、Pydantic数据验证、SQLite/Redis状态存储。3.1 环境准备与项目结构首先创建一个清晰的项目目录结构这是保持代码可维护性的第一步。anova_x/ ├── app.py # Flask应用主入口 ├── config.py # 配置文件 ├── requirements.txt # 项目依赖 ├── agents/ # 智能体相关模块 │ ├── __init__.py │ ├── planner.py # LLM规划器 │ ├── registry.py # 执行器注册表 │ └── executors/ # 具体执行器 │ ├── __init__.py │ ├── calendar.py │ ├── email.py │ └── system.py ├── core/ # 核心逻辑 │ ├── __init__.py │ ├── state_manager.py # 状态管理 │ └── recovery.py # 恢复逻辑 ├── services/ # 外部服务封装 │ ├── stt.py # 语音识别服务 │ └── tts.py # 语音合成服务 └── utils/ # 工具函数 └── validation.py安装核心依赖requirements.txtflask2.3.0 pydantic2.0.0 openai1.0.0 # 或其他LLM SDK如 llama-cpp-python langchain0.1.0 # 用于简化LLM调用和工具管理 speechrecognition3.10.0 # 语音识别库调用本地或在线服务 pyttsx32.90 # 本地文本转语音 pymysql # 如果需要连接MySQL redis4.5.0 # 用于分布式状态存储可选3.2 实现执行器注册与类型化调用这是系统的基石。我们先在agents/registry.py中创建一个全局的执行器注册表。# agents/registry.py from typing import Dict, Any, Callable, Optional from pydantic import BaseModel, ValidationError import inspect class ExecutorRegistry: def __init__(self): self._executors: Dict[str, Dict[str, Any]] {} def register(self, name: str, input_model: Optional[BaseModel] None, output_model: Optional[BaseModel] None): 装饰器用于注册一个执行器函数。 def decorator(func: Callable): self._executors[name] { “func”: func, “input_model”: input_model, “output_model”: output_model, “description”: func.__doc__ or “” # 函数的文档字符串可作为描述供LLM参考 } return func return decorator async def execute(self, action: str, parameters: Dict[str, Any]) - Any: 查找并执行指定的动作。 if action not in self._executors: raise ValueError(f“Executor ‘{action}’ not found.”) executor_info self._executors[action] func executor_info[“func”] input_model executor_info[“input_model”] # 1. 参数验证与转换 validated_params parameters if input_model: try: validated_params input_model(**parameters).dict() except ValidationError as e: raise ValueError(f“Invalid parameters for ‘{action}’: {e}”) # 2. 执行函数 # 检查函数是否是异步的 if inspect.iscoroutinefunction(func): result await func(**validated_params) else: result func(**validated_params) # 3. 输出验证可选 output_model executor_info[“output_model”] if output_model and result is not None: try: result output_model(**result) if isinstance(result, dict) else output_model(result) except ValidationError: # 如果输出不符合模型可以记录日志但不一定抛出异常 pass return result def get_executor_descriptions(self) - str: 生成所有执行器的描述用于构造LLM的系统提示词。 descriptions [] for name, info in self._executors.items(): desc f“- {name}: {info[‘description’]}” if info[‘input_model’]: # 简单展示输入字段更复杂的可以生成JSON Schema fields list(info[‘input_model’].__fields__.keys()) desc f“ 输入字段: {fields}” descriptions.append(desc) return “\n”.join(descriptions) # 全局注册表实例 registry ExecutorRegistry()然后我们实现几个具体的执行器。例如在agents/executors/system.py中# agents/executors/system.py from agents.registry import registry from pydantic import BaseModel from datetime import datetime import os class GetTimeInput(BaseModel): format: str “%Y-%m-%d %H:%M:%S” class GetTimeOutput(BaseModel): current_time: str registry.register(“get_current_time”, input_modelGetTimeInput, output_modelGetTimeOutput) def get_current_time(format: str “%Y-%m-%d %H:%M:%S”) - dict: “”“获取当前系统时间。”“” now datetime.now() return {“current_time”: now.strftime(format)} class ListFilesInput(BaseModel): directory_path: str registry.register(“list_files”, input_modelListFilesInput) def list_files(directory_path: str) - list: “”“列出指定目录下的文件。如果目录不存在返回空列表。”“” try: files os.listdir(directory_path) return files except FileNotFoundError: return []通过这样的设计我们实现了严格的输入输出契约。LLM规划器在生成计划时可以参考registry.get_executor_descriptions()返回的工具描述确保生成的parameters字段名与Pydantic模型匹配。3.3 构建LLM规划器与Flask API接口接下来我们实现规划器。这里使用LangChain来简化与LLM的交互。我们在agents/planner.py中实现# agents/planner.py import json from langchain.chains import LLMChain from langchain.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate from langchain.chat_models import ChatOpenAI # 示例使用OpenAI可替换为其他模型 from agents.registry import registry class LLMPlanner: def __init__(self, llm_model_name“gpt-3.5-turbo”): # 初始化LLM。对于本地模型可以使用 ChatOllama (Llama2) 或直接调用 llama-cpp-python self.llm ChatOpenAI(model_namellm_model_name, temperature0.1) # 构建系统提示词 system_template “““你是一个高级任务规划助手。你的目标是将用户的自然语言指令分解为一系列可执行的步骤。 你可以使用的工具执行器如下 {executor_descriptions} 请根据用户指令生成一个JSON格式的任务计划。计划是一个列表每个元素是一个步骤对象。 每个步骤对象必须包含以下字段 - step_id: 步骤序号从1开始。 - action: 要执行的动作名称必须来自上述工具列表。 - parameters: 一个字典包含传递给该动作的参数。参数名必须与工具描述中的输入字段匹配。 - depends_on: (可选)一个列表包含此步骤所依赖的step_id。默认为空。 如果指令不明确或无法用现有工具完成请返回一个包含单个步骤的列表其action为“request_clarification”并在parameters中提供“message”字段说明需要什么信息。 只输出JSON不要有其他任何解释。 示例指令“告诉我现在几点然后列出桌面文件。” 示例输出 [ {{“step_id”: 1, “action”: “get_current_time”, “parameters”: {{}}}}, {{“step_id”: 2, “action”: “list_files”, “parameters”: {{“directory_path”: “~/Desktop”}}, “depends_on”: [1]}} ] ”“” system_prompt SystemMessagePromptTemplate.from_template(system_template) human_prompt HumanMessagePromptTemplate.from_template(“用户指令{user_input}”) self.prompt ChatPromptTemplate.from_messages([system_prompt, human_prompt]) self.chain LLMChain(llmself.llm, promptself.prompt) def plan(self, user_input: str) - list: “”“根据用户输入生成任务计划。”“” # 获取最新的工具描述 executor_descriptions registry.get_executor_descriptions() # 调用LLM生成计划 result self.chain.run(executor_descriptionsexecutor_descriptions, user_inputuser_input) # 解析JSON输出。LLM的输出可能包含markdown代码块需要处理。 try: # 尝试提取JSON部分 if “json” in result: result result.split(“json”)[1].split(“”)[0].strip() elif “” in result: result result.split(“”)[1].split(“”)[0].strip() plan json.loads(result) if not isinstance(plan, list): plan [plan] return plan except json.JSONDecodeError as e: print(f“LLM返回了非JSON内容{result}”) # 返回一个请求澄清的默认计划 return [{“step_id”: 1, “action”: “request_clarification”, “parameters”: {“message”: “我无法解析您的指令请重新表述。”}}]现在我们将所有部分整合到Flask应用中。在app.py中# app.py from flask import Flask, request, jsonify from agents.planner import LLMPlanner from agents.registry import registry from core.state_manager import StateManager # 假设我们有一个状态管理器 import asyncio import uuid app Flask(__name__) planner LLMPlanner() state_manager StateManager() # 可以用内存字典或Redis实现 app.route(‘/api/process_command’, methods[‘POST’]) def process_command(): “”“接收文本指令生成并开始执行计划。”“” data request.json user_input data.get(‘text’) if not user_input: return jsonify({“error”: “Missing ‘text’ in request body”}), 400 # 1. 生成任务ID和初始状态 task_id str(uuid.uuid4()) state_manager.create_task(task_id, user_input) # 2. LLM规划 try: plan planner.plan(user_input) state_manager.update_plan(task_id, plan) except Exception as e: state_manager.set_task_status(task_id, “FAILED”, str(e)) return jsonify({“task_id”: task_id, “status”: “FAILED”, “error”: str(e)}), 500 # 3. 异步执行计划在实际生产中应使用Celery、RQ等任务队列 asyncio.create_task(execute_plan(task_id, plan)) return jsonify({“task_id”: task_id, “status”: “PLANNED”, “plan”: plan}) async def execute_plan(task_id: str, plan: list): “”“异步执行一个任务计划。”“” state_manager.set_task_status(task_id, “EXECUTING”) results {} for step in sorted(plan, keylambda x: x[‘step_id’]): # 检查依赖是否满足 depends_on step.get(‘depends_on’, []) if any(dep not in results or results[dep].get(‘status’) ! ‘SUCCESS’ for dep in depends_on): state_manager.set_step_status(task_id, step[‘step_id’], “BLOCKED”) continue state_manager.set_step_status(task_id, step[‘step_id’], “RUNNING”) try: # 调用执行器 result await registry.execute(step[‘action’], step.get(‘parameters’, {})) results[step[‘step_id’]] {“status”: “SUCCESS”, “result”: result} state_manager.set_step_status(task_id, step[‘step_id’], “SUCCESS”, result) except Exception as e: results[step[‘step_id’]] {“status”: “FAILED”, “error”: str(e)} state_manager.set_step_status(task_id, step[‘step_id’], “FAILED”, errorstr(e)) # 这里可以触发恢复逻辑 # await handle_recovery(task_id, step, e) # 简单起见我们先标记整个任务失败 state_manager.set_task_status(task_id, “FAILED”, f“Step {step[‘step_id’]} failed: {e}”) break if state_manager.get_task_status(task_id) ! “FAILED”: state_manager.set_task_status(task_id, “SUCCEEDED”) # 可以在这里汇总最终结果或通过TTS播报 app.route(‘/api/task_status/task_id’, methods[‘GET’]) def get_task_status(task_id): “”“查询任务执行状态。”“” status state_manager.get_task_info(task_id) if not status: return jsonify({“error”: “Task not found”}), 404 return jsonify(status) if __name__ ‘__main__’: # 在实际应用中需要导入所有执行器模块以确保它们被注册 import agents.executors.system import agents.executors.calendar # 假设有 import agents.executors.email # 假设有 app.run(debugTrue, port5000)至此一个最核心的、具备规划与多步执行能力的后端服务就搭建完成了。用户可以通过向/api/process_command发送JSON请求{“text”: “帮我列出桌面文件并告诉我时间”}来触发任务并通过/api/task_status/task_id查询进度。4. 实现语音交互与自适应恢复机制一个完整的语音助手还需要“耳朵”和“嘴巴”以及应对错误的智慧。4.1 集成语音识别STT与合成TTS我们创建两个服务模块。对于STT可以使用离线的Vosk或Whisper或者在线的服务。这里以speech_recognition库为例它支持多种后端。# services/stt.py import speech_recognition as sr import io class SpeechToTextService: def __init__(self, use_onlineTrue): self.recognizer sr.Recognizer() self.use_online use_online # 在线使用Google Web API需网络离线需配置Vosk等 def listen_and_transcribe(self, audio_source, language“zh-CN”): “”“从音频源麦克风或文件识别语音。”“” try: if isinstance(audio_source, io.BytesIO): with sr.AudioFile(audio_source) as source: audio self.recognizer.record(source) else: # 假设是麦克风 with sr.Microphone() as source: print(“Listening...”) audio self.recognizer.listen(source, timeout5, phrase_time_limit10) if self.use_online: text self.recognizer.recognize_google(audio, languagelanguage) else: # 离线识别例如使用Vosk # text self.recognizer.recognize_vosk(audio) raise NotImplementedError(“Offline STT not configured.”) return text except sr.UnknownValueError: return “”或返回请求重说的指令 except sr.RequestError as e: return f“STT服务错误{e}” except Exception as e: return f“未知错误{e}”对于TTS可以使用离线的pyttsx3或edge-tts。# services/tts.py import pyttsx3 import threading class TextToSpeechService: def __init__(self): self.engine pyttsx3.init() # 配置语音参数 self.engine.setProperty(‘rate’, 150) # 语速 self.engine.setProperty(‘volume’, 0.9) # 音量 def speak(self, text: str, blockFalse): “”“朗读文本。如果block为True则阻塞直到朗读完成。”“” def _speak(): self.engine.say(text) self.engine.runAndWait() t threading.Thread(target_speak) t.start() if block: t.join()然后在Flask中添加一个语音端点# app.py (新增端点) from services.stt import SpeechToTextService from services.tts import TextToSpeechService stt_service SpeechToTextService(use_onlineFalse) # 根据情况选择 tts_service TextToSpeechService() app.route(‘/api/voice_command’, methods[‘POST’]) def voice_command(): “”“接收音频数据识别并处理指令。”“” if ‘audio’ not in request.files: return jsonify({“error”: “No audio file provided”}), 400 audio_file request.files[‘audio’] # 将文件转换为字节流供STT服务使用 audio_bytes io.BytesIO(audio_file.read()) # 语音识别 user_input stt_service.listen_and_transcribe(audio_bytes) if not user_input or user_input.startswith(“STT服务错误”): return jsonify({“error”: “Speech recognition failed”, “detail”: user_input}), 500 # 后续逻辑与 /api/process_command 相同... # 生成任务ID规划异步执行... # 可以立即返回一个“正在处理”的响应并通过WebSocket或轮询通知客户端结果。 # 这里简单起见直接处理并返回最终文本结果。 try: plan planner.plan(user_input) # 同步执行仅用于演示复杂任务应异步 final_result “任务完成。” # 实际应执行plan并汇总结果 # 语音合成回复 # tts_service.speak(final_result) return jsonify({“text”: user_input, “reply”: final_result}) except Exception as e: return jsonify({“error”: str(e)}), 5004.2 设计自适应恢复策略恢复机制是系统的“智能”体现。我们需要在core/recovery.py中实现一个恢复处理器。它的逻辑比核心执行循环更复杂这里给出一个框架# core/recovery.py from agents.planner import LLMPlanner from agents.registry import registry from core.state_manager import state_manager class RecoveryHandler: def __init__(self, planner: LLMPlanner): self.planner planner async def handle_failure(self, task_id: str, failed_step: dict, error: str): “”“处理步骤执行失败。 策略 1. 分析错误类型网络、权限、资源不存在、参数错误等。 2. 根据错误类型和当前任务上下文决定恢复策略。 3. 执行恢复策略重试、调整参数、重新规划、人工求助。 ”“” step_id failed_step[‘step_id’] action failed_step[‘action’] original_parameters failed_step.get(‘parameters’, {}) # 策略1: 瞬时错误重试例如网络超时 if “timeout” in error.lower() or “connection” in error.lower(): max_retries 3 for i in range(max_retries): try: result await registry.execute(action, original_parameters) state_manager.set_step_status(task_id, step_id, “SUCCESS”, result) return True # 恢复成功 except Exception as retry_error: error str(retry_error) # 重试多次后仍失败进入下一策略 # 策略2: 将错误反馈给LLM请求新的计划或调整参数 # 获取任务历史上下文 task_info state_manager.get_task_info(task_id) original_user_input task_info[‘user_input’] execution_history task_info[‘steps’] # 假设state_manager记录了步骤历史 recovery_prompt f“““ 原始用户指令{original_user_input} 我们在执行步骤 {step_id} (动作{action}, 参数{original_parameters}) 时失败了。 错误信息{error} 已成功完成的步骤{[s for s in execution_history if s[‘status’]‘SUCCESS’]} 请分析错误并提供以下之一 A) 一组新的参数来重试这个动作。 B) 一个替代的动作必须是可用工具之一来达到相似目的。 C) 一个澄清问题以向用户询问更多信息。 请以JSON格式回复例如 {{“strategy”: “retry”, “new_parameters”: {{…}}}} 或 {{“strategy”: “alternative”, “new_step”: {{“action”: “…”, “parameters”: {{…}}}} }} 或 {{“strategy”: “clarify”, “question”: “…”}} ”“” # 调用一个专用的“恢复规划”LLM链或者复用主规划器但用不同的提示词 recovery_decision await self._ask_llm_for_recovery(recovery_prompt) if recovery_decision[‘strategy’] ‘retry’: new_params recovery_decision.get(‘new_parameters’, original_parameters) try: result await registry.execute(action, new_params) state_manager.set_step_status(task_id, step_id, “SUCCESS”, result) return True except Exception as e: # 如果调整参数后仍失败可能进入策略3或标记为最终失败 pass elif recovery_decision[‘strategy’] ‘alternative’: new_step recovery_decision[‘new_step’] # 插入一个新的替代步骤并更新后续步骤的依赖关系 # 这是一个复杂的图操作需要修改state_manager中的计划 # state_manager.insert_step(task_id, step_id, new_step) # 然后重新调度执行 # return await self.reschedule(task_id) pass elif recovery_decision[‘strategy’] ‘clarify’: question recovery_decision[‘question’] # 通过TTS向用户提问并等待下一轮语音输入 # 将任务状态置为 AWAITING_CLARIFICATION并存储问题 state_manager.set_task_status(task_id, “AWAITING_CLARIFICATION”, {“question”: question}) # 前端/客户端应捕获此状态并提示用户 return False # 恢复未完成等待用户输入 # 如果所有自动恢复策略都失败标记为需要人工干预 state_manager.set_task_status(task_id, “NEEDS_HUMAN”, {“failed_step”: step_id, “error”: error}) return False async def _ask_llm_for_recovery(self, prompt: str) - dict: # 简化实现实际可能需要更复杂的提示工程和解析 # 这里可以调用另一个LLM实例或者使用同一个但不同的提示模板 # 返回解析后的JSON决策 pass这个恢复处理器需要与主执行循环紧密集成。当步骤失败时主循环不应直接跳出而应调用recovery_handler.handle_failure并根据返回结果决定是继续执行、等待还是彻底失败。5. 进阶优化、安全考量与扩展方向一个基础系统跑起来后我们还需要考虑更多生产级的问题。5.1 性能与稳定性优化异步与并发Flask默认是同步的处理长任务会阻塞。对于生产环境务必使用异步框架如Quart、FastAPI或将耗时任务LLM调用、网络IO交给任务队列如Celery、RQ、Dramatiq。主API只负责接收请求和返回任务ID执行完全异步化。状态持久化内存中的state_manager在服务重启后会丢失所有状态。必须使用外部存储如Redis适合快速读写和过期设置或SQL数据库如PostgreSQL。每个任务和步骤的状态变更都应及时持久化。LLM调用优化缓存对相似的指令进行规划结果可能相同。可以缓存(指令, 工具集)到规划结果的映射减少LLM调用次数和成本。流式输出对于需要长时间思考的复杂规划可以使用LLM的流式响应让用户感知进度。备用模型可以配置多个LLM供应商OpenAI, Anthropic, 本地Llama作为备选在主供应商故障或限速时自动切换。执行器超时与隔离每个执行器都应设置超时时间防止某个执行器挂起导致整个任务卡死。可以考虑使用子进程或线程池来运行执行器实现资源隔离。5.2 安全与隐私强化输入验证与沙箱这是本地助手的生命线。LLM生成的action和parameters必须经过白名单验证即注册表。对于执行文件操作、系统命令等高危动作的执行器其参数必须被严格过滤如禁止../路径穿越。考虑在沙箱环境如Docker容器中运行不可信的执行器代码。权限控制可以为不同的执行器定义权限等级如“基础信息查询”、“文件读取”、“系统控制”。在用户首次使用或执行高危操作前明确请求授权。数据本地化确保STT/TTS也使用本地模型。所有与外部服务的通信如果必须有应使用加密连接并且API密钥等敏感信息妥善存储。提示词注入防护用户输入在拼接进LLM提示词前需要进行适当的清洗和转义防止用户通过精心构造的输入“劫持”系统提示词让LLM执行非预期操作。5.3 功能扩展与生态构建技能市场将执行器设计为可插拔的“技能包”。用户可以自行编写或从社区下载skill.py文件放到指定目录系统启动时自动加载注册。这能让你的助手能力无限扩展。上下文记忆让助手记住之前的对话和操作。可以为每个用户会话维护一个向量数据库如ChromaDB,FAISS存储历史交互。当新指令到来时先检索相关历史将其作为上下文提供给LLM规划器实现连贯的多轮对话。图形化界面为助手开发一个简单的本地Web UI显示任务执行状态、历史记录并提供手动触发和干预的按钮。硬件集成结合Home Assistant、IFTTT等平台你的语音助手可以成为智能家居的控制中心。通过添加对应的执行器实现“打开客厅灯”、“调整空调温度”等操作。构建像AnovaX这样的系统最大的挑战往往不在于单个技术点而在于如何将这些组件语音、LLM、多智能体、状态管理优雅、可靠地整合在一起并处理好边界情况。它更像是一个微型的操作系统调度器。从这个小项目出发你可以深入探索Agent系统的每一个方面例如更复杂的任务规划算法如基于图的规划、更强大的恢复策略如强化学习训练恢复策略、以及如何评估整个系统的可靠性和用户体验。这个过程中积累的经验对于理解下一代人机交互和自动化系统的构建将是无价的。 构建本地化多智能体语音助手：基于LLM规划与类型化执行器的实践
1. 项目概述为什么我们需要一个“本地化、多智能体”的语音助手如果你和我一样对市面上的语音助手总感觉“差那么点意思”那这个项目可能就是你一直在找的答案。我们习惯了那些需要联网、反应迟缓、功能固定且无法深度定制的语音助手。它们要么听不懂复杂的指令要么听懂了也做不了什么更别提在断网环境下完全瘫痪了。今天要聊的AnovaX就是一次彻底打破这些限制的尝试。它不是一个简单的“语音转文本再查天气”的工具而是一个运行在你本地电脑上的、由多个“专家”智能体Agent协同工作的、具备高级规划和自我纠错能力的语音指挥中心。简单来说AnovaX 的核心价值在于三个关键词本地Local、多智能体Multi-Agent、规划与执行Planning Execution。本地化意味着你的所有语音数据、对话历史和任务执行过程都留在你自己的机器上隐私和安全得到最大程度的保障。多智能体架构则像组建了一个“专家顾问团”——当你下达一个模糊指令如“帮我安排下周的会议并写个总结”系统会先让一个“规划师”智能体基于LLM拆解任务第一步访问日历第二步起草邮件第三步生成会议纪要。然后它会调用对应的“执行器”智能体Typed Executor去完成每一步的具体操作。而“自适应恢复”能力则让这个系统具备了“韧性”——当某个步骤出错比如日历API暂时不可用它不会直接崩溃而是尝试分析错误原因调整执行路径或者向你请求更多信息。从技术栈来看它巧妙地融合了当前最热门的技术方向用Python作为粘合剂Flask提供轻量级的Web服务接口JSON作为智能体间通信和数据交换的标准格式而这一切的大脑则是一个本地部署或通过API调用的大语言模型LLM。这不仅仅是又一个“玩具项目”它为我们展示了如何将前沿的AI能力LLM的规划与推理与传统的、可靠的软件工程实践类型化接口、错误处理相结合构建出真正实用、可控的自动化工具。接下来我将带你深入AnovaX的内部看看它是如何工作的以及我们如何能亲手搭建并定制一个属于自己的智能语音助手。2. 架构核心拆解LLM规划器、类型化执行器与状态机要理解AnovaX我们必须先抛开“语音助手”这个表象把它看作一个任务驱动的自动化系统。它的核心工作流可以抽象为感知语音输入→ 理解与规划LLM→ 分发与执行多智能体→ 反馈与调整自适应恢复。下面我们来逐一拆解这三个核心组件。2.1 LLM作为“总指挥”从模糊指令到可执行计划LLM在这里扮演的角色是“首席规划官”。它的输入是你的语音转文字后的自然语言指令输出是一个结构化的、可执行的任务计划Plan。这个计划通常是一个JSON数组其中每个元素代表一个子任务。为什么是LLM传统的规则引擎或意图识别模型在处理“帮我查一下明天天气如果下雨就提醒我带伞并取消下午的户外跑步计划”这类嵌套、多步骤、带条件的指令时会变得极其复杂和脆弱。LLM凭借其强大的语义理解和上下文推理能力可以自然地理解这种复杂指令并将其分解为逻辑步骤。一个典型的工作流程如下语音识别STT将用户的音频流实时转换为文本。这一步可以使用本地模型如Vosk、Whisper.cpp或云服务API。指令理解与规划转换后的文本被送入LLM。我们通过精心设计的系统提示词System Prompt来引导LLM的行为。这个提示词会定义系统的角色“你是一个任务规划助手”、可用的工具集“你可以使用以下功能查询天气、管理日历、发送邮件…”以及输出格式规范“请以如下JSON格式输出你的计划…”。计划生成LLM根据指令和可用工具生成一个计划。例如对于指令“安排一个明天下午3点与团队的会议主题是项目评审”LLM可能输出[ { “step_id”: 1, “action”: “check_calendar_availability”, “parameters”: { “time”: “tomorrow 15:00”, “duration”: “1h”, “attendees”: [“team”] } }, { “step_id”: 2, “action”: “create_calendar_event”, “parameters”: { “title”: “项目评审会议”, “start_time”: “2023-10-27T15:00:00”, “end_time”: “2023-10-27T16:00:00”, “attendees”: [“aliceexample.com”, “bobexample.com”] }, “depends_on”: [1] } ]这个JSON计划清晰地定义了动作序列、参数和依赖关系第二步依赖于第一步的成功。提示设计一个稳定可靠的系统提示词是成功的关键。它需要明确边界LLM不能做什么、格式化输出要求并包含足够的示例Few-shot Learning。一个常见的技巧是让LLM在无法理解或缺少信息时输出一个特定的“请求澄清request_clarification”动作而不是胡乱猜测。2.2 类型化执行器让每个“专家”可靠地干活计划中的每一个action如check_calendar_availability都需要一个对应的执行器Executor来具体实现。AnovaX强调“类型化Typed”这是其工程健壮性的基石。什么是类型化执行器它意味着每个执行器都有严格定义的输入和输出接口Schema。这通常通过Python的Pydantic库来实现。例如创建日历事件执行器的输入模型可能定义为from pydantic import BaseModel from datetime import datetime from typing import List class CreateEventInput(BaseModel): title: str start_time: datetime end_time: datetime attendees: List[str] description: str “”而它的输出模型可能包含一个event_id字段。这样做的好处是自动验证在代码执行前系统就能确保传入的参数类型和结构是正确的避免了运行时因数据格式错误导致的崩溃。清晰契约每个执行器做什么、需要什么、返回什么一目了然便于开发和维护。安全过滤可以防止LLM生成的计划中包含恶意或意外的参数。执行器如何工作每个执行器被注册到一个中央执行器注册表中。当调度器需要执行action: “create_calendar_event”时它会从注册表中找到对应的执行器函数将parameters字典反序列化成定义好的Pydantic模型然后调用该函数。函数执行完毕后再将结果序列化传递给下一个步骤或作为最终结果返回。注意执行器的实现应尽量保持“无状态”和“幂等”。即多次执行相同参数应产生相同的结果且执行器内部不依赖全局可变状态。这有助于错误恢复和重试机制的实现。2.3 自适应恢复状态机系统的“免疫系统”这是AnovaX区别于普通脚本最酷的部分。多步骤任务难免会出错网络波动、API限流、资源不存在、权限不足……一个脆弱的系统会直接报错退出。而AnovaX的“自适应恢复”机制试图让系统像人一样遇到挫折后尝试其他办法。其核心是一个状态机State Machine它管理着每个任务的生命周期PENDING-PLANNING-EXECUTING-SUCCEEDED/FAILED/NEEDS_RECOVERY。当某个步骤执行失败时系统不会立即将整个任务标记为FAILED而是进入NEEDS_RECOVERY状态。此时恢复处理器Recovery Handler被触发。恢复策略可以有多层简单重试对于瞬时错误如网络超时自动重试几次。参数调整将错误信息如“未找到文件A”反馈给LLM规划器请求其重新规划或调整参数“请使用文件B代替”生成一个新的子计划。备选路径如果某个执行器彻底失败如服务不可用LLM可能会被询问是否有替代方案“无法通过Google日历创建事件是否改为发送一封会议邀请邮件”。人工介入当自动恢复尝试多次后仍失败系统可以生成一个清晰的摘要并通过TTS语音或界面提示向用户请求帮助。这个循环执行→失败→分析→重规划/调整→再执行构成了“自适应”的核心。它要求系统不仅能执行计划还能理解执行结果包括错误并具备一定的元认知能力来调整自己的行为。3. 从零开始搭建技术栈选型与核心模块实现理解了架构我们就可以动手搭建了。这里我会基于常见的开源工具和库给出一个可运行的实现方案。我们的技术栈如下Python主语言、FlaskWeb服务框架、LangChain/LlamaIndexLLM应用框架可选但推荐、Pydantic数据验证、SQLite/Redis状态存储。3.1 环境准备与项目结构首先创建一个清晰的项目目录结构这是保持代码可维护性的第一步。anova_x/ ├── app.py # Flask应用主入口 ├── config.py # 配置文件 ├── requirements.txt # 项目依赖 ├── agents/ # 智能体相关模块 │ ├── __init__.py │ ├── planner.py # LLM规划器 │ ├── registry.py # 执行器注册表 │ └── executors/ # 具体执行器 │ ├── __init__.py │ ├── calendar.py │ ├── email.py │ └── system.py ├── core/ # 核心逻辑 │ ├── __init__.py │ ├── state_manager.py # 状态管理 │ └── recovery.py # 恢复逻辑 ├── services/ # 外部服务封装 │ ├── stt.py # 语音识别服务 │ └── tts.py # 语音合成服务 └── utils/ # 工具函数 └── validation.py安装核心依赖requirements.txtflask2.3.0 pydantic2.0.0 openai1.0.0 # 或其他LLM SDK如 llama-cpp-python langchain0.1.0 # 用于简化LLM调用和工具管理 speechrecognition3.10.0 # 语音识别库调用本地或在线服务 pyttsx32.90 # 本地文本转语音 pymysql # 如果需要连接MySQL redis4.5.0 # 用于分布式状态存储可选3.2 实现执行器注册与类型化调用这是系统的基石。我们先在agents/registry.py中创建一个全局的执行器注册表。# agents/registry.py from typing import Dict, Any, Callable, Optional from pydantic import BaseModel, ValidationError import inspect class ExecutorRegistry: def __init__(self): self._executors: Dict[str, Dict[str, Any]] {} def register(self, name: str, input_model: Optional[BaseModel] None, output_model: Optional[BaseModel] None): 装饰器用于注册一个执行器函数。 def decorator(func: Callable): self._executors[name] { “func”: func, “input_model”: input_model, “output_model”: output_model, “description”: func.__doc__ or “” # 函数的文档字符串可作为描述供LLM参考 } return func return decorator async def execute(self, action: str, parameters: Dict[str, Any]) - Any: 查找并执行指定的动作。 if action not in self._executors: raise ValueError(f“Executor ‘{action}’ not found.”) executor_info self._executors[action] func executor_info[“func”] input_model executor_info[“input_model”] # 1. 参数验证与转换 validated_params parameters if input_model: try: validated_params input_model(**parameters).dict() except ValidationError as e: raise ValueError(f“Invalid parameters for ‘{action}’: {e}”) # 2. 执行函数 # 检查函数是否是异步的 if inspect.iscoroutinefunction(func): result await func(**validated_params) else: result func(**validated_params) # 3. 输出验证可选 output_model executor_info[“output_model”] if output_model and result is not None: try: result output_model(**result) if isinstance(result, dict) else output_model(result) except ValidationError: # 如果输出不符合模型可以记录日志但不一定抛出异常 pass return result def get_executor_descriptions(self) - str: 生成所有执行器的描述用于构造LLM的系统提示词。 descriptions [] for name, info in self._executors.items(): desc f“- {name}: {info[‘description’]}” if info[‘input_model’]: # 简单展示输入字段更复杂的可以生成JSON Schema fields list(info[‘input_model’].__fields__.keys()) desc f“ 输入字段: {fields}” descriptions.append(desc) return “\n”.join(descriptions) # 全局注册表实例 registry ExecutorRegistry()然后我们实现几个具体的执行器。例如在agents/executors/system.py中# agents/executors/system.py from agents.registry import registry from pydantic import BaseModel from datetime import datetime import os class GetTimeInput(BaseModel): format: str “%Y-%m-%d %H:%M:%S” class GetTimeOutput(BaseModel): current_time: str registry.register(“get_current_time”, input_modelGetTimeInput, output_modelGetTimeOutput) def get_current_time(format: str “%Y-%m-%d %H:%M:%S”) - dict: “”“获取当前系统时间。”“” now datetime.now() return {“current_time”: now.strftime(format)} class ListFilesInput(BaseModel): directory_path: str registry.register(“list_files”, input_modelListFilesInput) def list_files(directory_path: str) - list: “”“列出指定目录下的文件。如果目录不存在返回空列表。”“” try: files os.listdir(directory_path) return files except FileNotFoundError: return []通过这样的设计我们实现了严格的输入输出契约。LLM规划器在生成计划时可以参考registry.get_executor_descriptions()返回的工具描述确保生成的parameters字段名与Pydantic模型匹配。3.3 构建LLM规划器与Flask API接口接下来我们实现规划器。这里使用LangChain来简化与LLM的交互。我们在agents/planner.py中实现# agents/planner.py import json from langchain.chains import LLMChain from langchain.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate from langchain.chat_models import ChatOpenAI # 示例使用OpenAI可替换为其他模型 from agents.registry import registry class LLMPlanner: def __init__(self, llm_model_name“gpt-3.5-turbo”): # 初始化LLM。对于本地模型可以使用 ChatOllama (Llama2) 或直接调用 llama-cpp-python self.llm ChatOpenAI(model_namellm_model_name, temperature0.1) # 构建系统提示词 system_template “““你是一个高级任务规划助手。你的目标是将用户的自然语言指令分解为一系列可执行的步骤。 你可以使用的工具执行器如下 {executor_descriptions} 请根据用户指令生成一个JSON格式的任务计划。计划是一个列表每个元素是一个步骤对象。 每个步骤对象必须包含以下字段 - step_id: 步骤序号从1开始。 - action: 要执行的动作名称必须来自上述工具列表。 - parameters: 一个字典包含传递给该动作的参数。参数名必须与工具描述中的输入字段匹配。 - depends_on: (可选)一个列表包含此步骤所依赖的step_id。默认为空。 如果指令不明确或无法用现有工具完成请返回一个包含单个步骤的列表其action为“request_clarification”并在parameters中提供“message”字段说明需要什么信息。 只输出JSON不要有其他任何解释。 示例指令“告诉我现在几点然后列出桌面文件。” 示例输出 [ {{“step_id”: 1, “action”: “get_current_time”, “parameters”: {{}}}}, {{“step_id”: 2, “action”: “list_files”, “parameters”: {{“directory_path”: “~/Desktop”}}, “depends_on”: [1]}} ] ”“” system_prompt SystemMessagePromptTemplate.from_template(system_template) human_prompt HumanMessagePromptTemplate.from_template(“用户指令{user_input}”) self.prompt ChatPromptTemplate.from_messages([system_prompt, human_prompt]) self.chain LLMChain(llmself.llm, promptself.prompt) def plan(self, user_input: str) - list: “”“根据用户输入生成任务计划。”“” # 获取最新的工具描述 executor_descriptions registry.get_executor_descriptions() # 调用LLM生成计划 result self.chain.run(executor_descriptionsexecutor_descriptions, user_inputuser_input) # 解析JSON输出。LLM的输出可能包含markdown代码块需要处理。 try: # 尝试提取JSON部分 if “json” in result: result result.split(“json”)[1].split(“”)[0].strip() elif “” in result: result result.split(“”)[1].split(“”)[0].strip() plan json.loads(result) if not isinstance(plan, list): plan [plan] return plan except json.JSONDecodeError as e: print(f“LLM返回了非JSON内容{result}”) # 返回一个请求澄清的默认计划 return [{“step_id”: 1, “action”: “request_clarification”, “parameters”: {“message”: “我无法解析您的指令请重新表述。”}}]现在我们将所有部分整合到Flask应用中。在app.py中# app.py from flask import Flask, request, jsonify from agents.planner import LLMPlanner from agents.registry import registry from core.state_manager import StateManager # 假设我们有一个状态管理器 import asyncio import uuid app Flask(__name__) planner LLMPlanner() state_manager StateManager() # 可以用内存字典或Redis实现 app.route(‘/api/process_command’, methods[‘POST’]) def process_command(): “”“接收文本指令生成并开始执行计划。”“” data request.json user_input data.get(‘text’) if not user_input: return jsonify({“error”: “Missing ‘text’ in request body”}), 400 # 1. 生成任务ID和初始状态 task_id str(uuid.uuid4()) state_manager.create_task(task_id, user_input) # 2. LLM规划 try: plan planner.plan(user_input) state_manager.update_plan(task_id, plan) except Exception as e: state_manager.set_task_status(task_id, “FAILED”, str(e)) return jsonify({“task_id”: task_id, “status”: “FAILED”, “error”: str(e)}), 500 # 3. 异步执行计划在实际生产中应使用Celery、RQ等任务队列 asyncio.create_task(execute_plan(task_id, plan)) return jsonify({“task_id”: task_id, “status”: “PLANNED”, “plan”: plan}) async def execute_plan(task_id: str, plan: list): “”“异步执行一个任务计划。”“” state_manager.set_task_status(task_id, “EXECUTING”) results {} for step in sorted(plan, keylambda x: x[‘step_id’]): # 检查依赖是否满足 depends_on step.get(‘depends_on’, []) if any(dep not in results or results[dep].get(‘status’) ! ‘SUCCESS’ for dep in depends_on): state_manager.set_step_status(task_id, step[‘step_id’], “BLOCKED”) continue state_manager.set_step_status(task_id, step[‘step_id’], “RUNNING”) try: # 调用执行器 result await registry.execute(step[‘action’], step.get(‘parameters’, {})) results[step[‘step_id’]] {“status”: “SUCCESS”, “result”: result} state_manager.set_step_status(task_id, step[‘step_id’], “SUCCESS”, result) except Exception as e: results[step[‘step_id’]] {“status”: “FAILED”, “error”: str(e)} state_manager.set_step_status(task_id, step[‘step_id’], “FAILED”, errorstr(e)) # 这里可以触发恢复逻辑 # await handle_recovery(task_id, step, e) # 简单起见我们先标记整个任务失败 state_manager.set_task_status(task_id, “FAILED”, f“Step {step[‘step_id’]} failed: {e}”) break if state_manager.get_task_status(task_id) ! “FAILED”: state_manager.set_task_status(task_id, “SUCCEEDED”) # 可以在这里汇总最终结果或通过TTS播报 app.route(‘/api/task_status/task_id’, methods[‘GET’]) def get_task_status(task_id): “”“查询任务执行状态。”“” status state_manager.get_task_info(task_id) if not status: return jsonify({“error”: “Task not found”}), 404 return jsonify(status) if __name__ ‘__main__’: # 在实际应用中需要导入所有执行器模块以确保它们被注册 import agents.executors.system import agents.executors.calendar # 假设有 import agents.executors.email # 假设有 app.run(debugTrue, port5000)至此一个最核心的、具备规划与多步执行能力的后端服务就搭建完成了。用户可以通过向/api/process_command发送JSON请求{“text”: “帮我列出桌面文件并告诉我时间”}来触发任务并通过/api/task_status/task_id查询进度。4. 实现语音交互与自适应恢复机制一个完整的语音助手还需要“耳朵”和“嘴巴”以及应对错误的智慧。4.1 集成语音识别STT与合成TTS我们创建两个服务模块。对于STT可以使用离线的Vosk或Whisper或者在线的服务。这里以speech_recognition库为例它支持多种后端。# services/stt.py import speech_recognition as sr import io class SpeechToTextService: def __init__(self, use_onlineTrue): self.recognizer sr.Recognizer() self.use_online use_online # 在线使用Google Web API需网络离线需配置Vosk等 def listen_and_transcribe(self, audio_source, language“zh-CN”): “”“从音频源麦克风或文件识别语音。”“” try: if isinstance(audio_source, io.BytesIO): with sr.AudioFile(audio_source) as source: audio self.recognizer.record(source) else: # 假设是麦克风 with sr.Microphone() as source: print(“Listening...”) audio self.recognizer.listen(source, timeout5, phrase_time_limit10) if self.use_online: text self.recognizer.recognize_google(audio, languagelanguage) else: # 离线识别例如使用Vosk # text self.recognizer.recognize_vosk(audio) raise NotImplementedError(“Offline STT not configured.”) return text except sr.UnknownValueError: return “”或返回请求重说的指令 except sr.RequestError as e: return f“STT服务错误{e}” except Exception as e: return f“未知错误{e}”对于TTS可以使用离线的pyttsx3或edge-tts。# services/tts.py import pyttsx3 import threading class TextToSpeechService: def __init__(self): self.engine pyttsx3.init() # 配置语音参数 self.engine.setProperty(‘rate’, 150) # 语速 self.engine.setProperty(‘volume’, 0.9) # 音量 def speak(self, text: str, blockFalse): “”“朗读文本。如果block为True则阻塞直到朗读完成。”“” def _speak(): self.engine.say(text) self.engine.runAndWait() t threading.Thread(target_speak) t.start() if block: t.join()然后在Flask中添加一个语音端点# app.py (新增端点) from services.stt import SpeechToTextService from services.tts import TextToSpeechService stt_service SpeechToTextService(use_onlineFalse) # 根据情况选择 tts_service TextToSpeechService() app.route(‘/api/voice_command’, methods[‘POST’]) def voice_command(): “”“接收音频数据识别并处理指令。”“” if ‘audio’ not in request.files: return jsonify({“error”: “No audio file provided”}), 400 audio_file request.files[‘audio’] # 将文件转换为字节流供STT服务使用 audio_bytes io.BytesIO(audio_file.read()) # 语音识别 user_input stt_service.listen_and_transcribe(audio_bytes) if not user_input or user_input.startswith(“STT服务错误”): return jsonify({“error”: “Speech recognition failed”, “detail”: user_input}), 500 # 后续逻辑与 /api/process_command 相同... # 生成任务ID规划异步执行... # 可以立即返回一个“正在处理”的响应并通过WebSocket或轮询通知客户端结果。 # 这里简单起见直接处理并返回最终文本结果。 try: plan planner.plan(user_input) # 同步执行仅用于演示复杂任务应异步 final_result “任务完成。” # 实际应执行plan并汇总结果 # 语音合成回复 # tts_service.speak(final_result) return jsonify({“text”: user_input, “reply”: final_result}) except Exception as e: return jsonify({“error”: str(e)}), 5004.2 设计自适应恢复策略恢复机制是系统的“智能”体现。我们需要在core/recovery.py中实现一个恢复处理器。它的逻辑比核心执行循环更复杂这里给出一个框架# core/recovery.py from agents.planner import LLMPlanner from agents.registry import registry from core.state_manager import state_manager class RecoveryHandler: def __init__(self, planner: LLMPlanner): self.planner planner async def handle_failure(self, task_id: str, failed_step: dict, error: str): “”“处理步骤执行失败。 策略 1. 分析错误类型网络、权限、资源不存在、参数错误等。 2. 根据错误类型和当前任务上下文决定恢复策略。 3. 执行恢复策略重试、调整参数、重新规划、人工求助。 ”“” step_id failed_step[‘step_id’] action failed_step[‘action’] original_parameters failed_step.get(‘parameters’, {}) # 策略1: 瞬时错误重试例如网络超时 if “timeout” in error.lower() or “connection” in error.lower(): max_retries 3 for i in range(max_retries): try: result await registry.execute(action, original_parameters) state_manager.set_step_status(task_id, step_id, “SUCCESS”, result) return True # 恢复成功 except Exception as retry_error: error str(retry_error) # 重试多次后仍失败进入下一策略 # 策略2: 将错误反馈给LLM请求新的计划或调整参数 # 获取任务历史上下文 task_info state_manager.get_task_info(task_id) original_user_input task_info[‘user_input’] execution_history task_info[‘steps’] # 假设state_manager记录了步骤历史 recovery_prompt f“““ 原始用户指令{original_user_input} 我们在执行步骤 {step_id} (动作{action}, 参数{original_parameters}) 时失败了。 错误信息{error} 已成功完成的步骤{[s for s in execution_history if s[‘status’]‘SUCCESS’]} 请分析错误并提供以下之一 A) 一组新的参数来重试这个动作。 B) 一个替代的动作必须是可用工具之一来达到相似目的。 C) 一个澄清问题以向用户询问更多信息。 请以JSON格式回复例如 {{“strategy”: “retry”, “new_parameters”: {{…}}}} 或 {{“strategy”: “alternative”, “new_step”: {{“action”: “…”, “parameters”: {{…}}}} }} 或 {{“strategy”: “clarify”, “question”: “…”}} ”“” # 调用一个专用的“恢复规划”LLM链或者复用主规划器但用不同的提示词 recovery_decision await self._ask_llm_for_recovery(recovery_prompt) if recovery_decision[‘strategy’] ‘retry’: new_params recovery_decision.get(‘new_parameters’, original_parameters) try: result await registry.execute(action, new_params) state_manager.set_step_status(task_id, step_id, “SUCCESS”, result) return True except Exception as e: # 如果调整参数后仍失败可能进入策略3或标记为最终失败 pass elif recovery_decision[‘strategy’] ‘alternative’: new_step recovery_decision[‘new_step’] # 插入一个新的替代步骤并更新后续步骤的依赖关系 # 这是一个复杂的图操作需要修改state_manager中的计划 # state_manager.insert_step(task_id, step_id, new_step) # 然后重新调度执行 # return await self.reschedule(task_id) pass elif recovery_decision[‘strategy’] ‘clarify’: question recovery_decision[‘question’] # 通过TTS向用户提问并等待下一轮语音输入 # 将任务状态置为 AWAITING_CLARIFICATION并存储问题 state_manager.set_task_status(task_id, “AWAITING_CLARIFICATION”, {“question”: question}) # 前端/客户端应捕获此状态并提示用户 return False # 恢复未完成等待用户输入 # 如果所有自动恢复策略都失败标记为需要人工干预 state_manager.set_task_status(task_id, “NEEDS_HUMAN”, {“failed_step”: step_id, “error”: error}) return False async def _ask_llm_for_recovery(self, prompt: str) - dict: # 简化实现实际可能需要更复杂的提示工程和解析 # 这里可以调用另一个LLM实例或者使用同一个但不同的提示模板 # 返回解析后的JSON决策 pass这个恢复处理器需要与主执行循环紧密集成。当步骤失败时主循环不应直接跳出而应调用recovery_handler.handle_failure并根据返回结果决定是继续执行、等待还是彻底失败。5. 进阶优化、安全考量与扩展方向一个基础系统跑起来后我们还需要考虑更多生产级的问题。5.1 性能与稳定性优化异步与并发Flask默认是同步的处理长任务会阻塞。对于生产环境务必使用异步框架如Quart、FastAPI或将耗时任务LLM调用、网络IO交给任务队列如Celery、RQ、Dramatiq。主API只负责接收请求和返回任务ID执行完全异步化。状态持久化内存中的state_manager在服务重启后会丢失所有状态。必须使用外部存储如Redis适合快速读写和过期设置或SQL数据库如PostgreSQL。每个任务和步骤的状态变更都应及时持久化。LLM调用优化缓存对相似的指令进行规划结果可能相同。可以缓存(指令, 工具集)到规划结果的映射减少LLM调用次数和成本。流式输出对于需要长时间思考的复杂规划可以使用LLM的流式响应让用户感知进度。备用模型可以配置多个LLM供应商OpenAI, Anthropic, 本地Llama作为备选在主供应商故障或限速时自动切换。执行器超时与隔离每个执行器都应设置超时时间防止某个执行器挂起导致整个任务卡死。可以考虑使用子进程或线程池来运行执行器实现资源隔离。5.2 安全与隐私强化输入验证与沙箱这是本地助手的生命线。LLM生成的action和parameters必须经过白名单验证即注册表。对于执行文件操作、系统命令等高危动作的执行器其参数必须被严格过滤如禁止../路径穿越。考虑在沙箱环境如Docker容器中运行不可信的执行器代码。权限控制可以为不同的执行器定义权限等级如“基础信息查询”、“文件读取”、“系统控制”。在用户首次使用或执行高危操作前明确请求授权。数据本地化确保STT/TTS也使用本地模型。所有与外部服务的通信如果必须有应使用加密连接并且API密钥等敏感信息妥善存储。提示词注入防护用户输入在拼接进LLM提示词前需要进行适当的清洗和转义防止用户通过精心构造的输入“劫持”系统提示词让LLM执行非预期操作。5.3 功能扩展与生态构建技能市场将执行器设计为可插拔的“技能包”。用户可以自行编写或从社区下载skill.py文件放到指定目录系统启动时自动加载注册。这能让你的助手能力无限扩展。上下文记忆让助手记住之前的对话和操作。可以为每个用户会话维护一个向量数据库如ChromaDB,FAISS存储历史交互。当新指令到来时先检索相关历史将其作为上下文提供给LLM规划器实现连贯的多轮对话。图形化界面为助手开发一个简单的本地Web UI显示任务执行状态、历史记录并提供手动触发和干预的按钮。硬件集成结合Home Assistant、IFTTT等平台你的语音助手可以成为智能家居的控制中心。通过添加对应的执行器实现“打开客厅灯”、“调整空调温度”等操作。构建像AnovaX这样的系统最大的挑战往往不在于单个技术点而在于如何将这些组件语音、LLM、多智能体、状态管理优雅、可靠地整合在一起并处理好边界情况。它更像是一个微型的操作系统调度器。从这个小项目出发你可以深入探索Agent系统的每一个方面例如更复杂的任务规划算法如基于图的规划、更强大的恢复策略如强化学习训练恢复策略、以及如何评估整个系统的可靠性和用户体验。这个过程中积累的经验对于理解下一代人机交互和自动化系统的构建将是无价的。