Function Calling в AI: Как LLM взаимодействуют с внешними инструментами

import requests

def call_external_api(function_name, parameters):
    """
    Универсальный обработчик внешних вызовов
    :param function_name: Название функции (соответствует endpoint'у API)
    :param parameters: Словарь параметров
    :return: Данные из API или ошибка
    """
    try:
        # Формирование запроса в едином формате
        payload = {
            "function": function_name,
            "parameters": parameters
        }

        # Отправка запроса с обработкой ошибок
        response = requests.post(
            "https://api.your-service.com/v1/functions",
            json=payload,
            timeout=10
        )
        response.raise_for_status()  # Проверка HTTP-статуса

        # Парсинг и валидация ответа
        result = response.json()
        if "error" in result:
            raise ValueError(f"API Error: {result['error']}")
        return result["data"]

    except requests.exceptions.RequestException as e:
        return {"error": f"Network error: {str(e)}"}
    except ValueError as e:
        return {"error": str(e)}

# Шаг 1: Формирование запроса
user_query = "Какая погода сегодня в Париже?"
function_call = {
    "name": "get_weather",
    "arguments": {"city": "Paris", "units": "metric"}
}

# Шаг 2: Вызов API через обработчик
weather_data = call_external_api(
    function_call["name"],
    function_call["arguments"]
)

# Шаг 3: Обработка результата с проверкой
if "error" in weather_data:
    final_response = f"Не удалось получить данные: {weather_data['error']}"
else:
    # Использование ДАННЫХ ИЗ API, а не хардкода!
    temp = weather_data.get("temperature", "N/A")
    description = weather_data.get("description", "без описания")
    final_response = (
        f"Сейчас в Париже {temp}°C, {description}. "
        f"Влажность: {weather_data.get('humidity', '?')}%"
    )

print(final_response)

def analyze_sales():
    # Получение данных из базы
    sales_data = call_external_api("query_sales_db", {
        "period": "2023-Q4",
        "region": "EU"
    })

    if "error" in sales_data:
        return f"Ошибка базы данных: {sales_data['error']}"

    # Анализ с проверкой данных
    total = sum(sales_data["sales"])
    avg = total / len(sales_data["sales"]) if sales_data["sales"] else 0

    # Формирование отчета с контекстом
    return (
        f"Анализ продаж за 2023-Q4 (Европа):\n"
        f"- Общий объем: ${total:,.2f}\n"
        f"- Средний чек: ${avg:,.2f}\n"
        f"- Лучший день: {sales_data['best_day']}"
    )

print(analyze_sales())

import openai

# Описание функции для OpenAI
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "Получить текущую погоду в указанном городе",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "Город и страна, например: Москва, Россия"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "default": "celsius"
                    }
                },
                "required": ["location"]
            },
            "returns": {
                "type": "object",
                "properties": {
                    "temperature": {"type": "number"},
                    "unit": {"type": "string"},
                    "description": {"type": "string"}
                }
            }
        }
    }
]

response = openai.chat.completions.create(
    model="gpt-4-turbo",
    messages=[
        {"role": "user", "content": "Какая погода в Токио?"}
    ],
    tools=tools,
    tool_choice="auto"  # Модель сама решает, когда вызывать функцию
)

# Проверка, требуется ли вызов функции
if response.choices[0].message.tool_calls:
    tool_call = response.choices[0].message.tool_calls[0]

    # Извлечение параметров
    function_args = eval(tool_call.function.arguments)
    location = function_args["location"]

    # Реальный вызов API (ваш код!)
    weather_data = call_external_api(
        "get_current_weather", 
        {"location": location, "unit": "celsius"}
    )

    # Отправка результата обратно в модель
    second_response = openai.chat.completions.create(
        model="gpt-4-turbo",
        messages=[
            {"role": "user", "content": "Какая погода в Токио?"},
            response.choices[0].message,  # Первый ответ с tool_call
            {
                "role": "tool",
                "content": str(weather_data),
                "tool_call_id": tool_call.id
            }
        ]
    )

    print("Окончательный ответ:", second_response.choices[0].message.content)

# Шаблон описания функции для OpenAI
def define_tool(name, description, parameters, returns):
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": description,
            "parameters": {
                "type": "object",
                "properties": parameters,
                "required": [k for k, v in parameters.items() if v.get("required", False)]
            },
            "returns": returns
        }
    }

# Пример использования
weather_tool = define_tool(
    name="get_current_weather",
    description="Получить текущую погоду",
    parameters={
        "location": {
            "type": "string",
            "description": "Город и страна",
            "required": True
        },
        "unit": {
            "type": "string",
            "enum": ["celsius", "fahrenheit"],
            "default": "celsius"
        }
    },
    returns={
        "type": "object",
        "properties": {
            "temperature": {"type": "number"},
            "unit": {"type": "string"},
            "conditions": {"type": "string"}
        }
    }
)

def safe_api_call(function_name, params, max_retries=2):
    for attempt in range(max_retries + 1):
        try:
            # Ваш код вызова API
            return call_external_api(function_name, params)
        except Exception as e:
            if attempt == max_retries:
                log_error(f"API failed after {max_retries} attempts: {str(e)}")
                return {"error": "Сервис временно недоступен"}
            time.sleep(1 * (attempt + 1))  # Экспоненциальная задержка

from functools import lru_cache

@lru_cache(maxsize=100)
def cached_weather(city, units="metric"):
    """Кэширование погодных данных на 10 минут"""
    return call_external_api("get_weather", {"city": city, "units": units})

# Использование
print(cached_weather("London"))  # Первый вызов - реальный запрос
print(cached_weather("London"))  # Второй вызов - из кэша

def validate_parameters(params, schema):
    """Проверка параметров перед вызовом API"""
    for field, rules in schema.items():
        if rules.get("required") and field not in params:
            raise ValueError(f"Missing required parameter: {field}")
        if "enum" in rules and params.get(field) not in rules["enum"]:
            raise ValueError(f"Invalid value for {field}")

# Никогда не храните ключи в коде!
import os
API_KEY = os.getenv("WEATHER_API_KEY")

# Используйте секрет-менеджеры в production
from google.cloud import secretmanager
def get_secret(secret_name):
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/my-project/secrets/{secret_name}/versions/latest"
    return client.access_secret_version(name=name).payload.data.decode("UTF-8")

# Пример unit-теста для обработчика API
def test_weather_api():
    mock_response = {"data": {"temperature": 22, "description": "ясно"}}
    with patch("requests.post", return_value=MockResponse(mock_response)):
        result = call_external_api("get_weather", {"city": "Test"})
        assert result["temperature"] == 22