尧图网页设计 
1. 项目概述当LLM智能体学会“编程式”思考最近在折腾LLM智能体LLM Agents的朋友估计都遇到过同一个头疼的问题让一个智能体去执行一个稍微复杂点的任务比如“帮我分析一下这个季度的销售数据生成一份报告并给市场部发一封邮件”结果往往是灾难性的。智能体要么在数据格式上卡壳要么在逻辑判断上出错要么就是生成一堆看似合理但完全无法执行的“伪代码”。问题的根源在于传统的基于自然语言指令驱动的智能体其“技能”是模糊、不稳定且难以精确控制的。这就像你让一个刚学编程的新手去写一个复杂的业务系统他可能知道每个单词的意思但完全无法将它们组织成正确、高效且健壮的程序。这正是“Formal Skill: Programmable Runtime Skills for Efficient and Accurate LLM Agents”这个项目要解决的核心痛点。它不是一个具体的工具库而是一种方法论和架构思想。其核心是引入“形式化技能”Formal Skill和“可编程运行时技能”Programmable Runtime Skills这两个概念旨在为LLM智能体赋予一种类似编程语言的、结构化的能力定义和执行方式。简单来说就是把原来用自然语言描述的、模糊的“技能”变成用结构化数据如JSON Schema或领域特定语言DSL精确定义的、可以在运行时被动态组合和调用的“程序单元”。为什么这很重要想象一下你有一个智能体负责处理客户工单。一个“处理退款”的技能如果只用自然语言描述智能体可能会误解步骤顺序、遗漏必要的验证如检查订单状态是否可退款或者生成不符合财务系统API要求的请求体。而一个“形式化技能”则会明确定义输入工单ID、用户信息、输出退款流水号、执行步骤1. 验证订单状态 - 2. 调用财务API - 3. 更新工单状态 - 4. 通知用户、错误处理逻辑以及所需的权限。这样智能体的行为就变得可预测、可审计、可复用。从网络热词中频繁出现的“Python”、“JSON”、“JSON格式”可以看出社区正在积极寻找将LLM能力与结构化编程和数据交换结合的方法。无论是“text2jsontext2sql”的尝试还是各种用Python处理JSON、CSV数据的教程都反映了大家对于“让LLM的输出更结构化、更可编程”的强烈需求。这个项目正是将这种需求系统化、理论化并提供了一套可行的工程实践思路。2. 核心理念拆解从“自然语言驱动”到“程序化编排”要理解Formal Skill我们必须先跳出“让LLM生成代码然后执行”的简单思维。那种方式更像是“代码生成器”其质量严重依赖LLM的代码生成能力且生成的代码是黑盒难以调试和保证安全。Formal Skill追求的是更高层次的抽象将技能本身作为一种可被声明、组合和推理的一等公民First-class Citizen。2.1 什么是“形式化技能”Formal Skill形式化技能的核心在于“形式化”Formal即用精确的、无歧义的数学或逻辑语言来描述一个技能。在工程实践中这通常体现为一种模式Schema或契约Contract。一个典型的形式化技能定义可能包含以下部分通常用JSON Schema来表述{ “skill_name”: “fetch_weather”, “description”: “根据城市名称获取当前天气信息”, “input_schema”: { “type”: “object”, “properties”: { “city”: { “type”: “string”, “description”: “城市名称例如‘北京’、‘New York’” } }, “required”: [“city”] }, “output_schema”: { “type”: “object”, “properties”: { “temperature”: {“type”: “number”, “description”: “摄氏度”}, “condition”: {“type”: “string”, “description”: “天气状况如‘晴’、‘多云’”}, “humidity”: {“type”: “number”, “description”: “湿度百分比”} } }, “execution”: { “type”: “http_request”, “config”: { “url”: “https://api.weather.com/v3/current”, “method”: “GET”, “params_mapping”: {“q”: “{input.city}”} } }, “error_handling”: { “on_http_error”: “retry_twice_then_fail”, “on_data_invalid”: “return_default_and_log” } }关键点解析input_schema output_schema: 这定义了技能的“类型签名”。它明确规定了技能需要什么以及会返回什么。这允许在技能组合时进行静态或动态的类型检查避免将错误类型的数据传递给技能。execution: 这里定义了技能如何被执行。它可以是HTTP请求、数据库查询、调用一个Python函数甚至是触发另一个LLM的思考过程。关键是执行逻辑被封装和抽象了。error_handling: 明确定义了在各种异常情况下的行为策略这使得智能体的行为更加健壮。注意这里的形式化定义并不要求技能的实现本身是“形式化证明”的而是强调其接口和行为是清晰、明确、可被机器理解和校验的。这是工程实践与纯理论研究的折中。2.2 什么是“可编程运行时技能”Programmable Runtime Skills“可编程”意味着技能不是固定的、写死的而是可以在智能体运行的过程中根据上下文、历史、用户目标被动态地创建、修改、组合和调度。“运行时”则强调这种能力是在智能体执行任务的生命周期内生效的。这带来了几个革命性的优势动态技能组合智能体可以将多个基础技能像乐高积木一样拼接起来形成复杂的“复合技能”Composite Skill。例如一个“生成季度报告”的复合技能可能由“fetch_sales_data”、“analyze_trend”、“generate_chart”、“format_to_pdf”四个基础技能按特定顺序和逻辑组合而成。这个组合逻辑本身也可以被定义和存储。条件逻辑与循环技能的执行可以包含条件判断if-else和循环for/while。例如“监控服务器状态”技能可以定义为每5分钟执行一次“check_server_health”技能如果返回状态为“unhealthy”则触发“send_alert”技能。上下文感知与参数化技能可以从当前对话历史、环境变量、用户偏好中动态获取参数。比如一个“预订餐厅”的技能其“地理位置”参数可以自动从用户消息中提取或默认使用用户个人资料中的地址。技能的学习与演化智能体在运行过程中如果发现某个技能组合模式频繁使用且有效可以将其“固化”为一个新的复合技能加入技能库供未来直接调用。实操心得实现“可编程运行时技能”的关键是设计一个轻量级的“技能编排引擎”。这个引擎的核心是一个解释器或执行器能够解析基于JSON或YAML定义的技能流程图DAG并按照定义调用相应的技能实现。Python的asyncio库非常适合用来构建这种异步、并发的技能执行引擎。3. 架构设计与核心组件实现构建一个支持Formal Skill的LLM智能体系统需要一套清晰的架构。下面是一个参考架构包含核心组件和它们之间的交互。3.1 核心组件详解技能注册表Skill Registry职责系统的核心目录存储所有可用技能的形式化定义元数据。它不存储技能的具体实现代码只存储如何找到和调用它的信息。实现可以是一个简单的内存字典、数据库表或者更复杂的版本控制系统。每个技能条目包含skill_id、name、description、input_schema、output_schema、executor_type如python_function、http_endpoint和executor_config如函数路径、API地址。技巧为技能添加标签tags如“data_fetch”,“calculation”,“notification”便于智能体根据任务类型快速检索相关技能。技能执行器Skill Executor职责负责实际运行一个技能。根据技能定义中的executor_type和config调用对应的后端。类型本地函数执行器调用Python环境中定义的函数。需要处理好函数参数的注入从input_schema映射到函数参数。HTTP请求执行器调用外部RESTful API。需要处理认证、重试、超时和错误码转换。LLM调用执行器将技能描述和输入作为Prompt的一部分调用LLM如GPT-4来生成输出。这常用于那些难以用固定逻辑实现但可以用自然语言清晰描述的“软技能”如“文本润色”、“创意生成”。复合技能执行器专门用于执行那些由其他技能组合而成的技能其本身是一个小型的流程引擎。技能编排引擎Orchestration Engine职责这是实现“可编程”的关键。它解析用户目标或规划器Planner生成的技能执行计划一个DAG并协调各个技能执行器按顺序、并行或条件分支执行。核心逻辑依赖解析分析技能之间的数据流依赖A技能的输出是B技能的输入。调度执行决定执行顺序对于无依赖的技能可以并行执行以提高效率。数据传递管理技能间输入输出的传递和格式转换例如确保数据符合下一个技能的input_schema。状态管理跟踪整个复合技能的执行状态进行中、成功、失败、中间结果和错误信息。实现建议可以考虑使用像Prefect或Airflow这类工作流引擎的思想但实现一个更轻量、更专注于LLM智能体场景的版本。核心是一个状态机。规划与决策模块Planner职责理解用户意图并将其分解成一个由技能组成的可执行计划。这是LLM大显身手的地方。工作流程技能检索根据用户查询从技能注册表中检索最相关的技能列表可以通过向量数据库进行语义检索。计划生成LLM根据检索到的技能及其输入输出模式生成一个初步的技能调用序列图或列表。Prompt模板可以设计为“给定用户目标‘{query}’和以下可用技能{skills_list}请生成一个分步执行计划。”计划验证与优化编排引擎可以可选地对LLM生成的计划进行静态检查比如检查技能间的输入输出类型是否匹配是否存在循环依赖等。这一步可以显著提高计划的可行性。上下文管理器Context Manager职责维护对话和任务执行的上下文。它为技能提供额外的输入信息如之前的对话历史、用户个人资料、环境变量并存储技能执行的历史记录供后续技能或规划器参考。关键数据conversation_history,user_profile,session_variables,skill_execution_logs。3.2 数据流与交互流程一个完整的任务处理流程如下用户输入用户提出请求如“帮我比较一下产品A和产品B在过去一个月的社交媒体声量”。规划阶段规划模块接收请求从技能注册表检索相关技能如fetch_social_media_posts,perform_sentiment_analysis,aggregate_metrics,generate_comparison_chart。LLM基于检索结果生成一个初步执行计划可能是一个JSON结构描述了技能调用顺序和数据流。验证与编译编排引擎接收计划进行依赖分析和类型校验。如果计划无效则反馈错误给规划模块进行修正。执行阶段编排引擎根据计划按序或并行地调用技能执行器。技能执行器根据技能定义执行具体操作调用API、运行函数等并将结果返回给编排引擎。编排引擎将结果传递给下一个技能或存储在上下文中。结果整合与输出所有技能执行完毕后编排引擎将最终结果可能是复合技能的输出或由LLM对多个结果进行总结返回给用户。学习与更新可选系统记录本次成功的技能组合和参数可用于优化未来的规划或创建新的复合技能。4. 实战用Python构建一个最小可行系统理论说再多不如动手。我们来构建一个最简单的Formal Skill系统原型只包含核心的注册、执行和简单编排功能。我们将使用Python和Pydantic用于数据验证和设置管理。4.1 定义技能模型Skill Model首先我们需要用Pydantic定义技能的数据结构。from typing import Any, Dict, List, Optional, Callable from pydantic import BaseModel, Field from enum import Enum class ExecutorType(str, Enum): PYTHON_FUNCTION “python_function” HTTP_REQUEST “http_request” LLM_GENERATION “llm_generation” class SkillDefinition(BaseModel): “”“形式化技能定义”“” id: str Field(..., description“技能唯一标识符”) name: str Field(..., description“技能名称”) description: str Field(..., description“技能功能描述”) input_schema: Dict[str, Any] Field(..., description“输入参数JSON Schema”) output_schema: Dict[str, Any] Field(..., description“输出结果JSON Schema”) executor_type: ExecutorType executor_config: Dict[str, Any] Field(default_factorydict, description“执行器配置如函数名、URL等”) tags: List[str] Field(default_factorylist, description“技能标签用于检索”)4.2 实现技能注册表与执行器class SkillRegistry: “”“简单的内存技能注册表”“” def __init__(self): self._skills: Dict[str, SkillDefinition] {} def register(self, skill_def: SkillDefinition): if skill_def.id in self._skills: raise ValueError(f“Skill with id ‘{skill_def.id}’ already exists.”) self._skills[skill_def.id] skill_def print(f“Registered skill: {skill_def.name} ({skill_def.id})”) def get(self, skill_id: str) - Optional[SkillDefinition]: return self._skills.get(skill_id) def search_by_tag(self, tag: str) - List[SkillDefinition]: return [s for s in self._skills.values() if tag in s.tags] class SkillExecutor: “”“技能执行器根据类型分发”“” def __init__(self): # 这里可以注入各种客户端如HTTP客户端、LLM客户端等 self._http_client None # 实际项目中会用aiohttp或httpx self._llm_client None # 实际项目中会用openai等库 async def execute(self, skill_def: SkillDefinition, input_data: Dict[str, Any]) - Dict[str, Any]: “”“执行一个技能”“” if skill_def.executor_type ExecutorType.PYTHON_FUNCTION: return await self._execute_python_function(skill_def, input_data) elif skill_def.executor_type ExecutorType.HTTP_REQUEST: return await self._execute_http_request(skill_def, input_data) elif skill_def.executor_type ExecutorType.LLM_GENERATION: return await self._execute_llm_generation(skill_def, input_data) else: raise ValueError(f“Unsupported executor type: {skill_def.executor_type}”) async def _execute_python_function(self, skill_def: SkillDefinition, input_data: Dict): # 从配置中获取函数路径并动态导入执行 func_path skill_def.executor_config.get(“function_path”) module_name, func_name func_path.rsplit(‘.’, 1) module __import__(module_name, fromlist[func_name]) func getattr(module, func_name) # 简单地将input_data作为关键字参数传入实际需要更精细的映射 result func(**input_data) return result async def _execute_http_request(self, skill_def: SkillDefinition, input_data: Dict): # 简化示例实际需要处理method, headers, params, body等 url skill_def.executor_config[“url”] # 将input_data映射到请求参数这里简化处理 # 实际项目应使用更健壮的模板系统如Jinja2 params input_data # 假设使用httpx async with httpx.AsyncClient() as client: resp await client.get(url, paramsparams) resp.raise_for_status() return resp.json() async def _execute_llm_generation(self, skill_def: SkillDefinition, input_data: Dict): prompt_template skill_def.executor_config.get(“prompt_template”) # 使用输入数据渲染提示词 # 实际项目中会用到更复杂的提示词工程 prompt prompt_template.format(**input_data) # 调用LLM API (示例) # response await self._llm_client.chat.completions.create(...) # return {“text”: response.choices[0].message.content} return {“text”: f“Simulated LLM output for input: {input_data}”}4.3 定义并注册几个示例技能让我们定义两个简单的技能一个获取天气一个进行单位转换。# 首先定义技能的具体实现函数这些函数可以放在其他模块 def get_current_temperature(city: str) - dict: “”“模拟获取天气的函数”“” # 这里应该是真实的API调用例如调用和风天气、OpenWeatherMap等 print(f“Fetching temperature for {city}...”) # 模拟数据 mock_data { “beijing”: {“temperature”: 22, “condition”: “sunny”}, “shanghai”: {“temperature”: 25, “condition”: “cloudy”}, } return mock_data.get(city.lower(), {“temperature”: 20, “condition”: “unknown”}) def celsius_to_fahrenheit(celsius: float) - dict: “”“摄氏度转华氏度”“” fahrenheit (celsius * 9/5) 32 return {“fahrenheit”: round(fahrenheit, 2)} # 初始化注册表并注册技能 registry SkillRegistry() executor SkillExecutor() # 注册“获取温度”技能 weather_skill SkillDefinition( id“fetch_temp”, name“Get City Temperature”, description“获取指定城市的当前温度摄氏度和天气状况”, input_schema{ “type”: “object”, “properties”: {“city”: {“type”: “string”}}, “required”: [“city”] }, output_schema{ “type”: “object”, “properties”: { “temperature”: {“type”: “number”}, “condition”: {“type”: “string”} } }, executor_typeExecutorType.PYTHON_FUNCTION, executor_config{“function_path”: “__main__.get_current_temperature”}, # 注意路径 tags[“weather”, “data_fetch”] ) registry.register(weather_skill) # 注册“温度转换”技能 conversion_skill SkillDefinition( id“c_to_f”, name“Celsius to Fahrenheit Converter”, description“将摄氏度转换为华氏度”, input_schema{ “type”: “object”, “properties”: {“celsius”: {“type”: “number”}}, “required”: [“celsius”] }, output_schema{ “type”: “object”, “properties”: {“fahrenheit”: {“type”: “number”}} }, executor_typeExecutorType.PYTHON_FUNCTION, executor_config{“function_path”: “__main__.celsius_to_fahrenheit”}, tags[“conversion”, “calculation”] ) registry.register(conversion_skill)4.4 实现一个简单的线性编排引擎现在我们实现一个最简单的编排引擎它只能顺序执行一个技能列表。class LinearOrchestrator: “”“简单的线性技能编排器”“” def __init__(self, registry: SkillRegistry, executor: SkillExecutor): self.registry registry self.executor executor async def execute_plan(self, plan: List[Dict]) - List[Dict]: “”“执行一个线性计划。计划格式[{‘skill_id’: ‘id1’, ‘input’: {...}}, ...]”“” results [] context {} # 简单的上下文存储上一步的输出 for step in plan: skill_id step[‘skill_id’] skill_def self.registry.get(skill_id) if not skill_def: raise ValueError(f“Skill not found: {skill_id}”) # 构建输入可以来自步骤显式定义也可以来自上一步的结果这里做简单映射 # 实际项目需要更复杂的数据绑定逻辑如使用Jinja2模板从context中取值 step_input step.get(‘input’, {}) # 一个简单的规则如果input是字符串‘prev_result’则使用上一步的整个结果 if step_input ‘prev_result’ and results: step_input results[-1] print(f“Executing skill: {skill_def.name} with input {step_input}”) try: result await self.executor.execute(skill_def, step_input) results.append(result) context[‘last_result’] result # 更新上下文 except Exception as e: print(f“Failed to execute skill {skill_id}: {e}”) results.append({“error”: str(e)}) # 简单的错误处理终止流程 break return results4.5 运行一个复合任务最后让我们模拟一个用户请求“获取北京的温度并转换成华氏度是多少”import asyncio async def main(): # 1. 创建组件 registry SkillRegistry() executor SkillExecutor() orchestrator LinearOrchestrator(registry, executor) # 2. 注册技能这里省略了重复的注册代码假设技能已注册 # ... [上面注册技能的代码放在这里] ... # 3. 定义执行计划这个计划可以由LLM生成 # 计划先获取北京温度再将获取到的温度值摄氏度转换为华氏度 execution_plan [ { “skill_id”: “fetch_temp”, “input”: {“city”: “beijing”} # 第一步的输入来自用户请求 }, { “skill_id”: “c_to_f”, “input”: {“celsius”: “prev_result.temperature”} # 第二步的输入需要从第一步的结果中提取 } ] # 4. 执行计划 final_results await orchestrator.execute_plan(execution_plan) print(“\n Execution Results ”) for i, res in enumerate(final_results): print(f“Step {i1} result: {res}”) # 5. 整合最终答案这里可以再用一个LLM技能来生成自然语言回答 if len(final_results) 2: temp_c final_results[0].get(“temperature”) temp_f final_results[1].get(“fahrenheit”) condition final_results[0].get(“condition”) print(f“\n最终答案北京当前天气{condition}气温{temp_c}°C即{temp_f}°F。”) if __name__ “__main__”: asyncio.run(main())运行结果可能如下Registered skill: Get City Temperature (fetch_temp) Registered skill: Celsius to Fahrenheit Converter (c_to_f) Executing skill: Get City Temperature with input {‘city’: ‘beijing’} Fetching temperature for beijing... Executing skill: Celsius to Fahrenheit Converter with input {‘celsius’: ‘prev_result.temperature’} Execution Results Step 1 result: {‘temperature’: 22, ‘condition’: ‘sunny’} Step 2 result: {‘fahrenheit’: 71.6} 最终答案北京当前天气sunny气温22°C即71.6°F。实操心得上面的例子极度简化尤其是第二步输入“prev_result.temperature”的处理。在实际系统中你需要一个数据绑定Data Binding或表达式求值Expression Evaluation模块。这个模块能够解析像“prev_result.temperature”、“context.user.name”这样的字符串并从执行上下文中提取出真正的值。可以使用jsonpath库如jsonpath-ng或自定义的模板引擎如Jinja2的一个子集来实现这是连接不同技能数据流的关键。5. 进阶话题与工程化挑战构建一个原型很有趣但要将其投入生产环境还需要解决一系列工程化挑战。5.1 技能依赖管理与版本控制当技能数量成百上千时管理它们之间的依赖和版本变得至关重要。依赖声明一个技能可能依赖于特定的Python包版本、外部服务的特定API版本。需要在SkillDefinition中增加dependencies字段。版本化技能本身也需要版本号如fetch_weather:v1.2.0。注册表应支持同一技能的多版本共存智能体可以根据兼容性要求选择特定版本。技能仓库借鉴Docker Registry或PyPI建立中心化的技能仓库支持技能的发布、发现、拉取和更新。5.2 复杂的流程控制条件、循环与错误处理线性执行只是基础。真实的业务逻辑需要分支和循环。条件分支在技能定义或计划中引入condition字段其值是一个基于上下文的布尔表达式。编排引擎根据表达式结果决定执行哪条分支。{ “type”: “conditional”, “condition”: “{context.last_result.temperature} 30”, “true_branch”: [{“skill_id”: “send_heat_alert”, ...}], “false_branch”: [{“skill_id”: “log_normal_temp”, ...}] }循环支持for_each和while循环。例如对一个用户列表循环执行“发送通知”技能。错误处理与重试在技能定义或计划节点中定义丰富的错误处理策略如重试带退避算法、降级执行备用技能、补偿执行回滚操作。5.3 技能的动态发现与组合这是“可编程”的终极体现。智能体不仅可以使用预定义的技能还能在运行时发现新的技能例如从技能仓库搜索甚至通过LLM自动生成新的复合技能草图。语义检索使用技能的名称、描述、标签和输入输出Schema构建向量嵌入存入向量数据库如ChromaDB, Weaviate。当用户提出新请求时用LLM提取关键意图并向量检索最相关的技能。自动规划与代码生成LLM根据检索到的技能和用户目标直接生成一个可执行的技能编排计划可能是JSON或YAML格式。更进一步的LLM可以为一些简单、通用的逻辑自动生成对应技能的Python实现代码并动态注册到系统中。5.4 监控、可观测性与调试当智能体由数十个技能动态编排而成时出了问题调试就像大海捞针。分布式追踪为每个用户请求生成唯一的trace_id并贯穿所有技能调用。记录每个技能的输入、输出、开始时间、结束时间和状态成功/失败。使用OpenTelemetry这样的标准来收集数据。技能性能指标监控每个技能的调用次数、成功率、平均耗时、错误类型。这有助于发现性能瓶颈和不可靠的技能。可视化调试界面提供一个界面可以回放任意请求的完整执行流程图查看每个节点的输入输出和日志这对于排查复杂问题不可或缺。5.5 安全与权限控制技能可能执行危险操作如删除数据、发送消息、调用付费API。技能权限模型为每个技能定义所需的权限级别如read_db,write_file,send_email。用户/角色权限定义不同用户或角色所能使用的技能集合。运行时权限检查在技能执行前由编排引擎或一个专门的授权服务检查当前会话是否具备执行该技能的权限。输入输出数据可能也需要进行净化Sanitization和过滤。6. 常见问题与避坑指南在实际开发和运用Formal Skill模式时我踩过不少坑这里总结几个最常见的。6.1 技能粒度设计多细才算合适这是最常被问到的问题。技能不是越细越好也不是越粗越好。过细的坏处技能数量爆炸管理成本高技能间通信开销大LLM规划器需要组合的步骤太多容易出错。过粗的坏处技能复用性差内部逻辑复杂难以测试和调试不符合单一职责原则。设计原则单一职责一个技能只做一件事并且做好。例如“验证用户邮箱”和“发送验证码”应该是两个技能。功能完整性一个技能应该完成一个对上游调用者LLM或其他技能有独立价值的完整功能单元。例如“创建用户账户”可能包含验证输入、检查重复、写入数据库、发送欢迎邮件等多个步骤但对于外部来说这是一个完整的业务操作适合封装为一个技能。复用性评估思考这个功能是否会在多个不同的任务流中被用到。如果是它就是一个好的技能候选。变更频率将变化频率不同的部分拆分开。例如调用某个第三方API的细节URL、参数可能经常变但业务逻辑获取数据-处理相对稳定可以考虑拆成“调用XX API”和“处理XX数据”两个技能。6.2 如何处理技能间的数据格式不匹配技能A输出{“user_id”: 123, “name”: “Alice”}技能B需要输入{“uid”: 123, “username”: “Alice”}。字段名和结构都对不上。方案一适配器技能Adapter Skill专门编写一个轻量的“数据转换”技能负责将A的输出格式映射为B的输入格式。这增加了技能数量但保持了基础技能的纯粹性。方案二在编排层进行映射在编排引擎的计划定义中显式指定数据映射规则。例如{ “skill_id”: “skill_b”, “input_mapping”: { “uid”: “{steps.skill_a.output.user_id}”, “username”: “{steps.skill_a.output.name}” } }推荐方案二因为它将数据转换逻辑集中在编排定义中更清晰且不影响基础技能。实现时需要一套灵活的表达式语言来支持这种映射。6.3 LLM生成的计划不靠谱怎么办LLM可能会生成无法执行的计划比如技能不存在、输入输出类型不匹配、有循环依赖。后置验证与修复在LLM生成计划后增加一个“计划验证器”模块。它根据技能注册表中的元数据对计划进行静态分析检查上述问题。如果发现问题可以将错误信息反馈给LLM让其重新生成计划。这形成了一个“规划-验证-修正”的循环。提供更丰富的上下文在给LLM的Prompt中不仅提供技能列表还可以提供一些成功的计划示例、技能组合的模式Pattern、以及明确的约束如“不要创建循环依赖”、“确保技能B的输入来自技能A的输出”。分层规划不要让LLM一次性生成所有细节。可以先让它生成一个高级别的目标分解如“1. 获取数据2. 分析数据3. 生成报告”然后针对每个子目标再利用技能库进行更细粒度的规划。6.4 技能执行失败如何保证整体任务不崩溃这是分布式系统常见的可靠性问题。技能级别的重试与超时在每个技能的executor_config中配置重试次数、重试间隔建议使用指数退避和超时时间。对于非幂等的操作如创建订单重试要格外小心。流程级别的错误处理策略在复合技能或计划定义中为每个步骤或整个流程定义错误处理策略。例如“on_failure”: “continue”忽略当前步骤错误继续执行后续步骤。“on_failure”: “retry_step”重试当前步骤N次。“on_failure”: “fallback_to”: “alternative_skill_id”执行一个备用的降级技能。“on_failure”: “compensate”触发一个补偿事务尝试回滚之前步骤造成的影响。断路器模式Circuit Breaker对于频繁失败的外部服务技能在一段时间内停止调用它直接返回失败或降级结果避免雪崩效应。6.5 如何测试Formal Skill测试分为多个层次单元测试测试每个技能实现函数本身的正确性与普通函数测试无异。集成测试测试技能在注册、被执行器调用时的端到端流程。需要Mock外部依赖如HTTP请求、数据库。编排测试测试特定的技能组合计划是否能正确执行并产生预期结果。可以编写YAML或JSON格式的测试用例定义输入、计划和期望输出。LLM规划测试这是最复杂的。需要构建一个测试集包含各种用户查询然后运行完整的系统LLM规划器编排引擎评估最终结果的正确性和计划的合理性。这更多是评估整个系统的“智能”水平。Formal Skill和Programmable Runtime Skills不是银弹它引入了一定的复杂性但换来的是智能体行为的确定性、可维护性、可复用性和可观测性的巨大提升。它本质上是一种软件工程思想在LLM应用领域的实践将智能体的“智能”更多地体现在高层的规划和编排上而将确定性的、可靠的操作下沉到一个个定义清晰、测试充分的“技能”中。对于构建严肃的、用于生产环境的LLM智能体应用我认为这是一条必经之路。从简单的技能注册表开始逐步引入编排、规划、监控你的智能体系统会像搭积木一样变得越来越强大和稳健。 LLM智能体编程化：Formal Skill与可编程运行时技能架构解析
1. 项目概述当LLM智能体学会“编程式”思考最近在折腾LLM智能体LLM Agents的朋友估计都遇到过同一个头疼的问题让一个智能体去执行一个稍微复杂点的任务比如“帮我分析一下这个季度的销售数据生成一份报告并给市场部发一封邮件”结果往往是灾难性的。智能体要么在数据格式上卡壳要么在逻辑判断上出错要么就是生成一堆看似合理但完全无法执行的“伪代码”。问题的根源在于传统的基于自然语言指令驱动的智能体其“技能”是模糊、不稳定且难以精确控制的。这就像你让一个刚学编程的新手去写一个复杂的业务系统他可能知道每个单词的意思但完全无法将它们组织成正确、高效且健壮的程序。这正是“Formal Skill: Programmable Runtime Skills for Efficient and Accurate LLM Agents”这个项目要解决的核心痛点。它不是一个具体的工具库而是一种方法论和架构思想。其核心是引入“形式化技能”Formal Skill和“可编程运行时技能”Programmable Runtime Skills这两个概念旨在为LLM智能体赋予一种类似编程语言的、结构化的能力定义和执行方式。简单来说就是把原来用自然语言描述的、模糊的“技能”变成用结构化数据如JSON Schema或领域特定语言DSL精确定义的、可以在运行时被动态组合和调用的“程序单元”。为什么这很重要想象一下你有一个智能体负责处理客户工单。一个“处理退款”的技能如果只用自然语言描述智能体可能会误解步骤顺序、遗漏必要的验证如检查订单状态是否可退款或者生成不符合财务系统API要求的请求体。而一个“形式化技能”则会明确定义输入工单ID、用户信息、输出退款流水号、执行步骤1. 验证订单状态 - 2. 调用财务API - 3. 更新工单状态 - 4. 通知用户、错误处理逻辑以及所需的权限。这样智能体的行为就变得可预测、可审计、可复用。从网络热词中频繁出现的“Python”、“JSON”、“JSON格式”可以看出社区正在积极寻找将LLM能力与结构化编程和数据交换结合的方法。无论是“text2jsontext2sql”的尝试还是各种用Python处理JSON、CSV数据的教程都反映了大家对于“让LLM的输出更结构化、更可编程”的强烈需求。这个项目正是将这种需求系统化、理论化并提供了一套可行的工程实践思路。2. 核心理念拆解从“自然语言驱动”到“程序化编排”要理解Formal Skill我们必须先跳出“让LLM生成代码然后执行”的简单思维。那种方式更像是“代码生成器”其质量严重依赖LLM的代码生成能力且生成的代码是黑盒难以调试和保证安全。Formal Skill追求的是更高层次的抽象将技能本身作为一种可被声明、组合和推理的一等公民First-class Citizen。2.1 什么是“形式化技能”Formal Skill形式化技能的核心在于“形式化”Formal即用精确的、无歧义的数学或逻辑语言来描述一个技能。在工程实践中这通常体现为一种模式Schema或契约Contract。一个典型的形式化技能定义可能包含以下部分通常用JSON Schema来表述{ “skill_name”: “fetch_weather”, “description”: “根据城市名称获取当前天气信息”, “input_schema”: { “type”: “object”, “properties”: { “city”: { “type”: “string”, “description”: “城市名称例如‘北京’、‘New York’” } }, “required”: [“city”] }, “output_schema”: { “type”: “object”, “properties”: { “temperature”: {“type”: “number”, “description”: “摄氏度”}, “condition”: {“type”: “string”, “description”: “天气状况如‘晴’、‘多云’”}, “humidity”: {“type”: “number”, “description”: “湿度百分比”} } }, “execution”: { “type”: “http_request”, “config”: { “url”: “https://api.weather.com/v3/current”, “method”: “GET”, “params_mapping”: {“q”: “{input.city}”} } }, “error_handling”: { “on_http_error”: “retry_twice_then_fail”, “on_data_invalid”: “return_default_and_log” } }关键点解析input_schema output_schema: 这定义了技能的“类型签名”。它明确规定了技能需要什么以及会返回什么。这允许在技能组合时进行静态或动态的类型检查避免将错误类型的数据传递给技能。execution: 这里定义了技能如何被执行。它可以是HTTP请求、数据库查询、调用一个Python函数甚至是触发另一个LLM的思考过程。关键是执行逻辑被封装和抽象了。error_handling: 明确定义了在各种异常情况下的行为策略这使得智能体的行为更加健壮。注意这里的形式化定义并不要求技能的实现本身是“形式化证明”的而是强调其接口和行为是清晰、明确、可被机器理解和校验的。这是工程实践与纯理论研究的折中。2.2 什么是“可编程运行时技能”Programmable Runtime Skills“可编程”意味着技能不是固定的、写死的而是可以在智能体运行的过程中根据上下文、历史、用户目标被动态地创建、修改、组合和调度。“运行时”则强调这种能力是在智能体执行任务的生命周期内生效的。这带来了几个革命性的优势动态技能组合智能体可以将多个基础技能像乐高积木一样拼接起来形成复杂的“复合技能”Composite Skill。例如一个“生成季度报告”的复合技能可能由“fetch_sales_data”、“analyze_trend”、“generate_chart”、“format_to_pdf”四个基础技能按特定顺序和逻辑组合而成。这个组合逻辑本身也可以被定义和存储。条件逻辑与循环技能的执行可以包含条件判断if-else和循环for/while。例如“监控服务器状态”技能可以定义为每5分钟执行一次“check_server_health”技能如果返回状态为“unhealthy”则触发“send_alert”技能。上下文感知与参数化技能可以从当前对话历史、环境变量、用户偏好中动态获取参数。比如一个“预订餐厅”的技能其“地理位置”参数可以自动从用户消息中提取或默认使用用户个人资料中的地址。技能的学习与演化智能体在运行过程中如果发现某个技能组合模式频繁使用且有效可以将其“固化”为一个新的复合技能加入技能库供未来直接调用。实操心得实现“可编程运行时技能”的关键是设计一个轻量级的“技能编排引擎”。这个引擎的核心是一个解释器或执行器能够解析基于JSON或YAML定义的技能流程图DAG并按照定义调用相应的技能实现。Python的asyncio库非常适合用来构建这种异步、并发的技能执行引擎。3. 架构设计与核心组件实现构建一个支持Formal Skill的LLM智能体系统需要一套清晰的架构。下面是一个参考架构包含核心组件和它们之间的交互。3.1 核心组件详解技能注册表Skill Registry职责系统的核心目录存储所有可用技能的形式化定义元数据。它不存储技能的具体实现代码只存储如何找到和调用它的信息。实现可以是一个简单的内存字典、数据库表或者更复杂的版本控制系统。每个技能条目包含skill_id、name、description、input_schema、output_schema、executor_type如python_function、http_endpoint和executor_config如函数路径、API地址。技巧为技能添加标签tags如“data_fetch”,“calculation”,“notification”便于智能体根据任务类型快速检索相关技能。技能执行器Skill Executor职责负责实际运行一个技能。根据技能定义中的executor_type和config调用对应的后端。类型本地函数执行器调用Python环境中定义的函数。需要处理好函数参数的注入从input_schema映射到函数参数。HTTP请求执行器调用外部RESTful API。需要处理认证、重试、超时和错误码转换。LLM调用执行器将技能描述和输入作为Prompt的一部分调用LLM如GPT-4来生成输出。这常用于那些难以用固定逻辑实现但可以用自然语言清晰描述的“软技能”如“文本润色”、“创意生成”。复合技能执行器专门用于执行那些由其他技能组合而成的技能其本身是一个小型的流程引擎。技能编排引擎Orchestration Engine职责这是实现“可编程”的关键。它解析用户目标或规划器Planner生成的技能执行计划一个DAG并协调各个技能执行器按顺序、并行或条件分支执行。核心逻辑依赖解析分析技能之间的数据流依赖A技能的输出是B技能的输入。调度执行决定执行顺序对于无依赖的技能可以并行执行以提高效率。数据传递管理技能间输入输出的传递和格式转换例如确保数据符合下一个技能的input_schema。状态管理跟踪整个复合技能的执行状态进行中、成功、失败、中间结果和错误信息。实现建议可以考虑使用像Prefect或Airflow这类工作流引擎的思想但实现一个更轻量、更专注于LLM智能体场景的版本。核心是一个状态机。规划与决策模块Planner职责理解用户意图并将其分解成一个由技能组成的可执行计划。这是LLM大显身手的地方。工作流程技能检索根据用户查询从技能注册表中检索最相关的技能列表可以通过向量数据库进行语义检索。计划生成LLM根据检索到的技能及其输入输出模式生成一个初步的技能调用序列图或列表。Prompt模板可以设计为“给定用户目标‘{query}’和以下可用技能{skills_list}请生成一个分步执行计划。”计划验证与优化编排引擎可以可选地对LLM生成的计划进行静态检查比如检查技能间的输入输出类型是否匹配是否存在循环依赖等。这一步可以显著提高计划的可行性。上下文管理器Context Manager职责维护对话和任务执行的上下文。它为技能提供额外的输入信息如之前的对话历史、用户个人资料、环境变量并存储技能执行的历史记录供后续技能或规划器参考。关键数据conversation_history,user_profile,session_variables,skill_execution_logs。3.2 数据流与交互流程一个完整的任务处理流程如下用户输入用户提出请求如“帮我比较一下产品A和产品B在过去一个月的社交媒体声量”。规划阶段规划模块接收请求从技能注册表检索相关技能如fetch_social_media_posts,perform_sentiment_analysis,aggregate_metrics,generate_comparison_chart。LLM基于检索结果生成一个初步执行计划可能是一个JSON结构描述了技能调用顺序和数据流。验证与编译编排引擎接收计划进行依赖分析和类型校验。如果计划无效则反馈错误给规划模块进行修正。执行阶段编排引擎根据计划按序或并行地调用技能执行器。技能执行器根据技能定义执行具体操作调用API、运行函数等并将结果返回给编排引擎。编排引擎将结果传递给下一个技能或存储在上下文中。结果整合与输出所有技能执行完毕后编排引擎将最终结果可能是复合技能的输出或由LLM对多个结果进行总结返回给用户。学习与更新可选系统记录本次成功的技能组合和参数可用于优化未来的规划或创建新的复合技能。4. 实战用Python构建一个最小可行系统理论说再多不如动手。我们来构建一个最简单的Formal Skill系统原型只包含核心的注册、执行和简单编排功能。我们将使用Python和Pydantic用于数据验证和设置管理。4.1 定义技能模型Skill Model首先我们需要用Pydantic定义技能的数据结构。from typing import Any, Dict, List, Optional, Callable from pydantic import BaseModel, Field from enum import Enum class ExecutorType(str, Enum): PYTHON_FUNCTION “python_function” HTTP_REQUEST “http_request” LLM_GENERATION “llm_generation” class SkillDefinition(BaseModel): “”“形式化技能定义”“” id: str Field(..., description“技能唯一标识符”) name: str Field(..., description“技能名称”) description: str Field(..., description“技能功能描述”) input_schema: Dict[str, Any] Field(..., description“输入参数JSON Schema”) output_schema: Dict[str, Any] Field(..., description“输出结果JSON Schema”) executor_type: ExecutorType executor_config: Dict[str, Any] Field(default_factorydict, description“执行器配置如函数名、URL等”) tags: List[str] Field(default_factorylist, description“技能标签用于检索”)4.2 实现技能注册表与执行器class SkillRegistry: “”“简单的内存技能注册表”“” def __init__(self): self._skills: Dict[str, SkillDefinition] {} def register(self, skill_def: SkillDefinition): if skill_def.id in self._skills: raise ValueError(f“Skill with id ‘{skill_def.id}’ already exists.”) self._skills[skill_def.id] skill_def print(f“Registered skill: {skill_def.name} ({skill_def.id})”) def get(self, skill_id: str) - Optional[SkillDefinition]: return self._skills.get(skill_id) def search_by_tag(self, tag: str) - List[SkillDefinition]: return [s for s in self._skills.values() if tag in s.tags] class SkillExecutor: “”“技能执行器根据类型分发”“” def __init__(self): # 这里可以注入各种客户端如HTTP客户端、LLM客户端等 self._http_client None # 实际项目中会用aiohttp或httpx self._llm_client None # 实际项目中会用openai等库 async def execute(self, skill_def: SkillDefinition, input_data: Dict[str, Any]) - Dict[str, Any]: “”“执行一个技能”“” if skill_def.executor_type ExecutorType.PYTHON_FUNCTION: return await self._execute_python_function(skill_def, input_data) elif skill_def.executor_type ExecutorType.HTTP_REQUEST: return await self._execute_http_request(skill_def, input_data) elif skill_def.executor_type ExecutorType.LLM_GENERATION: return await self._execute_llm_generation(skill_def, input_data) else: raise ValueError(f“Unsupported executor type: {skill_def.executor_type}”) async def _execute_python_function(self, skill_def: SkillDefinition, input_data: Dict): # 从配置中获取函数路径并动态导入执行 func_path skill_def.executor_config.get(“function_path”) module_name, func_name func_path.rsplit(‘.’, 1) module __import__(module_name, fromlist[func_name]) func getattr(module, func_name) # 简单地将input_data作为关键字参数传入实际需要更精细的映射 result func(**input_data) return result async def _execute_http_request(self, skill_def: SkillDefinition, input_data: Dict): # 简化示例实际需要处理method, headers, params, body等 url skill_def.executor_config[“url”] # 将input_data映射到请求参数这里简化处理 # 实际项目应使用更健壮的模板系统如Jinja2 params input_data # 假设使用httpx async with httpx.AsyncClient() as client: resp await client.get(url, paramsparams) resp.raise_for_status() return resp.json() async def _execute_llm_generation(self, skill_def: SkillDefinition, input_data: Dict): prompt_template skill_def.executor_config.get(“prompt_template”) # 使用输入数据渲染提示词 # 实际项目中会用到更复杂的提示词工程 prompt prompt_template.format(**input_data) # 调用LLM API (示例) # response await self._llm_client.chat.completions.create(...) # return {“text”: response.choices[0].message.content} return {“text”: f“Simulated LLM output for input: {input_data}”}4.3 定义并注册几个示例技能让我们定义两个简单的技能一个获取天气一个进行单位转换。# 首先定义技能的具体实现函数这些函数可以放在其他模块 def get_current_temperature(city: str) - dict: “”“模拟获取天气的函数”“” # 这里应该是真实的API调用例如调用和风天气、OpenWeatherMap等 print(f“Fetching temperature for {city}...”) # 模拟数据 mock_data { “beijing”: {“temperature”: 22, “condition”: “sunny”}, “shanghai”: {“temperature”: 25, “condition”: “cloudy”}, } return mock_data.get(city.lower(), {“temperature”: 20, “condition”: “unknown”}) def celsius_to_fahrenheit(celsius: float) - dict: “”“摄氏度转华氏度”“” fahrenheit (celsius * 9/5) 32 return {“fahrenheit”: round(fahrenheit, 2)} # 初始化注册表并注册技能 registry SkillRegistry() executor SkillExecutor() # 注册“获取温度”技能 weather_skill SkillDefinition( id“fetch_temp”, name“Get City Temperature”, description“获取指定城市的当前温度摄氏度和天气状况”, input_schema{ “type”: “object”, “properties”: {“city”: {“type”: “string”}}, “required”: [“city”] }, output_schema{ “type”: “object”, “properties”: { “temperature”: {“type”: “number”}, “condition”: {“type”: “string”} } }, executor_typeExecutorType.PYTHON_FUNCTION, executor_config{“function_path”: “__main__.get_current_temperature”}, # 注意路径 tags[“weather”, “data_fetch”] ) registry.register(weather_skill) # 注册“温度转换”技能 conversion_skill SkillDefinition( id“c_to_f”, name“Celsius to Fahrenheit Converter”, description“将摄氏度转换为华氏度”, input_schema{ “type”: “object”, “properties”: {“celsius”: {“type”: “number”}}, “required”: [“celsius”] }, output_schema{ “type”: “object”, “properties”: {“fahrenheit”: {“type”: “number”}} }, executor_typeExecutorType.PYTHON_FUNCTION, executor_config{“function_path”: “__main__.celsius_to_fahrenheit”}, tags[“conversion”, “calculation”] ) registry.register(conversion_skill)4.4 实现一个简单的线性编排引擎现在我们实现一个最简单的编排引擎它只能顺序执行一个技能列表。class LinearOrchestrator: “”“简单的线性技能编排器”“” def __init__(self, registry: SkillRegistry, executor: SkillExecutor): self.registry registry self.executor executor async def execute_plan(self, plan: List[Dict]) - List[Dict]: “”“执行一个线性计划。计划格式[{‘skill_id’: ‘id1’, ‘input’: {...}}, ...]”“” results [] context {} # 简单的上下文存储上一步的输出 for step in plan: skill_id step[‘skill_id’] skill_def self.registry.get(skill_id) if not skill_def: raise ValueError(f“Skill not found: {skill_id}”) # 构建输入可以来自步骤显式定义也可以来自上一步的结果这里做简单映射 # 实际项目需要更复杂的数据绑定逻辑如使用Jinja2模板从context中取值 step_input step.get(‘input’, {}) # 一个简单的规则如果input是字符串‘prev_result’则使用上一步的整个结果 if step_input ‘prev_result’ and results: step_input results[-1] print(f“Executing skill: {skill_def.name} with input {step_input}”) try: result await self.executor.execute(skill_def, step_input) results.append(result) context[‘last_result’] result # 更新上下文 except Exception as e: print(f“Failed to execute skill {skill_id}: {e}”) results.append({“error”: str(e)}) # 简单的错误处理终止流程 break return results4.5 运行一个复合任务最后让我们模拟一个用户请求“获取北京的温度并转换成华氏度是多少”import asyncio async def main(): # 1. 创建组件 registry SkillRegistry() executor SkillExecutor() orchestrator LinearOrchestrator(registry, executor) # 2. 注册技能这里省略了重复的注册代码假设技能已注册 # ... [上面注册技能的代码放在这里] ... # 3. 定义执行计划这个计划可以由LLM生成 # 计划先获取北京温度再将获取到的温度值摄氏度转换为华氏度 execution_plan [ { “skill_id”: “fetch_temp”, “input”: {“city”: “beijing”} # 第一步的输入来自用户请求 }, { “skill_id”: “c_to_f”, “input”: {“celsius”: “prev_result.temperature”} # 第二步的输入需要从第一步的结果中提取 } ] # 4. 执行计划 final_results await orchestrator.execute_plan(execution_plan) print(“\n Execution Results ”) for i, res in enumerate(final_results): print(f“Step {i1} result: {res}”) # 5. 整合最终答案这里可以再用一个LLM技能来生成自然语言回答 if len(final_results) 2: temp_c final_results[0].get(“temperature”) temp_f final_results[1].get(“fahrenheit”) condition final_results[0].get(“condition”) print(f“\n最终答案北京当前天气{condition}气温{temp_c}°C即{temp_f}°F。”) if __name__ “__main__”: asyncio.run(main())运行结果可能如下Registered skill: Get City Temperature (fetch_temp) Registered skill: Celsius to Fahrenheit Converter (c_to_f) Executing skill: Get City Temperature with input {‘city’: ‘beijing’} Fetching temperature for beijing... Executing skill: Celsius to Fahrenheit Converter with input {‘celsius’: ‘prev_result.temperature’} Execution Results Step 1 result: {‘temperature’: 22, ‘condition’: ‘sunny’} Step 2 result: {‘fahrenheit’: 71.6} 最终答案北京当前天气sunny气温22°C即71.6°F。实操心得上面的例子极度简化尤其是第二步输入“prev_result.temperature”的处理。在实际系统中你需要一个数据绑定Data Binding或表达式求值Expression Evaluation模块。这个模块能够解析像“prev_result.temperature”、“context.user.name”这样的字符串并从执行上下文中提取出真正的值。可以使用jsonpath库如jsonpath-ng或自定义的模板引擎如Jinja2的一个子集来实现这是连接不同技能数据流的关键。5. 进阶话题与工程化挑战构建一个原型很有趣但要将其投入生产环境还需要解决一系列工程化挑战。5.1 技能依赖管理与版本控制当技能数量成百上千时管理它们之间的依赖和版本变得至关重要。依赖声明一个技能可能依赖于特定的Python包版本、外部服务的特定API版本。需要在SkillDefinition中增加dependencies字段。版本化技能本身也需要版本号如fetch_weather:v1.2.0。注册表应支持同一技能的多版本共存智能体可以根据兼容性要求选择特定版本。技能仓库借鉴Docker Registry或PyPI建立中心化的技能仓库支持技能的发布、发现、拉取和更新。5.2 复杂的流程控制条件、循环与错误处理线性执行只是基础。真实的业务逻辑需要分支和循环。条件分支在技能定义或计划中引入condition字段其值是一个基于上下文的布尔表达式。编排引擎根据表达式结果决定执行哪条分支。{ “type”: “conditional”, “condition”: “{context.last_result.temperature} 30”, “true_branch”: [{“skill_id”: “send_heat_alert”, ...}], “false_branch”: [{“skill_id”: “log_normal_temp”, ...}] }循环支持for_each和while循环。例如对一个用户列表循环执行“发送通知”技能。错误处理与重试在技能定义或计划节点中定义丰富的错误处理策略如重试带退避算法、降级执行备用技能、补偿执行回滚操作。5.3 技能的动态发现与组合这是“可编程”的终极体现。智能体不仅可以使用预定义的技能还能在运行时发现新的技能例如从技能仓库搜索甚至通过LLM自动生成新的复合技能草图。语义检索使用技能的名称、描述、标签和输入输出Schema构建向量嵌入存入向量数据库如ChromaDB, Weaviate。当用户提出新请求时用LLM提取关键意图并向量检索最相关的技能。自动规划与代码生成LLM根据检索到的技能和用户目标直接生成一个可执行的技能编排计划可能是JSON或YAML格式。更进一步的LLM可以为一些简单、通用的逻辑自动生成对应技能的Python实现代码并动态注册到系统中。5.4 监控、可观测性与调试当智能体由数十个技能动态编排而成时出了问题调试就像大海捞针。分布式追踪为每个用户请求生成唯一的trace_id并贯穿所有技能调用。记录每个技能的输入、输出、开始时间、结束时间和状态成功/失败。使用OpenTelemetry这样的标准来收集数据。技能性能指标监控每个技能的调用次数、成功率、平均耗时、错误类型。这有助于发现性能瓶颈和不可靠的技能。可视化调试界面提供一个界面可以回放任意请求的完整执行流程图查看每个节点的输入输出和日志这对于排查复杂问题不可或缺。5.5 安全与权限控制技能可能执行危险操作如删除数据、发送消息、调用付费API。技能权限模型为每个技能定义所需的权限级别如read_db,write_file,send_email。用户/角色权限定义不同用户或角色所能使用的技能集合。运行时权限检查在技能执行前由编排引擎或一个专门的授权服务检查当前会话是否具备执行该技能的权限。输入输出数据可能也需要进行净化Sanitization和过滤。6. 常见问题与避坑指南在实际开发和运用Formal Skill模式时我踩过不少坑这里总结几个最常见的。6.1 技能粒度设计多细才算合适这是最常被问到的问题。技能不是越细越好也不是越粗越好。过细的坏处技能数量爆炸管理成本高技能间通信开销大LLM规划器需要组合的步骤太多容易出错。过粗的坏处技能复用性差内部逻辑复杂难以测试和调试不符合单一职责原则。设计原则单一职责一个技能只做一件事并且做好。例如“验证用户邮箱”和“发送验证码”应该是两个技能。功能完整性一个技能应该完成一个对上游调用者LLM或其他技能有独立价值的完整功能单元。例如“创建用户账户”可能包含验证输入、检查重复、写入数据库、发送欢迎邮件等多个步骤但对于外部来说这是一个完整的业务操作适合封装为一个技能。复用性评估思考这个功能是否会在多个不同的任务流中被用到。如果是它就是一个好的技能候选。变更频率将变化频率不同的部分拆分开。例如调用某个第三方API的细节URL、参数可能经常变但业务逻辑获取数据-处理相对稳定可以考虑拆成“调用XX API”和“处理XX数据”两个技能。6.2 如何处理技能间的数据格式不匹配技能A输出{“user_id”: 123, “name”: “Alice”}技能B需要输入{“uid”: 123, “username”: “Alice”}。字段名和结构都对不上。方案一适配器技能Adapter Skill专门编写一个轻量的“数据转换”技能负责将A的输出格式映射为B的输入格式。这增加了技能数量但保持了基础技能的纯粹性。方案二在编排层进行映射在编排引擎的计划定义中显式指定数据映射规则。例如{ “skill_id”: “skill_b”, “input_mapping”: { “uid”: “{steps.skill_a.output.user_id}”, “username”: “{steps.skill_a.output.name}” } }推荐方案二因为它将数据转换逻辑集中在编排定义中更清晰且不影响基础技能。实现时需要一套灵活的表达式语言来支持这种映射。6.3 LLM生成的计划不靠谱怎么办LLM可能会生成无法执行的计划比如技能不存在、输入输出类型不匹配、有循环依赖。后置验证与修复在LLM生成计划后增加一个“计划验证器”模块。它根据技能注册表中的元数据对计划进行静态分析检查上述问题。如果发现问题可以将错误信息反馈给LLM让其重新生成计划。这形成了一个“规划-验证-修正”的循环。提供更丰富的上下文在给LLM的Prompt中不仅提供技能列表还可以提供一些成功的计划示例、技能组合的模式Pattern、以及明确的约束如“不要创建循环依赖”、“确保技能B的输入来自技能A的输出”。分层规划不要让LLM一次性生成所有细节。可以先让它生成一个高级别的目标分解如“1. 获取数据2. 分析数据3. 生成报告”然后针对每个子目标再利用技能库进行更细粒度的规划。6.4 技能执行失败如何保证整体任务不崩溃这是分布式系统常见的可靠性问题。技能级别的重试与超时在每个技能的executor_config中配置重试次数、重试间隔建议使用指数退避和超时时间。对于非幂等的操作如创建订单重试要格外小心。流程级别的错误处理策略在复合技能或计划定义中为每个步骤或整个流程定义错误处理策略。例如“on_failure”: “continue”忽略当前步骤错误继续执行后续步骤。“on_failure”: “retry_step”重试当前步骤N次。“on_failure”: “fallback_to”: “alternative_skill_id”执行一个备用的降级技能。“on_failure”: “compensate”触发一个补偿事务尝试回滚之前步骤造成的影响。断路器模式Circuit Breaker对于频繁失败的外部服务技能在一段时间内停止调用它直接返回失败或降级结果避免雪崩效应。6.5 如何测试Formal Skill测试分为多个层次单元测试测试每个技能实现函数本身的正确性与普通函数测试无异。集成测试测试技能在注册、被执行器调用时的端到端流程。需要Mock外部依赖如HTTP请求、数据库。编排测试测试特定的技能组合计划是否能正确执行并产生预期结果。可以编写YAML或JSON格式的测试用例定义输入、计划和期望输出。LLM规划测试这是最复杂的。需要构建一个测试集包含各种用户查询然后运行完整的系统LLM规划器编排引擎评估最终结果的正确性和计划的合理性。这更多是评估整个系统的“智能”水平。Formal Skill和Programmable Runtime Skills不是银弹它引入了一定的复杂性但换来的是智能体行为的确定性、可维护性、可复用性和可观测性的巨大提升。它本质上是一种软件工程思想在LLM应用领域的实践将智能体的“智能”更多地体现在高层的规划和编排上而将确定性的、可靠的操作下沉到一个个定义清晰、测试充分的“技能”中。对于构建严肃的、用于生产环境的LLM智能体应用我认为这是一条必经之路。从简单的技能注册表开始逐步引入编排、规划、监控你的智能体系统会像搭积木一样变得越来越强大和稳健。