1. 从300行到30行一次Agent技能开发的效率革命最近在集中开发一批AI Agent的技能Skill当写到第10个的时候我停下来看了看代码仓库。一个让我自己都感到惊讶的事实摆在眼前最初几个技能每个的核心逻辑部分都洋洋洒洒写了近300行代码结构臃肿重复度极高。而到了最近完成的几个技能同样的功能核心代码被压缩到了30行左右并且结构清晰可维护性大大提升。这不仅仅是代码行数的减少更是一次开发范式、设计思路和工程效率的全面升级。如果你也在进行Agent技能开发或者任何具有相似模式输入-处理-输出的模块化功能开发那么这次从“手工作坊”到“标准化流水线”的演进过程或许能给你带来一些直接的启发和可以“抄作业”的解决方案。Agent技能本质上是一个个封装好的能力单元。它接收来自Agent核心的请求通常是一个明确的用户指令或目标调用必要的工具或API处理数据最后返回一个结构化的结果。听起来很简单对吧但在实际开发中魔鬼藏在细节里。每个技能似乎都有其特殊性有的需要调用外部搜索API并解析网页有的需要查询数据库并做数据聚合有的则是执行一个复杂的计算流程。最初我很自然地针对每个技能的特殊性“量身定制”了全套代码——从参数解析、错误处理、工具调用、结果格式化到最终返回。这就导致了大量重复的“样板代码”Boilerplate Code真正独特的业务逻辑反而被淹没其中调试和扩展都成了噩梦。2. 拆解原始300行重复代码的“重灾区”在哪里要解决问题首先得看清问题。我把最初几个技能的近300行代码打印出来用不同颜色的笔标出了功能区块。一番分析后我发现重复代码主要集中在以下几个“重灾区”这些地方几乎在每个技能中都以高度相似的形式出现但之前却被我视作“理所当然”的成本。2.1 输入参数的验证与解析每个技能都需要接收输入。比如一个“天气查询”技能需要city和date参数一个“股票查询”技能需要stock_code参数。在最原始的写法里我会在每个技能的开头写一大段类似这样的代码def weather_skill(params): # 参数提取与验证 city params.get(city) if not city: return {error: Missing required parameter: city} if not isinstance(city, str): return {error: Parameter city must be a string} date params.get(date, today) # 验证date格式是否为YYYY-MM-DD或‘today’/‘tomorrow’ if date not in [today, tomorrow]: try: datetime.strptime(date, %Y-%m-%d) except ValueError: return {error: Parameter date must be today, tomorrow or in YYYY-MM-DD format} # ... 后续业务逻辑这段代码干了三件事1. 检查参数是否存在2. 检查参数类型3. 对某些参数进行格式或取值验证。问题在于除了参数名和具体的验证规则如日期格式不同整个代码的结构、错误返回的格式一个包含error键的字典完全一样。10个技能我就把几乎相同的代码复制粘贴了10遍只改了改变量名和正则表达式。这不仅浪费时间更致命的是一旦我想统一错误信息的格式比如从{error: ...}改成{code: 400, message: ...}就需要修改10个文件极易出错。2.2 外部服务调用的错误处理与重试技能免不了要调用外部API、数据库或其它服务。网络波动、服务暂时不可用、接口限流等情况时有发生。原始代码中我会在每次调用后手动进行异常捕获和重试import requests from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def call_weather_api(city, date): try: response requests.get(fhttps://api.weather.com/v1/{city}?date{date}, timeout5) response.raise_for_status() # 检查HTTP状态码 return response.json() except requests.exceptions.Timeout: raise Exception(Weather API request timeout) except requests.exceptions.HTTPError as e: if e.response.status_code 404: raise Exception(fCity {city} not found) else: raise Exception(fWeather API error: {e}) except requests.exceptions.RequestException as e: raise Exception(fNetwork error: {e})这段代码包含了1. 装饰器实现的自动重试机制2. 对超时、HTTP错误如404、网络异常等不同情况的精细化捕获和处理3. 统一的异常转换将各类异常转换为带明确信息的Exception。思考一下对于“股票API”、“新闻搜索API”、“数据库查询”这段代码的骨架是不是几乎一样变化的仅仅是URL、参数、以及针对特定HTTP状态码如404可能意味着股票代码错误的错误信息。重复编写这些健壮性代码是另一个主要的效率瓶颈和潜在bug来源。2.3 结果数据的标准化格式化与返回技能处理完数据后需要以Agent核心能理解的固定格式返回。假设我们要求返回{success: bool, data: any, message: str}这样的结构。原始代码中每个技能的最后都会这样写# ... 业务逻辑处理得到 result_data if some_error_condition: return {success: False, data: None, message: Specific error message} else: return {success: True, data: result_data, message: Query successful}同样除了result_data的内容和特定的错误信息这个返回结构的构建代码又是一模一样的。更复杂的是有时需要对result_data进行后处理比如过滤掉敏感信息、将日期时间对象序列化为字符串、或者当数据为空时提供一个默认值。这些格式化逻辑如果散落在各个技能里就会导致标准不一致后续的Agent核心处理起来会很麻烦。2.4 技能元信息的声明为了让Agent核心能够动态发现和调用技能每个技能通常需要提供一个元信息metadata字典说明自己的名称、描述、所需参数等。skill_metadata { name: get_weather, description: Get the weather forecast for a specific city and date., parameters: { city: {type: string, description: The city name, required: True}, date: {type: string, description: Date in YYYY-MM-DD format or today/tomorrow, required: False, default: today} } }这部分信息是静态的但为每个技能手动编写这个字典并确保parameters的定义与技能函数内部的参数解析逻辑保持一致是一项枯燥且易错的工作。当你想修改技能描述或增加一个可选参数时需要在两个地方元信息和函数逻辑同步更新。提示当你发现自己在多个地方编写结构高度相似、只有细节不同的代码时这就是一个强烈的抽象信号。抽象的目标不是追求极致的代码简短而是将“不变的流程”和“变化的部分”分离。3. 构建技能框架抽象“不变”隔离“变化”看清了重复模式解决方案就呼之欲出了我们需要一个基础的技能框架Skill Framework或者叫技能基类Base Skill Class。这个框架负责处理所有那些“不变”的通用逻辑而将“变化”的业务逻辑留给具体的技能去实现。这是软件工程中“模板方法模式”和“依赖倒置原则”的经典应用。我构建的这个框架核心包含以下几个部分3.1 基于Pydantic的声明式参数验证我放弃了手动编写if-else进行参数验证的方式转而使用Pydantic。Pydantic是一个利用Python类型注解进行数据验证和设置管理的库。我为每个技能定义一个继承自pydantic.BaseModel的参数模型InputModel。from pydantic import BaseModel, Field, validator from datetime import datetime class WeatherSkillInput(BaseModel): city: str Field(..., descriptionThe city name) date: str Field(defaulttoday, descriptionDate in YYYY-MM-DD format or today/tomorrow) validator(date) def validate_date(cls, v): if v in [today, tomorrow]: return v try: datetime.strptime(v, %Y-%m-%d) return v except ValueError: raise ValueError(date must be today, tomorrow or in YYYY-MM-DD format)这样做的好处是巨大的验证与解析合一只需一行input_data WeatherSkillInput(**params)就完成了参数的存在性、类型、自定义格式的验证。如果验证失败Pydantic会抛出包含详细错误信息的ValidationError异常。自文档化Field中的description直接可以作为技能元信息的一部分保证了文档和代码的一致性。类型安全在IDE中可以获得良好的类型提示和自动补全。在技能基类中我只需要提供一个通用的方法来加载输入模型并执行验证所有技能都复用这一套机制。3.2 统一的客户端与可配置的重试策略我将所有对外部服务的调用抽象成一个个“客户端”Client。例如WeatherAPIClient、StockAPIClient、DatabaseClient。这些客户端继承自一个BaseClient。from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import requests class BaseClient: def __init__(self, base_url: str, timeout: int 5): self.base_url base_url self.timeout timeout self.session requests.Session() # 使用session保持连接提升性能 retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10), retryretry_if_exception_type((requests.exceptions.Timeout, requests.exceptions.ConnectionError)) ) def _request(self, method, endpoint, **kwargs): url f{self.base_url}{endpoint} kwargs.setdefault(timeout, self.timeout) try: resp self.session.request(method, url, **kwargs) resp.raise_for_status() return resp except requests.exceptions.HTTPError as e: # 这里可以记录日志或根据状态码抛出更具体的业务异常 raise Exception(fAPI Error [{e.response.status_code}]: {e.response.text}) # Timeout和ConnectionError会被tenacity自动重试然后具体的客户端只需关注如何构建自己领域的请求参数和解析响应class WeatherAPIClient(BaseClient): def get_forecast(self, city: str, date: str): # 构建特定于天气API的请求逻辑 params {city: city, date: date} response self._request(GET, /v1/forecast, paramsparams) return self._parse_forecast_response(response.json()) def _parse_forecast_response(self, raw_data): # 解析API响应返回结构化的数据 # 例如{temperature: 22, condition: Sunny} pass关键点重试策略、超时设置、基础错误处理全部在BaseClient中定义一次。不同的技能只需注入配置好的客户端实例即可。如果未来需要调整重试次数比如从3次改为2次或者增加对某种新异常的重试只需要修改BaseClient一处。3.3 标准化的执行流程与结果包装这是技能基类的核心。我定义了一个BaseSkill类它规定了技能执行的固定流程from abc import ABC, abstractmethod from pydantic import ValidationError class BaseSkill(ABC): # 由子类定义的属性 name: str description: str input_model: type[BaseModel] # 指向对应的Pydantic模型类 def execute(self, raw_input: dict) - dict: 技能执行的统一入口 # 1. 参数验证 try: validated_input self.input_model(**raw_input) except ValidationError as e: return self._format_error(fInput validation failed: {e.errors()}) # 2. 执行业务逻辑 try: result_data self._execute_logic(validated_input) except Exception as e: # 这里可以记录详细的异常日志便于排查 return self._format_error(fSkill execution error: {str(e)}) # 3. 格式化并返回成功结果 return self._format_success(result_data) abstractmethod def _execute_logic(self, validated_input: BaseModel) - any: 子类必须实现的抽象方法具体的业务逻辑 pass def _format_success(self, data): 统一格式化成功响应 return { success: True, data: self._post_process_data(data), message: Operation completed successfully } def _format_error(self, error_msg): 统一格式化错误响应 return { success: False, data: None, message: error_msg } def _post_process_data(self, data): 可选的数据后处理钩子如序列化、过滤等 # 默认实现直接返回数据。子类可重写。 return data def get_metadata(self): 自动生成技能元信息基于input_model的schema schema self.input_model.schema() parameters {} for prop_name, prop_info in schema.get(properties, {}).items(): parameters[prop_name] { type: prop_info.get(type, string), description: prop_info.get(description, ), required: prop_name in schema.get(required, []) } return { name: self.name, description: self.description, parameters: parameters }这个BaseSkill类做了几件关键事情流程控制execute方法定义了“验证输入 - 执行业务 - 格式化输出”的标准流程。错误隔离将输入验证错误和业务逻辑错误分开处理并统一格式。模板方法_execute_logic是抽象方法强制子类只关注最核心的业务代码。元信息自动化get_metadata方法利用Pydantic模型的schema()自动生成参数定义彻底避免了手动维护元信息字典的麻烦。4. 新范式实战30行实现一个健壮的天气技能现在让我们看看在新的框架下实现一个完整的“天气查询”技能有多么简洁。假设我们已经有了一个配置好的WeatherAPIClient实例。# weather_skill.py from pydantic import BaseModel, Field, validator from datetime import datetime from .base_skill import BaseSkill from .clients import weather_client # 导入预配置的客户端 # 1. 定义输入模型约10行 class WeatherInput(BaseModel): city: str Field(..., descriptionThe name of the city to query) date: str Field(defaulttoday, descriptionDate as today, tomorrow, or YYYY-MM-DD) validator(date) def validate_date_format(cls, v): if v in [today, tomorrow]: return v try: datetime.strptime(v, %Y-%m-%d) return v except ValueError: raise ValueError(Invalid date format. Use today, tomorrow, or YYYY-MM-DD.) # 2. 实现技能类约20行 class WeatherSkill(BaseSkill): name get_weather description Fetches the weather forecast for a given city and date. input_model WeatherInput # 关联输入模型 def __init__(self): # 可以在这里注入依赖比如不同的天气API客户端 self.client weather_client def _execute_logic(self, validated_input: WeatherInput) - dict: 核心业务逻辑调用API并解析数据 # 调用客户端所有重试、错误处理已在客户端内部完成 forecast_data self.client.get_forecast( cityvalidated_input.city, datevalidated_input.date ) # 对原始API返回的数据进行必要的业务转换 # 例如只提取我们关心的字段 formatted_result { location: f{validated_input.city}, date: validated_input.date, temperature: f{forecast_data[temp]}°C, condition: forecast_data[weather][0][description], humidity: f{forecast_data[humidity]}% } return formatted_result # 可选如果需要特殊的数据后处理可以重写_post_process_data # def _post_process_data(self, data): # data[fetch_time] datetime.now().isoformat() # return data代码行数分析输入模型定义 (WeatherInput)约10行包含验证器。技能类定义 (WeatherSkill)约20行。其中name、description、input_model的声明占3行__init__占2行核心的_execute_logic方法约15行。总计约30行。这30行代码做了什么它实现了一个具备完整参数验证、自动重试、统一错误处理、标准化返回格式的健壮技能。所有繁琐的“脚手架”代码都被基类和客户端承载了。作为开发者你只需要关心两件事1. 我的技能需要哪些参数定义Pydantic模型 2. 拿到验证好的参数后我的核心业务逻辑是什么实现_execute_logic。5. 效率提升之外的隐性收益与进阶思考将代码从300行压缩到30行最直观的收益是开发速度的飞跃。但这次重构带来的价值远不止于此一些隐性的、长期的收益对于项目维护和团队协作更为重要。5.1 一致性与可维护性修改只需一处当产品经理提出“所有技能的错误信息里都要加上错误发生的时间戳”这个需求时在旧模式下我需要检查并修改10个文件。而在新框架下我只需要修改BaseSkill._format_error方法可能只需要增加一行代码def _format_error(self, error_msg): import time return { success: False, data: None, message: error_msg, timestamp: int(time.time()) # 新增 }所有技能立即生效。这种“单一修改点”的特性极大地降低了维护成本和出错概率。同样如果想升级重试库tenacity的版本或者调整默认的超时时间也只需要在BaseClient中修改一次。5.2 可测试性的质变在旧模式下测试一个技能非常困难。你需要模拟整个执行流程包括参数验证、API调用、错误处理等。测试代码往往和技能代码一样冗长。在新框架下测试变得极其聚焦和简单。由于输入验证、错误处理等都被框架接管并视为可靠你可以集中精力测试_execute_logic这个纯函数或近似纯函数因为它依赖注入的客户端。你可以轻松地用模拟Mock对象替换真实的weather_client从而对业务逻辑进行各种边界条件和异常场景的单元测试。# 测试样例 def test_weather_skill_logic_success(): skill WeatherSkill() skill.client Mock() # 模拟客户端 skill.client.get_forecast.return_value {temp: 22, weather: [{description: clear sky}] humidity: 65} input_data WeatherInput(cityBeijing, datetoday) result skill._execute_logic(input_data) assert result[temperature] 22°C assert result[condition] clear sky skill.client.get_forecast.assert_called_once_with(cityBeijing, datetoday)5.3 技能的动态发现与注册由于每个技能都提供了标准的get_metadata方法Agent核心可以非常方便地实现技能的动态发现和加载。你可以写一个简单的注册机制class SkillRegistry: def __init__(self): self._skills {} def register(self, skill_class): instance skill_class() self._skills[instance.name] instance def get_skill(self, name): return self._skills.get(name) def list_skills(self): return [skill.get_metadata() for skill in self._skills.values()] # 自动发现并注册所有技能 registry SkillRegistry() for skill_class in [WeatherSkill, StockSkill, NewsSearchSkill]: # 可以自动扫描模块 registry.register(skill_class) # Agent核心可以这样调用 skill_name get_weather params {city: Shanghai} skill registry.get_skill(skill_name) if skill: result skill.execute(params)这使得你的Agent系统变得非常灵活新增一个技能只需要开发并注册核心调度代码完全不用改动。5.4 进阶优化中间件与执行钩子当框架稳定后你可以进一步引入“中间件”或“钩子”机制来处理横切关注点。例如日志记录在每个技能执行前后自动记录日志。性能监控自动统计每个技能的耗时。权限校验在执行前检查调用者是否有权限使用该技能。输入/输出审计记录原始输入和最终输出注意脱敏。这可以通过在BaseSkill.execute方法中插入钩子函数或使用装饰器模式来实现。这再次证明了良好抽象带来的强大扩展能力。从10个技能、3000行重复代码的泥潭中走出来构建一个30行实现一个技能的标准化框架这个过程让我深刻体会到“磨刀不误砍柴工”的真谛。前期在设计和抽象上投入的时间在第三个、第四个技能开发时就开始产生巨大的回报。更重要的是它让代码库变得清晰、健壮和易于扩展。如果你也在进行类似的多功能模块开发不妨停下来审视一下那些重复的“样板代码”它们很可能就是你下一个效率突破的关键。