Настройка и разработка кастомного AI ассистента на базе OpenAI и ChatGPT

"Знаете, на протяжении многих лет я занимаюсь работой с AI в области крипты. Большинство полагает что индивидуальный ассистент OpenAI - это только ChatGPT в красивой обертке. Но на самом деле - нет. Настроенный ассистент с его собственными инструментами - это даже не коллега-аналитик который никогда не спит." Но только в том случае, если его правильно подключить. И сейчас, пожалуй, я расскажу, как это сделать так, чтобы действительно работало."

Введение в кастомизированные ассистенты OpenAI

Кастомизированным ассистентом OpenAI называется AI-агент, который построен на GPT-моделях и заточен под конкретные задачи с помощью настраиваемых инструментов и баз знаний. В отличие от простого ChatGPT, который универсален, кастомный же помощник работает в строго определенных рамках. А доступ к внешним данным, и выполнение операций по вашим бизнес-правилам. Если представить для наглядности ChatGPT в роли универсального консультанта, который знает потихоньку обо всем, то кастомный ассистент - это узкопрофильный эксперт, имеющий прямой доступ к вашим системам, базам и процессам. Основные ключевые возможности в себе содержат:

• Функциональный вызов - запуск необходимых функций через API

• Поисковое знание - работа с загруженными документами и структурированными данными

• Интерпретация кода - выполнение Python скриптов для обработки информации

• Интеграция кастомных инструментов - интеграция внешних сервисов и также Базисами

• Диалоговая память - это сохранение контекста на всем протяжении общения

Криптоассистента именно по этим принципам и создала команда ASCN.AI. Мы не полагаемся на общедоступные интернетные данные - мы прикрутили инструменты, которые анализируют ончейн, DEX-метрики и обрабатывают настроения из Telegram. Ассистент не просто отвечает на заданные вопросы — он обращается к реальному времени блокчейна, анализирует активность кошельков, агрегирует сигналы из разных источников. Ключевое различие состоит в том, что кастомные помощники из реактивных чатботов трансформируются в проактивных агентов, которые имеют возможность исполнять задачи и принимать решения на базе актуальных данных.

Согласно имеющейся документации OpenAI, помощники с должным образом настроенными инструментами выполняют задачи в 3-4 раза более эффективно, чем просто по текстовым подсказкам. Почему? Причина в том, что они не злоупотребляют выдумкой ответов — они проверяют данные через вызовы функций.

Основные компоненты и инструменты для создания AI помощника

Итак, чтобы сделать рабочий OpenAI ассистент нам конечно нужны все основные базовые компоненты и их четыре.

Уровень конфигурации ассистента

В этом уровне задаются "личность" и ограничения. Находитесь в процессе конфигурирования: Системных инструкций (то есть того, каким образом должен вести себя ассистент и что имеет возможность с помощью своих специализированных навыков выполнять); Выбором модели (серверу GPT-4, GPT-4 Turbo и т. д.).)

• Температура и стиль ответов

• Права на доступ к файлам

Фреймворк интеграции инструментов

Инструменты расширяют возможности робота за пределами текста:

Система управления потоками (потоками)

Это поддержка контекста диалога — все сообщения в рамках одного потока сохраняются, ассистент может ссылаться на предыдущее.

Движок выполнения (run execution engine)

С каждым новым запросом ассистент создает так назваемый "run" - цепочку обработки:

• анализ входа

• определение необходимых инструментов

• вызываем по очереди функции

• собираем финальный ответ

Конкретный пример: для ончейн-анализа ASCN интегрированы 3 инструмента:

• API Ethereum-нода для настоящих транзакций

• DEX-агрегатор с ценами

• Парсер настроений Telegram-каналов

Ассистент сам решит, какой инструмент вызвать: если про активность китов - он ончейн-инструмент, если мнение рынка - Telegram-данные. С технической стороны OpenAI Assistants API - это тот механизм, который управляет всем этим. Ваша задача - описать инструменты и их схемы, а задача модели - решать, когда и как их задействовать. И это совсем не то, что традиционные чатботы с жёстким сценарным кодом.

Пошаговая инструкция по настройке и разработке кастомного ассистента на базе OpenAI API

Прежде чем писать код, создайте среду разработки и получите доступ к API.

Шаг 1: Создайте аккаунт OpenAI и получите API-ключи

Перейдите на platform.openai.com, зарегистрируйтесь и сгенерируйте секретный ключ в разделе API. Этот ключ подтверждает вашу личность при запросах.

Важно для безопасности: не храните ключи прямо в коде. Используйте переменные окружения или секретные менеджеры.

# Пример для .env (не коммитьте в открытый репозиторий)

OPENAI_API_KEY=sk-proj-xxxxxxxxxxxxxxxxxxxxxxxx

Шаг 2: Настройка среды разработки

pip install openai

Инициализируйте клиент:

from openai import OpenAI

import os

