[PYTHON/REQUESTS] Session 클래스 : OLLAMA 서버 통신하기

■ Session 클래스를 사용해 OLLAMA 서버와 통신하는 방법을 보여준다.

▶ main.py


import json
import requests

from typing import Any

class OllamaClient:
    def __init__(self, baseURL : str = "http://localhost:11434"):
        self.baseURL = baseURL.rstrip('/')
        self.session = requests.Session()

    def generate(self, model : str, prompt : str,  stream : bool = False, **keywardArgumentDictionary) -> Any:
        url = f"{self.baseURL}/api/generate"

        data = {
            "model"  : model,
            "prompt" : prompt,
            "stream" : stream,
            **keywardArgumentDictionary
        }

        try:
            if stream:
                return self.session.post(url, json = data, stream = True)
            else:
                response = self.session.post(url, json=data)
                response.raise_for_status()
                return response.json()
        except requests.exceptions.RequestException as exception:
            print(f"API 요청 중 오류 발생 : {exception}")
            raise

def processStreamResponse(response):
    for line in response.iter_lines():
        if line:
            try:
                yield json.loads(line)
            except json.JSONDecodeError as exception:
                print(f"JSON 파싱 오류 : {exception}")
                continue

def main():
    ollamaClient = OllamaClient()

    model  = "bnksys/eeve-yanolja-v1:latest"
    prompt = "여기어때와 야놀자의 차이점을 알려주세요."

    try:
        print("=== 일반 응답 ===")
        response1 = ollamaClient.generate(model = model, prompt = prompt, stream = False)
        print(f"질문 : {prompt}")
        print(f"응답 : {response1['response']}\n")

        print("=== 스트리밍 응답 ===")
        print(f"질문 : {prompt}")
        print( "응답 :", end = " ", flush = True)

        response2 = ollamaClient.generate(model = model, prompt = prompt, stream = True)
        for chunkDictionary in processStreamResponse(response2):
            if chunkDictionary.get("response"):
                print(chunkDictionary["response"], end = "", flush = True)
        print()
    except Exception as exception:
        print(f"오류 발생 : {exception}")

if __name__ == "__main__":
    main()

import json

import requests

from typing import Any

class OllamaClient:

def __init__(self, baseURL : str = "http://localhost:11434"):

self.baseURL = baseURL.rstrip('/')

self.session = requests.Session()

def generate(self, model : str, prompt : str, stream : bool = False, **keywardArgumentDictionary) -> Any:

url = f"{self.baseURL}/api/generate"

data = {

"model" : model,

"prompt" : prompt,

"stream" : stream,

**keywardArgumentDictionary

}

try:

if stream:

return self.session.post(url, json = data, stream = True)

else:

response = self.session.post(url, json=data)

response.raise_for_status()

return response.json()

except requests.exceptions.RequestException as exception:

print(f"API 요청 중 오류 발생 : {exception}")

raise

def processStreamResponse(response):

for line in response.iter_lines():

if line:

try:

yield json.loads(line)

except json.JSONDecodeError as exception:

print(f"JSON 파싱 오류 : {exception}")

continue

def main():

ollamaClient = OllamaClient()

model = "bnksys/eeve-yanolja-v1:latest"

prompt = "여기어때와 야놀자의 차이점을 알려주세요."

try:

print("=== 일반 응답 ===")

response1 = ollamaClient.generate(model = model, prompt = prompt, stream = False)

print(f"질문 : {prompt}")

print(f"응답 : {response1['response']}\n")

print("=== 스트리밍 응답 ===")

print(f"질문 : {prompt}")

print( "응답 :", end = " ", flush = True)

response2 = ollamaClient.generate(model = model, prompt = prompt, stream = True)

for chunkDictionary in processStreamResponse(response2):

if chunkDictionary.get("response"):

print(chunkDictionary["response"], end = "", flush = True)

print()

except Exception as exception:

print(f"오류 발생 : {exception}")

if __name__ == "__main__":

main()

※ pip install requests 명령을 실행했다.