Building a Multi-Tool Gemma 4 Agent with Error Recovery

在本文中，您将学习如何将基本的工具调用脚本转换为一个具有弹性的代理，能够优雅地处理工具故障、模型输出格式错误以及不可用服务的情况。

我们将涵盖的主题包括：

• 如何构建一个带有迭代次数上限的迭代代理循环。
• 代理在调用工具时遇到的四种不同类型的故障及其处理方法。
• 如何设计工具错误消息，使模型能够学会恢复，减少无效的迭代次数。

构建一个具有错误恢复功能的多工具 Gemma 4 代理

引言

在上一篇文章中，我们使用 Ollama 的工具调用 API 将 Gemma 4 连接到几个 Python 函数。这给了我们一个工作的单轮分发器：模型选择一个工具，我们的代码运行它，模型回答。这是一个有用的起点，但离真正的代理还很远。

将工具调用演示转变为实际代理的一个关键因素是如何处理出现问题的情况。工具会失败。模型会幻想一个函数名，或者传递一个字符串而不是数字，或者询问一个您的查找表从未听说过的城市。上游 API 超时。缺少必需的参数。在之前的教程中，这些情况中的任何一个要么会导致脚本崩溃，要么会被一个打印消息并放弃的 try/except 吞掉。这对于单路径演示来说是可以接受的。但对于您希望让它一直运行的任何东西来说，这是不行的。

本文重新构建了代理，假设事情会出错，并展示了当它们出错时如何优雅地恢复。模式很简单：在边界捕获错误，将它们转换为模型可以读取的消息，将其发送回模型，并让模型决定是重试、绕过问题还是向用户解释失败。我们还将所有内容包装在一个具有迭代次数上限的适当迭代代理循环中。

完整脚本可以在这里找到。本文将介绍重要的部分。

重新思考工具循环

原始分发器运行一轮：发送用户查询，收集工具调用，运行它们，发送结果回模型，打印模型的回答。这是一个一次性交互。当模型的第一响应正确回答用户的问题时，它工作得很好，但在出现问题时无处可去。如果一个工具失败，模型有一次机会反应，然后我们就结束了。如果模型在看到第一个结果后想调用另一个工具，太糟糕了；我们已经退出了。

一个合适的代理循环是迭代的。结构很简单：

1. 将当前消息历史发送给模型。
2. 如果模型产生工具调用，请执行每个调用，将每个结果附加到历史记录中，并再次循环。
3. 如果模型产生纯文本响应，那就是最终答案。返回。
4. 在 MAX_ITERATIONS 处限制循环，以防止困惑的模型永远占用您的 CPU。

最后一个要点是不可协商的。小型模型偶尔会反复调用同一个工具，或者在两个工具之间振荡，而没有什么比回到终端发现笔记本电脑风扇尖叫更令人沮丧的了，因为 Gemma 决定连续三十次查询伦敦的天气。

这里是循环：

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45 def run_agent(user_query):

messages=[{"role":"user","content":user_query}]

for iteration in range(1,MAX_ITERATIONS+1):

payload={

"model":MODEL_NAME,

"messages":messages,

"tools":available_tools,

"stream":False,

}

print(f"[EXECUTION — iteration {iteration}]")

print("● Querying model...\n")

try:

response_data=call_ollama(payload)

except Exception as e:

print(f"└─ [ERROR] Error calling Ollama API: {e}")

print(f"└─ Make sure Ollama is running and {MODEL_NAME} is pulled.")

return

message=response_data.get("message",{})

tool_calls=message.get("tool_calls")or[]

分支 A：模型想要使用工具

if tool_calls:

print(f"[TOOL EXECUTION — {len(tool_calls)} call(s)]")

messages.append(message)

tool_messages=print_tool_calls(tool_calls)

messages.extend(tool_messages)

print()

continue

分支 B：模型产生了最终答案

print("[RESPONSE]")

print(message.get("content","")+"\n")

return

安全措施：我们在没有最终答案的情况下达到了 MAX_ITERATIONS

print("[RESPONSE]")

print( f"Hit the {MAX_ITERATIONS}-iteration cap without a final answer. " "This usually means the model is stuck in a tool-calling loop. " "Try simplifying the query.\n" )

这种模式值得记住，因为它出现在您将要阅读的每一个代理框架中：消息历史是状态。对于每次迭代，我们将整个对话（原始用户查询、模型的工具调用请求、我们的工具结果、任何后续模型消息）发送回模型。模型是无状态的；列表是代理的记忆。

这种迭代结构也是使错误恢复成为可能的原因。当一个工具失败并将错误作为工具消息发送回去时，模型可以看到该错误并在下一次迭代中对其做出反应。如果没有循环，就没有什么可以反应进入。

构建工具注册表

