Python에서 올라마 API 사용

이 문서에서는 다음과 같이 간단하게 Ollama 간단한 채팅 대화, 스트리밍 응답을 이용한 빅 데이터 작업, 모델 생성, 복사, 삭제 등을 로컬에서 수행하고자 하는 경우 이 문서가 도움이 될 수 있습니다. 또한, 사용자 정의 클라이언트와 비동기 프로그래밍을 사용하여 애플리케이션의 성능을 최적화하는 방법을 보여드리며, Ollama를 처음 사용하든 숙련된 개발자이든 이 글은 Python에서 Ollama API를 보다 효율적으로 사용하는 데 도움이 될 것입니다.

이 튜토리얼은 또한 주피터 노트북 개선할 수 있는 예시.

 

환경 준비

Python을 사용하여 올라마 API와 상호 작용하기 전에 개발 환경이 다음 조건을 충족하는지 확인하세요:

• Python: Python 3.8 이상을 설치합니다.
• pip: 파이썬 패키지 관리 도구인 pip가 설치되어 있는지 확인합니다.
• 올라마 라이브러리: 올라마 API와 더 쉽게 상호 작용할 수 있도록 하는 데 사용됩니다. 설치 명령은 다음과 같습니다:

pip install ollama

 

사용법

from ollama import chat
from ollama import ChatResponse
response: ChatResponse = chat(model='llama3.1', messages=[
{
'role': 'user',
'content': '为什么天空是蓝色的？',
},
])
print(response['message']['content'])
print(response.message.content)

 

스트리밍 응답

이 작업은 stream=True 응답 스트리밍을 활성화하여 함수 호출이 각 부분이 스트림의 객체인 Python 제너레이터를 반환하도록 합니다.

from ollama import chat
stream = chat(
model='llama3.1',
messages=[{'role': 'user', 'content': '为什么天空是蓝色的？'}],
stream=True,
)
for chunk in stream:
print(chunk['message']['content'], end='', flush=True)

 

구조화된 출력

• 일반 출력(비정형 출력)
  • 자연어 텍스트를 직접 생성합니다.
  • 사람이 읽는 데는 적합하지만 프로그래밍 구문 분석이나 자동 처리에는 적합하지 않습니다.
  • 예시:

    这是一只黑色的小猫，它正在草地上玩耍。
    

• 구조화된 출력
  • 기계가 쉽게 구문 분석하고 사용할 수 있도록 JSON, YAML, XML 또는 기타 형식으로 데이터를 반환합니다.
  • API, 자동화된 워크플로 및 데이터 저장에 이상적입니다.
  • 예시:

    {
    "description": "这是一只黑色的小猫",
    "activity": "正在草地上玩耍"
    }
    

구조화된 출력의 장점

(1) 취급의 용이성

• 이 기계는 다음과 같은 특정 필드를 쉽게 추출할 수 있습니다. description 어쩌면 activity NLP는 자연어 처리 없이 일반 텍스트를 구문 분석하는 새로운 방식입니다.

(2) 제어 가능성 향상

• 구조화된 형식을 사용하면 개발자가 모델 출력을 정확하게 제어하고 길거나 예측할 수 없는 답변을 피할 수 있습니다.
• 예를 들어 AI가 코드를 생성할 때를 들 수 있습니다:

  {
  "language": "Python",
  "code": "print('Hello, World!')"
  }
  

(3) 간편한 저장 및 분석

• 구조화된 데이터는 쿼리 및 분석이 용이하도록 데이터베이스에 저장하는 것이 더 적합합니다.
• 예시:

  {
  "date": "2025-01-20",
  "summary": "今天的销售额增长了10%。"
  }
  

from pydantic import BaseModel, Field
from ollama import chat
import json
class CountryInfo(BaseModel):
capital: str = Field(..., alias="首都")
number: str = Field(..., alias="人口")
area: str = Field(..., alias="占地面积")  
response = chat(
model='llama3.1',
messages=[{
'role': 'user',
'content': "请介绍美国的首都、人口、占地面积信息，并以 JSON 格式返回。"
}],
format="json",  
options={'temperature': 0}, 
)
response_content = response["message"]["content"]
if not response_content:
raise ValueError("Ollama 返回的 JSON 为空")
json_response = json.loads(response_content)  
print(json_response)
friends_response = CountryInfo.model_validate(json_response)  
print(friends_response)

 

API

올라마 파이썬 라이브러리는 올라마와의 상호작용을 간소화하는 다양한 인터페이스 세트를 제공합니다. 이러한 인터페이스는 직관적이고 쉽게 통합할 수 있도록 설계되었으며, 개발자가 모델을 보다 쉽게 호출하고 관리할 수 있도록 돕기 위한 것입니다. 기본 구현과 전체 API 엔드포인트 정보에 대한 자세한 내용을 보려면 Ollama API 사용 가이드.

채팅

ollama.chat(model='llama3.1', messages=[{'role': 'user', 'content': '为什么天空是蓝色的？'}])

생성

ollama.generate(model='llama3.1', prompt='为什么天空是蓝色的？')

로컬 모델 목록

ollama.list()

모델 정보 표시

ollama.show('llama3.1')

모델 만들기

modelfile='''
FROM llama3.1
SYSTEM 你是超级马里奥兄弟中的马里奥。
'''
ollama.create(model='example', modelfile=modelfile)

복제 모델

ollama.copy('llama3.1', 'user/llama3.1')

모델 삭제

ollama.delete('llama3.1')

풀 모델

ollama.pull('llama3.1')

푸시 모델

ollama.push('user/llama3.1')

임베딩 생성

ollama.embeddings(model='llama3.1', prompt='天空是蓝色的因为瑞利散射')

# 批量生成embedding
ollama.embed(model='llama3.1', input=['天空是蓝色的', '草是绿色的'])

