Function Call学习
1、概述
大模型根据语言理解,调用外部工具,比如我们自己定义的函数,或工具,把参数传给函数。
【作用】
1、扩展模型的能力
模型本身无法直接操作外部系统,比如:数据库等。通过调用预设函数(Function call),就可以完成发送邮件、获取天气等功能,
就打通了与外部系统的交互,突破模型本身的能力边界。
2、结构话输出
模型将用户自然语言格式化成结构话参数,传递给函数,比如:
用户的自然语言请求(如 “明天北京天气如何?”)解析并格式化为函数可直接使用的结构化参数(如 location="北京", date="2025-05-06"),让非结构化的语言变成机器能执行的指令。
3、模型可根据上下文决定是否/何时调用函数,甚至可以链式调用多个函数
【流程步骤】
步骤 1:定义工具 / 函数 schema(明确函数能做什么、参数要什么)
步骤 2:大模型解析用户自然语言,匹配对应函数
步骤 3:生成结构化的函数调用参数(如 get_weather("北京", "2026-03-30"))
步骤 4:外部程序执行函数,获取返回结果
步骤 5:大模型结合结果,生成自然语言回答
2、细节点学习
1、qwen-agent(Function-call)工具类定义步骤
1、使用 @register_tool(工具名) 注册工具
2、定义类继承BaseTool,自定义工具类必须继承这个基类,基类提供了call方法规范、参数解析、异常捕获能力
3、description定义工具的描述,这是给大模型看的说明书,模型根据描述判断什么时候调用这个工具
4、def call 是模型真正调用的函数,工具的执行入口,params模型传过来的参数(字典或json字符串),返回值必须是str
示例代码:
@register_tool('exc_sql') class ExcSQLTool(BaseTool): """ SQL查询工具,执行传入的SQL语句并返回结果。 """ description = '对于生成的SQL,进行SQL查询' parameters = [{ 'name': 'sql_input', 'type': 'string', 'description': '生成的SQL语句', 'required': True }] def call(self, params: str, **kwargs) -> str: import json args = json.loads(params) sql_input = args['sql_input'] database = args.get('database', 'ubr') # 创建数据库连接 engine = create_engine( f'mysql+mysqlconnector://xxxxxxxxxxxxxxxxxxxxxxxxxxx/{database}?charset=utf8mb4', connect_args={'connect_timeout': 10}, pool_size=10, max_overflow=20 ) try: df = pd.read_sql(sql_input, engine) # 返回前10行,防止数据过多 return df.head(10).to_markdown(index=False) except Exception as e: return f"SQL执行出错: {str(e)}"2、使用function call 构建agent demo
需要安装依赖:pip install qwen-agent
from asyncio import streams import os from urllib import response import dashscope import json from qwen_agent.agents import Assistant from qwen_agent.tools.base import BaseTool, register_tool # 配置密钥 dashscope.api_key = os.getenv("DASHSCOPE_API_KEY", "") # ====================== # Step1: 注册工具名 # ====================== @register_tool("calculator") # ====================== # Step2: 继承 BaseTool # ====================== class CalculatorTool(BaseTool): # ====================== # Step3: 描述 + 参数定义 # ====================== name = "calculator" description = "用于进行加减乘除数学计算" parameters = [ { "name": "a", "type": "number", "description": "数字a", "required": True }, { "name": "b", "type": "number", "description": "数字b", "required": True }, { "name": "op", "type": "string", "description": "运算符:+ - * /", "required": True } ] # ====================== # Step4: call 执行入口 # ====================== def call(self, params: dict, **kwargs) -> str: # 关键修复:如果 params 是字符串,先解析为字典 if isinstance(params, str): params = json.loads(params) a = params["a"] b = params["b"] op = params["op"] try: if op == "+": res = a + b elif op == "-": res = a - b elif op == "*": res = a * b elif op == "/": if b == 0: return "错误:除数不能为0" res = a / b else: return "不支持的运算符" return f"计算结果:{a} {op} {b} = {res}" except Exception as e: return f"计算错误:{str(e)}" # ====================== # 初始化 Agent # ====================== def init_agent(): llm_cfg = { "model": "qwen-max", "stream": False } bot = Assistant( llm=llm_cfg, system_message="你是一个擅长计算的助手,遇到数学问题请调用 calculator 工具求解", function_list=["calculator"], # 必须和注册名一致 ) return bot # ====================== # 测试对话 # ====================== if __name__ == "__main__": # 初始化助手 bot = init_agent() messages = [{"role": "user", "content": "3 加 5 等于多少?"}] response = list(bot.run(messages)) for res in response: # 判断是否为字典类型 if isinstance(res, dict): if res.get("role") == "assistant" and res.get("content"): print("\nBot 返回:", res["content"]) break # 找到第一个有效回答就停止 elif isinstance(res, list): # 如果是列表,遍历其中的每个字典项 for item in res: if isinstance(item, dict) and item.get("role") == "assistant" and item.get("content"): print("\nBot 返回:", item["content"]) break