Creating an AI agent using Yandex Cloud Functions

Learn how to use a function from Cloud Functions with the OpenAI Agents SDK to create an AI agent, i.e., a personalized assistant integrated with Yandex Cloud AI Studio text generation models to provide generative responses.

Follow this guide to create a simple AI agent that uses a function from Cloud Functions to get weather information, interact with the Yandex Cloud AI Studio language model, and, to add a touch of fun, respond to the user in haiku form.

On the diagram:

1. The user sends a request to the AI agent. The Cloud Functions handler receives the request.
2. The Cloud Functions handler creates and runs the AI agent.
3. The AI agent requests additional weather data for a given city using the custom get_weather() function.
4. The AI agent sends the user request together with the extra context to the AI Studio text generation model.
5. A service account authorizes the AI agent to access the AI Studio Text Generation API using an API key.
6. The service account grants access to the Yandex Lockbox secret with the service account API key to the function in Cloud Functions.
7. The function in Cloud Functions retrieves the service account API key from the Yandex Lockbox secret.
8. The AI Studio model sends the generated response to the AI agent.
9. The AI agent returns the response to the user.

The AI agent you create in this tutorial will be deployed in Yandex Cloud using Yandex Cloud Functions. You will also create resources to arrange access to the AI Studio model: a service account, the service account API key, the Yandex Lockbox secret, and a function in Cloud Functions.

You can create these resources automatically or manually.

What an AI agent is and how to use itWhat an AI agent is and how to use it

An AI agent is an AI-powered software assistant which can follow instructions, give answers to questions, and interact with users or other systems within a given context. Unlike standard generative models, AI agents can:

• Support personalized instructions and have a personality.
• Use external sources and third-party tools to gather information.
• Maintain conversation context.
• Perform multi-step actions to solve complex tasks.

Why use serverless functions to work with AI agentsWhy use serverless functions to work with AI agents

Functions offered by Cloud Functions provide multiple benefits when deploying AI agents:

• Scalability: Automatic scaling to accommodate the load.
• Cost-effectiveness: You only pay for the actual execution time.
• No infrastructure management required: You do not need to configure or maintain servers.
• Fast deployment: AI agents are easy to create and update.
• Multiple integration options: Simple connection to APIs and other Yandex Cloud services.

Get your cloud readyGet your cloud ready

Sign up for Yandex Cloud and create a billing account:

1. Navigate to the management console and log in to Yandex Cloud or create a new account.
2. On the Yandex Cloud Billing page, make sure you have a billing account linked and it has the ACTIVE or TRIAL_ACTIVE status. If you do not have a billing account, create one and link a cloud to it.

If you have an active billing account, you can create or select a folder for your infrastructure on the cloud page.

Learn more about clouds and folders here.

Required paid resourcesRequired paid resources

The infrastructure support cost for this tutorial includes:

• Fee for using the function (see Yandex Cloud Functions pricing).
• Text generation fee (see Yandex Cloud AI Studio pricing).
• Fee for storing the secret and operations with it (see Yandex Lockbox pricing).
• Fee for logging operations and data storage in a log group (see Yandex Cloud Logging pricing) when using Cloud Logging.

Automated AI agent deployment using SourceCraft CI/CDAutomated AI agent deployment using SourceCraft CI/CD

Run a CI/CD workflow in SourceCraft to automatically create all the required resources in your selected folder and deploy a ready-to-use AI agent for you to experiment with.

Why SourceCraft CI/CD is the fastest and most secure way to deploy an AI agentWhy SourceCraft CI/CD is the fastest and most secure way to deploy an AI agent

SourceCraft provides multiple benefits when deploying AI agents:

• You do not need to set up a local development environment. All the required settings are configured remotely when running CI/CD workflows in SourceCraft.
• Deployment is secure as you do not need to store service account secrets locally. In SourceCraft, you can use a service connection to request short-living IAM tokens for access to Yandex Cloud services.

To create an AI agent automatically using SourceCraft CI/CD:

1. Fork the repository with scripts for AI agent creation.

   The repository contains scripts for creating the following AI agent types:

   Simple AI agent

   Advanced AI agent

   Translator agent

   The create-simple-ai-agent-workflow will create a simple AI agent which, when launched, returns information about the weather in Tokyo in haiku form.

   The create-advanced-ai-agent-workflow will create an advanced AI agent which responds to user requests in haiku form. When there is no user request, the agent returns a response to the default request.

   The create-complex-ai-agent-workflow will create a translator agent which can handle user requests for text translation to the specified language while maintaining the style and tone.