client = OpenAI(api_key=os.environ.get(""OPENAI_API_KEY""))

Шаг 3: Проверьте доступ к API

response = client.chat.completions.create(

    model=""gpt-4-turbo-preview"",

    messages=[{""role"": ""user"", ""content"": ""Проверка подключения""}]

)

print(response.choices[0].message.content)

Если получили ответ — всё готово. Ошибки на аутентификацию — попробуйте сгенерировать ключ заново.

Типичные проблемы при настройке:

• Использование старого формата ключа (начинается с sk-, но устаревший)

• Малый баланс или отсутствующие кредиты — проверьте биллинг

• Региональные ограничения — иногда помогает VPN

С ASCN.AI мы стартовали на бесплатном тарифе с лимитом 3 запроса в минуту. Для реальных задач с потоковыми данными подняли план до «по факту» с лимитом 10,000 запросов в минуту. Вывод — продумывайте инфраструктуру заранее.

Создание базового ассистента: параметры и концепции

Раз среда готова, приступаем к созданию ассистента.

Шаг 1: Определите цели ассистента

Перед кодом продумайте:

• Какие задачи решит ассистент?

• Какие источники данных ему нужны?

• Что должно быть автономным, а что по контролю человека?

У нас, например: «Анализ криптопроектов по ончейн-данным, социальным сигналам и рыночным метрикам — без инвестиционных советов».

Шаг 2: Создайте ассистента через API

assistant = client.beta.assistants.create(

    name=""Crypto Analysis Assistant"",

    instructions=""""""Ты — эксперт по криптовалютным проектам. 

Анализируй данные ончейна, социальные настроения, метрики рынка.

Не даёшь инвестиционных советов или прогнозов цен.

Всегда указывай источники данных и уровень уверенности."""""",

    model=""gpt-4-turbo-preview"",

    tools=[

        {""type"": ""code_interpreter""},

        {""type"": ""retrieval""}

    ]

)

print(f""Создан ассистент с ID: {assistant.id}"")

Ключевые параметры:

Шаг 3: Настройка поведения ассистента

Поле instructions — главный управляющий элемент. В нем прописываете:

• Область ответственности (что делать и чего не делать)

• Формат ответа

• Правила обработки данных

• Границы риска

Пример из ASCN, рабочий вариант:

instructions = """"""

Роль: аналитик криптопроектов

Источники данных: ончейн метрики, каналы Telegram, данные DEX

Формат вывода: структурированный анализ с указанием источников

Ограничения: ни финансовых советов, ни прогнозов цен

Обработка рисков: помечать данные с низкой уверенностью, требовать проверки по юридическим вопросам

Ассистент следует строгим ограничениям и не отклоняется от них, пока не измените конфиг. Здесь важно понимать — это не просто промпт, а архитектурный контроль.

Типичные ошибки в настройке:

• Слишком широкие инструкции, типа «анализируй всё о крипте»

• Отсутствие правил безопасности и верификации данных

• Неясная иерархия использования источников — конфликтующие данные средств

Настройка OpenAI assistant: шаги и рекомендации

После создания ассистента нужно настроить управление диалогами и обработку сообщений.

Шаг 1: Создайте поток (thread) для текущего диалога

Поток хранит историю и контекст беседы. На каждого пользователя создавайте отдельный поток:

thread = client.beta.threads.create()

print(f""Создан поток: {thread.id}"")

Сохраняйте ID потока в базе, чтобы обращаться при каждом запросе этого пользователя.

Шаг 2: Отправляйте сообщения в поток

Например, пользователь спрашивает:

message = client.beta.threads.messages.create(

    thread_id=thread.id,

    role=""user"",

    content=""Проанализируй ончейн-активность Ethereum-адреса 0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb""

)

Шаг 3: Запускайте цикл обработки (run)

Это запускает обработку ассистентом:

run = client.beta.threads.runs.create(

    thread_id=thread.id,

    assistant_id=assistant.id

)

Шаг 4: Ожидайте завершения run

Так как обработка асинхронна, проверяйте статус:

import time

while run.status in [""queued"", ""in_progress""]:

    run = client.beta.threads.runs.retrieve(

        thread_id=thread.id,

        run_id=run.id

    )

    time.sleep(1)

if run.status == ""completed"":

    messages = client.beta.threads.messages.list(thread_id=thread.id)

    print(messages.data[0].content[0].text.value)

Внимание на производительность:

• В зависимости от вызова инструментов run занимает от 5 до 30 секунд

• Вызовы внешних API добавляют задержки

• Интерпретатор кода тайм-аутит через 120 секунд

Оптимальная архитектура:

Запрос → Создать сообщение → Запустить run → Проверять статус →

    ↓

Если требуется действие → Выполнить функцию → Отправить результат → Продолжить run

    ↓

Если завершено → Получить ответ → Вернуть пользователю

