一、前言

随着 Agent 智能体开发热度攀升,Tool Calling(工具调用) 已经是构建具备业务执行能力大模型应用的核心技术。单纯的对话大模型只能做文本生成,而通过绑定自定义工具(函数、数据库、接口),智能体可以自主判断何时调用工具、解析参数、执行业务逻辑,实现从 “聊天” 到 “干活” 的质变。

本文基于智谱 AI GLM-4 Plus完整落地两套实战场景:

  1. 基础工具调用:自定义 CRM 订单查询工具,手把手拆解 Tool Schema 定义、模型调用、tool_calls 解析全链路;
  2. 进阶 Text-to-SQL 智能体:基于内存 SQLite 数据库,让大模型读懂数据表结构,自然语言自动生成可执行 SQL 并查询业务数据。

全部代码可直接复制运行,配套环境初始化、日志规范、异常捕获、模拟数据库,适合零基础入门 Agent 工具开发,同时包含工业级项目规范。

环境依赖:Python3.9+、python-dotenv、zhipuai、sqlite3(内置无需额外安装)

二、项目整体架构分层(第一阶段:智能体内核与协议标准)

本项目分为三层核心模块,符合工业级 Agent 开发分层规范:

  1. 内核初始化层:加载环境变量、日志配置、智谱 AI 客户端全局实例,统一管理模型参数;
  2. Tool 协议标准层:遵循 OpenAI 兼容 Function Calling Schema 规范,标准化定义工具入参、描述、必填项;
  3. 业务执行层:分为两类业务:CRM 函数工具调用、本地 SQLite 文本转 SQL 查询。

2.1 全局内核初始化代码(统一客户端与日志)

这是整个项目的基础文件,统一管理 API 密钥、日志输出、模型名称,所有后续工具调用均复用该client实例。

import os
import json
import logging
from typing import Dict, Any
from dotenv import load_dotenv
from zhipuai import ZhipuAI

# ====================== 工业级日志格式化配置 ======================
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# ====================== 加载环境变量 ======================
load_dotenv()
# 从.env文件读取智谱密钥与模型名称
ZHIPUAI_API_KEY = os.getenv("ZHIPUAI_API_KEY")
MODEL_NAME = os.getenv("CHAT_MODEL", "glm-4-plus")

# 密钥校验,缺失直接抛出异常阻断程序
if not ZHIPUAI_API_KEY:
    logger.error("环境变量 ZHIPUAI_API_KEY 未找到,请检查 .env 配置!")
    raise ValueError("Missing ZHIPUAI_API_KEY")

# 全局唯一智谱AI客户端实例
client = ZhipuAI(api_key=ZHIPUAI_API_KEY)
logger.info(f"✅ 环境初始化完成!当前模型:{MODEL_NAME}")

# 全局tools列表,用于存放所有自定义工具定义
tools = []

配套.env文件配置(项目根目录新建):

ZHIPUAI_API_KEY=你的智谱平台API密钥
CHAT_MODEL=glm-4-plus

三、核心知识点:Tool Schema 协议标准详解

大模型无法直接识别 Python 函数,必须通过一套标准化 JSON Schema 格式给模型 “写说明书”,也就是tools参数,这是 Tool Calling 的核心协议。

3.1 Tool 标准结构拆解

demo_tools = [
    {
        "type": "function",  # 固定值,代表函数工具类型
        "function": {
            "name": "get_weather",  # 函数名,模型调用时会携带该标识
            "description": "获取指定城市的天气信息",  # 工具用途描述,引导模型判断调用时机
            "parameters": {  # 参数JSON Schema定义
                "type": "object",
                "properties": {  # 所有入参字段
                    "city": {
                        "type": "string",
                        "description": "需要查询的城市名称"
                    },
                    "date": {
                        "type": "string",
                        "description": "查询日期,可选,不填默认今天"
                    }
                },
                "required": ["city"]  # 必填参数列表,模型调用时必须携带
            }
        }
    }
]

字段核心作用总结:

表格

字段 作用
type 固定function,标识这是函数调用工具
name 函数唯一标识,模型返回 tool_calls 时通过该字段匹配本地业务函数
description 关键!决定模型什么时候会调用工具,描述越清晰,模型调用准确率越高
parameters.properties 定义每个入参的类型、说明,约束模型输出参数格式
required 强制必填参数,模型不会生成缺失该字段的调用请求

3.2 工具调用完整交互流程

  1. 组装messages对话上下文 + tools工具列表传入大模型接口;
  2. 模型自主判断:是否需要调用工具,需要则返回tool_calls数组;不需要则直接返回文本回答;
  3. 解析tool_calls:提取函数名、JSON 格式参数;
  4. 本地执行对应 Python 业务函数,拿到工具返回结果;
  5. 将工具执行结果追加进messages,二次调用大模型整合结果输出给用户。

四、实战 1:CRM 订单查询自定义 Tool 完整实现

4.1 第一步:本地业务函数(模拟数据库)

