企业官网建设流程全解析

本文围绕 Function Calling（函数调用）展开，系统讲解了大语言模型调用外部工具的核心机制。文章首先说明 Function Calling 存在的根本原因——普通大模型只能生成自然语言、推理过程不透明且知识在训练时固定，因此需要让模型像程序一样调用外部 API。随后通过完整流程图梳理了「用户提问 → 模型决定调用工具 → 本地执行函数 → 返回结果 → 模型生成最终答案」的调用链路，并结合 Python 动态特性与元编程思想，演示了如何通过函数映射表把模型返回的函数名和参数落地为真实调用。最后给出一个基于 OpenAI 兼容协议、以员工数据为背景的完整可运行代码示例，覆盖工具 Schema 定义、函数分发、ReAct 循环简化版等关键环节，是构建 Agent 与 ChatBI 应用的基础骨架。

内容参考于：图灵AI大模型全栈

如下图红框的代码，它是用来绑定工具的代码，模型它仅仅是只能绑定工具，并不会调用工具，这个 Function Calling 就是模型调用工具的核心

Function Calling存在的原因

普通的大语言模型它只能生成自然语言，它并不会也并不知道怎样才能调用工具

普通的大模型的推理过程不是透明的，就是说我们人类并不知道，大模型是怎样推理的，也无法控制它的推理过程

大模型拥有的东西在训练的时候都固定了下来，当它某个东西不存在时它就没办法完成，所以Function Calling核心需求就是 让大模型像程序一样可以调用外部的API，就跟程序员缺什么东西就引入缺少的库，然后调用库里面提供的函数

Function Calling是OpenAI提出的，如下地址

OpenAI Function Calling：https://platform.openai.com/docs/guides/function-calling

Function 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_KEY=sk-xxxx # DASHSCOPE_BASE_URL=https://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_key=API_KEY, base_url=BASE_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(orient='index') # 转成 {部门: {字段: 值}} 的字典格式 return json.dumps(result, ensure_ascii=False) # ensure_ascii=False 保留中文 except Exception as e: return json.dumps({"error": str(e)}) def find_employees_by_criteria(min_salary=None, max_age=None, department=None): """ 按条件筛选员工（可选条件：最低薪资、最大年龄、部门）。 参数默认都是 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(orient='records') return json.dumps({"count": len(result), "data": result}, ensure_ascii=False) 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_choice="auto" 表示"由模型自己决定要不要调用工具" response = client.chat.completions.create( model=MODEL_NAME, messages=messages, tools=tools, tool_choice="auto" ) # 取出模型返回的消息对象（可能包含 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_salary=80000, department="IT") 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(f"Error: 函数 {fn_name} 未定义") # -------- 第 2 次调用 LLM -------- # 此时 messages 里包含了：system + user + assistant(tool_calls) + tool(结果...) # 模型基于这些上下文，用自然语言总结出最终答案 final_response = client.chat.completions.create( model=MODEL_NAME, messages=messages ) 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)