Создание AI-агента с помощью функции Yandex Cloud Functions

Узнайте, как с помощью функции Cloud Functions с использованием OpenAI Agents SDK создать AI-агента — персонализированного помощника, реализующего сценарий генеративного ответа на основе моделей генерации текста Yandex AI Studio.

В данном руководстве вы создадите простого AI-агента, который через функцию Cloud Functions будет получать информацию о погоде, взаимодействовать с языковой моделью Yandex AI Studio и, чтобы было интереснее, отвечать пользователю в форме хайку.

На схеме:

1. Пользователь отправляет запрос к AI-агенту. Запрос поступает в обработчик функции Cloud Functions.
2. Обработчик функции Cloud Functions создает и запускает AI-агента.
3. AI-агент запрашивает дополнительные данные о погоде в нужном городе в пользовательской функции get_weather().
4. AI-агент отправляет расширенный дополнительным контекстом запрос пользователя в модель генерации текста AI Studio.
5. Сервисный аккаунт с помощью API-ключа авторизует для AI-агента доступ к Text Generation API AI Studio.
6. Сервисный аккаунт предоставляет функции Cloud Functions доступ к секрету Yandex Lockbox, в котором хранится API-ключ сервисного аккаунта.
7. Функция Cloud Functions получает из секрета Yandex Lockbox API-ключ сервисного аккаунта.
8. Модель AI Studio передает AI-агенту сгенерированный ответ.
9. AI-агент возвращает полученный ответ пользователю.

Созданный в этом руководстве AI-агент будет развернут в Yandex Cloud с помощью Yandex Cloud Functions. При этом вы создадите необходимые для доступа к модели AI Studio ресурсы: сервисный аккаунт, API-ключ сервисного аккаунта, секрет Yandex Lockbox и функцию Cloud Functions.

Вы можете создать эти ресурсы автоматически или вручную.

Что такое AI-агент и зачем он нуженЧто такое AI-агент и зачем он нужен

AI-агент — это программный помощник на базе искусственного интеллекта, который может выполнять инструкции, отвечать на вопросы и взаимодействовать с пользователями или другими системами в заданном контексте. В отличие от обычных моделей генеративного искусственного интеллекта, AI-агенты могут:

• обладать персонализированными инструкциями и «личностью»;
• использовать внешние источники информации и сторонние инструменты, позволяющие получить дополнительную информацию;
• поддерживать контекст диалога;
• выполнять последовательности действий для решения комплексных задач.

Почему бессерверные функции подходят для работы с AI-агентамиПочему бессерверные функции подходят для работы с AI-агентами

Функции Cloud Functions при развертывании AI-агентов предоставляют несколько преимуществ:

• Масштабируемость — автоматическое масштабирование в зависимости от нагрузки.
• Экономичность — плата только за фактическое время выполнения.
• Отсутствие необходимости управлять инфраструктурой — не нужно настраивать и поддерживать серверы.
• Быстрое развертывание — простота создания и обновления AI-агента.
• Интеграция с другими сервисами — простота подключения к API и другим сервисам Yandex Cloud.

Подготовьте облако к работеПодготовьте облако к работе

Зарегистрируйтесь в Yandex Cloud и создайте платежный аккаунт:

1. Перейдите в консоль управления, затем войдите в Yandex Cloud или зарегистрируйтесь.
2. На странице Yandex Cloud Billing убедитесь, что у вас подключен платежный аккаунт, и он находится в статусе ACTIVE или TRIAL_ACTIVE. Если платежного аккаунта нет, создайте его и привяжите к нему облако.

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

Подробнее об облаках и каталогах.

Необходимые платные ресурсыНеобходимые платные ресурсы

В стоимость поддержки инфраструктуры для этого практического руководства входят:

• Плата за использование функции (см. тарифы Yandex Cloud Functions).
• Плата за генерацию текста (см. тарифы Yandex AI Studio).
• Плата за хранение секрета и операции с ним (см. тарифы Yandex Lockbox).
• Плата за запись и хранение данных в лог-группе (см. тарифы Yandex Cloud Logging), если вы используете сервис Cloud Logging.

Автоматизированное развертывание AI-агента с использованием SourceCraft CI/CDАвтоматизированное развертывание AI-агента с использованием SourceCraft CI/CD

Запустите в SourceCraft CI/CD-процесс, который автоматически создаст в выбранном вами каталоге все необходимые ресурсы и развернет готового AI-агента, с которым вы сможете поэкспериментировать.

