如何在 LangChain 中实现函数调用 Function Calling？（详解Agent工具集成与动态函数执行）

在构建智能AI应用时，单纯的语言模型能力往往不足以满足复杂的业务需求。用户可能需要查询实时天气、执行数学计算、访问数据库、调用API接口或控制外部系统，这些功能超出了语言模型的固有知识范围。LangChain框架中的Function Calling（函数调用）机制正是为解决这一问题而设计的核心功能，它让AI应用能够像人类一样，在需要时主动调用合适的工具来完成特定任务。

Function Calling的基本概念与工作原理

从静态响应到动态工具调用的演进

传统的语言模型应用只能基于训练数据提供静态响应，无法获取实时信息或执行具体操作。Function Calling机制改变了这一局限，它允许语言模型在推理过程中动态决定是否需要调用外部函数，并根据函数的执行结果继续生成响应。

工作原理可以概括为三个阶段：

1. 意图识别：模型分析用户输入，判断是否需要调用外部函数
2. 参数提取：如果需要调用函数，模型提取必要的参数信息
3. 结果整合：执行函数后，将结果作为上下文继续生成最终响应

这种机制使得AI应用具备了”工具使用”的能力，大大扩展了其实际应用场景和价值。

LangChain中的工具抽象层

LangChain通过BaseTool抽象类和@tool装饰器提供了统一的工具定义接口。无论底层函数多么复杂，都可以通过标准化的方式暴露给AI模型使用：

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

@tool
def multiply(a: int, b: int) -> int:
    """将两个整数相乘"""
    return a * b

@tool  
def get_weather(city: str) -> str:
    """获取指定城市的当前天气信息"""
    # 模拟天气API调用
    weather_data = {
        "北京": "晴天，25°C",
        "上海": "多云，28°C", 
        "广州": "雷阵雨，30°C"
    }
    return weather_data.get(city, "未知城市")

# 创建支持函数调用的LLM
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)

# 将工具绑定到LLM
llm_with_tools = llm.bind_tools([multiply, get_weather])

# 测试函数调用
messages = [HumanMessage(content="北京今天的天气怎么样？")]
response = llm_with_tools.invoke(messages)
print(response.tool_calls)  # 查看模型是否决定调用工具

工具定义的多种方式与最佳实践

使用@tool装饰器定义简单工具

对于功能单一、参数简单的工具，@tool装饰器是最简洁的定义方式。它自动处理参数解析、类型验证和文档生成：

from typing import Optional
from langchain_core.tools import tool

@tool
def search_web(query: str, max_results: Optional[int] = 5) -> str:
    """在互联网上搜索相关信息
    
    Args:
        query: 搜索关键词
        max_results: 最大返回结果数量，默认为5
    """
    # 模拟搜索结果
    results = [
        f"搜索结果1: 关于{query}的详细信息...",
        f"搜索结果2: {query}的最新动态...",
        f"搜索结果3: {query}相关的技术文章..."
    ]
    return "\n".join(results[:max_results])

@tool
def calculate_expression(expression: str) -> float:
    """安全地计算数学表达式
    
    Args:
        expression: 要计算的数学表达式，如"2+3*4"
    """
    # 安全的表达式求值（实际应用中应使用更安全的方法）
    allowed_chars = set("0123456789+-*/(). ")
    if not all(c in allowed_chars for c in expression):
        raise ValueError("表达式包含不允许的字符")
    return eval(expression)

这种方式的优势在于代码简洁、易于维护，特别适合快速原型开发和简单工具集成。

继承BaseTool创建复杂工具

对于需要复杂状态管理、异步操作或自定义验证逻辑的工具，继承BaseTool类提供了更大的灵活性：

from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field
import asyncio
import aiohttp

class WeatherInput(BaseModel):
    city: str = Field(description="城市名称")
    unit: str = Field(default="celsius", description="温度单位，可选'celsius'或'fahrenheit'")

class AdvancedWeatherTool(BaseTool):
    name: str = "advanced_weather"
    description: str = "获取指定城市的详细天气信息，支持多种温度单位"
    args_schema: type[BaseModel] = WeatherInput
    
    async def _arun(self, city: str, unit: str = "celsius") -> dict:
        """异步执行天气查询"""
        async with aiohttp.ClientSession() as session:
            # 模拟异步API调用
            await asyncio.sleep(0.1)
            weather_info = {
                "city": city,
                "temperature": 25 if unit == "celsius" else 77,
                "condition": "sunny",
                "humidity": 60,
                "unit": unit
            }
            return weather_info
    
    def _run(self, city: str, unit: str = "celsius") -> dict:
        """同步执行（通常调用异步版本）"""
        return asyncio.run(self._arun(city, unit))

# 使用复杂工具
weather_tool = AdvancedWeatherTool()

通过继承BaseTool，可以实现：

• 自定义参数验证逻辑
• 异步/同步双模式支持
• 复杂的状态管理和资源清理
• 详细的错误处理和重试机制

工具组合与工具链构建

在实际应用中，往往需要多个工具协同工作。LangChain支持将多个工具组合成工具链：

from langchain_core.tools import Tool

# 创建工具列表
tools = [
    multiply,
    get_weather,
    search_web,
    calculate_expression,
    AdvancedWeatherTool()
]

# 工具注册表（便于动态选择）
tool_registry = {tool.name: tool for tool in tools}

def execute_tool(tool_name: str, **kwargs) -> any:
    """执行指定工具"""
    if tool_name not in tool_registry:
        raise ValueError(f"未知工具: {tool_name}")
    
    tool = tool_registry[tool_name]
    return tool.invoke(kwargs)

