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='ollama3.1', messages=[
{

'content': 'どうして空は青いの？,
},
])
print(response['message']['content'])
print(response.message.content)

ストリーミング対応

これは stream=True レスポンスのストリーミングを有効にして、関数呼び出しがストリームの各パートをオブジェクトとするPythonジェネレーターを返すようにします。

from ollama import chat
stream = chat(
model='ollama3.1', [{'role': 'user', 'content': 'どうして空は青いの？
messages=[{'role': 'user', 'content': 'どうして空は青いの？'}] , [{'role': 'user', 'content': 'どうして空は青いの？,
stream=True, )
)
for chunk in stream.
print(chunk['message']['content'], end='', flush=True)

構造化出力

通常の出力（構造化されていない出力）
- 自然言語テキストの直接生成。
- 人間が読むには適しているが、プログラムによる解析や自動処理には適していない。
- 例
```
草むらで遊んでいた黒い子猫だった。
```
構造化出力
- JSON、YAML、XML、またはその他のフォーマットでデータを返し、機械が解析して使用しやすくします。
- API、自動ワークフロー、データストレージに最適。
- 例
```
{
"description": "黒いタビーの猫です、
"activity": "草むらで遊んでいます。"
}
```

構造化アウトプットの利点

(1) 取り扱いの容易さ

このマシンは、以下のような特定のフィールドを簡単に抽出することができる。 記述 もしかしたら 活動 NLPは、NLPを必要とせずに通常のテキストを解析する新しい方法である。

(2) 制御性の向上

構造化されたフォーマットにより、開発者はモデルの出力を正確に制御し、長かったり予測不可能な回答を避けることができる。

例えば、AIがコードを生成するとき：

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

(3) 保存と分析が容易

構造化されたデータは、クエリーや分析を容易にするためにデータベースに保存するのに適している。

例

{
"date": "2025-01-20".
"要約": "本日の売上は10%増加"
}

from pydantic import BaseModel, Field
from ollama import チャット
インポート json
class CountryInfo(BaseModel).
capital: str = Field(...エイリアス="capital")
number: str = フィールド(... , エイリアス="人口")別名="人口")
area: str = Field(... , alias="面積")別名="面積")
response = chat(
model='llama3.1'、
messages=[{
'role': 'user', 'content': '紹介する', 'content'.
'content': "アメリカの首都、人口、フットプリントに関する情報を提示し、JSON形式で返してください。"
}]
format="json"、
options={'temperature': 0}, )
)
response_content = response["message"]["content"].
if not response_content: raise ValueError("Ollama Return JSR")
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='ollama3.1', messages=[{'role': 'user', 'content': 'どうして空は青いの？'}])

生成

ollama.generate(model='ollama3.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='空が青いのはレイリー散乱のせいです')

# エンベッディングのバッチ生成
ollama.embed(model='llama3.1', input=['sky is blue', 'grass is green'])

コース

ollama.ps()

カスタム顧客

これは オラマ インスタンス化 クライアント もしかしたら 非同期クライアント を使用してカスタムクライアントを作成します。

カスタムクライアントは、以下のフィールドを使用して作成できます：

ホスト接続するOllamaホスト
タイムアウト:: リクエストタイムアウト

すべてのキーワード引数についてはhttpx.クライアント.

同期クライアント

同期クライアントが使用される (クライアント） を呼び出すと クライアントチャット メソッドを呼び出すと、プログラムはリクエストの完了を待って結果を返し、その後に続くコードを実行する。このアプローチはより直感的でシンプルであり、より直線的なフローを持ち、多数の同時タスクを処理する必要のないアプリケーションを書くのに適している。

from ollama import Client
クライアント = クライアント(
host='http://localhost:11434'、
headers={'x-some-header': 'some-value'}.
)
response = client.chat(model='ollama3.1', messages=[
{
'role': 'user', 'content': 'why-sky'.
'content': 'なぜ空は青いの？,
},
])
print(response)

非同期クライアント

このコードでは、非同期クライアント (非同期クライアント) を定義し、非同期関数 チャット .awaitキーワードを使うと、関数の実行を AsyncClient().chat() リクエストは完了するが、その間に他の処理をブロックすることはない。これは 入出力 操作（例えば、ネットワークリクエスト）や、複数のタスクを同時に実行したいアプリケーションのために使用される。さらに asyncio.run(chat()) この非同期関数を実行する。

インポート asyncio
from ollama import AsyncClient
インポート 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 非同期ジェネレータを返すように関数を変更します：

インポート asyncio
from ollama import AsyncClient
インポート nest_asyncio
nest_asyncio.apply()
async def chat().
message = {'role': 'user', 'content': 'どうして空は青いの？'}.
async for part in await AsyncClient().chat(model='llama3.1', message=[message], stream=True)：
print(part['message']['content'], end='', flush=True)
asyncio.run(chat())

同期クライアントと非同期クライアントの異なるコール数による時間消費比較テスト

以下のコードは、同期クライアント・リピートと非同期クライアント・リピートをそれぞれ呼び出す。 テスト番号 ユーザは、テストの以下のパラメータを変更することで、小テストプロセスの1セッションに要する時間と総所要時間を比較することができます：

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

インポート時間
インポート asyncio
from ollama import Client, AsyncClient
インポート nest_asyncio
# Jupyterで非同期操作をサポートするためにnest_asyncioを適用する
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()
開始時刻 = time.time()
#同期リクエスト return
response = client.chat(model=model_name, messages=messages)
except Exception as e.
print(f "同期リクエストに失敗しました: {e}")
response = なし
end_time = time.time()
duration = end_time - start_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()
start_time = time.time(): start_time = time.time()
#の非同期リクエストを返す
response = await client.chat(model=model_name, messages=messages)
except Exception as e.
print(f "非同期リクエストに失敗しました: {e}")
response = なし
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)
結果を返す
# syncテストを実行する
def sync_test(model_name, messages, test_num): total_time = 0
total_time = 0
for i in range(test_num): _, duration = request_example = 0
_, 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.
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
モデル名 = '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} 秒")

不正確

リクエストがエラーステータスを返すか、ストリーミング中にエラーが検出されると、エラーが発生する。

インポート ollama
model = '存在しない'
試す
ollama.chat(model)
except ollama.ResponseError as e:: print('Error:', e.error)
print('Error:', e.error)
print('Error:', e.error) if e.status_code == 404: ollama.pull(model): ollama.
ollama.pull(model)

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