Почему SourceCraft CI/CD — самый быстрый и безопасный способ для развертывания AI-агентаПочему SourceCraft CI/CD — самый быстрый и безопасный способ для развертывания AI-агента

В SourceCraft при развертывании AI-агентов предоставляется несколько преимуществ:

• Отсутствие необходимости настраивать среду разработки локально — все необходимые настройки происходят удаленно в процессе запуска CI/CD-процессов в SourceCraft.
• Безопасность — для развертывания не нужно хранить секреты от сервисных аккаунтов локально. SourceCraft через сервисное подключение предоставляет безопасный способ запроса короткоживущих IAM-токенов для доступа к сервисам Yandex Cloud.

Чтобы создать AI-агента в автоматическом режиме с использованием SourceCraft CI/CD:

1. Сделайте ответвление (fork) для репозитория со скриптами для создания AI-агента.

   Репозиторий содержит скрипты для создания следующих AI-агентов:

   Простой AI-агент

   Продвинутый AI-агент

   Агент-переводчик

   Процесс create-simple-ai-agent-workflow создаст простого AI-агента, который при запуске возвращает информацию о погоде в Токио в форме хайку.

   Процесс create-advanced-ai-agent-workflow создаст продвинутого AI-агента, который отвечает на пользовательские запросы в форме хайку. При отсутствии пользовательского запроса агент возвращает ответ на запрос по умолчанию.

   Процесс create-complex-ai-agent-workflow создаст Агента-переводчика, которому можно передавать пользовательские запросы на перевод текста на нужный язык в нужном стиле.

2. Настройте сервисное подключение в SourceCraft для доступа к вашему облаку Yandex Cloud.

3. Запустите рабочий процесс, например create-simple-ai-agent-workflow.

4. Протестируйте работу AI-агента.

   Вы можете увидеть результат работы функции, посмотрев логи кубика SourceCraft CI/CD, например, для create-simple-ai-agent-workflow этот кубик называется deploy-simple-ai-agent. Дополнительно вы можете протестировать работу функции в консоли управления Yandex Cloud.

5. Если созданные ресурсы вам больше не нужны, удалите их.

Автоматизированное создание AI-агентаАвтоматизированное создание AI-агента

Запустите скрипт, который автоматически создаст в выбранном вами каталоге все необходимые ресурсы и развернет готового AI-агента, с которым вы сможете поэкспериментировать.

Чтобы создать AI-агента в автоматическом режиме:

1. Установите Yandex Cloud CLI и аутентифицируйтесь.

2. Скачайте и установите Python.

3. Склонируйте репозиторий со скриптами для создания AI-агента:

   git clone https://git@git.sourcecraft.dev/yandex-cloud-examples/yc-serverless-ai-agent.git
   

   Репозиторий содержит скрипты для создания трех вариантов AI-агента:

   Простой AI-агент

   Продвинутый AI-агент

   Агент-переводчик

   Скрипт create-simple-ai-agent.py создаст простого AI-агента, который при запуске возвращает информацию о погоде в Токио в форме хайку.

   Скрипт create-advanced-ai-agent.py создаст продвинутого AI-агента, который отвечает на пользовательские запросы в форме хайку. При отсутствии пользовательского запроса агент возвращает ответ на запрос по умолчанию.

   Скрипт create-complex-ai-agent.py создаст Агента-переводчика, которому можно передавать пользовательские запросы на перевод текста на нужный язык в нужном стиле.

4. Сделайте файл скрипта исполняемым, например:

   chmod +x create-simple-ai-agent.py
   

5. Выполните скрипт, например:

   ./create-simple-ai-agent.py
   

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

6. Протестируйте работу AI-агента.

   Простой AI-агент

   Продвинутый AI-агент

   Агент-переводчик

   Протестируйте работу функции в консоли управления.

   Протестируйте работу функции в консоли управления.

   Дополнительно вы можете протестировать работу функции, передав AI-агенту пользовательский запрос с помощью Yandex Cloud CLI. Для этого выполните команду, указав полученный при создании AI-агента идентификатор функции:

   echo '{"query": "What's the weather in Paris?"}' > request.json \
     yc serverless function invoke <идентификатор_функции> --data-file request.json
   

   Протестируйте работу функции в консоли управления.

   Дополнительно вы можете протестировать работу функции, передав AI-агенту пользовательский запрос с помощью Yandex Cloud CLI. Для этого выполните команду, указав полученный при создании AI-агента идентификатор функции:

   echo '{"text": "Hello, world!", "target_language": "русский", "tone": "дружеский"}' > translate.json \
     yc serverless function invoke <идентификатор_функции> --data-file translate.json
   