Наш опыт: в первой версии ASCN забыл про тайм-ауты, и при обвале рынка runs зависали — пользователи думали, что система сломалась. Добавили 45-секундный тайм-аут с плавным деградированием — при падении инструментов выдаём кеш с предупреждением.

Разработка с OpenAI API: ключевые вызовы и примеры кода

Перейдём к созданию реального помощника, который умеет вызывать функции и справляться с комплексными задачами.

Полный рабочий пример: анализатор криптокошельков

from openai import OpenAI

import os

import json

import requests

client = OpenAI(api_key=os.environ.get(""OPENAI_API_KEY""))

def get_wallet_balance(address):

    """"""Запрашивает баланс Ethereum-кошелька через Etherscan API""""""

    api_key = os.environ.get(""ETHERSCAN_API_KEY"")

    url = f""https://api.etherscan.io/api?module=account&action=balance&address={address}&tag=latest&apikey={api_key}""

    response = requests.get(url)

    data = response.json()

    if data[""status""] == ""1"":

        balance_wei = int(data[""result""])

        balance_eth = balance_wei / 1e18

        return {""address"": address, ""balance_eth"": balance_eth}

    else:

        return {""error"": ""Некорректный адрес или ошибка API""}

assistant = client.beta.assistants.create(

    name=""Wallet Analyzer"",

    instructions=""Ты анализируешь Ethereum кошельки по ончейн-данным. Всегда приводишь точный баланс, избегая спекуляций."",

    model=""gpt-4-turbo-preview"",

    tools=[{

        ""type"": ""function"",

        ""function"": {

            ""name"": ""get_wallet_balance"",

            ""description"": ""Получает ETH баланс по адресу кошелька"",

            ""parameters"": {

                ""type"": ""object"",

                ""properties"": {

                    ""address"": {

                        ""type"": ""string"",

                        ""description"": ""Ethereum адрес кошелька""

                    }

                },

                ""required"": [""address""]

            }

        }

    }]

)




thread = client.beta.threads.create()

message = client.beta.threads.messages.create(

    thread_id=thread.id,

    role=""user"",

    content=""Каков баланс кошелька 0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb?""

)




run = client.beta.threads.runs.create(

    thread_id=thread.id,

    assistant_id=assistant.id

)




import time




while run.status in [""queued"", ""in_progress"", ""requires_action""]:

    run = client.beta.threads.runs.retrieve(

        thread_id=thread.id,

        run_id=run.id

    )

    

    if run.status == ""requires_action"":

        tool_calls = run.required_action.submit_tool_outputs.tool_calls

        tool_outputs = []

        

        for tool_call in tool_calls:

            function_name = tool_call.function.name

            arguments = json.loads(tool_call.function.arguments)

            

            if function_name == ""get_wallet_balance"":

                output = get_wallet_balance(arguments[""address""])

                tool_outputs.append({

                    ""tool_call_id"": tool_call.id,

                    ""output"": json.dumps(output)

                })

        

        run = client.beta.threads.runs.submit_tool_outputs(

            thread_id=thread.id,

            run_id=run.id,

            tool_outputs=tool_outputs

        )

    

    time.sleep(1)




messages = client.beta.threads.messages.list(thread_id=thread.id)

print(messages.data[0].content[0].text.value)

Объяснение:

1. Ассистент получает вопрос о балансе кошелька

2. GPT-4 понимает, что нужно вызвать функцию get_wallet_balance

3. Статус run меняется на requires_action

4. Запускается вызов функции с аргументами

5. Результат возвращается в run

6. Ассистент строит ответ на основе актуальных данных

Продвинутые улучшения:

# Обработка ошибок

def safe_function_call(func, *args, **kwargs):

    try:

        return func(*args, **kwargs)

    except Exception as e:

        return {""error"": str(e), ""fallback_data"": None}

# Потоковая передача ответов

from openai import AssistantEventHandler