先实现底层业务逻辑,模拟 CRM 订单数据库,提供get_order_status查询接口:

def get_order_status(order_id: str) -> str:
    """
    从模拟的中央数据库中获取订单状态
    Args:
        order_id (str): 订单ID,例如 'ORD1001'
    Returns:
        str: 标准化的JSON字符串结果
    """
    # 模拟数据库存储订单数据
    mock_db: Dict[str, str] = {
        "ORD1001": "已发货",
        "ORD1002": "处理中",
        "ORD1003": "已退款"
    }
    # 查询订单,无匹配则返回未知订单
    status = mock_db.get(order_id, "未知订单")
    result: Dict[str, Any] = {
        "order_id": order_id,
        "status": status
    }
    # 返回JSON格式化字符串,方便后续传给大模型
    return json.dumps(result, ensure_ascii=False)

# 本地测试函数
logger.info(f"🔍 测试函数输出: {get_order_status('ORD1001')}")

4.2 第二步:为函数定义标准 Tool Schema

按照协议规范,给get_order_status生成模型可识别的工具描述:

# CRM订单查询工具Schema定义
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_order_status",
            "description": "从 CRM 系统中获取指定订单的状态信息,传入订单ID即可查询发货、退款、处理中状态",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "订单ID,格式示例 'ORD1001'"
                    }
                },
                "required": ["order_id"]
            }
        }
    }
]

4.3 第三步:调用大模型并解析 tool_calls 完整代码

发送用户问题,接收模型返回、判断是否调用工具、提取参数执行业务:

# 用户对话上下文
messages = [{"role": "user", "content": "帮我查一下订单ORD1002现在是什么状态?"}]

try:
    # 调用智谱大模型,传入工具列表,tool-choice=auto自动决策是否调用工具
    response = client.chat.completions.create(
        model=MODEL_NAME,
        messages=messages,
        tools=tools,
        tool_choice="auto"
    )
    response_message = response.choices[0].message

    # 判断模型是否触发工具调用
    if response_message.tool_calls:
        logger.info("🤖 模型决定调用工具!")
        for tool_call in response_message.tool_calls:
            # 提取工具名称与参数
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)
            logger.info(f"🔧 被调用的函数: {func_name}")
            logger.info(f"📥 抽取的参数: {func_args}")

            # 匹配本地函数执行逻辑
            if func_name == "get_order_status":
                order_res = get_order_status(func_args["order_id"])
                logger.info(f"📤 工具执行结果: {order_res}")
    else:
        # 模型直接回答,无需工具
        logger.info("💬 模型直接回复了,没有调用工具。")
        logger.info(f"回复内容: {response_message.content}")

except Exception as e:
    logger.error(f"❌ 调用大模型出错: {e}")

4.4 运行效果说明

  1. 用户提问帮我查一下订单ORD1002现在是什么状态?,模型识别需要调用get_order_status
  2. 返回tool_calls,参数{"order_id": "ORD1002"}
  3. 本地函数执行,返回{"order_id":"ORD1002","status":"处理中"}
  4. 实际生产场景中,会把该结果追加进消息列表,再次请求大模型生成自然语言总结回复用户。

五、实战 2:进阶 Text-to-SQL 智能体(本地 SQLite 内存数据库)

单纯调用 Python 函数只能处理简单逻辑,复杂多表查询、数据统计场景需要文本转 SQL能力,结合 Tool Calling 可实现全自动数据查询 Agent。

5.1 第一步:初始化内存 SQLite 数据库与业务表

使用内存数据库,无需本地文件,程序结束自动销毁,适合演示:

import sqlite3

# 数据库表结构描述(传给大模型的Schema说明书)
schema_info = """
Table: users
- id (INTEGER PRIMARY KEY)
- name (TEXT)
- email (TEXT)

Table: orders
- order_id (TEXT PRIMARY KEY)
- user_id (INTEGER, FOREIGN KEY references users.id)
- product_name (TEXT)
- amount (REAL)
- status (TEXT)
"""

def init_mock_db():
    """初始化内存SQLite,创建用户、订单表并插入测试数据"""
    conn = sqlite3.connect(":memory:")
    cursor = conn.cursor()
    # 创建用户表
    cursor.execute("""
        CREATE TABLE users (
            id INTEGER PRIMARY KEY,
            name TEXT,
            email TEXT
        )
    """)
    # 创建订单表,关联用户外键
    cursor.execute("""
        CREATE TABLE orders (
            order_id TEXT PRIMARY KEY,
            user_id INTEGER,
            product_name TEXT,
            amount REAL,
            status TEXT,
            FOREIGN KEY(user_id) REFERENCES users(id)
        )
    """)
    # 插入测试用户数据
    cursor.executemany("INSERT INTO users VALUES (?, ?, ?)", [
        (1, "张三", "zhangsan@example.com"),
        (2, "李四", "lisi@example.com")
    ])
    # 插入测试订单数据
    cursor.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", [
        ("ORD1001", 1, "笔记本电脑", 5999.0, "已发货"),
        ("ORD1002", 1, "机械键盘", 399.0, "处理中"),
        ("ORD1003", 2, "鼠标", 199.0, "已退款")
    ])
    conn.commit()
    return conn