这种工具组合模式使得应用能够根据具体需求动态选择最合适的工具，提高了系统的灵活性和可扩展性。

Agent驱动的智能工具调用

创建工具调用Agent

虽然可以直接将工具绑定到LLM，但使用Agent模式能够实现更智能的工具选择和多轮对话管理：

from langchain import hub
from langchain.agents import create_openai_functions_agent, AgentExecutor

# 获取预定义的Agent提示词
prompt = hub.pull("hwchase17/openai-functions-agent")

# 创建Agent
agent = create_openai_functions_agent(llm, tools, prompt)

# 创建Agent执行器
agent_executor = AgentExecutor(
    agent=agent, 
    tools=tools, 
    verbose=True,
    handle_parsing_errors=True
)

# 测试Agent
result = agent_executor.invoke({"input": "计算25乘以48的结果，然后告诉我北京的天气"})
print(result["output"])

Agent模式的优势在于：

• 多工具协调：能够在一个请求中调用多个工具
• 上下文保持：在多轮对话中保持工具调用的上下文
• 错误恢复：当工具调用失败时能够进行重试或选择替代方案
• 自然语言交互：用户可以用自然语言描述需求，无需了解具体的工具细节

自定义Agent行为

通过自定义提示词和Agent逻辑，可以控制工具调用的行为：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

custom_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个智能助手，可以使用以下工具来帮助用户:\n{tools}"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")
])

# 自定义Agent
custom_agent = create_openai_functions_agent(llm, tools, custom_prompt)

# 带有对话历史的Agent执行器
from langchain_core.chat_history import InMemoryChatMessageHistory

chat_history = InMemoryChatMessageHistory()

def run_agent_with_history(user_input: str) -> str:
    result = agent_executor.invoke({
        "input": user_input,
        "chat_history": chat_history.messages
    })
    chat_history.add_user_message(user_input)
    chat_history.add_ai_message(result["output"])
    return result["output"]

这种自定义能力使得开发者可以根据具体业务需求调整Agent的行为，比如限制某些工具的使用、优先选择特定工具，或者在特定条件下触发特殊逻辑。

实际应用场景与高级技巧

实时数据查询与API集成

Function Calling最常见的应用场景是实时数据查询：

import requests
from datetime import datetime

@tool
def get_stock_price(symbol: str) -> dict:
    """获取股票实时价格信息"""
    try:
        # 调用真实的股票API（这里用模拟数据）
        response = requests.get(f"https://api.example.com/stock/{symbol}")
        data = response.json()
        return {
            "symbol": symbol,
            "price": data["price"],
            "change": data["change"],
            "timestamp": datetime.now().isoformat()
        }
    except Exception as e:
        return {"error": f"获取股票价格失败: {str(e)}"}

@tool  
def get_news_headlines(category: str = "general", limit: int = 5) -> list:
    """获取新闻头条"""
    # 模拟新闻API调用
    headlines = [
        "重大科技突破：新型AI芯片发布",
        "全球经济形势分析报告",
        "体育赛事最新结果汇总"
    ]
    return headlines[:limit]

通过这些工具，AI应用可以提供最新的市场信息、新闻动态和其他实时数据，大大提升了实用价值。

数据库操作与业务系统集成

Function Calling也可以用于集成现有的业务系统：

import sqlite3
from contextlib import contextmanager

@contextmanager
def get_db_connection():
    conn = sqlite3.connect('business.db')
    try:
        yield conn
    finally:
        conn.close()

@tool
def query_customer_info(customer_id: str) -> dict:
    """查询客户基本信息"""
    with get_db_connection() as conn:
        cursor = conn.cursor()
        cursor.execute(
            "SELECT name, email, phone, address FROM customers WHERE id = ?", 
            (customer_id,)
        )
        result = cursor.fetchone()
        if result:
            return {
                "name": result[0],
                "email": result[1], 
                "phone": result[2],
                "address": result[3]
            }
        return {"error": "客户不存在"}

@tool
def create_support_ticket(customer_id: str, issue: str, priority: str = "medium") -> dict:
    """创建客户支持工单"""
    with get_db_connection() as conn:
        cursor = conn.cursor()
        cursor.execute(
            "INSERT INTO support_tickets (customer_id, issue, priority, created_at) VALUES (?, ?, ?, datetime('now'))",
            (customer_id, issue, priority)
        )
        conn.commit()
        return {"ticket_id": cursor.lastrowid, "status": "created"}

这种集成方式使得AI应用能够直接操作企业内部的数据和业务流程，成为真正的生产力工具。

安全性和权限控制

在生产环境中，Function Calling的安全性至关重要：

from functools import wraps

def require_permission(permission_level: str):
    """权限装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 从上下文中获取用户权限信息
            user_permission = kwargs.get('user_permission', 'guest')
            
            permission_hierarchy = {
                'admin': 3,
                'staff': 2, 
                'user': 1,
                'guest': 0
            }
            
            if permission_hierarchy.get(user_permission, 0) < permission_hierarchy.get(permission_level, 0):
                raise PermissionError(f"需要{permission_level}权限才能执行此操作")
            
            return func(*args, **kwargs)
        return wrapper
    return decorator

@tool
@require_permission('staff')
def delete_user_account(user_id: str, user_permission: str = 'guest') -> dict:
    """删除用户账户（需要员工权限）"""
    # 删除逻辑
    return {"status": "success", "message": f"用户{user_id}已删除"}

通过权限控制、输入验证、速率限制等安全措施，可以确保Function Calling在提供强大功能的同时保持系统的安全性。