class EventHandler(AssistantEventHandler):

    def on_text_delta(self, delta, snapshot):

        print(delta.value, end="""", flush=True)

with client.beta.threads.runs.stream(

    thread_id=thread.id,

    assistant_id=assistant.id,

    event_handler=EventHandler(),

) as stream:

    stream.until_done()

В бою ASCN не просто смотрит балансы, а анализирует паттерны транзакций, активность китов, DEX-операции и настроение Telegram. Для этого нужна цепочка вызовов функций в одном run.

Лучшие практики вызова функций:

• Каждая функция — одна задача

• Возвращайте структурированный JSON, а не текст

• Обрабатывайте ошибки — не отдавайте голый сбой

• Документируйте параметры детально

• Тестируйте на крайних случаях и невалидных данных

Интеграция пользовательских инструментов и плагинов в AI ассистента

OpenAI ассистенты поддерживают три стандартных типа инструментов и неограниченное количество кастомных функций. Чтобы добиться производительности и стабильности, важно понимать их назначение.

1. Интерпретатор кода

Выполняет Python в изолированной среде.

Задачи:

• Анализ и визуализация данных

• Математические вычисления

• Обработка файлов (CSV, JSON, Excel)

• Создание отчетов

Ограничения:

• Нет доступа к сети — нельзя вызывать внешние API

• Тайм-аут 120 секунд

• Доступны только стандартные и дата-сайенс библиотеки Python

Пример: расчет риск-метрик по портфелю.

2. Поиск знаний (retrieval)

Ищет ответы по загруженным документам, используя семантический поиск.

Задачи:

• Запросы к внутренней базе знаний

• Ответы на вопросы из документов

• Поиск процедур и политик

• Историческая справка

Ограничения:

• Максимум 20 файлов (512 МБ)

• Качество поиска зависит от контента

• Нет сложных запросов по базам данных

Пример: поиск технических деталей в whitepapers криптопроектов.

3. Вызов функций (custom tools)

Запуск собственного кода через API.

Возможности:

• Вызовы произвольных внешних API

• Запросы к базам данных

• Запуск бизнес-процессов

• Интеграция сторонних сервисов

Как ASCN использует:

Главный инсайт: вызов функций — мощнейший инструмент. Он вырывает ассистента из «пузыря» языковой модели. Ассистент не выдумывает, а проверяет факты в живых источниках.

Пошаговая инструкция интеграции кастомных инструментов

Создание кастомного инструмента — три этапа: задаете схему функции, пишете реализацию, управляете её выполнением в run.

Шаг 1: Определите схему функции

Описываете, что делает функция, какие параметры принимает:

tools = [{

    ""type"": ""function"",

    ""function"": {

        ""name"": ""get_token_metrics"",

        ""description"": ""Возвращает ончейн метрики токена: количество держателей, объём транзакций, ликвидность"",

        ""parameters"": {

            ""type"": ""object"",

            ""properties"": {

                ""token_address"": {

                    ""type"": ""string"",

                    ""description"": ""Адрес смарт-контракта токена (Ethereum формат 0x...)""

                },

                ""chain"": {

                    ""type"": ""string"",

                    ""enum"": [""ethereum"", ""bsc"", ""polygon""],

                    ""description"": ""Блокчейн сети""

                },

                ""metrics"": {

                    ""type"": ""array"",

                    ""items"": {

                        ""type"": ""string"",

                        ""enum"": [""holders"", ""volume_24h"", ""liquidity"", ""price""]

                    },

                    ""description"": ""Метрики для возврата (необязательно, все если пусто)""

                }

            },

            ""required"": [""token_address"", ""chain""]

        }

    }

}]

Принципы:

• Четкие описания — модель на их основе решает, когда вызвать функцию

• Используйте enum для ограничений значений

• Разделяйте обязательные и необязательные параметры

• В описаниях сложных параметров добавляйте примеры

Шаг 2: Реализуйте функцию

import requests

def get_token_metrics(token_address, chain, metrics=None):

    """"""

    Запрашивает данные токена у DEX агрегатора

    """"""

    if not token_address.startswith(""0x""):

        return {""error"": ""Неверный формат адреса""}

    

    api_url = f""https://api.dextools.io/v1/token/{chain}/{token_address}""

    

    try:

        response = requests.get(

            api_url,

            headers={""X-API-Key"": os.environ.get(""DEXTOOLS_API_KEY"")},

            timeout=10

        )

        response.raise_for_status()

        data = response.json()

        

        if metrics:

            filtered_data = {k: data.get(k) for k in metrics}

            return filtered_data

        

        return data

        

    except requests.exceptions.RequestException as e:

        return {

            ""error"": f""Ошибка API: {str(e)}"",

            ""fallback_data"": {""source"": ""кеш"", ""last_updated"": ""2025-01-10""}

        }

Что важно:

• Добавьте повторные попытки запросов при сбоях

• Ограничьте запросы по квоте, чтобы не сесть по API

• Кешируйте результаты ради скорости и экономии

• Отдавайте ошибки в структурированном виде для ассистента

Шаг 3: Управляйте выполнением в цикле run

def execute_assistant_run(thread_id, assistant_id):

    run = client.beta.threads.runs.create(

        thread_id=thread_id,

        assistant_id=assistant_id

    )

    while run.status in [""queued"", ""in_progress"", ""requires_action""]:

        run = client.beta.threads.runs.retrieve(

            thread_id=thread_id,

            run_id=run.id

        )

        

        if run.status == ""requires_action"":

            tool_calls = run.required_action.submit_tool_outputs.tool_calls

            tool_outputs = []

            

            for tool_call in tool_calls:

                function_name = tool_call.function.name

                arguments = json.loads(tool_call.function.arguments)

                

                if function_name == ""get_token_metrics"":

                    result = get_token_metrics(**arguments)

                    tool_outputs.append({

                        ""tool_call_id"": tool_call.id,

                        ""output"": json.dumps(result)

                    })

            

            run = client.beta.threads.runs.submit_tool_outputs(

                thread_id=thread_id,

                run_id=run.id,

                tool_outputs=tool_outputs

            )

        

        time.sleep(1)

    

    messages = client.beta.threads.messages.list(thread_id=thread_id)

    return messages.data[0].content[0].text.value

 

Обработка ошибок — отдельный блок:

def safe_tool_execution(tool_call):

    try:

        function_name = tool_call.function.name

        arguments = json.loads(tool_call.function.arguments)

        

        result = execute_function(function_name, arguments)

        

        return {

            ""tool_call_id"": tool_call.id,

            ""output"": json.dumps(result)

        }

        

    except TimeoutError:

        return {

            ""tool_call_id"": tool_call.id,

            ""output"": json.dumps({

                ""error"": ""Тайм-аут исполнения функции"",

                ""retry_suggested"": True

            })

        }

    except Exception as e:

        logging.error(f""Ошибка при вызове функции: {e}"")

        return {

            ""tool_call_id"": tool_call.id,

            ""output"": json.dumps({

                ""error"": ""Сбой функции"",

                ""details"": str(e),

                ""fallback_available"": False

            })

        }

При интеграции ончейн-аналитики мы заметили, что запросы к Ethereum-ноде при заторах могут доходить до 15-20 секунд. Без тайм-аутов пользователи видели зависший интерфейс. Решили так:

• Жёсткий тайм-аут 30 секунд на функцию

• Запасные кешированные ответы для типовых запросов

• Индикаторы прогресса на интерфейсе при долгих обработках

Управление контекстом и инструментами в API

Надёжность ассистента зависит от сохранения смысла и корректного использования инструментов.

Стратегии хранения контекста

Память на уровне треда
Вся история сообщений сохраняется в потоке. Можно ссылаться на предыдущие данные:
# Сообщение пользователя:

"Анализируй токен 0xabc123"

# Вопрос спустя минуту без повторных деталей:

"А как там с ликвидностью?"

# Ассистент помнит контекст и отвечает без уточнений

1. Метки метаданных
   Добавляйте в тред и сообщения дополнительную информацию для фильтров и персонализации:

   thread = client.beta.threads.create(

    metadata={

        ""user_id"": ""user_12345"",

        ""session_type"": ""portfolio_analysis"",

        ""risk_profile"": ""conservative""

    }

)

2. Файловый контекст для поиска знаний
Загружаете профильные документы один раз и используете их повторно:
file = client.files.create(

    file=open(""crypto_glossary.pdf"", ""rb""),

    purpose=""assistants""

)




assistant = client.beta.assistants.create(

    name=""Crypto Educator"",

    model=""gpt-4-turbo-preview"",

    tools=[{""type"": ""retrieval""}],

    file_ids=[file.id]

)

3. Паттерны оркестрации инструментов

Последовательный вызов:

Пользователь: ""Сравни балансы кошельков 0xabc и 0xdef""

    ↓

Вызов функции 1: get_wallet_balance(0xabc)

    ↓

Вызов функции 2: get_wallet_balance(0xdef)

    ↓

Ассистент формирует итоговый ответ

 

Параллельный вызов (функция GPT-4 Turbo):

{

    ""requires_action"": {

        ""tool_calls"": [

            {""function"": ""get_token_price"", ""arguments"": {""symbol"": ""ETH""}},

            {""function"": ""get_token_volume"", ""arguments"": {""symbol"": ""ETH""}},

            {""function"": ""get_market_sentiment"", ""arguments"": {""symbol"": ""ETH""}}

        ]

    }

}

Ограничения по контексту:

• GPT-4 Turbo поддерживает до 128K токенов (~100K слов)

• Пото́ки с 1000+ сообщениями усечут самый ранний контент

• Выходные данные функций учитываются в лимите контекста

Подход ASCN:

• Быстрые запросы — отдельный поток, 24 часа жизни

• Глубокий анализ — мультипотоки с перекрёстными ссылками

• Мониторинг — постоянный поток с ежедневными сводками

Примеры интеграции ChatGPT tools и расширение функционала AI

Рассмотрим рабочие кейсы, где ассистент выходит за рамки простых диалогов.

Пример 1: Мульти-источниковый сбор данных

Смешиваем ончейн данные, соцнастрой и рыночные метрики:

tools = [

    {

        ""type"": ""function"",

        ""function"": {

            ""name"": ""analyze_token_comprehensive"",

            ""description"": ""Комплексный анализ токена: ончейн, соцнастрой, рынок"",

            ""parameters"": {

                ""type"": ""object"",

                ""properties"": {

                    ""token_symbol"": {""type"": ""string"", ""description"": ""Символ токена (ETH, BTC)""},

                    ""analysis_depth"": {

                        ""type"": ""string"",

                        ""enum"": [""basic"", ""detailed"", ""comprehensive""],

                        ""description"": ""Глубина анализа""

                    }

                },

                ""required"": [""token_symbol""]

            }

        }

    }

]




def analyze_token_comprehensive(token_symbol, analysis_depth=""basic""):

    results = {}

    

    results[""on_chain""] = fetch_onchain_metrics(token_symbol)

    results[""sentiment""] = fetch_telegram_sentiment(token_symbol)

    results[""market""] = fetch_dex_metrics(token_symbol)

    

    if analysis_depth == ""comprehensive"":

        results[""whale_activity""] = fetch_whale_transactions(token_symbol)

        results[""holder_distribution""] = fetch_holder_analysis(token_symbol)

    

    return {

        ""token"": token_symbol,

        ""timestamp"": datetime.now().isoformat(),

        ""data"": results,

        ""confidence"": calculate_confidence_score(results)

    }

Пример 2: Автоматический запуск рабочих процессов

def create_alert_workflow(threshold_data):

    """"""

    Триггер при важном движении рынка

    """"""

    db.alerts.insert({

        ""type"": ""price_movement"",

        ""threshold"": threshold_data[""threshold""],

        ""actual"": threshold_data[""current_value""],

        ""timestamp"": datetime.now(),

        ""user_id"": threshold_data[""user_id""]

    })

    

    send_telegram_alert(

        user_id=threshold_data[""user_id""],

        message=f""Внимание: {threshold_data['token']} изменился на {threshold_data['change_percent']}%""

    )

    

    return {""alert_created"": True, ""notification_sent"": True}

 

Схема функции:

{

    ""name"": ""create_market_alert"",

    ""description"": ""Создаёт ценовое оповещение и уведомляет пользователя при достижении порога"",

    ""parameters"": {

        ""type"": ""object"",

        ""properties"": {

            ""token"": {""type"": ""string""},

            ""threshold_type"": {""type"": ""string"", ""enum"": [""above"", ""below""]},

            ""threshold_value"": {""type"": ""number""},

            ""current_value"": {""type"": ""number""}

        }

    }

}

Пример 3: Безкодовое решение (платформа ASCN.AI NoCode)

Для тех, кто не программист — визуальный конструктор сценариев на базе OpenAI.

Пример структуры:

[Триггер: Новое сообщение в Telegram]

    ↓

[AI Агент: Анализ намерений сообщения]

    ↓

[Логика: Если intent = ""token_analysis""]

    ↓

[Инструмент: Вызов OpenAI ассистента с кастомными функциями]

    ↓

[Инструмент: Получение ончейн данных]

    ↓

[Инструмент: Формирование отчёта]

    ↓

[Действие: Отправить ответ в Telegram]

"Платформа NoCode позволяет за минуты настроить сложные AI автосценарии без программиста." — Команда ASCN.AI NoCode

Пользователи строят цепочки из готовых блоков для триггеров, AI-агентов, логики, HTTP-запросов и месседжинга.

Наш реальный кейс: автоматический трекинг портфеля занял 15 минут — вместо нескольких дней разработки.

Функция вызова (function calling) в GPT-4: как использовать и настраивать

Function calling переводит GPT из текстового генератора в исполнитель конкретных действий. Модель формирует JSON со структурой функции и аргументами, ваш сервер выполняет вызов и возвращает ответ. GPT вставляет данные в итоговый ответ.

Основной процесс:

Ввод пользователя → GPT анализирует → Определяет функцию → 

Возвращает JSON с названием + параметрами → Ваш код выполняет → 

Возвращает результат → GPT строит ответ

Области применения: запросы данных, автоматизация процессов, системы принятия решений, многослойный AI.

Почему function calling лучше других методов?

Реальный пример: вопрос «Кит ли кошелёк 0xabc?» Без function calling модель угадывает, а с ним — вызывает get_wallet_balance, получает точный баланс и отвечает фактами.

Реализация и оптимизация вызовов функций GPT-4

Шаг 1: Чётко опишите схему функции

tools = [{

    ""type"": ""function"",

    ""function"": {

        ""name"": ""execute_token_swap"",

        ""description"": ""Обменивает токены на DEX. ВЫЗЫВАТЬ только после подтверждения пользователя с осознанием рисков."",

        ""parameters"": {

            ""type"": ""object"",

            ""properties"": {

                ""from_token"": {

                    ""type"": ""string"",

                    ""description"": ""Токен для продажи (например, 'ETH')""

                },

                ""to_token"": {

                    ""type"": ""string"",

                    ""description"": ""Токен для покупки (например, 'USDC')""

                },

                ""amount"": {

                    ""type"": ""number"",

                    ""description"": ""Количество from_token для обмена""

                },

                ""slippage_tolerance"": {

                    ""type"": ""number"",

                    ""description"": ""Максимально допустимое проскальзывание (%)"",

                    ""default"": 0.5

                },

                ""user_confirmation"": {

                    ""type"": ""boolean"",

                    ""description"": ""Явное подтверждение от пользователя (должен быть true)""

                }

            },

            ""required"": [""from_token"", ""to_token"", ""amount"", ""user_confirmation""]

        }

    }

}]

Шаг 2: Создайте надёжные обработчики функций

import logging

from functools import wraps

from typing import Dict, Any




def function_error_handler(func):

    @wraps(func)

    def wrapper(*args, **kwargs):

        try:

            result = func(*args, **kwargs)

            return {

                ""success"": True,

                ""data"": result,

                ""error"": None

            }

        except ValueError as e:

            logging.warning(f""Ошибка валидации в {func.__name__}: {e}"")

            return {

                ""success"": False,

                ""data"": None,

                ""error"": f""Неверные данные: {str(e)}"",

                ""retry_suggested"": False

            }

        except requests.exceptions.Timeout:

            logging.error(f""Тайм-аут в {func.__name__}"")

            return {

                ""success"": False,

                ""data"": None,

                ""error"": ""Тайм-аут сервиса. Попробуйте позже."",

                ""retry_suggested"": True

            }

        except Exception as e:

            logging.error(f""Непредвиденная ошибка в {func.__name__}: {e}"")

            return {

                ""success"": False,

                ""data"": None,

                ""error"": ""Внутренняя ошибка"",

                ""retry_suggested"": False

            }

    return wrapper

Шаг 3: Оптимизируйте цикл исполнения run

def execute_run_with_functions(thread_id: str, assistant_id: str, max_iterations: int = 10):

    run = client.beta.threads.runs.create(

        thread_id=thread_id,

        assistant_id=assistant_id

    )

    

    iterations = 0

    

    while run.status in [""queued"", ""in_progress"", ""requires_action""]:

        if iterations >= max_iterations:

            logging.error(f""Достигнуто максимальное число итераций для run {run.id}"")

            return {""error"": ""Тайм-аут обработки — слишком много вызовов""}

        

        run = client.beta.threads.runs.retrieve(

            thread_id=thread_id,

            run_id=run.id

        )

        

        if run.status == ""requires_action"":

            iterations += 1

            tool_calls = run.required_action.submit_tool_outputs.tool_calls

            tool_outputs = []

            

            for tool_call in tool_calls:

                function_name = tool_call.function.name

                arguments = json.loads(tool_call.function.arguments)

                

                logging.info(f""Выполняем {function_name} с аргументами: {arguments}"")

                

                if function_name in FUNCTION_MAP:

                    result = FUNCTION_MAP[function_name](**arguments)

                else:

                    result = {""error"": f""Неизвестная функция: {function_name}""}

                

                tool_outputs.append({

                    ""tool_call_id"": tool_call.id,

                    ""output"": json.dumps(result)

                })

            

            run = client.beta.threads.runs.submit_tool_outputs(

                thread_id=thread_id,

                run_id=run.id,

                tool_outputs=tool_outputs

            )

        

        time.sleep(0.5)

    

    if run.status == ""completed"":

        messages = client.beta.threads.messages.list(thread_id=thread_id)

        return messages.data[0].content[0].text.value

    else:

        return {""error"": f""Run завершился со статусом: {run.status}""}




FUNCTION_MAP = {

    ""get_token_price"": get_token_price,

    ""get_wallet_balance"": get_wallet_balance,

    ""analyze_token_comprehensive"": analyze_token_comprehensive,

    ""execute_token_swap"": execute_token_swap

}

Оптимизации по производительности:

1. Параллельное выполнение функций

import concurrent.futures

def execute_functions_parallel(tool_calls):

    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:

        futures = {}

        

        for tool_call in tool_calls:

            function_name = tool_call.function.name

            arguments = json.loads(tool_call.function.arguments)

            

            future = executor.submit(FUNCTION_MAP[function_name], **arguments)

            futures[future] = tool_call.id

        

        tool_outputs = []

        for future in concurrent.futures.as_completed(futures):

            tool_call_id = futures[future]

            result = future.result()

            tool_outputs.append({

                ""tool_call_id"": tool_call_id,

                ""output"": json.dumps(result)

            })

        

        return tool_outputs

2. Кеширование результатов

from functools import lru_cache

import hashlib




@lru_cache(maxsize=100)

def cached_token_price(symbol: str, chain: str) -> str:

    result = get_token_price(symbol, chain)

    return json.dumps(result)

Используйте функцию с кешем для сокращения вызовов

3. Стриминг ответов для моментального впечатления

from openai import AssistantEventHandler




class FunctionCallingHandler(AssistantEventHandler):

    def on_tool_call_created(self, tool_call):

        print(f""\n🔧 Вызов функции: {tool_call.function.name}"")

    

    def on_tool_call_done(self, tool_call):

        print(f""✅ Завершено: {tool_call.function.name}"")

    

    def on_text_delta(self, delta, snapshot):

        print(delta.value, end="""", flush=True)




with client.beta.threads.runs.stream(

    thread_id=thread_id,

    assistant_id=assistant_id,

    event_handler=FunctionCallingHandler()

) as stream:

    stream.until_done()

 

Пример оптимизации ASCN: изначальные поочередные вызовы занимали около 12 секунд. Параллелизация с кешированием сократила время до 5 секунд и уменьшила расходы на 40%.

Примеры, ошибки и лучшие практики

Паттерн 1: подтверждение для чувствительных операций

{""function"": ""preview_token_swap"",""arguments"": {""from_token"": ""ETH"",""to_token"": ""USDC"",""amount"": 1.5}}

Сначала возвращается оценка. Пользователь подтверждает. Только после этого вызывается execute_token_swap с user_confirmation=true.

Паттерн 2: постепенный сбор данных

Несколько последовательных вызовов функций, результаты которых направляют следующую логику и финальный отзыв.

Паттерн 3: цепочки с резервными источниками

Функции пытаются данные из нескольких источников с плавным переходом, если основной недоступен.

Типичные ошибки и их решения

Ошибка 1: бесконечный цикл вызовов — решается явными флагами успеха.

Ошибка 2: некорректный разбор аргументов — нужна строгая схема и валидация.

Ошибка 3: тайм-ауты в многоступенчатых сценариях — разбивайте логику на мелкие кешируемые функции.

Сводка лучших практик

1. Одна функция — одна задача

2. Возвращайте структурированный JSON с флагами статуса и ошибками

3. Пишите чёткие и подробные схемы с валидацией

4. Обрабатывайте ошибки и обеспечивайте плавный откат

5. Оптимизируйте скорость через кеширование, параллелизм и тайм-ауты

6. Защищайте входы и требуйте явное подтверждение для рискованных действий

По опыту ASCN: 40% ошибок связаны с непонятными схемами, 30% — с тайм-аутами (исправлено кешированием), 20% — с валидацией, 10% — редкие крайние случаи.

Автоматизация AI агентов и рабочих процессов с кастомными GPT и инструментами

Автоматизация переводит ассистента из реактивного режима в проактивный:

• Активаторы (триггеры) по времени, событию или условию

• Автономные деревья решений

• Оркестрация нескольких агентов

• Управление состоянием

К примеру, Python-фреймворк, который запускает workflow в зависимости от контекста и триггеров.

Пример ASCN: агент мониторинга портфеля периодически и при событиях дергает OpenAI, отправляет оповещения.

Паттерны автоматизации

1. Запланированный анализ: пакетная обработка с уведомлениями

2. Событийный рабочий процесс: ответ на транзакции блокчейна с отчетами

3. Автоматизация по запросу пользователя: уведомления и анализ под списки наблюдения

Преимущества кастомных GPT и инструментов

• Сокращение затрат: ИИ обходится в $200-500 в месяц вместо $12,000 или более на аналитика

• Точность благодаря живым данным

• Масштабируемость с минимальными дополнительными расходами

• Единые стандарты и последовательные оценки

• Быстрая реакция на изменения рынка

Примеры из реальной жизни

Агент автоматического ребалансинга портфеля

Следит за распределением активов и помогает или выполняет ребалансировку. Экономит время и улучшает дисциплину.

Система мониторинга рисков в реальном времени

Отслеживает ликвидность, активность китов, безопасность смарт-контрактов и новости. Ранжирует риски и оперативно предупреждает.

Бот для криптотрейдинга без кодов (ASCN NoCode)

Трейдеры без программирования создают ботов через визуальный интерфейс — триггеры, AI-анализ, логику, вызовы DEX API и нотификации.

Сравнение: традиционная разработка занимает дни, no-code — минуты, с сопоставимой функциональностью.

Часто задаваемые вопросы (FAQ)

Как расширить функциональность ассистента?

Добавляйте нативные инструменты, кастомные функции, интегрируйте сторонние сервисы, расширяйте базы знаний через документы. Растите функциональность по мере нужд пользователей.

Какие ограничения у OpenAI кастомных ассистентов?

Технические: лимиты по времени ответа, максимальной длительности run, размеру контекста, риски рекурсии в вызовах, лимиты API.

Функциональные: нет прямого доступа к базам данных; runs необязательно сохраняют состояние; вывод — только текст; ограниченное взаимодействие в реальном времени.

Стоимость: варьируется в зависимости от использования; оптимизируйте кешированием и правильным выбором моделей.

Безопасность: нет встроенной аутентификации; обязательна проверка и валидация входных данных; нет прямого доступа к устройствам пользователей или отправке писем.

Как тестировать и отлаживать интеграции?

Тестируйте многоуровнево:

1. Юнит-тесты функций

2. Тесты интеграции ассистента

3. Полные сценарии end-to-end

Ведите подробный лог, проверяйте ответы, трассируйте вызовы функций. Исправляйте типичные ошибки с помощью схем, тайм-аутов и улучшенного информирования об ошибках.