2. Configure a service connection in SourceCraft for access to your Yandex Cloud.

3. Run a workflow, e.g., create-simple-ai-agent-workflow.

4. Test the AI agent.

   You can view the function result in the logs of the SourceCraft CI/CD cube, e.g., for create-simple-ai-agent-workflow this cube is named deploy-simple-ai-agent. In addition, you can test the function in the Yandex Cloud management console.

5. If you no longer need the resources you created, delete them.

Creating an AI agent automaticallyCreating an AI agent automatically

Run a script to automatically create all the required resources in your selected folder and deploy a ready-to-use AI agent for you to experiment with.

To create an AI agent automatically:

1. Install the Yandex Cloud CLI and get authenticated.

2. Download and install Python.

3. Clone the repository with scripts for creating the AI agent:

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

   The repository contains scripts for creating three types of AI agents:

   Simple AI agent

   Advanced AI agent

   Translator agent

   The create-simple-ai-agent.py script will create a simple AI agent which returns information about the weather in Tokyo in haiku form when you run it.

   The create-advanced-ai-agent.py script will create an advanced AI agent which responds to user requests in haiku form. When there is no user request, the agent returns a response to the default request.

   The create-complex-ai-agent.py script will create a translator agent which can handle user requests for text translation to the specified language while maintaining the style and tone.

4. Make the script file executable. Here is an example:

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

5. Run the script:

   ./create-simple-ai-agent.py
   

   While the script is running, specify your cloud and folder details, then wait until the resources are created.

6. Test the AI agent.

   Simple AI agent

   Advanced AI agent

   Translator agent

   Test the function in the management console.

   Test the function in the management console.

   You can also test the function by sending a user request to the AI agent via the Yandex Cloud CLI. To do this, run this command and specify the function ID you obtained when creating the AI agent:

   echo '{"query": "What's the weather in Paris?"}' > request.json \
     yc serverless function invoke <function_ID> --data-file request.json
   

   Test the function in the management console.

   You can also test the function by sending a user request to the AI agent via the Yandex Cloud CLI. To do this, run this command and specify the function ID you obtained when creating the AI agent:

   echo '{"text": "Hello, world!", "target_language": "Russian", "tone": "friendly"}' > translate.json \
     yc serverless function invoke <function_ID> --data-file translate.json
   

7. If you no longer need the resources you created, delete them.

Creating an AI agent manuallyCreating an AI agent manually

To create an AI agent manually using a function from Cloud Functions:

1. Create a service account.
2. Create the service account API key.
3. Create a Yandex Lockbox secret.
4. Prepare a ZIP archive with the function code.
5. Create a function.
6. Test the function.

If you no longer need the resources you created, delete them.

Create a service accountCreate a service account

Create the service account API keyCreate the service account API key

Create a Yandex Lockbox secretCreate a Yandex Lockbox secret

Grant read permissions for the secret contents to the service accountGrant read permissions for the secret contents to the service account

Prepare a ZIP archive with the function codePrepare a ZIP archive with the function code

1. Save the following code to a file named index.py:

   Simple AI agent

   Advanced AI agent

   Translator agent

   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):
       # Extracting the user request from the event
       user_query = "What's the weather today?"  # Default request
       
       try:
           # Attempting to get the request from the event body
           if event and isinstance(event, dict):
               body = event.get('body')
               if body:
                   # If `body ` is provided as a string (which is often the case during API Gateway calls)
                   if isinstance(body, str):
                       try:
                           body_json = json.loads(body)
                           if 'query' in body_json:
                               user_query = body_json['query']
                       except json.JSONDecodeError:
                           pass  # Using the default request if `body` is not JSON
                   elif isinstance(body, dict) and 'query' in body:
                       user_query = body['query']
               # If the request was provided directly to `event`:
               elif 'query' in event:
                   user_query = event['query']
       except Exception as e:
           print(f"Request processing error: {e}")
           # Running the default request in case of an error
       
       print(f"Processed request: {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. Save the following code to a file named requirements.txt:

   openai-agents
   

3. Add the index.py and requirements.txt files into the openai-function.zip archive.

Create a functionCreate a function

Test the functionTest the function

How to delete the resources you createdHow to delete the resources you created

To stop paying for the resources you created:

1. Delete the function.
2. Delete the secret.
3. If you logged data to a log group, delete it.