5.2 第二步:构建系统 Prompt,引导模型生成标准 SQL

核心思路:把数据表结构全部交给大模型,限定输出格式,去除多余文本,只返回可执行 SQL:

# 用户自然语言查询需求
user_question = "查询张三所有已发货的订单产品名称和金额"

# 系统提示词,约束模型行为
system_prompt = f"""你是一个专业的 SQL 专家。请根据以下数据库 Schema 将用户的自然语言问题转换为标准的 SQLite SQL 语句。
只输出 SQL 语句,不要包含任何解释或 Markdown 格式标记(如 ```sql)。

Schema:
{schema_info}
"""

# 组装对话上下文
messages = [
    {"role": "system", "content": system_prompt},
    {"role": "user", "content": user_question}
]

5.3 第三步:调用模型生成 SQL + 本地执行查询

包含 SQL 清洗逻辑(去除 markdown 代码块标记)、异常捕获、结果打印:

try:
    # 请求大模型生成SQL
    response = client.chat.completions.create(
        model=MODEL_NAME,
        messages=messages
    )
    generated_sql = response.choices[0].message.content.strip()

    # 清洗模型输出的Markdown代码块标记
    if generated_sql.startswith("```sql"):
        generated_sql = generated_sql[6:-3].strip()
    elif generated_sql.startswith("```"):
        generated_sql = generated_sql[3:-3].strip()
    logger.info(f"🤖 生成的SQL: {generated_sql}")

    # 初始化数据库执行SQL
    conn = init_mock_db()
    cursor = conn.cursor()
    # 生产环境提示:此处必须增加SQL注入校验,本Demo仅演示
    cursor.execute(generated_sql)
    results = cursor.fetchall()
    logger.info(f"✅ 查询结果: {results}")
    conn.close()

except Exception as e:
    logger.error(f"❌ Text-to-SQL 执行出错: {e}")

5.4 生成 SQL 与查询结果演示

用户问题:查询张三所有已发货的订单产品名称和金额 模型输出 SQL:

SELECT o.product_name, o.amount
FROM users u
JOIN orders o ON u.id = o.user_id
WHERE u.name = '张三' AND o.status = '已发货'

程序执行后输出结果:[('笔记本电脑', 5999.0)]

六、工业级开发避坑总结

6.1 Tool Calling 常见问题

  1. 模型不调用工具description描述模糊,没有清晰说明工具使用场景;或tool_choice="none"强制关闭工具调用;
  2. 参数解析报错 JSON loads 失败:模型输出参数携带多余注释、换行,可增加参数清洗逻辑;
  3. client 未定义报错:代码执行顺序错误,先运行工具调用代码,再初始化客户端;
  4. 必填参数缺失:Schema 中required数组未配置完整,模型随机省略关键参数。

6.2 Text-to-SQL 生产安全规范

  1. 禁止直接执行模型生成 SQL:增加表名、字段白名单校验,拦截DROP/ALTER/DELETE危险语句,防止 SQL 注入;
  2. Schema 精简:不要把全量数据库表交给模型,仅传入当前业务需要的表结构,减少模型混淆;
  3. 错误重试机制:SQL 执行失败时,把报错信息传回大模型,自动修正 SQL 二次查询。

6.3 项目规范优化建议

  1. 环境变量统一管理,密钥禁止硬编码在代码中;
  2. 全局日志统一格式化,区分 INFO/ERROR 日志等级,方便线上排查;
  3. 工具函数、Schema、业务逻辑分层拆分文件,大型 Agent 项目解耦;
  4. 所有外部接口、数据库操作增加 try-except 全局异常捕获。

七、完整项目拓展方向

  1. 多工具联动:同时绑定天气查询、订单查询、数据库查询,模型自主选择对应工具;
  2. 工具返回结果二次总结:把工具执行结果追加进 messages,让大模型整合数据输出通顺自然语言;
  3. 持久化数据库:替换内存 SQLite 为 MySQL/PostgreSQL,对接真实业务库;
  4. 前端封装:基于 FastAPI 封装接口,Web 页面输入自然语言,展示工具调用链路与查询结果。

八、结语

Tool Calling 是 Agent 开发的基石,本文从底层内核初始化、标准 Schema 协议、简单函数工具、文本转 SQL 复杂数据库查询由浅入深完成落地。对比纯对话模型,具备工具调用能力的智能体可以打通企业 CRM、数据库、第三方 API,真正落地业务自动化场景。

本文全部代码基于智谱 GLM-4 Plus 实现,兼容 OpenAI 接口规范,替换其他大模型仅需修改客户端初始化代码,通用性极强。

Logo

汇聚全球AI编程工具,助力开发者即刻编程。

更多推荐