4. 工具调用核心概念
工具调用编排,说白了就是决定多个工具函数以什么顺序、什么条件、什么策略来执行。我见过不少团队,模型调得不错,工具也写了一大堆,结果编排逻辑一塌糊涂——要么死循环,要么漏掉关键步骤,要么重试策略把 API 打爆了。
这一章,我们就把三种核心编排模式讲透。你想想看,一个复杂的任务,比如「帮我查一下北京到上海的航班,再订一张最便宜的」,背后至少涉及查询、比价、下单三个工具。怎么串起来?出错了怎么办?这就是编排要解决的问题。
4.1 顺序调用:最基础也最常用
顺序调用,就是按固定顺序一个接一个执行工具。前一个的输出,作为后一个的输入。这是最直观的模式,也是我最早在项目中使用的模式。
举个例子,一个文档处理流水线:
# 伪代码:顺序调用
def document_pipeline(text):
# 第一步:提取关键信息
info = extract_info(text)
# 第二步:翻译成英文
translated = translate(info, target="en")
# 第三步:生成摘要
summary = summarize(translated)
return summary
这种模式适合什么场景?
- 数据流水线:清洗 → 转换 → 加载
- 多步验证:格式检查 → 内容审核 → 合规校验
- 级联推理:先拆解问题,再逐步求解
我的习惯:顺序调用时,我会在每一步之间加一个「检查点」。不是简单的传参,而是验证输出格式、类型、关键字段是否存在。曾经有一次,翻译工具返回了空字符串,后续步骤全崩了。从那以后,每个检查点我都加了非空校验。
4.2 条件分支调用:让模型自己选路
条件分支,就是根据某个条件判断结果,决定下一步调用哪个工具。这比顺序调用灵活得多,也是实现「智能路由」的基础。
我举个例子,一个客服系统:
# 伪代码:条件分支
def customer_service_router(query):
intent = classify_intent(query) # 意图识别
if intent == "退款":
return refund_tool(query)
elif intent == "物流":
return logistics_tracking(query)
elif intent == "投诉":
return escalate_to_human(query)
else:
return general_faq(query)
条件分支的核心在于「条件判断」本身。这个判断可以由模型做,也可以由规则做。
| 判断方式 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| 模型判断 | 灵活,能处理模糊语义 | 有延迟,可能误判 | 开放域对话、复杂意图 |
| 规则判断 | 快,确定性强 | 死板,维护成本高 | 固定格式、参数校验 |
| 混合判断 | 兼顾灵活与稳定 | 实现复杂 | 生产环境推荐 |
我曾经踩过的坑:条件分支里,如果判断条件本身依赖模型输出,一定要加「兜底分支」。有一次,模型对意图分类的置信度很低,结果所有分支都没命中,系统直接抛异常了。从那以后,每个条件分支我都加一个 else 兜底,哪怕只是返回「抱歉,我没理解您的意思」。
4.3 循环与重试机制:让系统更健壮
循环和重试,是应对「不确定性」的两大利器。模型调用不是每次都能成功,工具调用也不是每次都能返回正确结果。怎么办?重试。但重试不是无脑重复,得有策略。
4.3.1 重试策略
常见的重试策略有几种:
- 固定间隔重试:每隔 N 秒重试一次。简单,但容易造成资源浪费。
- 指数退避:每次重试间隔翻倍。比如 1s、2s、4s、8s……这是生产环境最常用的策略。
- 抖动退避:在指数退避的基础上加随机偏移,防止多个请求同时重试造成「惊群效应」。
# 伪代码:指数退避重试
import time
import random
def retry_with_backoff(func, max_retries=3, base_delay=1):
for attempt in range(max_retries):
try:
return func()
except Exception as e:
if attempt == max_retries - 1:
raise e
delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
time.sleep(delay)
return None
4.3.2 循环调用
循环调用不是简单的 for 循环。在工具编排里,循环通常用于:
- 批量处理:对列表中的每个元素调用同一个工具
- 迭代优化:反复调用某个工具,直到结果满足条件
- 多轮对话:模型与工具之间来回交互,逐步逼近最终答案
举个例子,一个「自动写文章并优化」的流程:
# 伪代码:迭代优化循环
def write_and_polish(topic, max_iterations=3):
draft = generate_outline(topic)
for i in range(max_iterations):
feedback = review_draft(draft)
if feedback["score"] >= 0.9:
break
draft = revise_draft(draft, feedback)
return draft
关键点:循环一定要有「终止条件」。要么达到目标,要么达到最大次数,要么超时。否则,一个 bug 就能让你的系统跑一整天,账单直接爆炸。
4.4 三种模式的组合实战
实际项目中,这三种模式很少单独使用。我参与过一个智能客服项目,它的编排逻辑是这样的:
- 顺序调用:先做意图识别,再做实体抽取
- 条件分支:根据意图路由到不同工具链
- 循环重试:每个工具调用都带重试,整个流程如果失败会重试 3 次
你看,这就是一个典型的组合编排。下面这张图展示了核心流程:
4.5 避坑指南与最佳实践
做了这么多年编排,我总结了几条铁律:
- 永远设超时:每个工具调用都要有超时时间。我习惯设 30 秒,超过就认为失败,触发重试。
- 重试要幂等:同一个请求重试多次,不能产生副作用。比如下单接口,重试前要检查是否已经下单成功。
- 日志要详细:每一步的输入、输出、耗时、重试次数,全都要记。出了问题才能快速定位。
- 熔断机制:如果某个工具连续失败超过阈值,暂时熔断,不再调用,避免雪崩。
我个人习惯:在编排引擎里加一个「可视化追踪」功能。每次调用都生成一个 trace ID,把整个调用链串起来。调试的时候,直接看 trace 图,哪一步慢了、哪一步失败了,一目了然。这个习惯帮我省了无数排查时间。
嗯,三种编排模式讲完了。顺序调用是骨架,条件分支是灵魂,循环重试是保障。把它们组合好,你的工具调用系统才能真正「智能」起来。