在这里，我们构建了四个工具，它们都是确定性的，并且离线运行。没有 API 密钥，没有网络调用，也没有不可靠的外部服务需要调试。本文的重点是错误处理架构，而不是工具本身，所以我们希望工具的行为是可预测的，这样我们就可以专注于围绕它们的框架，并且我们可以随意触发每种故障模式。

这些工具包括：

• get_weather(city)：在一个小的预定义天气数据字典中查找城市
• get_local_time(city)：使用 zoneinfo 计算该城市的当前时间
• convert_currency(amount, from_currency, to_currency)：根据硬编码的 USD 锚定汇率表进行计算
• get_city_population(city)：另一个对小字典的查找

静态数据位于文件顶部：

1

2

3

4

5

6

7

8

9

10

11

12

13

14 CITY_DATA={

"london":{"timezone":"Europe/London","population":8_982_000},

"tokyo":{"timezone":"Asia/Tokyo","population":13_960_000},

"sao paulo":{"timezone":"America/Sao_Paulo","population":12_330_000},

"paris":{"timezone":"Europe/Paris","population":2_161_000},

"new york":{"timezone":"America/New_York","population":8_336_000},

"sydney":{"timezone":"Australia/Sydney","population":5_312_000},

"mumbai":{"timezone":"Asia/Kolkata","population":20_410_000},

}

EXCHANGE_RATES={

"USD":1.00,"EUR":0.92,"GBP":0.79,"JPY":156.40,

"BRL":5.12,"CAD":1.37,"AUD":1.51,"INR":83.20,

}

这些函数故意简单，但它们会在输入无效时引发异常，而不是返回错误字符串。以下是 get_weather 的实现：

1

2

3

4

5

6

7

8

9 def get_weather(city:str)->str:

"""返回已知城市的当前天气状况。"""

key=city.lower().strip()

if key not in WEATHER_DATA:

raise ValueError(

f"未知城市: '{city}'。已知城市: {', '.join(sorted(WEATHER_DATA.keys()))}。"

)

data=WEATHER_DATA[key]

return f"{city.title()} 的天气是 {data['conditions']}，气温为 {data['temp_c']}°C。"

关于这个错误消息有两点需要注意。首先，它是具体的：它告诉调用者发生了什么以及有效的选项是什么。其次，工具会引发 ValueError 而不是将错误作为字符串返回。不要在工具内部捕获并格式化错误；相反，让它们传播。我们希望调度器能够在一个地方处理所有类型的失败，并且希望模型在输入无效时看到的消息足够详细，以便模型可以自行纠正。

get_local_time 做了唯一的真实工作——实际的时区感知日期时间运算——这也是我们稍后将用来演示模拟上游故障时优雅降级的工具：

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60 def get_local_time(city:str)->str:

"""返回城市的当前本地时间，带有缓存的回退。"""

key=city.lower().strip()

模拟可能不可靠的上游地理编码服务

if SIMULATE_GEOCODING_OUTAGE and random.random()<0.6:

if key in TIMEZONE_FALLBACK_CACHE:

tz_name=TIMEZONE_FALLBACK_CACHE[key]

now=datetime.datetime.now(ZoneInfo(tz_name))

return(

f"[cached] {city.title()} 的当前本地时间是 "

f"{now.strftime('%H:%M on %A, %B %d, %Y')} ({tz_name})。 "

"注意：地理编码服务目前不可用；此值来自本地缓存。"

)

raise ToolUnavailableError(

f"地理编码服务不可用且 '{city}' 不在本地缓存中。 "

"请稍后再试或使用缓存中的城市："

f"{', '.join(sorted(TIMEZONE_FALLBACK_CACHE.keys()))}。"

)

if key not in CITY_DATA:

raise ValueError(f"未知城市: '{city}'。已知城市: {', '.join(sorted(CITY_DATA.keys()))}。")

tz_name=CITY_DATA[key]["timezone"]

now=datetime.datetime.now(ZoneInfo(tz_name))

return f"{city.title()} 的当前本地时间是 {now.strftime('%H:%M on %A, %B %d, %Y')} ({tz_name})。"

这个 SIMULATE_GEOCODING_OUTAGE 标志让我们可以在不需要实际基础设施失败的情况下重现真实世界故障模式。我们稍后会回到这个标志。

工具模式与 上一个教程 中的风格相同：标准 Ollama 函数调用格式，每个工具的功能和预期参数都有清晰的描述。

<h2>四种错误恢复模式</h2>

是时候认真对待了。当代理与工具交互时，会有四种不同的故障模式，每种模式都需要自己的策略。它们在一个单独的调度函数中处理，但理解它们作为独立概念是有价值的。

<h3>模式 1：工具执行错误</h3>

