Functions Calling이란 사용자의 입력에 따라, 미리 정의된 소스 코드 상의 함수(Function)로부터 호출해야 할 함수와 그 함수에 전달하는 인자들을 Gemini가 구조화된 데이터로(JSON으로) 출력해주는 기능.
일종의 검색을 위한 프레임워크라고 생각하면 될 것 같다.
일반 RAG 기반 검색과의 차이점은?
"어떤 함수를 호출할지를 Gemini가 판단"
사용자 질문의 '의도'를 분류하는 작업을 의도별로 각각 프롬프트 엔지니어링을 수행하지 않아도 Gemini가 질문을 이해하고 이 질문에 답하기 위해 사용해야 할 함수들을 알아서 호출한다는 것이다.
또한 외부 API와의 상호작용도 가능하다.
import requests
import vertexai
from vertexai.preview.generative_models import (
Content,
FunctionDeclaration,
GenerativeModel,
Part,
Tool,
)
# Vertex AI Init
vertexai.init(project=PROJECT_ID, location=LOCATION)
model = GenerativeModel("gemini-1.5-flash-001")
# 현재 날씨를 얻는 함수
def get_current_weather(location):
if location == "Seoul":
return "sunny"
elif location == "Busan":
return "cloudy"
else:
return "rain"
이제 Gemini가 아래 함수의 description을 이해하여 질문에서 위치정보가 있으면 알아서 인식하여 해당 지역의 날씨를 반환할 것이다.
(이 테스트에선 서울, 부산일 때만 날씨를 반환할 것이다.)
# FunctionDeclaration 만들기
get_current_weather_func = FunctionDeclaration (
name= "get_current_weather" ,
description= "지정된 위치의 현재 날씨 가져오기" ,
parameters={
"type" : "object" ,
"properties" : {
"location" : {
"type" : "string" ,
"description" : "location"
}
}
},
)
# LLM이 호출하는 Tool 정의
weather_tool = Tool(
function_declarations=[get_current_weather_func],
)
# Function calling 함수 정의
def function_calling (prompt) -> dict :
response = model.generate_content(
prompt,
generation_config={ "temperature" : 0 },
tools=[weather_tool],
)
return response
prompt= "How is the weather in Seoul?"
response = function_calling(prompt)
print (response)
print ( "------" )
if response.candidates[ 0 ].content.parts[ 0 ].function_call.name:
print (f"가장 적합한 Function : {response.candidates[0].content.parts[0].function_call.name}" )
print (f"필요한 Arguments : {response.candidates[0].content.parts[0].function_call.args.get('location')}" )
else :
print (f"가장 적합한 Function : 아무것도 없음" )
결과
물론 여기선 날씨 관련 Function이 하나밖에 없긴 하지만 그 Function을 잘 불러왔고 Seoul이라는 location도 잘 인식한 것을 확인할 수 있다.
prompt= "What's the exchange rate?"
response = function_calling(prompt)
print (response)
print ( "------" )
if response.candidates[ 0 ].content.parts[ 0 ].function_call.name:
print (f"가장 적합한 Function : {response.candidates[0].content.parts[0].function_call.name}" )
print (f"필요한 Arguments : {response.candidates[0].content.parts[0].function_call.args.get('location')}" )
else :
print (f"가장 적합한 Function : 아무것도 없음" )
결과
필요한 내용을 찾을 수 없다고 추가 정보를 달라고 LLM의 답변이 나온다.
당연히 적합한 Function은 찾을 수 없었을 것이다.
prompt= "How is the weather in Busan?"
response = function_calling(prompt)
if response.candidates[ 0 ].content.parts[ 0 ].function_call:
# 적합한 Function name 얻기
function_name = response.candidates[ 0 ].content.parts[ 0 ].function_call.name
# 인수에 필요한 location 얻기
location = response.candidates[ 0 ].content.parts[ 0 ].function_call.args.get( "location" )
# 실행 가능한 Function
available_functions = {
"get_current_weather" : get_current_weather
}
exec_function = available_functions[function_name]
# 함수 실행
function_response = exec_function(location)
print (f"{location}의 날씨는 {function_response}입니다." )
else :
print (f"가장 적합한 Function : 아무것도 없음" )
결과
회사 정보 API는 Alpha Vantage에서 얻을 것이다.
정보를 입력하고 GET FREE API KEY를 누르면 API키가 나올 것이다. 복사해서 아래 Notebook 코드에 붙여넣자.
import requests
from IPython.display import display, Markdown
from vertexai.generative_models import (
FunctionDeclaration,
GenerativeModel,
GenerationConfig,
Part,
Tool,
)
API_KEY = '<복사한 API Key 입력>'
OpenAPI JSON 스키마에 따라 목적에 맞게 함수 선언
get_stock_price = FunctionDeclaration(
name="get_stock_price",
description="Fetch the current stock price of a given company",
parameters={
"type": "object",
"properties": {
"ticker": {
"type": "string",
"description": "Stock ticker symbol for a company",
}
},
},
)
get_company_overview = FunctionDeclaration(
name="get_company_overview",
description="Get company details and other financial data",
parameters={
"type": "object",
"properties": {
"ticker": {
"type": "string",
"description": "Stock ticker symbol for a company",
}
},
},
)
get_company_news = FunctionDeclaration(
name="get_company_news",
description="Get the latest news headlines for a given company.",
parameters={
"type": "object",
"properties": {
"tickers": {
"type": "string",
"description": "Stock ticker symbol for a company",
}
},
},
)
get_news_with_sentiment = FunctionDeclaration(
name="get_news_with_sentiment",
description="Gets live and historical market news and sentiment data",
parameters={
"type": "object",
"properties": {
"news_topic": {
"type": "string",
"description": """News topic to learn about. Supported topics
include blockchain, earnings, ipo,
mergers_and_acquisitions, financial_markets,
economy_fiscal, economy_monetary, economy_macro,
energy_transportation, finance, life_sciences,
manufacturing, real_estate, retail_wholesale,
and technology""",
},
},
},
)
Tool을 사용하여 하나로 묶어주고 Gemini가 이 Tool에서 사용자 질문의 의도를 파악하여 필요한 함수를 선별하여 호출하게 된다.
company_insights_tool = Tool(
function_declarations=[
get_stock_price,
get_company_overview,
get_company_news,
get_news_with_sentiment,
],
)
의도별로 사용할 외부 API 설정
def get_stock_price_from_api(content):
url = f"https://www.alphavantage.co/query?function=GLOBAL_QUOTE&symbol={content['ticker']}&apikey={API_KEY}"
api_request = requests.get(url)
return api_request.text
def get_company_overview_from_api(content):
url = f"https://www.alphavantage.co/query?function=OVERVIEW&symbol={content['ticker']}&apikey={API_KEY}"
api_response = requests.get(url)
return api_response.text
def get_company_news_from_api(content):
url = f"https://www.alphavantage.co/query?function=NEWS_SENTIMENT&tickers={content['tickers']}&limit=20&sort=RELEVANCE&apikey={API_KEY}"
api_response = requests.get(url)
return api_response.text
def get_news_with_sentiment_from_api(content):
url = f"https://www.alphavantage.co/query?function=NEWS_SENTIMENT&topics={content['news_topic']}&limit=20&sort=RELEVANCE&apikey={API_KEY}"
api_request = requests.get(url)
return api_request.text
function_handler = {
"get_stock_price": get_stock_price_from_api,
"get_company_overview": get_company_overview_from_api,
"get_company_news": get_company_news_from_api,
"get_news_with_sentiment": get_news_with_sentiment_from_api,
}
Gemini 모델 설정
gemini_model = GenerativeModel(
"gemini-1.5-flash-001",
generation_config=GenerationConfig(temperature=0),
tools=[company_insights_tool],
)
chat = gemini_model.start_chat()
답변 형식 설정
def send_chat_message(prompt):
display(Markdown("#### Prompt"))
print(prompt, "\n")
prompt += """
Give a concise, high-level summary. Only use information that you learn from
the API responses.
"""
# Gemini로 채팅 메시지 보내기
response = chat.send_message(prompt)
# 함수 호출
function_calling_in_process = True
while function_calling_in_process:
# 함수 호출의 응답에서 값 추출
function_call = response.candidates[0].content.parts[0].function_call
if function_call.name in function_handler.keys():
# 함수 호출의 응답에서 값 추출
function_call = response.candidates[0].content.parts[0].function_call
# Gemini가 판단하여 호출한 함수
function_name = function_call.name
display(Markdown("#### Predicted function name"))
print(function_name, "\n")
# 질문을 이해하고 Gemini가 추출한 parameter
params = {key: value for key, value in function_call.args.items()}
display(Markdown("#### Predicted function parameters"))
print(params, "\n")
# 외부 API를 호출하는 함수 호출
function_api_response = function_handler[function_name](params)[
:20000
] # Stay within the input token limit
display(Markdown("#### API response"))
print(function_api_response[:500], "...", "\n")
# 자연어 요약 또는 다른 함수 호출을 생성하는 Gemini로 API 응답을 다시 보냄.
response = chat.send_message(
Part.from_function_response(
name=function_name,
response={"content": function_api_response},
),
)
else:
function_calling_in_process = False
# 최종 LLM 요약
display(Markdown("#### Natural language response"))
display(Markdown(response.text.replace("$", "\\\$")))
send_chat_message("구글의 현재 주가가 얼마야?")
결과
send_chat_message("구글에 대한 회사 정보 알려줘")
결과
send_chat_message("Walmart랑 The Home Depot에 대한 회사 정보 알려줘.")
결과
이를 활용하면 앞서 포스팅했던 NL2SQL구현이 훨씬 간단해질 수도 있다. 이유는 위의 포스팅에서도 외부 API에 데이터도 스스로 이해하고 LLM이 해당 정보를 가져와서 답변을 한다. 이는 BigQuery API를 통해 Dataset및 Table 리스트와 스키마 정보들을 불러오는 함수들을 선언해 놓으면 LLM이 사용자 질문에 맞는 테이블 정보들을 스스로 찾은 뒤 필요한 작업을 진행하게도 할 수 있다는 뜻이다.
[Gemini Function Calling을 사용한 AI기반 Agent 구현 참고]