PythonでOllama APIを使う

中国初のAI IDE インテリジェント・プログラミング・ソフトウェア 🚀体験への招待 Trae 中国語版ダウンロードDeepSeek-R1とDoubao-proは無制限で利用できる！

の使い方を簡単に説明する。オーラマ簡単なチャット会話から、ストリーミング・レスポンスを使ったビッグデータ処理、ローカルでのモデル作成・コピー・削除まで、この記事でガイドします。さらに、カスタムクライアントや非同期プログラミングを使ってアプリケーションのパフォーマンスを最適化する方法も紹介します。Ollamaを初めて使う人も、経験豊富な開発者も、この記事を読めばPythonでOllama APIをより効率的に使えるようになるでしょう。

このチュートリアルではジュピター・ノートブック良くなるための例

環境準備

Pythonを使ってOllama APIとやり取りを始める前に、開発環境が以下の条件を満たしていることを確認してください：

Python: Python 3.8以降をインストールしてください。
pip: Pythonのパッケージ管理ツールであるpipがインストールされていることを確認する。
ollama ライブラリ: Ollama 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は、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

Ollama Python ライブラリは、Ollama とのインタラクションを簡素化する豊富なインターフェイスを提供します。これらのインターフェースは直感的で統合しやすいように設計されており、開発者がより簡単にモデルを呼び出して管理できるようにすることを目的としています。基本的な実装の詳細とAPIエンドポイントの完全な情報については、Ollama Pythonライブラリを参照してください。 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接続するOllamaホスト
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 Python 非同期ジェネレータを返すように関数を変更します：

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 ユーザは、テストの以下のパラメータを変更することで、小テストプロセスの1セッションに要する時間と総所要時間を比較することができます：

test_messages: テストデータ
test_num: テスト数
モデル名：テストモデル

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)

ドキュメントを参照してください：オラマ・パイソン