Ollama로 Mistral 7B의 함수 호출(Function Calling) REST API 사용하기
Ollama로 Mistral 7B의 함수 호출(Function Calling) REST API 사용하기
Ollama에서 실행하는 Mistral 모델로 함수 호출(Function calling)을 구현하는 방법을 배우는 문서예요. 예시로 Pet Store라는 외부 API를 흉내 내어, 모델이 함수 인자만 생성하고 실제 REST API 호출은 사용자(코드) 쪽에서 실행하는 4단계 흐름을 보여줘요.
출처: 문서
본문
함수 호출은 Mistral 모델이 외부 도구에 연결할 수 있게 해줘요. Mistral 모델을 사용자 정의 함수나 API 같은 외부 도구와 통합하면, 특정 사용 사례와 실용적인 문제에 맞는 애플리케이션을 쉽게 만들 수 있어요. 이 가이드에서는 예시로 Pet Store의 Pet과 User 정보를 추적하는 두 함수를 만들고, 이 두 도구로 펫 관련 질문에 답할 거예요.
함수 호출은 크게 네 단계로 이뤄져요:
- User: 도구와 쿼리를 지정한다.
- Model: 해당된다면 함수 인자를 생성한다.
- User: 함수를 실행해서 도구 결과를 얻는다.
- Model: 최종 답변을 생성한다.
이 가이드에서는 이 네 단계로 Mistral 모델이 함수 호출로 동작하는 간단한 예시를 살펴볼 거예요. 먼저 Pet store 정보로 구성된 OpenAPI 스펙 엔드포인트가 있다고 가정해요. 사용자가 이 API에 대해 질문하면, 특정 도구로 이 데이터에 대한 질문에 답할 수 있어요. 이것은 LLM이 직접 접근할 수 없는 외부 데이터베이스를 API로 흉내 내는 예시예요.
설치
!pip install --upgrade ollama mistral-common pandas
!pip install --upgrade prance openapi-spec-validator
import prance
from typing import List
from mistral_common.tokens.tokenizers.mistral import MistralTokenizer
from mistral_common.protocol.instruct.request import ChatCompletionRequest
from mistral_common.protocol.instruct.tool_calls import Function, Tool
import ollama
from mistral_common.protocol.instruct.messages import UserMessage
import json
import requests
import functools
import os
Step 1. 도구와 쿼리 지정 (User: specify tools and query)
REST API 호출을 하는 함수를 설정해요. Swagger Editor에서 가져온 pet store 예시를 사용해요. openapi.json 스펙을 다운로드하고, PetID로 Pet 정보, username으로 User 정보를 얻어요.
curl -X 'GET' \
'https://petstore3.swagger.io/api/v3/pet/1' \
-H 'accept: application/json'
curl -X 'GET' \
'https://petstore3.swagger.io/api/v3/user/user1' \
-H 'accept: application/json'
사용자는 자신의 사용 사례에 필요한 도구를 모두 정의할 수 있어요. 많은 경우 여러 도구를 쓸 수 있는데, 예를 들어 petID와 username으로 Pet·User 정보를 검색하는 retrieve_pet_info와 retreive_user_info 두 함수가 있다고 생각해 봐요. 그런 다음 두 함수를 키가 함수 이름이고 값이 df가 정의된 함수인 딕셔너리로 정리해서, 함수 이름으로 각 함수를 호출할 수 있게 해요.
def getPetById(petId: int) -> str:
try:
method = 'GET'
headers=None
data=None
url = 'https://petstore3.swagger.io/api/v3/pet/' + str(petId)
response = requests.request(method, url, headers=headers, data=data)
# Raise an exception if the response was unsuccessful
response.raise_for_status()
#response = make_api_call('GET', url + str(petId))
if response.ok :
json_response = response.json()
if petId == json_response['id']:
return json_response
return json.dumps({'error': 'Pet id not found.'})
except requests.exceptions.HTTPError as e:
if response.status_code == 404:
return json.dumps({'error': 'Pet id not found.'})
else:
return json.dumps({'error': 'Error with API.'})
def getUserByName(username: str) -> str:
try:
url = 'https://petstore3.swagger.io/api/v3/user/' + username
response = requests.get(url)
# Raise an exception if the response was unsuccessful
response.raise_for_status()
if response.ok :
json_response = response.json()
if username == json_response['username']:
return json_response
return json.dumps({'error': 'Username id not found.'})
except requests.exceptions.HTTPError as e:
if response.status_code == 404:
return json.dumps({'error': 'Username not found.'})
else:
return json.dumps({'error': 'Error with API.'})
names_to_functions = {
'getPetById': functools.partial(getPetById, petId=''),
'getUserByName': functools.partial(getUserByName, username='')
}
Mistral 모델이 함수를 이해하려면 JSON 스키마로 함수 명세를 정의해야 해요. 구체적으로 함수의 타입, 함수 이름, 함수 설명, 함수 파라미터, 필수 파라미터를 설명해야 해요. 함수가 두 개이므로 함수 명세 두 개를 리스트에 나열해요.
도구 생성기 (Tool Generator)
OpenAPI 스펙을 파싱해 동적으로 도구 정의를 만드는 함수예요. openai.json은 https://editor.swagger.io/ 에서 다운로드해요.
def generate_tools(objs, function_end_point)-> List[Tool]:
params = ['operationId', 'description', 'parameters']
parser = prance.ResolvingParser(function_end_point, backend='openapi-spec-validator')
spec = parser.specification
user_tools = []
for obj in objs:
resource, field = obj
path = '/' + resource + '/{' + field + '}'
function_name=spec['paths'][path]['get'][params[0]]
function_description=spec['paths'][path]['get'][params[1]]
function_parameters=spec['paths'][path]['get'][params[2]]
func_parameters = {
"type": "object",
"properties": {
function_parameters[0]['name']: {
"type": function_parameters[0]['schema']['type'],
"description": function_parameters[0]['description']
}
},
"required": [function_parameters[0]['name']]
}
user_function= Function(name = function_name, description = function_description, parameters = func_parameters, )
user_tool = Tool(function = user_function)
user_tools.append(user_tool)
return user_tools
사용자가 "내 Pet 1의 상태가 뭐야?"라고 물어본다고 가정해 봐요. 단독으로는 LLM이 이 질문에 답할 수 없어요. 필요한 데이터에 접근하려면 비즈니스 로직 백엔드를 질의해야 하거든요. 하지만 이 질문에 답할 정확한 도구가 있다면 답변을 제공할 수 있어요!
def get_user_messages(queries: List[str]) -> List[UserMessage]:
user_messages=[]
for query in queries:
user_message = UserMessage(content=query)
user_messages.append(user_message)
return user_messages
외부 ollama 엔드포인트를 사용한다면 환경 변수 OLLAMA_ENDPOINT를 설정해요.
export OLLAMA_ENDPOINT="YOUR-Ollama-IP"
Step 2-4 실행 (execute_generator)
메시지와 토큰을 만들고, Ollama에 프롬프트를 보내 모델이 도구 호출을 결정하게 하고, 결과를 처리하는 함수를 만들어요.
def execute_generator():
queries = ["What's the status of my Pet 1?", "Find information of user user1?" , "What's the status of my Store Order 3?"]
return_objs = [['pet','petId'], ['user', 'username'], ['store/order','orderId']]
function_end_point
user_messages=get_user_messages(queries)
user_tools = generate_tools(return_objs, function_end_point)
#create tokens for message and tools prompt
tokenizer = MistralTokenizer.v3()
completion_request = ChatCompletionRequest(tools=user_tools, messages=user_messages,)
tokenized = tokenizer.encode_chat_completion(completion_request)
_, text = tokenized.tokens, tokenized.text
ollama_endpoint_env = os.environ.get('OLLAMA_ENDPOINT')
model = "mistral:7b"
prompt = text
if ollama_endpoint_env is None:
ollama_endpoint_env = 'http://localhost:11434'
ollama_endpoint = ollama_endpoint_env + "/api/generate" # replace with localhost
response = requests.post(ollama_endpoint,
json={
'model': model,
'prompt': prompt,
'stream':False,
'raw': True
}, stream=False
)
response.raise_for_status()
result = response.json()
process_results(result, user_messages)
Step 3. 함수 실행 (User: Execute function to obtain tool results)
함수를 어떻게 실행할까요? 현재는 이 함수들을 실행하는 것이 사용자의 책임이고, 함수 실행은 사용자 쪽에서 이뤄져요. 나중에는 서버 쪽에서 실행할 수 있는 유용한 함수를 도입할 수도 있어요.
모델 응답에서 함수 이름(function_name)과 함수 파라미터(function_params) 같은 유용한 함수 정보를 추출해요. 여기서 Mistral 모델이 파라미터 petId를 1로 설정한 getPetId 함수를 사용하기로 선택한 게 분명해요.
def process_results(result, messages):
result_format = result['response'].split("\n\n")
result_tool_calls = result_format[0].replace("[TOOL_CALLS] ","")
tool_calls = json.loads(result_tool_calls)
index = 0
try:
for tool_call in tool_calls:
function_name = tool_call["name"]
function_params = (tool_call["arguments"])
print(messages[index].content)
function_result = names_to_functions[function_name](**function_params)
print(function_result)
index = index + 1
except:
print(function_name + " is not defined")
execute_generator()