本文围绕 Function Calling函数调用展开系统讲解了大语言模型调用外部工具的核心机制。文章首先说明 Function Calling 存在的根本原因——普通大模型只能生成自然语言、推理过程不透明且知识在训练时固定因此需要让模型像程序一样调用外部 API。随后通过完整流程图梳理了「用户提问 → 模型决定调用工具 → 本地执行函数 → 返回结果 → 模型生成最终答案」的调用链路并结合 Python 动态特性与元编程思想演示了如何通过函数映射表把模型返回的函数名和参数落地为真实调用。最后给出一个基于 OpenAI 兼容协议、以员工数据为背景的完整可运行代码示例覆盖工具 Schema 定义、函数分发、ReAct 循环简化版等关键环节是构建 Agent 与 ChatBI 应用的基础骨架。内容参考于图灵AI大模型全栈如下图红框的代码它是用来绑定工具的代码模型它仅仅是只能绑定工具并不会调用工具这个 Function Calling 就是模型调用工具的核心Function Calling存在的原因普通的大语言模型它只能生成自然语言它并不会也并不知道怎样才能调用工具普通的大模型的推理过程不是透明的就是说我们人类并不知道大模型是怎样推理的也无法控制它的推理过程大模型拥有的东西在训练的时候都固定了下来当它某个东西不存在时它就没办法完成所以Function Calling核心需求就是 让大模型像程序一样可以调用外部的API就跟程序员缺什么东西就引入缺少的库然后调用库里面提供的函数Function Calling是OpenAI提出的如下地址OpenAI Function Callinghttps://platform.openai.com/docs/guides/function-callingFunction Calling核心概念上图是 Function Calling 完整的流程图用户问题我们给大模型提出问题LLM决定调用工具大语言模型根据我们的问题确定要调用那些工具和工具的参数就是说这一步大模型会根据问题返回要调用的工具名和工具对应的入参执行函数我们自己手动调用函数就是说需要我们自己实现一个对应关系通过大模型的返回值调用对应的函数返回结果函数返回结果LLM生成最终答案把函数的返回结果给到大模型然后大模型根据函数的返回结果回答问题上方写了这么多它是怎样调用的函数呢如下简单实例如下图Python中的函数是可以给变量赋值的下图中把getName函数通过函数名赋值给了name变量然后name变量的值就会是一个函数它就可以被调用参数的传递下图中通过 ** 是Python的解包/收集运算符通过解包符号来解出fn_args的值当参数给到getName函数里如下图可以到三种传参方式它们做的事情都是一样的函数中文档字符串的获取通过函数名点__doc__就可以得到函数中的文档字符串了下方的代码就是LangChain调用函数的核心利用Python动态特性元编程通过上方简单实例我们就可以演变成下方的实例把函数放到一个字典里字典的key值就是函数的名字大模型就会返回这个名字然后大模型还会返回参数这样就能完成Agent中自动工具调用了利用Function Calling思想实现也就是带入大模型的代码示例上方Function Calling流程图的实现# # 【文件说明】 # 这是一个 LLM Function Calling函数调用 的完整示例代码。 # 核心流程用户提问 - 大模型决策要调用哪些函数 - 本地执行函数 - 结果回传给模型 - 模型生成最终自然语言回答。 # 这是构建 Agent智能体 和 ChatBI对话式数据分析 最基础的骨架。 # # pandas用于处理表格型数据类似 Excel这里作为模拟的数据库 import pandas as pd # OpenAI 官方 SDK本项目使用它来调用兼容 OpenAI 协议的大模型这里是阿里通义千问 # 只要模型服务商兼容 OpenAI 接口规范就能用同一个 client 调用 from openai import OpenAI # dotenv从 .env 文件里读取环境变量比如 API Key避免把密钥写死在代码里 from dotenv import load_dotenv # os读取环境变量 import os # json序列化/反序列化用于把函数返回值变成字符串、把模型返回的参数字符串变成字典 import json # numpy数值计算库此示例中未直接使用可作为后续扩展 import numpy as np # # 【第一步】加载 .env 文件中的环境变量 # 例如 .env 里写 # DASHSCOPE_API_KEYsk-xxxx # DASHSCOPE_BASE_URLhttps://dashscope.aliyuncs.com/compatible-mode/v1 # load_dotenv() # # 【第二步】配置大模型参数 # # 模型名称建议使用服务商推荐的、指令遵循能力强的模型 MODEL_NAME qwen3.7-flash # 从环境变量中读取 API Key不要硬编码到代码里 API_KEY os.getenv(DASHSCOPE_API_KEY) # 从环境变量中读取 API 的 Base URL兼容 OpenAI 的地址 BASE_URL os.getenv(DASHSCOPE_BASE_URL) # 创建 OpenAI 客户端实例后续所有请求都通过 client 发起 client OpenAI(api_keyAPI_KEY, base_urlBASE_URL) # # 【第三步】数据准备 # 用 DataFrame 模拟数据库表代码里所有函数都基于这份数据做分析 # df_employees pd.DataFrame({ Name: [Alice, Bob, Charlie, Diana, Eve, Frank, Grace, Hank], # 姓名 Age: [25, 30, 35, 28, 32, 45, 29, 40], # 年龄 Salary: [50000.0, 75000.5, 95000.75, 62000.0, 88000.25, 120000.0, 55000.0, 105000.0], # 年薪 Department: [IT, HR, IT, Finance, IT, Finance, HR, IT], # 部门 IsMarried: [True, False, True, False, True, True, False, True], # 婚否 YearsExperience: [3, 5, 8, 4, 7, 15, 4, 12] # 工作年限 }) def get_data_schema(): 生成数据集的 Schema结构描述用于注入到 System Prompt 里告诉模型 你手里有哪些数据、字段含义是什么、有哪些可选值。 ⚠️ 注意这里只描述结构不传递真实数据本身——既节省 Token又避免隐私泄露。 return f 数据集包含以下列 - Name (str): 员工姓名 - Age (int): 年龄 - Salary (float): 年薪 - Department (str): 部门 (包含: {, .join(df_employees[Department].unique())}) - IsMarried (bool): 婚姻状况 - YearsExperience (int): 工作年限 数据总行数: {len(df_employees)} # # 【第四步】业务函数定义 # 这些函数是真正干活的工具模型不会执行它们模型只会决定去调用哪个函数、传什么参数。 # 每个函数返回 json 字符串因为 messages 里的 content 必须是字符串。 # def calculate_salary_statistics(): 计算全公司薪资的统计指标均值、中位数、最大值、最小值 try: stats { average: round(df_employees[Salary].mean(), 2), # 平均薪资保留 2 位小数 median: round(df_employees[Salary].median(), 2), # 中位数 max: round(df_employees[Salary].max(), 2), # 最高薪资 min: round(df_employees[Salary].min(), 2) # 最低薪资 } # 转成 json 字符串返回tool message 的 content 必须是字符串 return json.dumps(stats) except Exception as e: # 出错时也返回 json方便模型理解发生了什么不要让异常直接冒泡中断流程 return json.dumps({error: str(e)}) def analyze_by_department(): 按部门分组统计人数、平均薪资、平均年龄 try: # groupby agg分组聚合类似 SQL 的 GROUP BY COUNT/AVG dept_stats df_employees.groupby(Department).agg({ Name: count, # 每个部门的人数用 Name 计数相当于 COUNT(*) Salary: mean, # 每个部门的平均薪资 Age: mean # 每个部门的平均年龄 }).round(2) # 保留 2 位小数 # 重命名列输出更语义化 result dept_stats.rename( columns{Name: count, Salary: avg_salary, Age: avg_age} ).to_dict(orientindex) # 转成 {部门: {字段: 值}} 的字典格式 return json.dumps(result, ensure_asciiFalse) # ensure_asciiFalse 保留中文 except Exception as e: return json.dumps({error: str(e)}) def find_employees_by_criteria(min_salaryNone, max_ageNone, departmentNone): 按条件筛选员工可选条件最低薪资、最大年龄、部门。 参数默认都是 None即用户没指定条件就不按该条件过滤。 try: df df_employees.copy() # 复制一份避免修改原始数据 if min_salary: # 薪资 min_salary df df[df[Salary] min_salary] if max_age: # 年龄 max_age df df[df[Age] max_age] if department: # 部门精确匹配 df df[df[Department] department] # 只输出关心的 4 个字段转成 list of dict result df[[Name, Department, Salary, Age]].to_dict(orientrecords) return json.dumps({count: len(result), data: result}, ensure_asciiFalse) except Exception as e: return json.dumps({error: str(e)}) def analyze_experience_salary_correlation(): 计算工作年限YearsExperience与薪资Salary之间的皮尔逊相关系数 try: # corr() 默认使用皮尔逊相关系数值域 [-1, 1] # 越接近 1 说明工龄与薪资越正相关越接近 0 说明几乎无线性相关 corr df_employees[YearsExperience].corr(df_employees[Salary]) return json.dumps({correlation_coefficient: round(corr, 4)}) except Exception as e: return json.dumps({error: str(e)}) # # 【第五步】函数映射表 # 模型返回的 tool_call 里只有函数名字符串和参数 JSON # 我们需要通过这张表把名字映射到真正的 Python 函数对象然后才能执行。 # 这是把模型决策落地为真实动作的关键一步本质就是一个 dispatcher分发器。 # FUNCTION_MAP { calculate_salary_statistics: calculate_salary_statistics, analyze_by_department: analyze_by_department, find_employees_by_criteria: find_employees_by_criteria, analyze_experience_salary_correlation: analyze_experience_salary_correlation, } # # 【第六步】Tools 定义Function Schema # 这份 schema 会随请求一起发给模型模型据此判断 # 用户的问题可以用哪个函数来解决需要传哪些参数 # 描述description写得越清晰模型选择工具就越准确。 # tools [ { type: function, # 固定写法表示这是一个可调用的函数 function: { # 函数的具体信息 name: calculate_salary_statistics, # 函数名必须和 FUNCTION_MAP 的 key 一致 description: 计算全公司员工薪资的统计指标平均值、中位数、最大最小, # 函数描述模型据此判断是否调用 parameters: {type: object, properties: {}, required: []} # 无参函数 } }, { type: function, function: { name: analyze_by_department, description: 按部门进行分组统计人数、平均薪资、平均年龄, parameters: {type: object, properties: {}, required: []} # 无参数 } }, { type: function, function: { name: find_employees_by_criteria, description: 筛选员工。如果不指定条件则不要传参。, parameters: { type: object, # 参数整体是一个对象即 key-value 结构 properties: { # 定义每个可选参数 min_salary: {type: number, description: 最低薪资}, max_age: {type: integer, description: 最大年龄}, department: {type: string, description: 部门名称} } # 没写 required说明三个参数都是可选的 } } }, { type: function, function: { name: analyze_experience_salary_correlation, description: 计算工作年限与薪资的相关系数, parameters: {type: object, properties: {}, required: []} } } ] # # 【第七步】核心执行逻辑ReAct 循环简化版Thought → Action → Observation → Answer # 完整的调用链 # 1) 把用户问题 tools schema 发给模型 # 2) 模型返回 tool_calls可能 0 个、1 个、多个 # 3) 本地逐个执行 tool_calls把结果作为 tool role 消息追加到 messages # 4) 再次把 messages 发给模型让模型基于工具结果生成最终自然语言回答 # def run_query(query): # 打印分隔线方便调试时看清楚是哪个问题的输出 print(f\n{ * 60}\n用户提问: {query}\n{ * 60}) # -------- 构造 messages -------- # System Prompt告诉模型它的角色、背景数据 Schema # User Prompt用户的原始提问 messages [ { role: system, content: f你是高级数据分析师。当前持有员工数据如下\n{get_data_schema()}\n请根据用户需求调用工具。 }, {role: user, content: query} ] # -------- 第 1 次调用 LLM -------- # 把 tools 一起传进去tool_choiceauto 表示由模型自己决定要不要调用工具 response client.chat.completions.create( modelMODEL_NAME, messagesmessages, toolstools, tool_choiceauto ) # 取出模型返回的消息对象可能包含 content也可能包含 tool_calls response_msg response.choices[0].message # 模型决定要调用的工具列表若为 None说明模型认为不需要调用工具 tool_calls response_msg.tool_calls # -------- 分支 A模型决定调用工具 -------- if tool_calls: print(f模型决定调用 {len(tool_calls)} 个工具...) # ⚠️ 关键一步必须把模型的这条回复含 tool_calls追加到 messages # 否则第 2 次请求时模型会找不到 tool_call_id 对应的上下文直接报错。 messages.append(response_msg) # 遍历所有 tool_calls这里演示的是同步、串行执行实际场景可以并行以加速 for tool_call in tool_calls: # 从 tool_call 中提取函数名 fn_name tool_call.function.name # 从 tool_call 中提取参数模型返回的是 JSON 字符串需要解析成 dict fn_args json.loads(tool_call.function.arguments) print(f - 执行工具: {fn_name} | 参数: {fn_args}) if fn_name in FUNCTION_MAP: # 根据函数名从映射表取出真正的函数对象并用 ** 解包参数调用它 # 等价于calculate_salary_statistics() 或 find_employees_by_criteria(min_salary80000, departmentIT) fn_result FUNCTION_MAP[fn_name](**fn_args) # 把执行结果作为一条 tool role 的消息加入 messages # - tool_call_id必须与上面的 tool_call.id 一一对应多工具并行时靠它区分 # - name函数名部分服务商非必填但带上更规范 # - content函数返回结果必须是字符串 messages.append({ role: tool, tool_call_id: tool_call.id, name: fn_name, content: fn_result }) print(f - 结果: {fn_result}) else: # 防御性编程模型幻觉编了一个不存在的函数名 print(fError: 函数 {fn_name} 未定义) # -------- 第 2 次调用 LLM -------- # 此时 messages 里包含了system user assistant(tool_calls) tool(结果...) # 模型基于这些上下文用自然语言总结出最终答案 final_response client.chat.completions.create( modelMODEL_NAME, messagesmessages ) print(f\n 最终回答:\n{final_response.choices[0].message.content}) # -------- 分支 B模型认为不需要调用工具直接回答 -------- else: print(f直接回答: {response_msg.content}) # # 【第八步】运行测试 # if __name__ __main__: # 测试用例 # 1) 简单查询应该调用 find_employees_by_criteria calculate_salary_statistics # 2) 复合查询可能触发并行调用一次返回多个 tool_calls queries [ IT部门有多少人他们的平均工资是多少, 帮我找一下工资高于8万的IT部门员工顺便算一下全公司的薪资相关性。, # 今天天气怎么样 # 可用于测试模型直接回答分支 ] # 遍历执行每个查询 for q in queries: run_query(q)