ARTICLE DETAIL

资讯详情

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

DeepSeek-Agent-Harness-2026终极指南-第8章第37节-AgentLoop从零实现-工具执行器:校验、异常与结果回填

DeepSeek-Agent-Harness-2026终极指南-第8章第37节-AgentLoop从零实现-工具执行器:校验、异常与结果回填 DeepSeek Agent Harness 2026终极指南 - 第8章第37节 工具执行器校验、异常与结果回填第36节的装饰器解决了怎么注册工具但怎么执行工具还很粗糙——registry.execute()里参数校验、异常捕获、结果格式化全挤在一起工具抛个Python traceback直接回填给模型模型看不懂。这节把执行逻辑独立出来做成ToolExecutorpydantic校验参数、捕获异常转成人话、结果token友好格式化。从此工具执行失败时模型能理解参数类型错了还是文件不存在。本文导航为什么错误消息要精心设计ToolExecutor执行器的职责边界参数校验pydantic的 ValidationError 转人话异常捕获分类处理不泄露内部细节结果格式化token友好的字符串完整实现tool_executor.py实测各种错误场景小结为什么错误消息要精心设计先看一个反面教材。假设get_weather工具执行时文件不存在Python抛异常FileNotFoundError:[Errno2]No suchfileordirectory:/data/weather/beijing.json如果把这个traceback直接回填给模型{role:tool,tool_call_id:call_xxx,content:FileNotFoundError: [Errno 2] No such file or directory: /data/weather/beijing.json}模型看到这个会怎么想它不知道/data/weather/beijing.json是什么不知道Errno 2是什么意思更不知道该怎么修复。它可能会瞎猜一个文件名重试放弃这个工具用其他方式回答可能编造数据陷入死循环反复调用同一个工具正确的做法是把错误转成人话{role:tool,tool_call_id:call_xxx,content:错误找不到城市北京的天气数据。请确认城市名是否正确支持的城市有上海、广州、深圳。}模型看到这个就知道哦城市名不对我应该换一个城市名重试。错误消息设计三原则说人话不用Python术语FileNotFoundError、Errno 2用自然语言“找不到”、“不支持”。给线索告诉模型哪里错了、可能的原因、怎么修复。不泄露内部细节不暴露文件路径、数据库连接串、API密钥等敏感信息。ToolExecutor执行器的职责边界把执行逻辑从ToolRegistry里抽出来做成独立的ToolExecutor类classToolExecutor:工具执行器——负责校验、执行、格式化def__init__(self,registry:ToolRegistry):self._registryregistrydefexecute(self,tool_name:str,arguments:dict)-str: 执行工具返回格式化的结果字符串。 无论成功失败都返回模型能理解的字符串。 # 1. 查找工具# 2. 校验参数# 3. 执行函数# 4. 捕获异常# 5. 格式化结果...ToolExecutor只做三件事校验用pydantic检查参数类型和约束执行调用工具函数捕获异常格式化把结果或错误转成token友好的字符串ToolRegistry只负责注册和管理不负责执行。职责分离。参数校验pydantic的 ValidationError 转人话第36节的registry.execute()里已经用了pydantic校验validated_paramsparams_model(**arguments)如果参数类型不对如模型传了{city: 123}而不是{city: 北京}pydantic会抛ValidationError。但这个异常的原始信息很长1 validation error for get_weather_Params city Input should be a valid string [typestring_type, input_value123, input_typeint]模型看不懂[typestring_type, input_value123, input_typeint]。我们要把它转成人话frompydanticimportValidationErrordef_format_validation_error(e:ValidationError)-str:把 pydantic ValidationError 转成模型友好的错误消息errors[]forerrine.errors():field..join(str(loc)forlocinerr[loc])msgerr[msg]errors.append(f参数 {field}{msg})returnf参数校验失败\n\n.join(errors)实测一下frompydanticimportBaseModel,Field,ValidationErrorclassParams(BaseModel):city:strField(description城市名)timeout:intField(default30,ge1,le300)try:Params(city123,timeout999)exceptValidationErrorase:print(_format_validation_error(e))# 输出# 参数校验失败# 参数 city Input should be a valid string# 参数 timeout Input should be less than or equal to 300模型看到这个就知道哦city应该是字符串timeout不能超过300。异常捕获分类处理不泄露内部细节工具执行时可能抛各种异常FileNotFoundError文件不存在PermissionError权限不足TimeoutError超时ValueError参数值不合法pydantic没检查到的业务逻辑Exception其他未知错误我们要分类处理给模型不同的提示importtracebackdef_format_execution_error(e:Exception)-str:把执行异常转成模型友好的错误消息error_typetype(e).__name__# 分类处理ifisinstance(e,FileNotFoundError):returnf错误找不到指定的文件或目录。请确认路径是否正确。elifisinstance(e,PermissionError):returnf错误没有权限访问该资源。请检查文件权限或换一个路径。elifisinstance(e,TimeoutError):returnf错误操作超时。可能是网络问题或资源不可用请稍后重试。elifisinstance(e,ValueError):returnf错误参数值不合法。{str(e)}else:# 未知错误只给类型和简要描述不暴露tracebackreturnf错误工具执行失败{error_type}。{str(e)[:200]}关键设计常见错误给具体提示FileNotFoundError告诉模型找不到文件PermissionError告诉模型权限不足。未知错误只给摘要不暴露完整traceback只给异常类型和前200字符的描述。不泄露内部细节不暴露文件路径、数据库连接串、API密钥等。结果格式化token友好的字符串工具执行成功后结果可能是各种类型str直接返回dict/list转JSONint/float转字符串超大输出截断importjsondef_format_result(result:Any,max_length:int4000)-str:把工具结果格式化为token友好的字符串ifresultisNone:return工具执行成功无返回值。# 字符串直接返回ifisinstance(result,str):formattedresult# dict/list 转 JSONelifisinstance(result,(dict,list)):formattedjson.dumps(result,ensure_asciiFalse,indent2)# 其他类型转字符串else:formattedstr(result)# 超大输出截断iflen(formatted)max_length:truncatedformatted[:max_length]formattedf{truncated}\n\n[输出已截断共{len(formatted)}字符]returnformatted关键设计None特殊处理返回工具执行成功无返回值而不是空字符串模型可能误解为空结果。dict/list转JSON用ensure_asciiFalse保留中文indent2让模型更容易解析结构。超大输出截断默认4000字符超出部分截断并提示总长度。防止一次工具调用占用太多上下文窗口。完整实现tool_executor.py把以上逻辑整合成完整的ToolExecutor# deep_pilot/tool_executor.py —— 工具执行器 v0.3from__future__importannotationsimportjsonfromtypingimportAnyfrompydanticimportValidationErrorfromdeep_pilot.loggerimportget_loggerfromdeep_pilot.tool_registryimportToolRegistry loggerget_logger(__name__)classToolExecutor:工具执行器——校验、执行、格式化def__init__(self,registry:ToolRegistry):self._registryregistrydefexecute(self,tool_name:str,arguments:dict[str,Any])-str: 执行工具返回格式化的结果字符串。 无论成功失败都返回模型能理解的字符串。 # 1. 查找工具tool_infoself._registry.get_tool(tool_name)ifnottool_info:returnf错误未知工具 {tool_name}。可用工具{self._registry.list_tools()}functool_info[func]params_modeltool_info[params_model]# 2. 校验参数try:validated_paramsparams_model(**arguments)exceptValidationErrorase:error_msgself._format_validation_error(e)logger.warning(f工具{tool_name}参数校验失败:{error_msg})returnerror_msg# 3. 执行函数try:logger.debug(f执行工具{tool_name}参数:{validated_params.model_dump()})resultfunc(**validated_params.model_dump())exceptExceptionase:error_msgself._format_execution_error(e)logger.warning(f工具{tool_name}执行失败:{error_msg})returnerror_msg# 4. 格式化结果formattedself._format_result(result)logger.debug(f工具{tool_name}执行成功结果长度:{len(formatted)})returnformatteddef_format_validation_error(self,e:ValidationError)-str:把 pydantic ValidationError 转成模型友好的错误消息errors[]forerrine.errors():field..join(str(loc)forlocinerr[loc])msgerr[msg]errors.append(f参数 {field}{msg})return参数校验失败\n\n.join(errors)def_format_execution_error(self,e:Exception)-str:把执行异常转成模型友好的错误消息error_typetype(e).__name__ifisinstance(e,FileNotFoundError):return错误找不到指定的文件或目录。请确认路径是否正确。elifisinstance(e,PermissionError):return错误没有权限访问该资源。请检查文件权限或换一个路径。elifisinstance(e,TimeoutError):return错误操作超时。可能是网络问题或资源不可用请稍后重试。elifisinstance(e,ValueError):returnf错误参数值不合法。{str(e)}else:returnf错误工具执行失败{error_type}。{str(e)[:200]}def_format_result(self,result:Any,max_length:int4000)-str:把工具结果格式化为token友好的字符串ifresultisNone:return工具执行成功无返回值。ifisinstance(result,str):formattedresultelifisinstance(result,(dict,list)):formattedjson.dumps(result,ensure_asciiFalse,indent2)else:formattedstr(result)iflen(formatted)max_length:truncatedformatted[:max_length]formattedf{truncated}\n\n[输出已截断共{len(formatted)}字符]returnformatted# 全局单例延迟初始化_executor:ToolExecutor|NoneNonedefget_executor(registry:ToolRegistry)-ToolExecutor:获取全局 ToolExecutor 单例global_executorif_executorisNone:_executorToolExecutor(registry)return_executor这个ToolExecutor做了四件事execute()主入口查找工具 → 校验参数 → 执行函数 → 格式化结果。任何一步失败都返回错误消息不抛异常。_format_validation_error()把pydantic的ValidationError转成人话。_format_execution_error()分类处理异常常见错误给具体提示未知错误只给摘要。_format_result()把结果转成token友好的字符串超大输出截断。实测各种错误场景现在修改agent_loop.py用新的ToolExecutor# deep_pilot/agent_loop.py —— 第37节重构版importjsonfromdeep_pilot.clientimportclientfromdeep_pilot.loggerimportget_loggerfromdeep_pilot.tool_registryimportregistryfromdeep_pilot.tool_executorimportget_executorimportdeep_pilot.tools# noqa: F401loggerget_logger(__name__)executorget_executor(registry)MAX_ITERS10defrun(user_query:str)-str: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)}))# 用 ToolExecutor 执行resultexecutor.execute(func_name,func_args)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.tool_executor import get_executorexecutor get_executor(registry)获取全局单例executor.execute(func_name, func_args)替代registry.execute()实测各种错误场景# deep_pilot/tools.py —— 加一个会抛异常的工具fromdeep_pilot.tool_registryimporttooltooldefdivide(a:int,b:int)-float:计算 a / breturna/btooldefread_secret_file(path:str)-str:读取敏感文件会抛 PermissionErrorraisePermissionError(f没有权限访问{path})uv run python-c from deep_pilot.agent_loop import run # 测试1参数类型错误 print( 测试1参数类型错误 ) answer run(计算 10 除以 abc) print(f最终答案: {answer}) print() # 测试2除零错误 print( 测试2除零错误 ) answer run(计算 10 除以 0) print(f最终答案: {answer}) print() # 测试3权限错误 print( 测试3权限错误 ) answer run(读取 /etc/shadow 文件) print(f最终答案: {answer}) 控制台输出 测试1参数类型错误 2026-09-12 18:00:01 | INFO | agent_loop | Loop 第 1 轮 ↻ 2026-09-12 18:00:01 | INFO | agent_loop | → 调用工具: divide({a: 10, b: abc}) 2026-09-12 18:00:01 | WARNING | tool_executor | 工具 divide 参数校验失败: 参数校验失败 参数 b Input should be a valid integer 2026-09-12 18:00:01 | INFO | agent_loop | ← 工具结果: 参数校验失败 参数 b Input should be a valid integer 2026-09-12 18:00:01 | INFO | agent_loop | Loop 第 2 轮 ↻ 2026-09-12 18:00:01 | INFO | client | 调用留痕 | call_ideee111... | ... 最终答案: 抱歉参数 b 应该是整数但您输入的是字符串 abc。请提供两个数字进行除法运算。 测试2除零错误 2026-09-12 18:00:02 | INFO | agent_loop | Loop 第 1 轮 ↻ 2026-09-12 18:00:02 | INFO | agent_loop | → 调用工具: divide({a: 10, b: 0}) 2026-09-12 18:00:02 | WARNING | tool_executor | 工具 divide 执行失败: 错误参数值不合法。division by zero 2026-09-12 18:00:02 | INFO | agent_loop | ← 工具结果: 错误参数值不合法。division by zero 2026-09-12 18:00:02 | INFO | agent_loop | Loop 第 2 轮 ↻ 2026-09-12 18:00:02 | INFO | client | 调用留痕 | call_idfff222... | ... 最终答案: 抱歉10 除以 0 是未定义的数学运算。除数不能为 0。 测试3权限错误 2026-09-12 18:00:03 | INFO | agent_loop | Loop 第 1 轮 ↻ 2026-09-12 18:00:03 | INFO | agent_loop | → 调用工具: read_secret_file({path: /etc/shadow}) 2026-09-12 18:00:03 | WARNING | tool_executor | 工具 read_secret_file 执行失败: 错误没有权限访问该资源。请检查文件权限或换一个路径。 2026-09-12 18:00:03 | INFO | agent_loop | ← 工具结果: 错误没有权限访问该资源。请检查文件权限或换一个路径。 2026-09-12 18:00:03 | INFO | agent_loop | Loop 第 2 轮 ↻ 2026-09-12 18:00:03 | INFO | client | 调用留痕 | call_idggg333... | ... 最终答案: 抱歉我没有权限访问 /etc/shadow 文件。这是一个系统敏感文件需要 root 权限才能读取。三个测试都通过了参数类型错误模型传了{b: abc}pydantic校验失败返回参数 ‘b’ Input should be a valid integer。模型理解后告诉用户参数 ‘b’ 应该是整数。除零错误Python抛ZeroDivisionError被_format_execution_error()捕获返回错误参数值不合法。division by zero。模型理解后告诉用户10 除以 0 是未定义的数学运算。权限错误工具主动抛PermissionError被分类处理返回错误没有权限访问该资源。模型理解后告诉用户我没有权限访问 /etc/shadow 文件。注意模型的回答都是自然语言不是把错误消息原样复述。它理解了错误的含义然后用用户能理解的方式表达。小结错误消息要精心设计说人话、给线索、不泄露内部细节。模型看不懂Python traceback。ToolExecutor职责分离校验、执行、格式化三件事ToolRegistry只负责注册。pydantic ValidationError转人话提取字段名和错误描述拼成参数 ‘xxx’ yyy格式。异常分类处理FileNotFoundError告诉模型找不到文件PermissionError告诉模型权限不足未知错误只给摘要。结果token友好格式化None特殊处理、dict/list转JSON、超大输出截断默认4000字符。不抛异常ToolExecutor.execute()无论成功失败都返回字符串Agent Loop不需要try/except。DeepPilot v0.3工具执行器完成——从粗糙的registry.execute()到精细的ToolExecutor错误处理大幅提升。下节预告工具执行器搞定了但现在Agent Loop是同步的——每次调模型都要等模型生成完才能继续。用户体验很差尤其是长回答要等很久。下一节做流式Agent用streamTrue逐token渲染回答同时处理tool_calls的分片重组难题一个工具调用可能拆成多个chunk。从此Agent回答像ChatGPT一样实时流出不再干等。如果觉得本文对你有帮助欢迎点赞、收藏、关注三连本系列持续更新中关注不迷路~
返回列表