Ollama API

import os
import requests

# Ollama REST API URL 설정
# os.environ["OLLAMA_HOST"] = "http://host.docker.internal:11434"  # Docker 컨테이너에서 호스트 접근
OLLAMA_BASE_URL = os.environ.get("OLLAMA_HOST", "http://localhost:11434")
print(f"OLLAMA_BASE_URL: {OLLAMA_BASE_URL}")

response = requests.get(OLLAMA_BASE_URL)
if response.status_code == 200:
    print(response.text)
else:
    print(f"Error: {response.status_code}")

import requests
import pandas as pd

url = f"{OLLAMA_BASE_URL}/api/tags"

response = requests.get(url)
if not response.status_code == 200:
    print('Error:', response.status_code)
else:
    data = response.json()
    # print(data)
    models = pd.DataFrame(data['models'])
    print('항목:', ', '.join(models.columns.tolist()))
    display(
        models
            .sort_values(by='size')
            .assign(**{"size (GB)": models['size'].apply(lambda b: f"{b / (1024**3):.2f} GB")})
            [["model", "size (GB)", 'modified_at']]
            .set_index('model')
    )

import requests

# 앞서 설정한 OLLAMA_BASE_URL을 사용
url = f"{OLLAMA_BASE_URL}/api/generate"
request_parameters = {
    "model": "gpt-oss",  # 모델 이름
    "prompt": "자기 소개",  # 프롬프트
    "stream": False  # 한 번에 전체 응답 반환
}

response = requests.post(url, json=request_parameters)
if response.status_code == 200:
    data = response.json()
    if 'thinking' in data:
        print(f'<think>{data["thinking"]}</think>')
    print(data['response'])
else: 
    print('Error:', response.status_code)
    print(response.json())

import requests

# 앞서 설정한 OLLAMA_BASE_URL을 사용
url = f"{OLLAMA_BASE_URL}/api/chat"
request_parameters = {
    "model": "qwen3:latest",
    "stream": False
}

request_parameters['messages'] = [
    {'role': 'system', 'content': '너는 허탕 아재 개그를 하는 모델이다.'},
    {'role': 'user', 'content': 'RAG가 뭐야?'}
]

response = requests.post(url, json=request_parameters)
print(response.json())

import requests

# 앞서 설정한 OLLAMA_BASE_URL을 사용
url = f"{OLLAMA_BASE_URL}/api/tags"

response = requests.get(url)
if response.ok:
    data = response.json()
    models = data.get('models', [])
    if not models:
        print("설치된 모델이 없습니다.")
    else:
        print(f"설치된 모델: {len(models)}개\n")
        for m in models:
            name = m.get('name') or 'Unknown'
            size_bytes = m.get('size') or 0
            size_gb = size_bytes / (1024**3)
            modified = m.get('modified_at') or ''
            digest = m.get('digest') or m.get('id') or ''
            print(f"- {name}")
            print(f"  크기: {size_gb:.2f} GB")
            if modified:
                print(f"  수정 시간: {modified}")
            if digest:
                print(f"  ID: {digest}")
            print()
else:
    print(f"요청 실패: {response.status_code} {response.text}")

from ollama import Client

# 앞서 설정한 OLLAMA_BASE_URL을 사용
client = Client(host=OLLAMA_BASE_URL)

# 서버 연결 확인
try:
    response = client.list()
    print("✓ Ollama 서버에 성공적으로 연결되었습니다.")
    print(f"\n설치된 모델 수: {len(response.get('models', []))}")
except Exception as e:
    print(f"✗ 연결 실패: {e}")
    print("Ollama 서버가 실행 중인지 확인하세요. (기본 포트: 11434)")

# 설치된 모델 목록 조회 (Python 라이브러리)
response = client.list()
models = response.get('models', [])

if models:
    print(f"설치된 모델: {len(models)}개\n")
    for m in models:
        name = m.get('name', 'Unknown')
        size_gb = (m.get('size') or 0) / (1024**3)
        modified = m.get('modified_at') or ''
        digest = m.get('digest') or m.get('id') or ''
        print(f"- {name}")
        print(f"  크기: {size_gb:.2f} GB")
        if modified:
            print(f"  수정 시간: {modified}")
        if digest:
            print(f"  ID: {digest}")
        print()
else:
    print("설치된 모델이 없습니다.")

# 예시: 사용 가능한 모델이 있는 경우
# 실제 모델명으로 변경하여 사용하세요

model_name = "llama2"  # 사용 중인 모델명으로 변경
user_message = "안녕하세요. 자기 소개 좀 해주시겠어요?"

try:
    response = client.generate(
        model=model_name,
        prompt=user_message,
        stream=False
    )
    
    print(f"사용자: {user_message}")
    print(f"\n{model_name}의 응답:")
    print(response.get('response', ''))
    
except Exception as e:
    print(f"오류: {e}")
    print(f"'{model_name}' 모델이 설치되어 있는지 확인하세요.")

# 스트리밍 응답 예시
model_name = "llama2"
user_message = "Python에서 리스트 컴프리헨션이란 무엇인가요?"

try:
    print(f"사용자: {user_message}")
    print(f"\n{model_name}의 응답:")
    print("-" * 50)
    
    # stream=True로 설정하여 스트리밍 수신
    response = client.generate(
        model=model_name,
        prompt=user_message,
        stream=True
    )
    
    for chunk in response:
        print(chunk.get('response', ''), end='', flush=True)
    
    print("\n" + "-" * 50)
    
except Exception as e:
    print(f"오류: {e}")

# 모델 정보 조회
model_name = "llama2"

try:
    response = client.show(model=model_name)
    
    print(f"모델: {model_name}")
    print()
    
    if 'modelfile' in response:
        print("Modelfile:")
        print(response['modelfile'])
        print()
    
    if 'details' in response:
        details = response['details']
        print("모델 상세 정보:")
        for key, value in details.items():
            print(f"  {key}: {value}")
            
except Exception as e:
    print(f"오류: {e}")

# 다중 턴 대화 예시
model_name = "llama2"

messages = [
    {
        "role": "system",
        "content": "당신은 한국 역사 전문가입니다."
    },
    {
        "role": "user",
        "content": "조선 시대는 몇 년 동안 지속되었나요?"
    }
]

try:
    response = client.chat(
        model=model_name,
        messages=messages,
        stream=False
    )
    
    print(f"시스템: {messages[0]['content']}")
    print(f"사용자: {messages[1]['content']}")
    print(f"\n{model_name}의 응답:")
    print(response['message']['content'])
    
except Exception as e:
    print(f"오류: {e}")

import numpy as np

# 임베딩 모델 (임베딩 전용 모델이 필요합니다)
embedding_model = "nomic-embed-text"  # 또는 다른 임베딩 모델

text = "딥러닝은 인공지능의 한 분야입니다."

try:
    response = client.embeddings(
        model=embedding_model,
        prompt=text
    )
    
    embedding = response.get('embedding', [])
    
    print(f"텍스트: {text}")
    print(f"\n임베딩 벡터 차원: {len(embedding)}")
    print(f"임베딩 벡터 (처음 10개): {embedding[:10]}")
    print(f"\n벡터 노름 (L2): {np.linalg.norm(embedding):.4f}")
    
except Exception as e:
    print(f"오류: {e}")
    print(f"'{embedding_model}' 모델이 설치되어 있는지 확인하세요.")