Интеграция OpenAI Function Calling в приложение

Интеграция OpenAI Function Calling в приложение

Function Calling в OpenAI API позволяет модели структурированно вызывать ваши функции. Модель определяет, когда и с какими аргументами вызвать функцию, на основе описания. Это основа для агентных систем, подключения к базам данных, API и любым внешним сервисам.

Базовый пример

from openai import OpenAI
import json

client = OpenAI()

# Определение функций
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_order_status",
            "description": "Получить статус заказа по его ID",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "Идентификатор заказа"
                    }
                },
                "required": ["order_id"],
                "additionalProperties": False,
            },
            "strict": True,  # Гарантирует точное соответствие схеме
        }
    },
    {
        "type": "function",
        "function": {
            "name": "cancel_order",
            "description": "Отменить заказ (только если статус pending или processing)",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"},
                    "reason": {"type": "string", "description": "Причина отмены"}
                },
                "required": ["order_id"],
                "additionalProperties": False,
            },
            "strict": True,
        }
    }
]

# Имплементации функций
def get_order_status(order_id: str) -> dict:
    # Запрос в БД
    return {"order_id": order_id, "status": "processing", "items": 3}

def cancel_order(order_id: str, reason: str = "") -> dict:
    return {"success": True, "message": f"Order {order_id} cancelled"}

FUNCTION_MAP = {
    "get_order_status": get_order_status,
    "cancel_order": cancel_order,
}

def run_with_tools(user_message: str, conversation_history: list = None) -> str:
    messages = conversation_history or []
    messages.append({"role": "user", "content": user_message})

    while True:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto",
            parallel_tool_calls=True,  # Разрешаем параллельные вызовы
        )

        message = response.choices[0].message

        if response.choices[0].finish_reason == "stop":
            return message.content

        # Обрабатываем вызовы функций
        messages.append(message.model_dump())

        for tool_call in message.tool_calls or []:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)

            result = FUNCTION_MAP[func_name](**func_args)

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False),
            })

Structured Outputs с функциями

from pydantic import BaseModel
from typing import Literal

class OrderAction(BaseModel):
    action: Literal["status_check", "cancellation", "escalation"]
    order_id: str
    priority: Literal["low", "medium", "high"]
    notes: str

# Принудительное использование конкретной функции с Pydantic
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Заказ #12345 завис в обработке уже 3 дня"}],
    tools=[openai.pydantic_function_tool(OrderAction)],
    tool_choice="required",
)

# Автоматически парсится в Pydantic модель
tool_call = response.choices[0].message.tool_calls[0]
action = OrderAction.model_validate_json(tool_call.function.arguments)
print(action.action, action.priority)

Параллельные вызовы для агрегации данных

async def aggregate_customer_data(customer_id: str) -> dict:
    """Параллельно запрашивает данные из нескольких источников"""
    # GPT-4o с parallel_tool_calls=True вызовет все инструменты одновременно

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": f"Собери полную информацию о клиенте {customer_id}: профиль, заказы, тикеты"
        }],
        tools=[get_profile_tool, get_orders_tool, get_tickets_tool],
        tool_choice="required",
        parallel_tool_calls=True,
    )

    # Модель вызывает все три функции в одном ответе
    # Затем агрегирует результаты
    ...

Практический кейс: служба поддержки e-commerce

Функции: get_order, track_shipment, process_refund, update_address, get_product_info.

Результат: 64% обращений обрабатываются автономно (статусы, трекинг, простые возвраты). Среднее время решения: 45 мин → 2 мин.

Сроки

• Базовый function calling loop: 1–2 дня
• Параллельные вызовы + Pydantic: 2–3 дня
• Production с error handling: 1 неделя