5 步让大模型串联 10+ 内部 API：Function Calling 进阶实战

用 Function Calling 把企业内部 API 串联成自动化工作流，附完整 Python 代码。

---

为什么需要 Function Calling 进阶？

基础版 Function Calling 只能调一个 API，但实际业务场景是：

查客户信息 → 调用 CRM API

查库存 → 调用 ERP API

创建工单 → 调用 OA API

发通知 → 调用企微 API

一个用户问题，需要串联多个内部系统。 这就是本文要解决的问题。

---

核心架构设计

用户提问
   ↓
大模型分析意图
   ↓
决定调用哪些 API（多步骤规划）
   ↓
依次执行 API 调用
   ↓
汇总结果，生成回复

---

Step 1：定义内部 API 工具集

假设我们有 3 个内部系统：CRM、库存管理、通知系统。

# internal_apis.py
import requests
import json

# 配置
API_BASE = "https://internal-api.company.com"
HEADERS = {"Authorization": "Bearer YOUR_TOKEN"}

def crm_get_customer(phone: str) -> dict:
    """根据手机号查询客户信息"""
    resp = requests.get(
        f"{API_BASE}/crm/customer",
        params={"phone": phone},
        headers=HEADERS,
        timeout=10
    )
    resp.raise_for_status()
    return resp.json()

def crm_create_order(customer_id: str, product: str, amount: float) -> dict:
    """为客户创建订单"""
    resp = requests.post(
        f"{API_BASE}/crm/order",
        json={"customer_id": customer_id, "product": product, "amount": amount},
        headers=HEADERS,
        timeout=10
    )
    resp.raise_for_status()
    return resp.json()

def inventory_check(product_id: str) -> dict:
    """查询库存数量"""
    resp = requests.get(
        f"{API_BASE}/inventory/check",
        params={"product_id": product_id},
        headers=HEADERS,
        timeout=10
    )
    resp.raise_for_status()
    return resp.json()

def notify_wechat(user_id: str, message: str) -> dict:
    """发送企微通知"""
    resp = requests.post(
        f"{API_BASE}/notify/wechat",
        json={"user_id": user_id, "message": message},
        headers=HEADERS,
        timeout=10
    )
    resp.raise_for_status()
    return resp.json()

# 把所有 API 封装成可调用的方法字典
AVAILABLE_FUNCTIONS = {
    "crm_get_customer": crm_get_customer,
    "crm_create_order": crm_create_order,
    "inventory_check": inventory_check,
    "notify_wechat": notify_wechat,
}

---

Step 2：定义 Function 描述（给大模型看）

# function_definitions.py
functions = [
    {
        "name": "crm_get_customer",
        "description": "根据手机号查询客户信息，返回客户ID、姓名、会员等级",
        "parameters": {
            "type": "object",
            "properties": {
                "phone": {
                    "type": "string",
                    "description": "客户手机号，如 13800138000"
                }
            },
            "required": ["phone"]
        }
    },
    {
        "name": "crm_create_order",
        "description": "为客户创建订单，需要客户ID、产品名称和金额",
        "parameters": {
            "type": "object",
            "properties": {
                "customer_id": {"type": "string", "description": "客户ID"},
                "product": {"type": "string", "description": "产品名称"},
                "amount": {"type": "number", "description": "订单金额"}
            },
            "required": ["customer_id", "product", "amount"]
        }
    },
    {
        "name": "inventory_check",
        "description": "查询指定产品的库存数量",
        "parameters": {
            "type": "object",
            "properties": {
                "product_id": {"type": "string", "description": "产品ID"}
            },
            "required": ["product_id"]
        }
    },
    {
        "name": "notify_wechat",
        "description": "向指定用户发送企微通知",
        "parameters": {
            "type": "object",
            "properties": {
                "user_id": {"type": "string", "description": "企微用户ID"},
                "message": {"type": "string", "description": "通知内容"}
            },
            "required": ["user_id", "message"]
        }
    }
]

