LLM function calling：让模型可靠地调你的工具（不是字符串解析）

起因

要做一个"自然语言查数据库"的功能。用户问"上周日北京下单的用户有几个？"
→ LLM 生成 SQL → 后端执行 → 返结果。

最原始做法是让 LLM 生成 SQL 字符串然后 regex 提取。痛点：

模型有时输出 sql ... markdown 包裹
有时多输出一段"分析这个 SQL..." 散文
有时 SQL 语法错（缺逗号、wrong table）
parse 失败要 try/except 重试

function calling（OpenAI 起的名，Anthropic 叫 tool use）让模型
直接结构化输出"调用什么函数 + 什么参数"，零解析。

解决方案：tool calling

定义工具

from openai import OpenAI

client = OpenAI()

tools = [
    {
        'type': 'function',
        'function': {
            'name': 'run_sql',
            'description': '在分析数据库上执行 SELECT SQL，返回最多 50 行。'
                          '只允许 SELECT；DELETE/UPDATE/INSERT 会被拒绝。',
            'parameters': {
                'type': 'object',
                'properties': {
                    'sql': {
                        'type': 'string',
                        'description': 'PostgreSQL 标准 SQL，必须以 SELECT 开头',
                    },
                    'explanation': {
                        'type': 'string',
                        'description': '一句话解释这个 SQL 在做什么',
                    },
                },
                'required': ['sql', 'explanation'],
            },
        },
    },
    {
        'type': 'function',
        'function': {
            'name': 'list_tables',
            'description': '列出可用表名',
            'parameters': {'type': 'object', 'properties': {}, 'required': []},
        },
    },
]

system_prompt = """
你是一个数据分析助手。回答用户问题前，可能需要：
1. 用 list_tables 看有哪些表
2. 用 run_sql 查数据

数据 schema：
- users(id, email, country, created_at, plan)
- orders(id, user_id, amount, city, ordered_at, status)
- products(id, name, category, price)

回答用户用中文。
"""

def chat(messages):
    return client.chat.completions.create(
        model='gpt-4o',
        messages=messages,
        tools=tools,
    )

调用循环

def run(user_question: str):
    messages = [
        {'role': 'system', 'content': system_prompt},
        {'role': 'user', 'content': user_question},
    ]
    while True:
        resp = chat(messages)
        msg = resp.choices[0].message

        # 模型决定调函数
        if msg.tool_calls:
            messages.append(msg)   # assistant turn
            for tc in msg.tool_calls:
                result = dispatch(tc.function.name, json.loads(tc.function.arguments))
                messages.append({
                    'role': 'tool',
                    'tool_call_id': tc.id,
                    'content': json.dumps(result),
                })
            # 继续下一轮，让模型看 tool 结果再决定
            continue

        # 模型给最终答案
        return msg.content

实现 dispatch（真的执行 SQL）

import psycopg

def dispatch(name, args):
    if name == 'list_tables':
        return list_tables()
    if name == 'run_sql':
        return run_sql(args['sql'])
    return {'error': f'unknown function: {name}'}

def list_tables():
    with psycopg.connect(DB_URL) as conn:
        cur = conn.execute("""
            SELECT table_name, obj_description(...)
            FROM information_schema.tables
            WHERE table_schema='public'
        """)
        return [{'table': r[0], 'desc': r[1]} for r in cur.fetchall()]

def run_sql(sql: str):
    if not sql.strip().upper().startswith('SELECT'):
        return {'error': 'only SELECT allowed'}
    try:
        with psycopg.connect(DB_URL, autocommit=False) as conn:
            conn.execute('SET statement_timeout=10000')   # 10s 限时
            cur = conn.execute(sql)
            rows = [dict(zip([c[0] for c in cur.description], r))
                    for r in cur.fetchmany(50)]
            return {'rows': rows, 'count': len(rows)}
    except Exception as e:
        return {'error': str(e)}

跑一下

print(run('上周日北京下单的用户有几个？'))
# 输出：
# 上周日（2024-05-19）在北京下单的用户共 47 个。

模型自动：

调 list_tables() 看有哪些
调 run_sql('SELECT COUNT(DISTINCT user_id) FROM orders WHERE city=...')
拿到结果后用自然语言回答