第一道防线是调度器本身。它将每个工具调用包装在一个结构化的 try/except 块中，并将每种类型的失败转换为代理循环可以传递回模型的 (status, content) 对：

<pre class="lang:default decode:true">def dispatch_tool_call(tool_call):

function_name = tool_call["function"]["name"]

arguments = tool_call["function"]["arguments"] or {}

防御措施 1：将工具名称与注册表进行验证

if function_name not in TOOL_FUNCTIONS:

return "error", (

f"未知工具 '{function_name}'。 "

f"有效工具是: {','.join(TOOL_FUNCTIONS.keys())}。"

)

func = TOOL_FUNCTIONS[function_name]

防御措施 2：捕获参数错误（类型错误、缺少或多余的参数）

try:

result = func(**arguments)

return "ok", str(result)

except TypeError as e:

return "error", f"{function_name} 的参数错误: {e}"

except ValueError as e:

return "error", str(e)

except ToolUnavailableError as e:

return "error", f"工具暂时不可用: {e}"

except Exception as e:

return "error", f"{function_name} 中的意外错误: {type(e).__name__}: {e}"

</pre>

关键洞察：将错误作为工具结果返回给模型，而不是将其抛回代理循环。模型可以读取错误信息，看到它请求的是“Atlantis”，而 Atlantis 不是一个已知的城市，然后转向另一个城市或向用户道歉。如果你抛出错误，你就剥夺了模型恢复的能力。

注意底部的四种不同异常类型和通用捕获块。每一种都对应一个实际的失败类别：领域错误（ValueError）、签名不匹配（TypeError）、基础设施故障（ToolUnavailableError），以及唐纳德·拉姆斯菲尔德所说的未知未知（Exception）。将它们分开可以得到更清晰的错误消息，这为模型提供了更好的恢复信号。

通用捕获块很重要，可能具有争议性。一些风格指南会告诉你永远不要捕获裸露的 Exception。在一个代理调度器中，替代方案——让意外的异常杀死循环——更糟糕。模型失去了恢复的机会，用户失去了响应，你失去了可以用来调试发生了什么的对话历史记录。更好的做法是捕获、记录并将消息传递给模型。

模式 2：来自模型的格式化工具调用

模型偶尔会幻想一个不存在的工具名称，或者将参数发送到错误的键下（例如 town 而不是 city）。上面的代码片段中的第一种防御措施处理了这种情况：在我们甚至尝试分发之前，我们检查名称是否在注册表中，并返回一个纠正消息，列出有效的名称。

错误参数的情况由第二种防御措施处理。Python 的 **arguments 解包会在模型发送函数不接受的关键字参数或省略必需参数时引发 TypeError。我们捕获 TypeError，将其格式化得更干净，并在下一次迭代中让模型得到有用的错误信息：

1[ERROR]:Bad arguments for get_weather:get_weather()got an unexpected keyword argument'town'

这条消息包含了模型需要纠正自己的所有信息：工具名称、有问题的参数，以及隐含的信号，即正确的名称是其他东西。实际上，模型通常会在下一次轮次中修复调用。

还有一种更微妙的与参数相关的失败：类型漂移。模型知道 amount 应该是一个数字，但在较长的对话中，它偶尔会将 "100" 发送为字符串。让 convert_currency 在这种情况下引发错误会迫使模型再进行一次轮次来纠正自己。更好的方法是在工具本身中进行防御性强制转换：

1

2

3

4

5

6

7 def convert_currency(amount:float,from_currency:str,to_currency:str)->str:

防御性类型强制转换：模型有时会将数字发送为字符串

try:

amount=float(amount)

except(TypeError,ValueError):

raise ValueError(f"'amount' must be a number, got: {amount!r}")

... 函数的其余部分

这种方法默默地修复了常见的案例（"100" 变为 100.0），同时仍然为真正损坏的案例（"fifty"）引发一个干净的错误。原则：对从模型接收到的内容宽松，对你要抱怨的内容严格。

模式 3：领域级别的错误

这些是工具本身在输入格式正确但请求无法满足时引发的错误，例如请求阿特兰蒂斯的天气，或者从不在汇率表中的货币进行转换。这些应该产生能够教会模型如何恢复的错误消息，而不仅仅是说“失败”。

比较这两个错误消息：

1 坏："未知城市。"

1 好："未知城市：'Atlantis'。已知城市：伦敦、孟买、纽约、巴黎、圣保罗、悉尼、东京。"

好的版本给了模型一切它需要的东西，要么重试一个有效的输入，要么向用户解释限制。坏的版本迫使模型猜测。工具函数中的每个错误消息都遵循这种模式：说出了什么出错，如果可能的话，列出了有效的替代方案。

