当前位置：首页 > news >正文

第八章：Error Handling

news 来源：原创 2025/4/29 9:51:29

Chapter 8: Error Handling

从服务器通信到错误处理：如何让AI像“快递客服”一样优雅应对问题？

在上一章的服务器/客户端通信中，我们已经能让AI通过网络远程协作。但你是否想过：如果天气API突然返回错误，或者用户输入了超过限制的文本，系统该如何应对？就像快递客服需要区分“包裹丢失”和“地址错误”一样，本章将介绍如何通过错误处理系统让AI优雅地识别和解决各种问题。

核心问题：如何让AI像“客服”一样分门别类处理问题？

想象你是一位快递客服，遇到不同问题需要不同解决方案：

工具错误：天气API返回“服务器忙”
参数错误：用户输入超过最大字符限制
通用错误：未知的网络问题

错误处理系统就像这位客服，通过以下机制实现：
✅ 自定义异常分类精准定位问题
✅ 统一错误格式方便开发者调试
✅ 自动记录错误日志供后续分析

核心概念分解

1. 自定义异常层次：为不同错误制作“工单分类”

通过继承基础异常类，为不同错误类型创建专用异常：

# app/exceptions.py中的异常定义
class ToolError(Exception):"""工具调用时的通用错误"""passclass TokenLimitExceeded(ToolError):"""文本长度超过限制时的专用错误"""def __init__(self, max_tokens):self.message = f"超过最大token限制({max_tokens})！请缩短输入内容"

2. 错误传播：像接力棒一样传递错误信息

在代理和工具间传递异常对象，保留原始错误信息：

# 示例：代理调用天气工具时捕获错误
try:response = weather_tool.get_weather()
except TokenLimitExceeded as e:raise e  # 直接抛出专用错误
except Exception as e:raise ToolError(f"工具调用失败：{str(e)}")  # 统一包装其他错误

3. 全局错误处理器：像“总客服”一样统一响应

在入口层统一处理所有错误，返回友好提示：

# run_mcp.py中的错误处理示例
async def main():try:await agent.run(user_prompt)except TokenLimitExceeded as e:print(f"⚠️ {e.message}")except ToolError as e:print(f"⚠️ 工具错误：{e}")except Exception as e:print(f"⚠️ 系统发生未知错误：{str(e)}")

如何用错误处理解决问题？

场景：用户输入过长导致token超限

步骤1：定义最大token限制

在config.toml中设置：

[llm.default]
max_tokens = 4096  # 最大允许4096个token

步骤2：工具层抛出专用错误

# weather_tool.py中的token检查逻辑
def get_weather(prompt):if len(prompt) > config.llm["default"].max_tokens:raise TokenLimitExceeded(config.llm["default"].max_tokens)# 继续调用真实API...

步骤3：代理捕获并传递错误

# agent.py中的调用示例
try:weather_data = weather_tool.get_weather(user_input)
except TokenLimitExceeded as e:# 将错误信息返回给用户raise e

最终用户会看到这样的提示：

⚠️ 超过最大token限制(4096)！请缩短输入内容

内部实现揭秘：错误处理的“工单流转流程”

流程分解（用简单序列图演示）

关键代码解析

1. 异常定义文件（app/exceptions.py）

class OpenManusError(Exception):"""所有OpenManus错误的基类"""passclass ToolError(OpenManusError):"""工具调用相关错误"""passclass TokenLimitExceeded(ToolError):"""文本长度超过限制时触发"""def __init__(self, max_tokens):self.max_tokens = max_tokensself.message = f"最大允许{max_tokens}个token！请精简输入内容"

2. 全局错误处理（run_mcp.py）

async def run_mcp():try:await runner.run()  # 执行主流程except TokenLimitExceeded as e:print(f"⚠️ {e.message}")except ToolError as e:print(f"⚠️ 工具调用失败：{str(e)}")except Exception as e:print(f"⚠️ 系统发生未知错误：{str(e)}")# 后续章节会介绍日志记录