聊聊我在AI工具链里踩的那些坑
晚上十一点半,终于把试婚纱的行程表排好了,我瘫在沙发上,打开电脑准备写点东西。明天还要早起赶地铁去公司,通勤一小时,路上正好可以复盘一下最近搞的那个AI工具链优化的事儿。
先说下背景吧,我是个在北京写代码的妹子,白天在公司跟各种需求斗智斗勇,晚上回到家还得筹备婚礼。没错,备婚中的程序媛,听起来就很刺激对吧?白天改bug,晚上选喜糖,周末还得去试婚纱、对流程,我对象说我现在是"双线并发,全栈人生"。行吧,他说得对。
事情是这么开始的
上个月,我们组接了个AI应用的活儿。产品经理(我们叫他老王吧)兴冲冲地跑过来说:"咱们要做个智能助手,能帮用户自动调用各种工具,查天气、订机票、查数据库,啥都能干!"
我当时第一反应是:这不就是Function Calling嘛。
说实话,Function Calling这个概念出来也有一阵子了,OpenAI最早搞的,后来各家大模型都跟进了。原理也不复杂,就是让大模型在对话过程中,能够识别用户的意图,然后输出一个结构化的函数调用请求,由我们的代码去执行这个函数,再把结果喂回给模型。
听起来很简单对吧?我当初也是这么想的。
第一版:裸奔的Function Calling
最开始我直接用的OpenAI的API,写了个demo,大概长这样:
import openai
import json
tools = [
{
"type": "function",
"function": {
"name": "query_database",
"description": "查询数据库中的用户信息",
"parameters": {
"type": "object",
"properties": {
"user_id": {
"type": "string",
"description": "用户ID"
},
"query_type": {
"type": "string",
"enum": ["profile", "orders", "balance"],
"description": "查询类型"
}
},
"required": ["user_id", "query_type"]
}
}
}
]
def chat_with_tools(user_message: str, history: list):
messages = history + [{"role": "user", "content": user_message}]
response = openai.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto"
)
message = response.choices[0].message
# 如果有工具调用
if message.tool_calls:
for tool_call in message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
# 执行函数
result = execute_function(func_name, func_args)
# 把结果喂回去
messages.append(message)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(result)
})
# 再次调用模型,让它根据工具结果生成回复
final_response = openai.chat.completions.create(
model="gpt-4o",
messages=messages
)
return final_response.choices[0].message.content
return message.content
跑起来确实能用,用户说"帮我查一下张三的订单",模型就能正确调用query_database函数,拿到结果后生成一段自然的回复。
我当时还挺得意的,觉得这事儿就这么简单嘛。
然后老王来了,说:"这个助手能不能复杂点?比如用户说'帮我查一下张三最近的订单,如果金额超过1000就给他发个优惠券',这种能处理吗?"
好的,多步工具调用。
踩坑一:多步调用的死循环
多步调用就是模型需要连续调用多个工具,中间可能还要做判断。比如上面那个场景,模型需要先查订单,判断金额,再决定是否发优惠券。
我加了个循环:
def chat_with_multi_tools(user_message: str, history: list, max_steps: int = 10):
messages = history + [{"role": "user", "content": user_message}]
for step in range(max_steps):
response = openai.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto"
)
message = response.choices[0].message
if not message.tool_calls:
return message.content
messages.append(message)
for tool_call in message.tool_calls:
result = execute_function(
tool_call.function.name,
json.loads(tool_call.function.arguments)
)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(result)
})
return "抱歉,处理步骤过多,请稍后再试。"
然后噩梦就开始了。
有一天晚上,我正在家试婚纱的配饰,手机突然疯狂报警。线上监控显示,有个用户的请求触发了死循环,模型一直在反复调用query_database,API调用量蹭蹭往上涨。
我当时真的想砸电脑。
排查了半天发现,是模型在某次调用后,返回的结果让它觉得"信息不够",于是又去查了一遍,然后又觉得不够……无限循环。更离谱的是,有时候模型会调用一个不存在的函数名,或者传一堆乱七八糟的参数,我们的代码直接抛异常。
这就是裸奔Function Calling的问题:你没法控制模型的行为边界。
踩坑二:工具描述的艺术
为了解决上面那些问题,我开始疯狂优化tools的定义。这里有个很大的坑——工具描述的质量直接决定了模型调用的准确率。
最开始我的工具描述写得跟代码注释一样:
{
"name": "query_database",
"description": "查询数据库",
"parameters": {
"user_id": {"type": "string"},
"query_type": {"type": "string"}
}
}
模型经常搞不清楚该传什么参数,或者把query_type传成"查询订单"这种自然语言,而不是我们定义的枚举值。
后来我学乖了,描述要写得像给一个"聪明但不懂业务的人"看的说明书:
{
"name": "query_database",
"description": "查询公司内部数据库中的用户信息。支持查询用户基本资料、历史订单和账户余额。注意:此工具只负责查询,不会修改任何数据。如果需要修改数据,请使用其他工具。",
"parameters": {
"type": "object",
"properties": {
"user_id": {
"type": "string",
"description": "用户的唯一标识符,格式为'U'加6位数字,例如'U000123'。如果用户提供的是姓名,请先使用search_user工具获取用户ID。"
},
"query_type": {
"type": "string",
"enum": ["profile", "orders", "balance"],
"description": "要查询的信息类型。profile=用户基本资料,orders=历史订单列表,balance=账户余额。只能从这三个值中选择。"
}
},
"required": ["user_id", "query_type"]
}
}
你看,加了格式说明、举例、使用场景、注意事项……描述越长越详细,模型越不容易犯错。但这又带来另一个问题:token消耗暴增。
我们当时算了一笔账,光工具定义就占了将近2000个token,加上system prompt,每次请求还没开始对话就已经花了大几千token的钱。对于C端产品来说,这个成本是不可接受的。
引入AutoGen:让Agent来管理Agent
就在我为这些事情头疼的时候,微软的AutoGen进入了我的视野。
说实话,一开始我是拒绝的。框架太多了,LangChain、LlamaIndex、Dify、Coze……每个都说自己能解决所有问题。但AutoGen的核心理念打动了我:多Agent协作。
它的思路是这样的:与其让一个大模型搞定所有事情,不如搞多个专门的Agent,每个Agent负责一类任务,然后让它们互相"对话"来完成任务。
比如我们的场景,可以拆成:
| Agent角色 | 职责 | 可用工具 |
|---|---|---|
| Planner | 理解用户意图,拆解任务步骤 | 无 |
| DataAgent | 负责所有数据查询操作 | query_database, search_user |
| ActionAgent | 负责执行操作(发优惠券、发通知等) | send_coupon, send_notification |
| Critic | 检查其他Agent的输出是否合理 | 无 |
这个架构的好处是,每个Agent的prompt和工具集都很精简,不会像之前那样一个大prompt塞所有东西。而且Critic Agent可以作为一个"安全阀",防止其他Agent犯傻。
来,看看实际代码怎么搞的:
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
# 配置LLM
llm_config = {
"config_list": [
{
"model": "gpt-4o",
"api_key": "your-api-key",
}
],
"temperature": 0.1, # 工具调用场景,温度调低
}
# Planner Agent:负责任务规划
planner = AssistantAgent(
name="Planner",
system_message="""你是一个任务规划专家。你的职责是:
1. 理解用户的需求
2. 将复杂需求拆解成具体的步骤
3. 为每个步骤指定负责的Agent
注意:你不需要执行任何操作,只需要输出执行计划。
输出格式为JSON数组,每个元素包含step、agent、action三个字段。""",
llm_config=llm_config,
)
# Data Agent:负责数据查询
data_agent = AssistantAgent(
name="DataAgent",
system_message="""你是数据查询专家。你只能使用以下工具:
- query_database: 查询用户数据
- search_user: 根据姓名搜索用户
你只能查询数据,不能修改任何内容。
查询结果必须以结构化的JSON格式返回。""",
llm_config=llm_config,
)
# Action Agent:负责执行操作
action_agent = AssistantAgent(
name="ActionAgent",
system_message="""你是操作执行专家。你只能使用以下工具:
- send_coupon: 发放优惠券
- send_notification: 发送通知
在执行操作前,你必须确认已经获得了足够的信息。
每次操作后,必须报告操作结果。""",
llm_config=llm_config,
)
# Critic Agent:负责审查
critic = AssistantAgent(
name="Critic",
system_message="""你是质量审查专家。你的职责是:
1. 检查每个Agent的输出是否合理
2. 检查是否存在安全风险
3. 如果发现问题,明确指出并要求修正
你有一票否决权。如果你觉得某个操作不合理,可以直接拒绝。""",
llm_config=llm_config,
)
# User Proxy:代表用户与Agent群交互
user_proxy = UserProxyAgent(
name="user",
human_input_mode="NEVER", # 自动化场景,不需要人工介入
max_consecutive_auto_reply=10,
code_execution_config=False,
)
# 注册工具(这里简化了)
data_agent.register_function(
function_map={
"query_database": query_database,
"search_user": search_user,
}
)
action_agent.register_function(
function_map={
"send_coupon": send_coupon,
"send_notification": send_notification,
}
)
# 创建群聊
groupchat = GroupChat(
agents=[user_proxy, planner, data_agent, action_agent, critic],
messages=[],
max_round=15,
speaker_selection_method="auto", # 自动选择下一个发言的Agent
)
manager = GroupChatManager(
groupchat=groupchat,
llm_config=llm_config,
)
# 开始对话
user_proxy.initiate_chat(
manager,
message="帮我查一下张三最近的订单,如果金额超过1000就给他发个优惠券"
)
踩坑三:AutoGen的那些"惊喜"
你以为引入框架就万事大吉了?天真。
坑1:Agent之间的"废话"太多
AutoGen的多Agent对话机制很灵活,但也很"话痨"。有时候Planner说了一句"我来分析一下这个任务",Critic非要回一句"好的,我同意你的分析",DataAgent又说一句"收到,我准备开始查询"……一来一回,光寒暄就消耗了好几千token。
解决办法是在system message里严格限制每个Agent的发言风格:
system_message="""你是数据查询专家。
规则:
1. 只输出必要的信息,不要寒暄
2. 不要重复别人说过的话
3. 直接给出查询结果或提出需要的信息
4. 每次发言不超过100字"""
坑2:循环对话停不下来
跟之前裸奔Function Calling的问题类似,Agent们有时候会陷入"我觉得不行""为什么不行""因为我觉得不行"的死循环。
AutoGen提供了max_round参数来限制最大对话轮数,但更好的做法是在Critic Agent里加一个退出条件判断:
critic = AssistantAgent(
name="Critic",
system_message="""你是质量审查专家。
...
重要规则:
- 如果任务已经完成且结果合理,你必须输出"[TASK_COMPLETE]"来结束对话
- 如果连续两次审查都通过,必须结束对话
- 不要为了审查而审查""",
llm_config=llm_config,
)
坑3:工具执行的错误处理
AutoGen默认的工具执行是同步的,而且错误处理比较粗糙。如果一个工具执行超时或者抛异常,整个Agent链路就断了。
我后来做了一层封装,加了超时控制和重试机制:
import functools
from concurrent.futures import ThreadPoolExecutor, TimeoutError
def tool_with_timeout(func, timeout=30, max_retries=2):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries + 1):
try:
with ThreadPoolExecutor(max_workers=1) as executor:
future = executor.submit(func, *args, **kwargs)
return future.result(timeout=timeout)
except TimeoutError:
if attempt == max_retries:
return json.dumps({"error": f"工具执行超时({timeout}s)"})
continue
except Exception as e:
if attempt == max_retries:
return json.dumps({"error": f"工具执行失败: {str(e)}"})
continue
return wrapper
# 使用
query_database = tool_with_timeout(query_database, timeout=10, max_retries=2)
这样即使某个工具挂了,也会返回一个错误信息给Agent,让它自己决定怎么处理,而不是直接把整个链路搞崩。
关于产品化的一些思考
搞技术的人容易陷入一个误区:觉得技术牛逼就行了。但老王经常教育我:"你这个东西用户能用吗?好用吗?"
好吧,他说得对。
AI应用的产品化,有几个点是我这段时间踩坑后总结的:
1. 延迟是第一杀手
多Agent协作听起来很美,但每多一个Agent,就多一轮LLM调用。用户提一个问题,后台可能跑了5、6轮,每轮1-2秒,加起来就是10秒+。用户的耐心是有限的。
我们的优化方案是并行执行无依赖的步骤。比如Planner规划出3个独立的查询任务,DataAgent可以一次性并行调用3个工具,而不是串行。
# 并行工具调用
import asyncio
async def parallel_tool_calls(tool_calls: list):
tasks = []
for call in tool_calls:
tasks.append(asyncio.to_thread(
execute_function,
call['name'],
call['arguments']
))
results = await asyncio.gather(*tasks)
return results
2. 可解释性很重要
用户不关心你后台跑了几个Agent,但他们想知道"你在干嘛"。我们加了一个流式输出的机制,把每个Agent的思考过程实时展示给用户:
🤔 正在分析您的需求... 📋 已制定执行计划:1.查询用户信息 2.查询订单 3.判断金额 🔍 正在查询张三的用户信息... 📊 找到订单3笔,最大金额1580元 🎁 满足条件,正在发放优惠券... ✅ 已完成!已为张三发放满1000减100优惠券
这种"过程透明"的设计,用户满意度明显提升。即使总时间没变,但用户觉得"系统在认真干活",而不是"卡住了"。
3. 兜底策略必须有
AI应用永远不可能100%可靠。我们设计了一套降级策略:
| 场景 | 降级方案 |
|---|---|
| LLM调用超时 | 返回预设话术 + 转人工 |
| 工具调用失败 | 跳过该步骤 + 告知用户 |
| Agent死循环 | 强制终止 + 返回已有结果 |
| 结果质量差 | Critic拦截 + 重新生成 |
最后说两句
搞了这一个多月,从裸奔Function Calling到AutoGen多Agent架构,踩的坑比我试婚纱踩的高跟鞋还多。但说实话,看到产品上线后用户反馈还不错,还是挺有成就感的。
技术选型这件事,没有银弹。Function Calling简单直接,适合轻量级场景;AutoGen适合复杂的多步骤任务,但引入的成本也不小。关键是要根据业务场景来选,不要为了用新技术而用新技术。
好了,不写了,明天还要早起。对了,下周末要去拍婚纱照, Hopefully那天不要有线上事故。
如果这篇文章对你有帮助,点个赞呗。有问题也欢迎评论区讨论,我会在通勤地铁上回复你们的(手动狗头)。
一个正在备婚的程序媛,于北京深夜


评论 0