这不仅仅是一个用户体验上的小细节。它直接影响代理循环在到达一个好答案之前会燃烧多少次迭代。模糊的错误可能会让你多花一个完整的往返时间，而模型在摸索修复方法。具体的错误通常在下一次轮次中得到纠正，或者当输入真正无法恢复时，让模型在根本不尝试的情况下生成一个干净的解释。

模式 4：不可用工具的优雅降级

最后一个模式是工具没有损坏，只是消失了的情况——地理编码服务宕机、API 配额耗尽、数据库遇到问题。这里你有三种选择，大致按你信任模型处理这种情况的程度排序：

1. 返回缓存或默认值，并在结果中标记。当工具的新鲜度不重要时最好。
2. 完全跳过工具，并返回一个关于无法提供的清晰消息。让模型决定是否重试或绕过它。
3. 通过让代理停止并询问指导来将故障暴露给用户。

get_local_time 演示了选项 1。当 SIMULATE_GEOCODING_OUTAGE 打开且随机检查触发时，工具首先尝试本地缓存：

1

2

3

4

5

6

7

8

9

10

11

12

13

14 if SIMULATE_GEOCODING_OUTAGE and random.random()<0.6:

if key in TIMEZONE_FALLBACK_CACHE:

tz_name=TIMEZONE_FALLBACK_CACHE[key]

now=datetime.datetime.now(ZoneInfo(tz_name))

return( f"[cached] The current local time in {city.title()} is " f"{now.strftime('%H:%M on %A, %B %d, %Y')} ({tz_name}). " "Note: geocoding service is currently unavailable; this value is from the local cache." )

raise ToolUnavailableError( f"Geocoding service is unavailable and '{city}' is not in the local cache. " "Please try again later or use a city from the cache: " f"{', '.join(sorted(TIMEZONE_FALLBACK_CACHE.keys()))}." )

如果城市在缓存中，工具会返回一个带有 [cached] 标签的成功结果，并附带一条说明，指出实时服务不可用。模型会看到一个完全可用的答案和一个小的注意事项，可以选择向用户提及。如果城市不在缓存中，工具会转到选项 2：抛出 ToolUnavailableError，并附带一条消息，列出哪些城市是可用的。

这个 ToolUnavailableError 故意是一个单独的异常类型，而不是 ValueError。调度器为它提供了一个独立的捕获分支，并带有不同的错误前缀（“工具暂时不可用”），这样模型可以区分“你请求了我没有的东西”和“服务现在宕机了”。这两种失败的适当响应是不同的——稍后再试或选择不同的输入——给模型一个清晰的信号有助于它做出正确的选择。

在生产环境中，你会在转到回退之前扩展这种模式，采用带有退避策略的重试。结构保持不变：调度器区分可恢复和不可恢复的失败，模型会被告知每种失败的足够信息，以便做出合理的下一步行动。

结合所有内容

是时候实际运行这个东西了。这里有一个查询，涵盖了所有内容——多个城市、多个工具以及故意的坏输入以触发飞行中的错误恢复：

1. python main.py "What's the weather in London, Tokyo, and Atlantis right now? And convert 50 GBP to JPY."

确切的迭代次数和工具调用顺序会根据 Gemma 如何安排工作而有所不同，但这里有一个代表性跟踪记录，稍微修剪了一下：

看看第三次迭代发生了什么。模型询问了亚特兰蒂斯，工具抛出了 ValueError，调度器将其转换为列出有效城市的错误消息，模型在第五次迭代时将这些信息整合到一个干净的响应中。它没有重试亚特兰蒂斯。它没有崩溃。它注意到失败，将其与成功结果结合起来，并生成了一个承认限制的答案。这就是错误恢复架构的全部回报。

要看到优雅降级的效果，请将 SIMULATE_GEOCODING_OUTAGE 设置为 True 并运行一个查询，询问本地时间：

1. python main.py "What's the local time in London and Paris?"

大约 60% 的时间你会在工具结果中看到 [cached] 前缀，并且模型会在最终响应中提到缓存源。其余时间工具会成功返回，缓存路径不会触发。无论哪种情况，循环都会完成，用户会得到一个答案。

结论

我们在第一篇教程的基础上构建了三件东西：一个带有硬迭代上限的迭代代理循环、一个捕获所有工具故障类别的分层调度器，以及工具函数，其错误消息教会模型如何恢复。它们一起是工具调用演示和实际希望无监督运行的代理之间的区别。

一些自然的下一步包括：

• 跨会话的持久记忆，以便代理可以记住上周学到的关于你的事情
• 对瞬态上游故障的带有退避策略的重试
• 将外部 API 重新纳入静态查找表的替代方案，这基本上意味着接受超时和速率限制成为正常故障表面的一部分

完整脚本在 GitHub 上。克隆它，运行它，故意破坏它以观看恢复过程，并结合上述下一步。