LangChain函数调用实战：构建可操作外部API的AI智能体系统

大模型开发中，函数调用（Function Calling）是让AI智能体具备操作外部工具能力的核心技术路径。通过LangChain框架的函数调用机制，大语言模型可以根据用户意图自动选择并执行预定义的函数，实现查数据库、调API、执行代码等操作。本文以构建一个可查询天气和执行数学计算的智能体为例，完整演示函数定义、工具绑定、调用链路搭建的全过程。

函数调用机制原理与适用场景

函数调用的本质是让大模型输出结构化的JSON参数，而非自然语言文本。模型根据函数的名称、参数描述和用户输入，判断是否需要调用函数以及应该传入什么参数。AI模型部署时，这个机制让模型从”对话工具”升级为”执行引擎”。

典型应用场景包括：

• 查询实时数据（天气、股价、物流状态）
• 操作数据库（增删改查业务数据）
• 调用企业内部API（ERP、CRM系统接口）
• 执行计算任务（数学运算、数据统计）

环境准备与依赖安装

pip install langchain langchain-openai langchain-community

# .env 文件配置
OPENAI_API_KEY=sk-your-api-key
OPENAI_API_BASE=https://api.openai.com/v1

使用OpenAI兼容接口时，如果接入国产大模型如通义千问、DeepSeek，只需修改API_BASE和模型名称即可。

定义工具函数：天气查询与数学计算

工具函数是智能体的”手”，需要用清晰的docstring描述函数用途和参数含义，模型据此判断何时调用。

from langchain_core.tools import tool
import requests

@tool
def get_weather(city: str) -> str:
    """查询指定城市的当前天气信息。
    
    Args:
        city: 城市名称，如"北京"、"上海"
    
    Returns:
        该城市当前的天气描述，包含温度和天气状况
    """
    # 实际项目中调用天气API
    # 这里用模拟数据演示
    weather_data = {
        "北京": "晴，气温32°C，湿度45%",
        "上海": "多云，气温29°C，湿度68%",
        "深圳": "雷阵雨，气温28°C，湿度85%"
    }
    return weather_data.get(city, f"暂无{city}的天气数据")

@tool
def calculate(expression: str) -> str:
    """执行数学表达式计算。
    
    Args:
        expression: 数学表达式字符串，如"3.14 * 5 * 5"、"100 / 7"
    
    Returns:
        计算结果
    """
    try:
        # 安全执行数学表达式
        allowed_names = {
            "abs": abs, "round": round, "pow": pow,
            "min": min, "max": max, "sum": sum
        }
        result = eval(expression, {"__builtins__": {}}, allowed_names)
        return f"计算结果: {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"

绑定工具到模型并创建调用链

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

# 初始化模型并绑定工具
llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0
)
llm_with_tools = llm.bind_tools([get_weather, calculate])

# 测试模型是否正确识别函数调用意图
response = llm_with_tools.invoke([
    HumanMessage(content="北京今天天气怎么样？")
])

print(f"模型输出: {response.content}")
print(f"工具调用: {response.tool_calls}")

# 输出示例:
# 模型输出: 
# 工具调用: [{'name': 'get_weather', 'args': {'city': '北京'}, 'id': 'call_xxx'}]

构建完整的智能体调用循环

智能体的核心是一个调用循环：模型判断是否调用工具 → 执行工具函数 → 将结果反馈给模型 → 模型生成最终回复。Prompt工程在这个环节至关重要，系统提示词决定了智能体的行为边界。

from langchain_core.messages import SystemMessage, AIMessage, ToolMessage

def run_agent(user_input: str, max_iterations: int = 5):
    """运行智能体，自动处理工具调用循环"""
    tools = {
        "get_weather": get_weather,
        "calculate": calculate
    }
    
    messages = [
        SystemMessage(content="你是一个智能助手，可以查询天气和进行数学计算。根据用户需求选择合适的工具，给出准确简洁的回答。"),
        HumanMessage(content=user_input)
    ]
    
    for i in range(max_iterations):
        response = llm_with_tools.invoke(messages)
        messages.append(response)
        
        # 如果没有工具调用，返回最终回复
        if not response.tool_calls:
            return response.content
        
        # 执行所有工具调用
        for tool_call in response.tool_calls:
            func_name = tool_call["name"]
            func_args = tool_call["args"]
            
            print(f"[调用工具] {func_name}({func_args})")
            
            # 执行对应的工具函数
            tool_result = tools[func_name].invoke(func_args)
            
            messages.append(ToolMessage(
                content=str(tool_result),
                tool_call_id=tool_call["id"]
            ))
            print(f"[工具返回] {tool_result}")
    
    return "达到最大调用次数限制"

# 测试多轮对话
print(run_agent("北京和深圳的天气分别是什么？哪个城市温度更高？"))

多工具并发调用与错误处理

当用户请求涉及多个工具调用时，现代大模型可以在单次推理中并发输出多个工具调用请求。上面的循环已支持这种情况——模型返回的tool_calls列表包含多个调用，代码逐个执行后将结果一次性反馈。

生产环境中必须处理的异常情况：

import asyncio
from functools import partial

async def run_tool_async(tool_func, args):
    """异步执行工具函数，避免阻塞"""
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(None, partial(tool_func.invoke, args))

async def run_agent_async(user_input: str):
    """支持异步并发的智能体"""
    tools_map = {
        "get_weather": get_weather,
        "calculate": calculate
    }
    
    messages = [
        SystemMessage(content="你是智能助手，可以查询天气和计算数学表达式。"),
        HumanMessage(content=user_input)
    ]
    
    response = llm_with_tools.invoke(messages)
    
    if response.tool_calls:
        # 并发执行所有工具调用
        tasks = []
        for tc in response.tool_calls:
            func = tools_map[tc["name"]]
            tasks.append(run_tool_async(func, tc["args"]))
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        for tc, result in zip(response.tool_calls, results):
            content = str(result) if not isinstance(result, Exception) else f"工具执行失败: {result}"
            messages.append(ToolMessage(
                content=content,
                tool_call_id=tc["id"]
            ))
        
        messages.append(response)
        final = llm_with_tools.invoke(messages)
        return final.content
    
    return response.content

常见问题诊断

问题1：模型不触发函数调用

检查函数的docstring是否清晰描述了用途和参数。模型依赖描述文字判断调用时机。把”查询天气”写成”get_weather”就无法触发——描述要面向自然语言理解。

问题2：参数格式不匹配

在函数签名中使用明确的类型标注（str、int、float），LangChain会自动生成JSON Schema供模型参考。避免使用Any或Union等模糊类型。

问题3：调用循环不终止

设置max_iterations限制最大调用轮次。模型在某些边界情况下可能反复调用工具而不生成最终回复，超过限制后直接返回已有信息。

部署到生产环境的注意事项

函数调用对延迟敏感——每次工具调用都会增加一轮模型推理。对于实时性要求高的场景，考虑以下优化：

• 工具函数实现幂等性，支持重试
• 对外部API调用设置超时和熔断
• 使用流式输出（streaming）减少首字延迟
• 对工具调用结果做缓存，避免重复请求

AI智能体的函数调用能力是把双刃剑——它赋予模型操作真实系统的能力，也带来了安全风险。务必对工具函数做权限控制和输入校验，避免模型被诱导执行危险操作。