코스

ollama.ps()

 

맞춤형 클라이언트

이 작업을 수행하려면 ollama 인스턴스화 Client 어쩌면 AsyncClient 를 클릭하여 사용자 지정 클라이언트를 만듭니다.

다음 필드를 사용하여 사용자 지정 클라이언트를 만들 수 있습니다:

• host연결할 올라마 호스트
• timeout:: 요청 시간 초과

모든 키워드 매개변수는 다음을 참조하세요.httpx.Client.

클라이언트 동기화

동기화 클라이언트 사용 (Client) 를 호출하면 client.chat() 메서드를 사용하면 프로그램은 요청이 완료될 때까지 기다렸다가 결과를 반환한 후 후속 코드를 계속 실행합니다. 이 접근 방식은 보다 직관적이고 간단하며, 보다 선형적인 흐름을 가지며 많은 수의 동시 작업을 처리할 필요가 없는 애플리케이션을 작성하는 데 적합합니다.

from ollama import Client
client = Client(
host='http://localhost:11434',
headers={'x-some-header': 'some-value'}
)
response = client.chat(model='llama3.1', messages=[
{
'role': 'user',
'content': '为什么天空是蓝色的？',
},
])
print(response)

비동기 클라이언트

이 코드는 비동기 클라이언트를 사용합니다. (AsyncClient) 비동기 함수를 정의합니다. chat() . await 키워드를 사용하면 함수 실행을 일시 중지할 수 있습니다. AsyncClient().chat() 요청이 완료되더라도 그 동안 다른 작업을 차단하지 않습니다. 이는 효율적으로 처리하는 데 유용합니다. I/O 작업(예: 네트워크 요청)을 수행하거나 여러 작업을 동시에 수행하려는 애플리케이션에 적합합니다. 또한 asyncio.run(chat()) 를 사용하여 이 비동기 함수를 실행합니다.

import asyncio
from ollama import AsyncClient
import nest_asyncio
nest_asyncio.apply()
async def chat():
message = {'role': 'user', 'content': '为什么天空是蓝色的？'}
response = await AsyncClient().chat(model='llama3.1', messages=[message])
print(response)
asyncio.run(chat())

설정 stream=True 파이썬 비동기 제너레이터를 반환하도록 함수를 수정합니다:

import asyncio
from ollama import AsyncClient
import nest_asyncio
nest_asyncio.apply()
async def chat():
message = {'role': 'user', 'content': '为什么天空是蓝色的？'}
async for part in await AsyncClient().chat(model='llama3.1', messages=[message], stream=True):
print(part['message']['content'], end='', flush=True)
asyncio.run(chat())

다양한 통화 수에 따른 동기식 및 비동기식 클라이언트 시간 소비 비교 테스트

다음 코드는 각각 동기식 및 비동기식 클라이언트 반복을 호출합니다. test_num 사용자는 테스트에 대한 다음 매개변수를 변경하여 총 소요 시간과 퀴즈 프로세스의 단일 세션에 소요되는 시간을 비교할 수 있습니다:

• test_messages: 테스트 데이터
• test_num: 테스트 횟수
• model_name: 테스트 모델

import time
import asyncio
from ollama import Client, AsyncClient
import nest_asyncio
# 应用nest_asyncio以支持Jupyter中的异步操作
nest_asyncio.apply()
# 初始化客户端
client = Client(host='http://localhost:11434')
async_client = AsyncClient(host='http://localhost:11434')
# 同步请求处理函数
def request_example(client, model_name, messages):
start_time = time.time()
try:
# 同步请求返回
response = client.chat(model=model_name, messages=messages)
except Exception as e:
print(f"同步请求失败: {e}")
response = None
end_time = time.time()
duration = end_time - start_time
print(f"同步请求时间: {duration}")
return response, duration
# 异步请求处理函数
async def async_request_example(client, model_name, messages):
start_time = time.time()
try:
# 异步请求返回
response = await client.chat(model=model_name, messages=messages)
except Exception as e:
print(f"异步请求失败: {e}")
response = None
end_time = time.time()
duration = end_time - start_time
print(f"异步请求时间: {duration}")
return response, duration
# 异步请求测试函数
async def async_client_test(test_num, model_name, messages):
tasks = [asyncio.create_task(async_request_example(async_client, model_name, messages)) 
for _ in range(test_num)]
results= await asyncio.gather(*tasks)
return results
# 运行同步测试
def sync_test(model_name, messages, test_num):
total_time = 0
for i in range(test_num):
_, duration = request_example(client, model_name, messages)
total_time += duration
return total_time / test_num
# 运行异步测试
async def async_test(model_name, messages, test_num):
start_time = time.time()
await async_client_test(test_num, model_name, messages)
end_time = time.time()
return (end_time - start_time) / test_num
# 准备测试数据
test_messages = [{'role': 'user', 'content': '为什么天空是蓝色的？'}]
test_num = 10
model_name = 'llama3.1'
# 运行同步测试并输出结果
print("运行同步测试")
sync_avg_time = sync_test(model_name, test_messages, test_num)
print(f"同步测试平均时间: {sync_avg_time:.2f} 秒")
# 运行异步测试并输出结果
print("运行异步测试")
async_avg_time = asyncio.run(async_test(model_name, test_messages, test_num))
print(f"异步测试平均时间: {async_avg_time:.2f} 秒")

 

잘못된

요청이 오류 상태를 반환하거나 스트리밍 중에 오류가 감지되면 오류가 발생합니다.

import ollama
model = 'does-not-yet-exist'
try:
ollama.chat(model)
except ollama.ResponseError as e:
print('错误:', e.error)
if e.status_code == 404:
ollama.pull(model)

 

설명서를 참조하세요:올라마 파이썬