ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

DeepSeek-Agent-Harness-2026终极指南-第8章第36节-AgentLoop从零实现-工具注册中心:装饰器与Schema自动生成

DeepSeek-Agent-Harness-2026终极指南-第8章第36节-AgentLoop从零实现-工具注册中心:装饰器与Schema自动生成 DeepSeek Agent Harness 2026终极指南 - 第8章第36节 工具注册中心装饰器与Schema自动生成第35节的Agent Loop跑通了但工具系统还很原始——手动维护TOOLS_MAP字典、手写TOOLS_SCHEMA JSON。每加一个工具就要写两份代码参数类型变了还要同步改JSON。这节用装饰器pydantic反射彻底重构tool装饰器自动注册函数、model_json_schema()自动生成JSON Schema。从此加工具只需要一个装饰器三行代码搞定。本文导航手工注册的三个痛点tool装饰器一行代码注册函数pydantic反射从类型注解到JSON SchemaToolRegistry全局注册中心完整实现tool_registry.py实测三行代码加新工具小结手工注册的三个痛点第35节的tools.py长这样# 第35节版本——手工注册defget_weather(city:str)-str:获取指定城市的天气returnf晴天28°CTOOLS_MAP{get_weather:get_weather,# 手动维护映射}TOOLS_SCHEMA[{type:function,function:{name:get_weather,description:获取指定城市今天的天气信息...,parameters:{type:object,properties:{city:{type:string,description:城市中文名例如北京、上海,}},required:[city],},},}]三个痛点痛点1重复劳动。函数名写一遍def get_weather、TOOLS_MAP写一遍get_weather: get_weather、TOOLS_SCHEMA再写一遍name: get_weather。改函数名要改三处漏改就崩。痛点2参数类型不同步。函数签名是city: strJSON Schema里是type: string。如果我改成city: int虽然不合理但假设JSON Schema不会自动更新模型传过来的参数类型就对不上。痛点3描述维护成本。函数的docstring是给人看的TOOLS_SCHEMA里的description是给模型看的。两套描述要分别维护写漏了模型就不知道工具干嘛。解决方案装饰器 pydantic反射。tool装饰器一行代码注册函数Python的装饰器本质是一个函数接收被装饰的函数作为参数返回一个新的函数。我们用装饰器做两件事注册把函数存到全局registry里元数据提取从函数签名和docstring提取工具描述先写一个最简版本感受一下# 最简装饰器——只做注册不提取元数据_registry{}deftool(func):装饰器注册函数到全局工具表_registry[func.__name__]funcreturnfunc# 原样返回不改函数行为tooldefget_weather(city:str)-str:获取指定城市的天气returnf晴天28°Cprint(_registry)# 输出{get_weather: function get_weather at 0x...}装饰器tool在函数定义时立即执行把函数对象存到_registry字典里。从此_registry[get_weather]就是get_weather函数本身。但这只解决了注册问题还没解决描述问题。模型需要知道工具名get_weather→ 从func.__name__取工具描述“获取指定城市的天气”→ 从func.__doc__取参数列表city: str→ 从func.__annotations__取参数JSON Schema → 需要pydantic反射pydantic反射从类型注解到JSON Schemapydantic的BaseModel有一个杀手级方法model_json_schema()。它能从类型注解自动生成符合JSON Schema标准的字典。frompydanticimportBaseModel,FieldclassGetWeatherParams(BaseModel):获取指定城市的天气参数city:strField(description城市中文名例如北京、上海)schemaGetWeatherParams.model_json_schema()print(schema)# 输出# {# type: object,# properties: {# city: {# type: string,# description: 城市中文名例如北京、上海# }# },# required: [city],# title: GetWeatherParams# }pydantic自动做了三件事从city: str推断type: string从Field(description...)提取描述从是否有默认值推断required列表但这里有个问题我不想为每个工具都写一个XxxParams类。get_weather要写GetWeatherParamsread_file要写ReadFileParams……又是重复劳动。解决方案动态生成pydantic模型。用create_model()从函数签名动态构建frompydanticimportcreate_modelimportinspectdefget_weather(city:str):获取指定城市的天气returnf晴天28°C# 从函数签名提取参数siginspect.signature(get_weather)fields{}forname,paraminsig.parameters.items():# param.annotation 是类型注解如 str# param.default 是默认值如有fields[name](param.annotation,...)# ... 表示必填# 动态创建 pydantic 模型ParamsModelcreate_model(f{get_weather.__name__}_Params,**fields)schemaParamsModel.model_json_schema()print(schema)# 输出# {# type: object,# properties: {# city: {type: string}# },# required: [city],# title: get_weather_Params# }create_model()的第一个参数是模型名第二个参数**fields是字段定义字典。每个字段的值是一个元组(type, default)...表示必填。这样我们就能从任意函数签名自动生成JSON Schema不需要手写pydantic类。ToolRegistry全局注册中心把装饰器、反射、注册整合成一个ToolRegistry类# deep_pilot/tool_registry.py —— 工具注册中心 v0.3from__future__importannotationsimportinspectimportjsonfromtypingimportAny,Callable,get_type_hintsfrompydanticimportBaseModel,Field,create_modelfromdeep_pilot.loggerimportget_logger loggerget_logger(__name__)classToolRegistry:工具注册中心——管理所有可用工具def__init__(self):self._tools:dict[str,dict[str,Any]]{}deftool(self,func:Callable)-Callable:装饰器注册函数为工具namefunc.__name__ descriptioninspect.getdoc(func)orf工具{name}# 从函数签名生成参数模型siginspect.signature(func)fields{}forparam_name,paraminsig.parameters.items():ifparam_namein(self,cls):continue# 类型注解默认 Anytype_hintparam.annotationifparam.annotation!inspect.Parameter.emptyelseAny# 默认值... 表示必填defaultparam.defaultifparam.default!inspect.Parameter.emptyelse...fields[param_name](type_hint,default)# 动态创建 pydantic 模型params_modelcreate_model(f{name}_Params,**fields)# 生成 JSON Schemaschemaparams_model.model_json_schema()# 移除 title 字段OpenAI 协议不需要schema.pop(title,None)forpropinschema.get(properties,{}).values():prop.pop(title,None)# 存储工具信息self._tools[name]{func:func,description:description,params_model:params_model,schema:schema,}logger.debug(f注册工具:{name}参数: {list(schema.get(properties, {}).keys())})returnfuncdefget_tool(self,name:str)-dict[str,Any]|None:获取工具信息returnself._tools.get(name)deflist_tools(self)-list[str]:列出所有已注册工具名returnlist(self._tools.keys())defto_openai_tools(self)-list[dict[str,Any]]:生成符合 OpenAI 协议的工具描述列表tools[]forname,infoinself._tools.items():tools.append({type:function,function:{name:name,description:info[description],parameters:info[schema],},})returntoolsdefexecute(self,name:str,arguments:dict[str,Any])-Any:执行工具tool_infoself._tools.get(name)ifnottool_info:raiseValueError(f未知工具:{name})functool_info[func]params_modeltool_info[params_model]# 用 pydantic 校验参数validated_paramsparams_model(**arguments)# 调用函数resultfunc(**validated_params.model_dump())returnresult# 全局单例registryToolRegistry()toolregistry.tool这个ToolRegistry做了四件事tool()装饰器从函数签名提取参数动态创建pydantic模型生成JSON Schema注册到_tools字典。to_openai_tools()生成符合OpenAI协议的工具描述列表直接传给client.chat(tools...)。execute()用pydantic校验参数调用函数返回结果。全局单例registry整个DeepPilot共用一个注册中心。完整实现tool_registry.py上面已经给出了完整代码约120行。这里补充几个细节参数类型推断type_hintparam.annotationifparam.annotation!inspect.Parameter.emptyelseAny如果函数签名有类型注解如city: str就用注解没有就用Any。必填 vs 可选defaultparam.defaultifparam.default!inspect.Parameter.emptyelse......Ellipsis在pydantic里表示必填字段。如果函数参数有默认值如timeout: int 30pydantic会自动把它从required列表移除。Schema清理schema.pop(title,None)forpropinschema.get(properties,{}).values():prop.pop(title,None)pydantic生成的Schema里每个字段都有title如title: City但OpenAI协议不需要这个字段手动移除。实测三行代码加新工具现在用新的注册中心重写第35节的get_weather再加一个新工具get_time# deep_pilot/tools.py —— v0.3 重构版fromdeep_pilot.tool_registryimporttoolfromdatetimeimportdatetimetooldefget_weather(city:str)-str:获取指定城市今天的天气信息。输入城市中文名返回天气描述。weather_db{北京:晴28°C湿度 45%北风 3 级,上海:多云32°C湿度 70%东南风 2 级,深圳:雷阵雨26°C湿度 85%西南风 4 级,}returnweather_db.get(city,f未找到{city}的天气数据)tooldefget_time()-str:获取当前系统时间。无需参数返回格式化的日期时间字符串。returndatetime.now().strftime(%Y-%m-%d %H:%M:%S)# 就这些。不需要维护 TOOLS_MAP不需要手写 TOOLS_SCHEMA。三行代码加一个新工具tool装饰器 函数定义 docstring。现在修改agent_loop.py用新的注册中心# deep_pilot/agent_loop.py —— 第36节重构版fromdeep_pilot.clientimportclientfromdeep_pilot.loggerimportget_loggerfromdeep_pilot.tool_registryimportregistry# 导入 tools.py 触发装饰器注册importdeep_pilot.tools# noqa: F401loggerget_logger(__name__)MAX_ITERS10defrun(user_query:str)-str:Agent Loop 主入口messages[{role:system,content:你是一个有用的助手。当需要查询实时数据时使用提供的工具。},{role:user,content:user_query},]# 从注册中心生成工具描述toolsregistry.to_openai_tools()foriterationinrange(1,MAX_ITERS1):logger.info(fLoop 第{iteration}轮 ↻)respclient.chat(messages,toolstools)ifnotresp.tool_calls:returnresp.contentormessages.append({role:assistant,content:resp.content,tool_calls:resp.tool_calls,})fortcinresp.tool_calls:func_nametc[function][name]func_argsjson.loads(tc[function][arguments])logger.info(f → 调用工具:{func_name}({json.dumps(func_args,ensure_asciiFalse)}))try:resultregistry.execute(func_name,func_args)ifnotisinstance(result,str):resultjson.dumps(result,ensure_asciiFalse)exceptExceptionase:resultf工具执行失败:{e}logger.info(f ← 工具结果:{result[:80]}{...iflen(result)80else})messages.append({role:tool,tool_call_id:tc[id],content:result,})logger.warning(fAgent Loop 达到最大迭代次数{MAX_ITERS})return[Agent] 抱歉处理超时。改动很小删掉from deep_pilot.tools import TOOLS_MAP, TOOLS_SCHEMA加from deep_pilot.tool_registry import registry和import deep_pilot.toolstools registry.to_openai_tools()替代硬编码的TOOLS_SCHEMAregistry.execute(func_name, func_args)替代手动查字典调用实测一下uv run python-c from deep_pilot.agent_loop import run # 测试天气工具 print( 测试 get_weather ) answer run(北京今天天气怎么样) print(f最终答案: {answer}) print() # 测试时间工具 print( 测试 get_time ) answer run(现在几点了) print(f最终答案: {answer}) print() # 测试多工具组合 print( 测试多工具组合 ) answer run(北京天气怎么样顺便告诉我现在几点了) print(f最终答案: {answer}) 控制台输出2026-09-12 17:00:01 | INFO | agent_loop | Loop 第 1 轮 ↻ 2026-09-12 17:00:01 | INFO | client | 调用留痕 | call_id777aaa... | ... 2026-09-12 17:00:01 | INFO | agent_loop | → 调用工具: get_weather({city: 北京}) 2026-09-12 17:00:01 | INFO | agent_loop | ← 工具结果: 晴28°C湿度 45%北风 3 级 2026-09-12 17:00:01 | INFO | agent_loop | Loop 第 2 轮 ↻ 2026-09-12 17:00:01 | INFO | client | 调用留痕 | call_id888bbb... | ... 最终答案: 北京今天天气晴28°C湿度 45%北风 3 级。 测试 get_time 2026-09-12 17:00:02 | INFO | agent_loop | Loop 第 1 轮 ↻ 2026-09-12 17:00:02 | INFO | client | 调用留痕 | call_id999ccc... | ... 2026-09-12 17:00:02 | INFO | agent_loop | → 调用工具: get_time({}) 2026-09-12 17:00:02 | INFO | agent_loop | ← 工具结果: 2026-09-12 17:00:02 2026-09-12 17:00:02 | INFO | agent_loop | Loop 第 2 轮 ↻ 2026-09-12 17:00:02 | INFO | client | 调用留痕 | call_idaaaddd... | ... 最终答案: 现在是 2026年9月12日 17:00:02。 测试多工具组合 2026-09-12 17:00:03 | INFO | agent_loop | Loop 第 1 轮 ↻ 2026-09-12 17:00:03 | INFO | client | 调用留痕 | call_idbbbeee... | ... 2026-09-12 17:00:03 | INFO | agent_loop | → 调用工具: get_weather({city: 北京}) 2026-09-12 17:00:03 | INFO | agent_loop | ← 工具结果: 晴28°C 2026-09-12 17:00:03 | INFO | agent_loop | Loop 第 2 轮 ↻ 2026-09-12 17:00:03 | INFO | client | 调用留痕 | call_idcccfff... | ... 2026-09-12 17:00:03 | INFO | agent_loop | → 调用工具: get_time({}) 2026-09-12 17:00:03 | INFO | agent_loop | ← 工具结果: 2026-09-12 17:00:03 2026-09-12 17:00:03 | INFO | agent_loop | Loop 第 3 轮 ↻ 2026-09-12 17:00:03 | INFO | client | 调用留痕 | call_iddddggg... | ... 最终答案: 北京今天天气晴28°C。现在是 2026年9月12日 17:00:03。三个测试都通过了单工具调用模型识别需要查天气调get_weather拿到结果后回答无参数工具get_time不需要参数模型传空字典{}pydantic校验通过多工具组合模型先查天气再查时间最后综合回答小结装饰器反射替代手工注册tool自动注册函数model_json_schema()自动生成JSON Schema消除重复劳动。动态创建pydantic模型用create_model()从函数签名动态构建参数模型不需要为每个工具写单独的类。ToolRegistry统一管理to_openai_tools()生成工具描述execute()校验参数并执行全局单例registry。三行代码加新工具tool装饰器 函数定义 docstring从此加工具不再手写JSON。参数类型自动同步函数签名改类型JSON Schema自动更新模型传参类型永远对得上。DeepPilot v0.3工具系统升级完成——从手工注册到装饰器反射开发体验大幅提升。下节预告工具注册中心搞定了但registry.execute()里的异常处理还很粗糙——工具抛异常就原样回填给模型模型看不懂Python的traceback。下一节做工具执行器用pydantic校验参数、捕获异常并转换为模型友好的错误消息、把工具结果格式化为token友好的字符串。从此工具执行失败时模型能理解参数类型错了还是文件不存在而不是看到一坨TypeError: ...。如果觉得本文对你有帮助欢迎点赞、收藏、关注三连本系列持续更新中关注不迷路~
返回列表