---

Step 3：实现多步 Function Calling 循环

# agent.py
import openai
import json
from internal_apis import AVAILABLE_FUNCTIONS
from function_definitions import functions

client = openai.OpenAI(api_key="YOUR_API_KEY")

def run_agent(user_query: str) -> str:
    """
    多步 Function Calling Agent
    支持连续调用多个 API，直到大模型认为可以回复用户
    """
    messages = [
        {
            "role": "system",
            "content": "你是企业智能助手，可以调用内部API帮用户查询客户信息、创建订单、查库存、发通知。如果缺少参数，主动向用户询问。"
        },
        {"role": "user", "content": user_query}
    ]
    
    max_iterations = 10  # 防止无限循环
    iteration = 0
    
    while iteration < max_iterations:
        iteration += 1
        
        # 调用大模型，启用 Function Calling
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=[{"type": "function", "function": f} for f in functions],
            tool_choice="auto"  # 让模型自己决定是否调用函数
        )
        
        message = response.choices[0].message
        
        # 情况1：大模型直接回复（不需要调用函数）
        if not message.tool_calls:
            return message.content
        
        # 情况2：大模型要求调用函数
        messages.append(message)  # 把模型的「我想调用函数」加入历史
        
        for tool_call in message.tool_calls:
            function_name = tool_call.function.name
            function_args = json.loads(tool_call.function.arguments)
            
            print(f"🔧 调用函数: {function_name}, 参数: {function_args}")
            
            # 执行对应的内部 API
            if function_name in AVAILABLE_FUNCTIONS:
                func = AVAILABLE_FUNCTIONS[function_name]
                try:
                    result = func(**function_args)
                    result_str = json.dumps(result, ensure_ascii=False)
                except Exception as e:
                    result_str = json.dumps({"error": str(e)}, ensure_ascii=False)
            else:
                result_str = json.dumps({"error": "函数不存在"}, ensure_ascii=False)
            
            # 把函数执行结果返回给大模型
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "name": function_name,
                "content": result_str
            })
    
    return "执行步骤过多，已终止"

# 测试
if __name__ == "__main__":
    result = run_agent("帮我用手机 13800138000 查一下这个客户的信息，然后通知他订单已发货")
    print("🤖 回复:", result)

---

Step 4：增加上下文记忆和错误处理

# agent_v2.py - 增强版
import sqlite3
from datetime import datetime

class FunctionCallAgent:
    def __init__(self, model="gpt-4o"):
        self.client = openai.OpenAI(api_key="YOUR_API_KEY")
        self.model = model
        self.db = sqlite3.connect("agent_history.db", check_same_thread=False)
        self.init_db()
    
    def init_db(self):
        """初始化历史记录数据库"""
        self.db.execute("""
        CREATE TABLE IF NOT EXISTS call_history (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT,
            user_query TEXT,
            function_name TEXT,
            function_args TEXT,
            result TEXT
        )
        """)
        self.db.commit()
    
    def log_call(self, query, func_name, args, result):
        """记录每次函数调用"""
        self.db.execute(
            "INSERT INTO call_history (timestamp, user_query, function_name, function_args, result) VALUES (?, ?, ?, ?, ?)",
            (datetime.now().isoformat(), query, func_name, json.dumps(args), str(result))
        )
        self.db.commit()
    
    def execute_function_with_retry(self, func_name, args, max_retry=3):
        """带重试的函数执行"""
        func = AVAILABLE_FUNCTIONS.get(func_name)
        if not func:
            return {"error": f"函数 {func_name} 不存在"}
        
        for attempt in range(max_retry):
            try:
                result = func(**args)
                return result
            except requests.exceptions.Timeout:
                if attempt == max_retry - 1:
                    return {"error": "API 超时，请稍后重试"}
                time.sleep(2 ** attempt)  # 指数退避
            except Exception as e:
                return {"error": str(e)}
    
    def run(self, user_query: str) -> str:
        messages = [
            {"role": "system", "content": "你是企业智能助手..."},
            {"role": "user", "content": user_query}
        ]
        
        while True:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=[{"type": "function", "function": f} for f in functions]
            )
            
            message = response.choices[0].message
            
            if not message.tool_calls:
                return message.content
            
            messages.append(message)
            
            for tool_call in message.tool_calls:
                func_name = tool_call.function.name
                func_args = json.loads(tool_call.function.arguments)
                
                result = self.execute_function_with_retry(func_name, func_args)
                self.log_call(user_query, func_name, func_args, result)
                
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "name": func_name,
                    "content": json.dumps(result, ensure_ascii=False)
                })

