Разработка AI-агента с Function Calling (Tool Use)

Разработка AI-агента с Function Calling (Tool Use)

Function Calling (Tool Use) — механизм, при котором LLM генерирует структурированный вызов функции вместо текстового ответа. Модель анализирует запрос, определяет нужный инструмент и его параметры, а хост-приложение выполняет вызов и возвращает результат. Это делает агента способным взаимодействовать с любыми внешними системами.

OpenAI Function Calling

from openai import OpenAI
import json

client = OpenAI()

# Схема инструментов
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_customer_info",
            "description": "Получить информацию о клиенте по ID или email",
            "parameters": {
                "type": "object",
                "properties": {
                    "customer_id": {"type": "string"},
                    "email": {"type": "string"},
                    "fields": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "Нужные поля: name, orders, balance, status"
                    }
                },
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "create_support_ticket",
            "description": "Создать тикет в службу поддержки",
            "parameters": {
                "type": "object",
                "properties": {
                    "customer_id": {"type": "string"},
                    "category": {"type": "string", "enum": ["billing", "technical", "account", "shipping"]},
                    "priority": {"type": "string", "enum": ["low", "medium", "high", "critical"]},
                    "description": {"type": "string"},
                },
                "required": ["customer_id", "category", "description"]
            }
        }
    },
]

# Реестр функций
def get_customer_info(customer_id=None, email=None, fields=None) -> dict:
    # Реальная реализация: запрос в CRM/БД
    return {"id": customer_id, "name": "Иванов И.И.", "balance": 15000, "status": "active"}

def create_support_ticket(customer_id: str, category: str, description: str, priority: str = "medium") -> dict:
    # Реальная реализация: Jira/Zendesk API
    return {"ticket_id": "TKT-12345", "status": "created"}

FUNCTION_MAP = {
    "get_customer_info": get_customer_info,
    "create_support_ticket": create_support_ticket,
}

# Агентный цикл с Function Calling
def run_support_agent(user_message: str) -> str:
    messages = [
        {"role": "system", "content": "Ты — агент службы поддержки. Используй инструменты для помощи клиентам."},
        {"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
        messages.append(message)

        if not message.tool_calls:
            return message.content

        # Выполняем все вызовы (параллельно если несколько)
        for tool_call in message.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)

            func = FUNCTION_MAP.get(func_name)
            if func:
                result = func(**func_args)
            else:
                result = {"error": f"Function {func_name} not found"}

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

Parallel Tool Calls

GPT-4o и Claude 3+ поддерживают параллельный вызов нескольких инструментов:

# Пример: агент может запросить данные из нескольких источников одновременно
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Покажи заказы и баланс клиента #12345"}],
    tools=tools,
    tool_choice="auto",
    parallel_tool_calls=True,
)

# Модель генерирует два параллельных вызова:
# get_orders(customer_id="12345") AND get_balance(customer_id="12345")

Anthropic Tool Use

import anthropic

client = anthropic.Anthropic()

tools = [
    {
        "name": "search_knowledge_base",
        "description": "Поиск в корпоративной базе знаний",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "category": {"type": "string"},
            },
            "required": ["query"],
        }
    }
]

def run_claude_agent(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        )

        if response.stop_reason == "end_turn":
            # Финальный текстовый ответ
            return next(b.text for b in response.content if hasattr(b, "text"))

        # Обработка tool_use блоков
        messages.append({"role": "assistant", "content": response.content})

        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                result = execute_tool(block.name, block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": json.dumps(result),
                })

        messages.append({"role": "user", "content": tool_results})

Практический кейс: агент для HR-запросов

Инструменты:

• get_employee_info(employee_id) — данные сотрудника
• check_vacation_balance(employee_id) — остаток отпуска
• submit_vacation_request(employee_id, start_date, end_date) — заявка на отпуск
• get_company_policy(policy_name) — поиск в базе HR-политик

Запрос: «Хочу взять отпуск с 15 апреля по 25 апреля. Есть ли у меня достаточно дней?»

Траектория агента:

1. get_employee_info(employee_id="emp_789") — получаем ID из контекста сессии
2. check_vacation_balance(employee_id="emp_789") — остаток: 14 дней
3. get_company_policy("vacation_approval") — читаем правила согласования
4. Финальный ответ: «У вас 14 дней отпуска. Период 15–25 апреля — 11 рабочих дней (с учётом праздников). Баланса хватает. Для оформления — submit_vacation_request. Заявку должен одобрить ваш руководитель в течение 3 рабочих дней согласно политике.»

Метрики за первый месяц:

• Запросов обработано автономно (без оператора): 84%
• Точность информации о балансе/политиках: 97%
• Среднее время ответа: 4.2с

Валидация вызовов инструментов

from pydantic import BaseModel, validator
from datetime import date

class VacationRequest(BaseModel):
    employee_id: str
    start_date: date
    end_date: date

    @validator("end_date")
    def end_after_start(cls, v, values):
        if "start_date" in values and v <= values["start_date"]:
            raise ValueError("end_date must be after start_date")
        return v

# Валидация до выполнения
def safe_tool_execution(func_name: str, func_args: dict) -> dict:
    try:
        if func_name == "submit_vacation_request":
            VacationRequest(**func_args)  # Валидация
        return FUNCTION_MAP[func_name](**func_args)
    except Exception as e:
        return {"error": str(e), "status": "validation_failed"}

Сроки

• Разработка агента с 3–7 инструментами: 2–4 недели
• Интеграция с корпоративными системами: 2–4 недели
• Тестирование и мониторинг: 1–2 недели
• Итого: 5–10 недель