FastAPI/Python 接入通义千问 Function Calling(工具调用)实战 Day3
关键词:FastAPI、通义千问、Function Calling、工具调用、Agent、Python
这是大模型接入实战的 Day3。Day1 跑通单轮非流式 / SSE 流式,Day2 用 Redis 做了多轮会话 + 上下文压缩。Day3 往前再走一步——让模型能「调用工具」:模型自己决定要不要查天气、查学历,程序员负责把工具跑完把结果喂回去,模型再生成最终回答。这是做 AI Agent 的基础能力。
一、Function Calling 是什么
大模型本身只会「说话」,不会查实时数据、不会调业务接口。Function Calling(工具调用)的机制是:
- 你先告诉模型:「你有哪些工具可用」(
tools列表,每个工具含名称 / 描述 / 参数 JSON Schema) - 用户提问后,模型判断是否需要调用工具、调用哪个、参数是什么
- 模型不直接回答,而是返回一个
tool_calls(指明函数名 + 参数) - 程序员执行对应工具,拿到结果
- 把工具结果以
role=tool的形式塞回messages,再调一次模型 - 模型这次基于工具结果,生成最终的自然语言回答
一句话:模型出「调用指令」,你出「执行 + 回填」,最后模型「总结」。
二、单工具:天气查询llm/case5.py
2.1 定义工具
importjsonimportosimportrandomfromopenaiimportOpenAIdefget_current_weather(arguments):weather_conditions=["晴天","阴天","雨天","雪天"]random_weather=random.choice(weather_conditions)location=arguments["location"]returnf"{location}今天是{random_weather}"tools=[{"type":"function","function":{"name":"get_current_weather","description":"当你想查询指定城市的天气时非常有用","parameters":{"type":"object","properties":{"location":{"type":"string","description":"城市或县区,比如北京市、杭州市、余杭区等",}},"required":["location"],},},}]工具定义就是一段JSON Schema:name是函数名,description告诉模型「什么时候该用它」,parameters描述入参。模型靠description来决策,所以描述要写清楚。
2.2 客户端与带 tools 的调用
client=OpenAI(api_key=os.environ["DASHSCOPE_API_KEY"],base_url="https://ws-ulkao56twirebft4.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",)messages=[]defget_ai_response(messages):completion=client.chat.completions.create(model="qwen-plus",messages=messages,temperature=0.75,tools=tools,# 把工具列表传进去)returncompletion关键点:tools=tools必须带上,否则模型不知道有工具可用。
2.3 判断是否需要调用工具 + 执行 + 回填
user_message={"role":"user","content":"北京今天的天气?"}messages.append(user_message)completion=get_ai_response(messages)messages.append(completion.choices[0].message)# 判断第一次返回:是否需要调用工具ifcompletion.choices[0].message.tool_callsisNone:print("不需要调用工具")print(completion.choices[0].message.content)else:print("需要调用工具")tool_calls=completion.choices[0].message.tool_callsfortool_callintool_calls:tool_id=tool_call.idfunc_name=tool_call.function.name func_arguments=tool_call.function.arguments function_mapping={"get_current_weather":get_current_weather}tool_result=function_mapping[func_name](json.loads(func_arguments))tool_message={"role":"tool","content":tool_result,"tool_call_id":tool_id,}messages.append(tool_message)completion=get_ai_response(messages)# 再调一次,拿到最终回答print(f"最终的结果是:{completion.choices[0].message.content}")流程要点:
tool_calls is None→ 模型直接回答了,不用调工具- 否则遍历
tool_calls(模型可能一次让你调多个工具) tool_call.function.arguments是JSON 字符串,要json.loads成 dict 再传给函数- 用
function_mapping字典映射函数名 → 函数对象,不要用eval(安全风险) - 回填的消息
role="tool"且必须带tool_call_id,与模型下发的 id 对应 - 再调一次
get_ai_response,模型才能基于工具结果生成自然语言回答
三、多工具:天气 + 学历验证llm/case6.py
在单工具基础上加第二个工具:学历验证。它要调外部 HTTP 接口,并用 Redis 缓存结果。
3.1 学历验证工具(外部 API + Redis 缓存)
importredisimportrequests redis_client=redis.Redis(host="localhost",port=6379,db=7,decode_responses=True,protocol=2,)defacademic_credential_verification(arguments):vcode=arguments["vcode"]key=f"boss:llm:academic_credential_verification:{vcode}"redis_verification_data=redis_client.get(key)ifredis_verification_dataisNone:BASE_URL="https://www.apimy.cn/api/xxw/bgcx"payload={"key":os.getenv("MY_XXW_BGCX_API_KEY"),"vcode":arguments["vcode"]}headers={"Content-Type":"application/json"}response=requests.post(BASE_URL,json=payload,headers=headers,timeout=30)response.raise_for_status()data=response.json()redis_client.set(key,json.dumps(data,ensure_ascii=False))returnjson.dumps(data,ensure_ascii=False)else:returnredis_verification_data要点:
- 先查 Redis 缓存:同一
vcode不重复打外部接口,省额度也更快 - 外部调用加
raise_for_status()+timeout:HTTP 异常和超时都要兜底,否则会卡死工具调用 - API Key 走环境变量
MY_XXW_BGCX_API_KEY,不硬编码
3.2 注册两个工具 + 映射
tools=[{"type":"function","function":{"name":"get_current_weather","description":"当你想查询指定城市的天气时非常有用","parameters":{"type":"object","properties":{"location":{"type":"string","description":"城市或县区"}},"required":["location"],},},},{"type":"function","function":{"name":"academic_credential_verification","description":"当你想查询学历或学历验证时非常有用","parameters":{"type":"object","properties":{"vcode":{"type":"string","description":"学历验证码"}},"required":["vcode"],},},},]function_mapping={"get_current_weather":get_current_weather,"academic_credential_verification":academic_credential_verification,}调用逻辑和单工具完全一致——遍历tool_calls,按function.name从function_mapping取函数执行。模型会根据用户问题自动选工具(问天气调天气、问学历验证码调学历)。
注意:
case5.py/case6.py目前是原生脚本(直接python跑),还没封装成 FastAPI 接口。脚本用来验证链路,下一步可像 Day1/Day2 那样包成POST /llm-day03/...路由对外提供。
四、踩坑清单(重点)
tools的parameters必须是合法 JSON Schema:type/properties/required缺一不可,required里列的字段模型才会保证给。description决定模型会不会调:描述模糊,模型容易「该调不调」或「乱调」。把触发场景写清楚。arguments是字符串不是 dict:tool_call.function.arguments是 JSON 字符串,必须json.loads才能当参数用。- 回填消息必须
role="tool"+tool_call_id:少一个模型会报错或忽略结果;tool_call_id要和模型下发的tool_call.id一一对应。 - 必须再调一次模型:调完工具把结果塞回
messages后,要再create一次,模型才会生成「基于工具结果的自然语言回答」——很多新手只执行工具就结束了,以为没输出。 - 一次可能调多个工具:
tool_calls是列表,记得for遍历,别只取[0]。 - 用字典映射代替
eval:function_mapping[name](args)安全;千万别eval(name)执行模型下发的字符串。 - 外部工具要加失败兜底:
raise_for_status()+timeout,并考虑接口超时 / 限流时给模型一个友好错误信息(作为 tool 结果回填)。 - 外部结果加缓存:像 case6 这样用 Redis 按业务 key 缓存,避免重复调用、省钱省时。
- API Key 走环境变量:
os.environ["DASHSCOPE_API_KEY"]缺省会抛KeyError(强制校验),生产建议os.getenv+ 显式判空返回友好错误。 - 温度建议调低:Function Calling 更看重参数格式正确,
temperature偏高(如 0.75)可能影响 JSON 参数稳定性,关键业务场景可调到 0.1~0.3。
五、小结
Day3 把「模型调用工具」跑通了:
- 单工具:天气查询,验证 Function Calling 全流程
- 多工具:天气 + 学历验证,后者接外部 API 并加 Redis 缓存
- 核心套路固定:定义 tools → 带 tools 调模型 → 判断 tool_calls → 执行工具 → role=tool 回填 → 再调一次拿最终回答
这是迈向AI Agent的第一步——模型从「只会聊天」变成「能动手查数据 / 调接口」。
后续可继续做:
- 把 Function Calling 封装成 FastAPI 流式接口(SSE + 工具执行进度)
- 多轮对话 + 工具调用结合(Day2 的 Redis 会话 + 今天工具)
- 工具执行异步化(外部 API 慢,用
httpx.AsyncClient+ 异步) - 工具结果做结构化校验(Pydantic 校验
arguments)
注:本文代码片段均来自当天真实提交的后端练习文件(
llm/case5.py、llm/case6.py),仅做脱敏(API Key 走环境变量)。case3.py/case4.py仍为全注释的多轮对话草稿,未纳入本篇。