整套流程无字符串解析——arguments 已经是 typed JSON。

几个重要细节

1. 多 tool 同时调

if msg.tool_calls:
    # 可能一次调 2-3 个函数
    for tc in msg.tool_calls:
        ...

模型可能并行调 list_tables + run_sql。要 loop 处理所有。

2. 防恶意 SQL

工具签名再严格也挡不住"DROP TABLE users; --" 写在 SQL 字符串里。
在 dispatch 层做实际验证：

只允许 SELECT 开头
用只读 DB 用户（无 DDL / DML 权限）
SET statement_timeout=N 限时长
ACL 限 schema / table 访问
用 SQLAlchemy text(sql) + 参数化（更难做）

或者更激进：sandbox 跑（DuckDB on read-only data copy）。

3. 限制循环次数

模型可能死循环调工具。限步数：

for step in range(10):
    resp = chat(messages)
    if not msg.tool_calls:
        return msg.content
    # ...
raise RuntimeError('exceeded 10 tool-use steps')

4. parallel tool call

OpenAI 默认开 parallel；要禁用：

resp = client.chat.completions.create(..., parallel_tool_calls=False)

复杂任务有 dependency 时禁用更稳。

5. tool_choice 强制

client.chat.completions.create(
    ...,
    tool_choice={'type': 'function', 'function': {'name': 'run_sql'}},
)

强制本轮调某个函数（不让模型 freestyle 直接答）。

Anthropic / Gemini / Ollama 也支持

API 风格略不同但概念一致。

# Anthropic
import anthropic
client = anthropic.Anthropic()
resp = client.messages.create(
    model='claude-sonnet-4-5',
    max_tokens=1024,
    tools=[{'name': 'run_sql', 'description': '...',
            'input_schema': {...}}],
    messages=[{'role': 'user', 'content': '...'}],
)
# stop_reason='tool_use' 时遍历 content blocks

# Ollama (qwen2.5 等支持 function calling)
import ollama
resp = ollama.chat(
    model='qwen2.5:7b',
    messages=[...],
    tools=[{'type': 'function', 'function': {...}}],
)

跨家 LLM 工具调用接口已经形成事实标准（OpenAI 格式被大部分模仿）。

实际应用场景

数据库 query agent（上面例子）
代码 review bot：tool 是 read_file / list_files / run_tests
客服 agent：lookup_order / refund / escalate
DevOps agent：check_deployment / rollback / fetch_logs
RAG with citations：search_docs / fetch_doc tool

任何"模型需要查外部信息再回答" 都适合。

与 LangChain / LlamaIndex 的关系

LangChain create_react_agent / create_tool_calling_agent 是上面
循环的封装：

from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain.tools import tool

@tool
def run_sql(sql: str) -> str:
    """Execute SELECT SQL, return rows."""
    return run_sql_impl(sql)

llm = ChatOpenAI(model='gpt-4o')
agent = create_tool_calling_agent(llm, [run_sql], prompt)
executor = AgentExecutor(agent=agent, tools=[run_sql], max_iterations=10)
executor.invoke({'input': '上周日北京...'})

封装方便但藏了细节。简单场景手写循环更可控；复杂 agent 用 LangChain
省事。

效果

我们的 SQL agent 上线后：

业务团队不再用 Metabase 拖拽，直接问中文
"为什么这个客户流失了" 类自由查询能 90% 准确给出 SQL + 结果
function calling 解析失败率：0%（结构化 output）
vs 之前用 regex 提取 SQL 字符串：~12% 失败要重试

踩过的坑

参数 schema 错：模型按 schema 生成参数，schema 不严会乱传。
required / enum / type 都明确写。
大 tool 数 → 模型 confused：超过 ~10 个 tool 后模型选错率
上升。分层：top-level "router" → 选 sub-agent → sub-agent 有
3-5 个 tool。
tool 实现 crash 抛异常：返 {'error': str(e)} 让模型看见，
模型会 retry / 换策略。直接 raise 就只能上层 catch。
token 成本：tool description + schema 占 system prompt 不少
token。每次请求都付。优化：精简 description / 用 short tool name。
流式（streaming）+ tool：streaming response 中 tool_call
chunks 是分段的，要 buffer 后再 parse。复杂场景非流式更稳。