---

Step 5：实战演示

# main.py
from agent_v2 import FunctionCallAgent

agent = FunctionCallAgent()

# 场景1：查询客户 + 通知
response = agent.run("手机号 13800138000 的客户是谁？查到后发企微通知他会员已升级")
print(response)

# 场景2：查库存 + 创建订单
response = agent.run("查一下产品 PROD-001 的库存，如果大于10就给客户 CUST-99 创建订单")
print(response)

执行流程（场景2）：

用户: 查一下产品 PROD-001 的库存，如果大于10就给客户 CUST-99 创建订单

🔧 第1步: inventory_check(product_id="PROD-001")
   ← 返回: {"stock": 25, "product_name": "腾讯云服务器3折券"}

🤖 大模型思考: 库存25 > 10，需要创建订单，但还缺客户信息...
🔧 第2步: crm_get_customer(phone=?) 
   ← 大模型发现缺参数，询问用户客户手机号

用户: 13800138000

🔧 第3步: crm_get_customer(phone="13800138000")
   ← 返回: {"customer_id": "CUST-99", "name": "张三", "level": "VIP"}

🔧 第4步: crm_create_order(customer_id="CUST-99", product="腾讯云服务器3折券", amount=99.0)
   ← 返回: {"order_id": "ORD-20240505-001", "status": "success"}

🤖 最终回复: 已为张三（VIP客户）创建订单 ORD-20240505-001，产品：腾讯云服务器3折券，金额：99元。

---

进阶技巧

1. 并行调用多个 API（提速）

# 使用 ThreadPoolExecutor 并行调用
from concurrent.futures import ThreadPoolExecutor

def parallel_function_calls(tool_calls):
    with ThreadPoolExecutor() as executor:
        futures = []
        for tc in tool_calls:
            args = json.loads(tc.function.arguments)
            futures.append(executor.submit(AVAILABLE_FUNCTIONS[tc.function.name], **args))
        return [f.result() for f in futures]

2. 函数调用结果缓存

import hashlib
import pickle
from pathlib import Path

cache_dir = Path("./function_cache")
cache_dir.mkdir(exist_ok=True)

def cached_call(func_name, args):
    key = hashlib.md5(f"{func_name}:{json.dumps(args, sort_keys=True)}".encode()).hexdigest()
    cache_file = cache_dir / key
    
    if cache_file.exists():
        return pickle.loads(cache_file.read_bytes())
    
    result = AVAILABLE_FUNCTIONS[func_name](**args)
    cache_file.write_bytes(pickle.dumps(result))
    return result

---

总结

功能	基础版	进阶版（本文）
——	——–	—————-
单 API 调用	✅	✅
多 API 串联	❌	✅
错误重试	❌	✅
调用历史记录	❌	✅
并行调用	❌	✅
结果缓存	❌	✅

核心要点：

1. 用循环实现多步 Function Calling

2. 每次把函数执行结果追加到 messages

3. 加强错误处理和重试机制

---

👤 作者简介

一枚在大中原腹地（河南）卖公有云的从业者，主营腾讯云/阿里云/火山云，曾踩坑无数，现专注AI大模型应用落地。关注公众号「公有云cloud」，围观AI前沿动态~

博客：yunduancloud.icu