A Beginner’s Journey into AI with Fetch.ai: Exploring the AI Agents(Part-2)
Welcome back! If you’ve made it to this page, it means you’re truly interested in how are revolutionising the current AI landscape. In the previous part, we discussed what AI agents are and the various components of the Fetch.ai ecosystem that implement uAgents in our daily lives. In this section, we will explore how to use a multi-agent system with primary and secondary functions. We will see how multiple agents with autonomous scripts can work together to achieve an objective.
Primary and Secondary Functions
When a user provides an objective to DeltaV, the AI engine (the backend for DeltaV) converts the human-readable objective into actionable commands. The AI Engine searches for the primary function that matches the objective and collects fields from the user or other secondary functions (helper functions) to accomplish the objective. Once all the fields for the functions are collected, the agent’s script is executed, and the response is sent back to the AI Engine.
To illustrate this better, we will implement an example where we use multiple agents to get stock prices, perform technical analysis, and analyse news sentiment for a given company.
Step 1: Create Individual agents and respective secondary functions.
Open Agentverse and go to “My Agents”. Click on “New Agent” and select “Blank Agent”. Create all three secondary agents one by one as shown below:
- Stocks Price Agent.
Save the below script in agent.
from ai_engine import UAgentResponse, UAgentResponseType
from uagents import Model, Context, Protocol
from pydantic import Field
import requests
import json
# Data model for stock price request
class StockPriceRequest(Model):
company_name: str
# Function to get the symbol for a company name
async def get_symbol(company_name):
url = f"https://www.alphavantage.co/query?function=SYMBOL_SEARCH&keywords={company_name}&apikey={API_KEY}"
response = requests.get(url)
data = response.json()
if 'bestMatches' in data and data['bestMatches']:
first_match = data['bestMatches'][0]
symbol = first_match['1. symbol']
return symbol
else:
return f"No symbol found for {company_name}."
# Function to get the stock price for a symbol
async def get_stock_price(symbol):
url = f"https://www.alphavantage.co/query?function=TIME_SERIES_INTRADAY&symbol={symbol}&interval=1min&apikey={API_KEY}"
response = requests.get(url)
data = response.json()
print(data)
if 'Time Series (1min)' in data:
latest_time = sorted(data['Time Series (1min)'].keys())[0]
latest_data = data['Time Series (1min)'][latest_time]
current_price = latest_data['1. open']
return current_price
else:
return "Error: Unable to fetch stock price."
# Define the protocol for the stock price agent
stock_price_agent = Protocol("Stock Price", version="1.1")
# Message handler for stock price requests
@stock_price_agent.on_message(model=StockPriceRequest, replies={UAgentResponse})
async def handle_request(ctx: Context, sender: str, msg: StockPriceRequest):
try:
symbol = await get_symbol(msg.company_name)
ctx.logger.info(f'Ticker symbol for given company name is {symbol}')
if "Error" in symbol:
raise Exception(symbol)
stock_price = await get_stock_price(symbol)
final_string = f'Stock price for {symbol} is $ {str(stock_price)}. '
ctx.logger.info(f"Success: {final_string}")
await ctx.send(sender, UAgentResponse(message=final_string, type=UAgentResponseType.FINAL))
except Exception as e:
error_message = f"Error: {str(e)}"
ctx.logger.error(error_message)
await ctx.send(sender, UAgentResponse(message=error_message, type=UAgentResponseType.FINAL))
# Include the protocol in the agent
agent.include(stock_price_agent)This script takes the company name as input, finds the ticker symbol, and retrieves the stock price for that ticker symbol using the Alpha Vantage API. Save the API_KEY in agent secrets. To get the API key, please visit here.
Run the script using the “RUN” button and go to the “Deploy” tab. Create a new function and fill in the details as shown below:
{
"Function Title": "Stocks price Function",
"Description for AI Engine": "This function provides stocks price for the provided company name.",
"Application": "Secondary function",
"Protocol": "Stock Price Protocol", // Auto populating from agent script
"Model": "StockPriceRequest", // Auto populating from agent script
"Fields": {
"company_name": {
"Description": "This describes the company name for which stocks price is required."
}
}
}2. Technical Analysis Agent
Below is the agent’s script.
from ai_engine import UAgentResponse, UAgentResponseType
from functions import get_indicator, calculate_signal, analyze_stock, summarize_signals
from uagents import Model, Protocol, Context
from pydantic import Field
import requests
import json
# Data model for technical analysis request
class TechAnalysisRequest(Model):
company_name: str
# Function to get the symbol for a company name
async def get_symbol(company_name):
url = f"https://www.alphavantage.co/query?function=SYMBOL_SEARCH&keywords={company_name}&apikey={API_KEY}"
response = requests.get(url)
data = response.json()
if 'bestMatches' in data and data['bestMatches']:
first_match = data['bestMatches'][0]
symbol = first_match['1. symbol']
return symbol
else:
return f"No symbol found for {company_name}."
# Function to get the technical summary for a symbol
async def get_technical_summary(symbol):
results = analyze_stock(symbol)
print(results)
summary = summarize_signals(results)
return summary
# Define the protocol for the technical analysis agent
tech_analysis_agent = Protocol("Technical Analysis", version="2.1")
# Message handler for technical analysis requests
@tech_analysis_agent.on_message(model=TechAnalysisRequest, replies={UAgentResponse})
async def handle_request(ctx: Context, sender: str, msg: TechAnalysisRequest):
ctx.logger.info(f"Got request from {sender} for technical analysis: {msg.company_name}")
try:
symbol = await get_symbol(msg.company_name)
ctx.logger.info(f"Symbol: {symbol}")
if "No symbol found" in symbol:
raise Exception(symbol)
summary = await get_technical_summary(symbol)
await ctx.send(sender, UAgentResponse(message=str(summary), type=UAgentResponseType.FINAL))
except Exception as ex:
ctx.logger.info(f"Exception: {str(ex)}")
await ctx.send(sender, UAgentResponse(message=str('function is unsuccessful'), type=UAgentResponseType.FINAL))
# Include the protocol in the agent
agent.include(tech_analysis_agent)Create a new file functions.py to save all the required functions in the file:
import requests
# Function to get indicator data
def get_indicator(symbol, interval, time_period, series_type, function):
url = f'https://www.alphavantage.co/query?function={function}&symbol={symbol}&interval={interval}&time_period={time_period}&series_type={series_type}&apikey={API_KEY}'
response = requests.get(url)
data = response.json()
return data
# Function to calculate the signal (buy, sell, hold)
def calculate_signal(latest_value, previous_value):
if latest_value > previous_value:
return 'BUY'
elif latest_value < previous_value:
return 'SELL'
else:
return 'HOLD'
# Function to analyze the stock using various indicators
def analyze_stock(symbol):
interval = 'daily'
time_period = 20
series_type = 'close'
indicators = [
"SMA", "EMA", "STOCH", "RSI", "CCI", "ADX", "AROON", "BBANDS", "AD", "VWAP", "MACD"
]
results = []
for indicator in indicators:
try:
data = get_indicator(symbol, interval, time_period, series_type, indicator)
key = f"Technical Analysis: {indicator}"
if key in data:
latest_key = list(data[key].keys())[0]
previous_key = list(data[key].keys())[1]
latest_value = float(data[key][latest_key][indicator])
previous_value = float(data[key][previous_key][indicator])
signal = calculate_signal(latest_value, previous_value)
results.append({'indicator': indicator, 'latest_value': latest_value, 'previous_value': previous_value, 'signal': signal})
except Exception as e:
print(f"Skipping indicator {indicator} due to error: {e}")
return results
# Function to summarize the signals
def summarize_signals(results):
buy_signals = 0
sell_signals = 0
hold_signals = 0
for result in results:
if result['signal'] == 'BUY':
buy_signals += 1
elif result['signal'] == 'SELL':
sell_signals += 1
elif result['signal'] == 'HOLD':
hold_signals += 1
summary = f"{buy_signals} indicators for 'BUY'. - {sell_signals} indicators for 'SELL'. - {hold_signals} indicators for 'HOLD'."
return summaryThis tells us about the indicators and whether to buy or sell the stock. Save the API_KEY in agent secrets. To get the API key, please visit here.
Run the script using the “RUN” button and go to the “Deploy” tab. Create a new function and fill in the details as shown below:
{
"Function Title": "Technical Analysis Function",
"Description for AI Engine": "This function gets technical analysis for the provided company name.",
"Application": "Secondary function",
"Protocol": "Technical Analysis Protocol", // Auto populating from agent script
"Model": "TechAnalysisRequest", // Auto populating from agent script
"Fields": {
"company_name": {
"Description": "This describes the company name for which stocks price is required."
}
}
}3. Financial News Sentiment Analysis Agent
Below is the script for our third agent.
from ai_engine import UAgentResponse, UAgentResponseType
from uagents import Model, Context, Protocol
from pydantic import Field
import requests
import json
# Data model for sentiment analysis request
class FinBertRequest(Model):
company_name: str
# Define the protocol for the sentiment analysis agent
finbert_agent = Protocol("FinBert Protocol", version="2.0.1")
# Function to get the symbol for a company name
async def get_symbol(company_name):
url = f"https://www.alphavantage.co/query?function=SYMBOL_SEARCH&keywords={company_name}&apikey={API_KEY}"
response = requests.get(url)
data = response.json()
if 'bestMatches' in data and data['bestMatches']:
first_match = data['bestMatches'][0]
symbol = first_match['1. symbol']
return symbol
else:
return f"No symbol found for {company_name}."
# Function to get news titles for a symbol
async def get_news_titles(symbol):
news_url = f"https://www.alphavantage.co/query?function=NEWS_SENTIMENT&tickers={symbol}&apikey={API_KEY}"
news_response = requests.get(news_url)
news_data = news_response.json()
if 'feed' not in news_data:
return "Error: Unable to fetch news articles."
news_articles = news_data['feed']
if not news_articles:
return "Error: No news articles found."
headlines = [article['title'] for article in news_articles]
combined_headlines = " ".join(headlines)
return combined_headlines
# Function to get sentiment analysis for news titles
async def get_news_sentiment(headlines):
API_URL = "https://api-inference.huggingface.co/models/ProsusAI/finbert"
headers = {"Authorization": f"Bearer {HF_KEY}"}
payload = {"inputs": headlines}
response = requests.post(API_URL, headers=headers, json=payload)
print(response)
sentiment_data = response.json()
sentiment_scores = sentiment_data[0]
overall_sentiment = max(sentiment_scores, key=lambda x: x['score'])['label']
formatted_scores = ", ".join([f"{score['label']}: {score['score']:.2f}" for score in sentiment_scores])
return f"Overall sentiment: {overall_sentiment}. with scores: {formatted_scores}."
# Message handler for sentiment analysis requests
@finbert_agent.on_message(model=FinBertRequest, replies={UAgentResponse})
async def handle_request(ctx: Context, sender: str, request: FinBertRequest):
ctx.logger.info(f"Got request from {sender} for text classification: {request.company_name}")
try:
symbol = await get_symbol(request.company_name)
ctx.logger.info(f"Got symbol: {symbol}")
if "No symbol found" in symbol:
raise Exception(symbol)
news_titles = await get_news_titles(symbol)
if "Error" in news_titles:
raise Exception(news_titles)
ctx.logger.info(f"Collected news titles: {news_titles}")
sentiment_score = await get_news_sentiment(news_titles[:200])
if "Error" in sentiment_score:
raise Exception(sentiment_score)
ctx.logger.info(f"Sentiment analysis result: {str(sentiment_score)}")
await ctx.send(sender, UAgentResponse(message=str(sentiment_score), type=UAgentResponseType.FINAL))
except Exception as ex:
ctx.logger.error(f"Exception: {str(ex)}")
await ctx.send(sender, UAgentResponse(message=str(ex), type=UAgentResponseType.ERROR))
# Include the protocol in the agent
agent.include(finbert_agent)This agent performs sentiment analysis of news related to the specified company. Save the API_KEY in agent secrets. To get the API key, please visit here.
Run the script using the “RUN” button and go to the “Deploy” tab. Create a new function and fill in the details as shown below:
{
"Function Title": "Financial News Sentiment Analysis",
"Description for AI Engine": "This function helps to get news sentiment for the provided company name.",
"Application": "Secondary function",
"Protocol": "FinBert Protocol", // Auto populating from agent script
"Model": "FinBertRequest", // Auto populating from agent script
"Fields": {
"company_name": {
"Description": "This describes the company name for which news sentiment is required. If not in objective ask to user"
}
}
}4. Stocks Details Agent
This will be our primary agent, which will provide us with all the stock details. It will trigger the secondary functions created above.
Script for the main agent:
import requests
from ai_engine import UAgentResponse, UAgentResponseType
from uagents import Model, Context, Protocol
# Data model for stock details request
class StocksRequest(Model):
company_name: str
stock_price: str
tech_analysis: str
sentiment: str
# Define the protocol for the stock details agent
stocks_protocol = Protocol('Stocks details protocol')
# Message handler for stock details requests
@stocks_protocol.on_message(model=StocksRequest, replies=UAgentResponse)
async def on_nutrient_request(ctx: Context, sender: str, msg: StocksRequest):
ctx.logger.info(f"Received Stocks details request from {sender} with company name: {msg.company_name}")
response = f'The stock price for {msg.company_name} is {msg.stock_price} \nIndicators: {msg.tech_analysis} \nOverall Sentiment: {msg.sentiment}'
ctx.logger.info(response)
await ctx.send(sender, UAgentResponse(message=response, type=UAgentResponseType.FINAL))
# Include the protocol in the agent
agent.include(stocks_protocol)Primary function for our stocks details objective:
{
"Function Title": "Stocks details function",
"Description for AI Engine": "This function provides stocks details to the user which includes tech analysis, stocks price and news sentiments.",
"Application": "Primary function",
"Protocol": "Stocks details protocol", // Auto populating from agent script
"Model": "StocksRequest", // Auto populating from agent script
"Fields": {
"company_name": {
"Description": "This describes the company name for which news sentiment is required. If not in objective ask to user"
},
"sentiment": {
"Description": "You MUST use secondary function to get the value for this field. Search for "get stocks sentiments for <company_name>""
},
"stock_price": {
"Description": "You MUST use secondary function to get the value for this field. Search for "get stocks price for <company_name>""
},
"tech_analysis": {
"Description": "You MUST use secondary function to get the value for this field. Search for "get stocks technical analysis for <company_name>""
}
}
}The first field, company_name, is provided by the user, and all other fields are populated using secondary functions. Once the primary function collects all the details, it passes them to the script of the primary agent. After the script is executed, the primary agent formats and sends the final response to DeltaV/AI engine, and the user receives the response.
Let’s check how our agent works on DeltaV. Below are the expected outputs on DeltaV UI:
- Provide your objective and select the “my functions” function group and “talkative v3” personality. Click “start”.
- The AI Engine will recognise the company name and start fetching all the fields.
- It executes all secondary functions and gets relevant information.
- All the fetched fields are passed to the primary function.
- The final response with the required details is provided.
This is how we use a multi-agent system to achieve an overall goal. In the next blog, we will learn more about how uAgents behave as TCP/IP for agents from different providers like Langchain and CrewAI. We can wrap those AI agents or tools around uAgents and perform chit-chat using dialogues from Fetch.ai’s tool stack.