7. Если созданные ресурсы вам больше не нужны, удалите их.

Создание AI-агента вручнуюСоздание AI-агента вручную

Чтобы вручную создать AI-агента с помощью функции Cloud Functions:

1. Создайте сервисный аккаунт.
2. Создайте API-ключ сервисного аккаунта.
3. Создайте секрет Yandex Lockbox.
4. Подготовьте ZIP-архив с кодом функции.
5. Создайте функцию.
6. Протестируйте работу функции.

Если созданные ресурсы вам больше не нужны, удалите их.

Создайте сервисный аккаунтСоздайте сервисный аккаунт

Создайте API-ключ сервисного аккаунтаСоздайте API-ключ сервисного аккаунта

Создайте секрет Yandex LockboxСоздайте секрет Yandex Lockbox

Назначьте сервисному аккаунту права на чтение содержимого секретаНазначьте сервисному аккаунту права на чтение содержимого секрета

Подготовьте ZIP-архив с кодом функцииПодготовьте ZIP-архив с кодом функции

1. Сохраните следующий код в файл с названием index.py:

   Простой AI-агент

   Продвинутый AI-агент

   Агент-переводчик

   import os
   
   from openai import AsyncOpenAI
   
   from agents import (
       Agent,
       OpenAIChatCompletionsModel,
       Runner,
       function_tool,
       set_tracing_disabled,
   )
   
   BASE_URL = os.getenv("BASE_URL")
   API_KEY = os.getenv("API_KEY")
   MODEL_NAME = os.getenv("MODEL_NAME")
   
   client = AsyncOpenAI(base_url=BASE_URL, api_key=API_KEY)
   set_tracing_disabled(disabled=True)
   
   
   @function_tool
   def get_weather(city: str):
       print(f"[debug] getting weather for {city}")
       return f"The weather in {city} is sunny."
   
   
   async def handler(event, context):
       agent = Agent(
           name="Assistant",
           instructions="You only respond in haikus.",
           model=OpenAIChatCompletionsModel(model=MODEL_NAME, openai_client=client),
           tools=[get_weather],
       )
   
       result = await Runner.run(agent, "What's the weather in Tokyo?")
   
       return {
           "statusCode": 200,
           "body": result.final_output,
       }
   

   import os
   import json
   
   from openai import AsyncOpenAI
   
   from agents import (
       Agent,
       OpenAIChatCompletionsModel,
       Runner,
       function_tool,
       set_tracing_disabled,
   )
   
   BASE_URL = os.getenv("BASE_URL")
   API_KEY = os.getenv("API_KEY")
   MODEL_NAME = os.getenv("MODEL_NAME")
   
   client = AsyncOpenAI(base_url=BASE_URL, api_key=API_KEY)
   set_tracing_disabled(disabled=True)
   
   
   @function_tool
   def get_weather(city: str):
       print(f"[debug] getting weather for {city}")
       return f"The weather in {city} is sunny."
   
   
   async def handler(event, context):
       # Извлекаем пользовательский запрос из события
       user_query = "What's the weather today?"  # Запрос по умолчанию
       
       try:
           # Пытаемся получить запрос из тела события
           if event and isinstance(event, dict):
               body = event.get('body')
               if body:
                   # Если body передан как строка (что часто бывает при вызове через API Gateway)
                   if isinstance(body, str):
                       try:
                           body_json = json.loads(body)
                           if 'query' in body_json:
                               user_query = body_json['query']
                       except json.JSONDecodeError:
                           pass  # Используем запрос по умолчанию, если body не является JSON
                   elif isinstance(body, dict) and 'query' in body:
                       user_query = body['query']
               # Если запрос передан напрямую в event
               elif 'query' in event:
                   user_query = event['query']
       except Exception as e:
           print(f"Ошибка при обработке запроса: {e}")
           # В случае ошибки используем запрос по умолчанию
       
       print(f"Обрабатываемый запрос: {user_query}")
       
       agent = Agent(
           name="Assistant",
           instructions="You only respond in haikus. Be creative and poetic.",
           model=OpenAIChatCompletionsModel(model=MODEL_NAME, openai_client=client),
           tools=[get_weather],
       )
   
       result = await Runner.run(agent, user_query)
   
       return {
           "statusCode": 200,
           "body": result.final_output,
           "headers": {
               "Content-Type": "application/json"
           }
       }
   

   import os
   import json
   import re
   from typing import Dict, List, Optional, Any
   
   from openai import AsyncOpenAI
   
   from agents import (
       Agent,
       OpenAIChatCompletionsModel,
       Runner,
       function_tool,
       set_tracing_disabled,
   )
   
   BASE_URL = os.getenv("BASE_URL")
   API_KEY = os.getenv("API_KEY")
   MODEL_NAME = os.getenv("MODEL_NAME")
   
   client = AsyncOpenAI(base_url=BASE_URL, api_key=API_KEY)
   set_tracing_disabled(disabled=True)
   
   # Dictionary with language codes to simplify input
   LANGUAGE_CODES = {
       "русский": "ru",
       "russian": "ru",
       "английский": "en",
       "english": "en",
       "французский": "fr",
       "french": "fr",
       "немецкий": "de",
       "german": "de",
       "испанский": "es",
       "spanish": "es",
       "итальянский": "it",
       "italian": "it",
       "китайский": "zh",
       "chinese": "zh",
       "японский": "ja",
       "japanese": "ja",
       "корейский": "ko",
       "korean": "ko",
       "арабский": "ar",
       "arabic": "ar",
       "португальский": "pt",
       "portuguese": "pt",
   }
   
   # Dictionary with tone codes
   TONE_TYPES = {
       "формальный": "formal",
       "formal": "formal",
       "неформальный": "informal",
       "informal": "informal",
       "дружеский": "friendly",
       "friendly": "friendly",
       "деловой": "business",
       "business": "business",
       "технический": "technical",
       "technical": "technical",
       "простой": "simple",
       "simple": "simple",
       "академический": "academic",
       "academic": "academic",
       "поэтический": "poetic",
       "poetic": "poetic",
       "разговорный": "conversational",
       "conversational": "conversational",
   }
   
   @function_tool
   def detect_language(text: str) -> dict:
       """Detects the language of the input text.
       
       Args:
           text: Text to analyze
           
       Returns:
           Dictionary with detected language and its code
       """
       # This function uses model capabilities to determine the language
       # In a real application, you could use specialized libraries for more accurate detection
       
       common_words = {
           "en": ["the", "and", "of", "to", "in", "a", "is", "that", "for", "it", "with", "as", "was", "on"],
           "ru": ["и", "в", "не", "на", "что", "с", "по", "это", "я", "он", "как", "из", "то", "а", "все", "так"],
           "fr": ["le", "la", "les", "un", "une", "et", "de", "des", "du", "dans", "en", "est", "que", "pour", "qui"],
           "de": ["der", "die", "das", "und", "in", "von", "mit", "den", "zu", "ist", "für", "auf", "dem", "nicht"],
           "es": ["el", "la", "los", "las", "un", "una", "y", "de", "en", "que", "a", "por", "con", "no", "es"],
           "it": ["il", "la", "i", "le", "e", "di", "un", "una", "in", "che", "per", "con", "è", "non", "sono"],
       }
       
       # Simple detection by characteristic symbols and words
       text_lower = text.lower()
       
       # Check for Cyrillic characters (for Russian)
       if re.search('[а-яА-Я]', text):
           return {"language": "Russian", "code": "ru"}
       
       # Check for characters (for Chinese and Japanese)
       if re.search('[\u4e00-\u9fff]', text):
           return {"language": "Chinese", "code": "zh"}
       
       if re.search('[\u3040-\u309F\u30A0-\u30FF]', text):
           return {"language": "Japanese", "code": "ja"}
       
       # Check by common words
       word_counts = {}
       for lang, words in common_words.items():
           count = 0
           for word in words:
               pattern = r'\b' + re.escape(word) + r'\b'
               count += len(re.findall(pattern, text_lower))
           word_counts[lang] = count
       
       # Determine the language with the most matches
       if word_counts:
           detected_code = max(word_counts.items(), key=lambda x: x[1])[0]
           language_names = {
               "en": "English",
               "ru": "Russian",
               "fr": "French",
               "de": "German",
               "es": "Spanish",
               "it": "Italian",
           }
           return {"language": language_names.get(detected_code, "Unknown"), "code": detected_code}
       
       # Default to English
       return {"language": "Unknown (assuming English)", "code": "en"}
   
   @function_tool
   def translate_text(text: str, target_language: str, source_language: Optional[str] = None, tone: Optional[str] = None) -> str:
       """Translates text to the specified language while preserving tone and style.
       
       Args:
           text: Text to translate
           target_language: Target language for translation (code or name)
           source_language: Source language of the text (code or name, optional)
           tone: Desired tone of translation (formal, informal, business, etc., optional)
           
       Returns:
           Translated text
       """
       # This function would use LLM for translation in reality
       # Here we just return information about the request for demonstration
       
       # Normalize language codes
       target_code = LANGUAGE_CODES.get(target_language.lower(), target_language.lower())
       source_code = None
       if source_language:
           source_code = LANGUAGE_CODES.get(source_language.lower(), source_language.lower())
       
       # Normalize tone
       tone_type = None
       if tone:
           tone_type = TONE_TYPES.get(tone.lower(), tone.lower())
       
       # Return information for LLM to perform translation
       # In a real application, you could call a specialized translation API here
       return {
           "text": text,
           "target_language_code": target_code,
           "source_language_code": source_code,
           "tone": tone_type,
           "length": len(text)
       }
   
   async def handler(event, context):
       """Function handler"""
       try:
           # Get text for translation and parameters from the request
           text = None
           target_language = None
           source_language = None
           tone = None
           
           # Process different formats of incoming data
           if event and isinstance(event, dict):
               body = event.get('body')
               
               if body:
                   # If body is passed as a string (common when called via API Gateway)
                   if isinstance(body, str):
                       try:
                           body_json = json.loads(body)
                           text = body_json.get('text')
                           target_language = body_json.get('target_language')
                           source_language = body_json.get('source_language')
                           tone = body_json.get('tone')
                       except json.JSONDecodeError:
                           # Maybe it's just text for translation
                           text = body
                   elif isinstance(body, dict):
                       text = body.get('text')
                       target_language = body.get('target_language')
                       source_language = body.get('source_language')
                       tone = body.get('tone')
               else:
                   # If parameters are passed directly in the event
                   text = event.get('text')
                   target_language = event.get('target_language')
                   source_language = event.get('source_language')
                   tone = event.get('tone')
           
           # Check required parameters
           if not text:
               return {
                   "statusCode": 400,
                   "body": json.dumps({"error": "Missing 'text' parameter (text to translate)"}),
                   "headers": {"Content-Type": "application/json"}
               }
           
           if not target_language:
               return {
                   "statusCode": 400,
                   "body": json.dumps({"error": "Missing 'target_language' parameter (target language)"}),
                   "headers": {"Content-Type": "application/json"}
               }
               
           # Create request for AI agent
           user_query = f"Translate the following text to {target_language}:"
           if source_language:
               user_query += f" from {source_language}"
           if tone:
               user_query += f". Preserve the {tone} tone and style of the original"
           user_query += f"\\n\\n{text}"
           
           # Set up AI agent
           agent = Agent(
               name="TranslatorAssistant",
               instructions="""You are a professional translator assistant who can translate text between different languages 
   while preserving the original tone, style, and context. 
   
   Your tools:
   1. detect_language - to identify the language of the input text
   2. translate_text - to translate text to the target language
   
   You should:
   - Understand the tone and style of the original text
   - Preserve formatting, including paragraphs, bullet points, and special symbols
   - Keep names, terms, and brand names unchanged unless translation is specifically required
   - When translating technical or specialized content, use appropriate terminology
   - Be sensitive to cultural context and adapt idioms appropriately
   - For formal documents, maintain formal language in the translation
   - For creative content, preserve the creative elements in the translation
   
   When responding:
   1. Provide the translation
   2. If requested or if there are important notes about the translation, briefly mention them
   3. Be concise and focus on accurate translation""",
               model=OpenAIChatCompletionsModel(model=MODEL_NAME, openai_client=client),
               tools=[detect_language, translate_text],
           )
   
           # Run the agent
           result = await Runner.run(agent, user_query)
           
           # Format the response
           return {
               "statusCode": 200,
               "body": json.dumps({
                   "translation": result.final_output,
                   "original_text": text,
                   "target_language": target_language,
                   "source_language": source_language,
                   "tone": tone
               }, ensure_ascii=False),
               "headers": {"Content-Type": "application/json; charset=utf-8"}
           }
           
       except Exception as e:
           print(f"Request processing error: {str(e)}")
           return {
               "statusCode": 500,
               "body": json.dumps({"error": f"Internal server error: {str(e)}"}),
               "headers": {"Content-Type": "application/json"}
           }
   

2. Сохраните следующий код в файл с названием requirements.txt:

   openai-agents
   

3. Добавьте файлы index.py и requirements.txt в ZIP-архив openai-function.zip.

Создайте функциюСоздайте функцию

Протестируйте работу функцииПротестируйте работу функции

Как удалить созданные ресурсыКак удалить созданные ресурсы

Чтобы перестать платить за созданные ресурсы:

1. Удалите функцию.
2. Удалите секрет.
3. Если вы записывали логи в лог-группу, удалите ее.