Ollamaの公式PythonライブラリでLLMを呼び出す方法｜pip install ollamaでストリーミング・非同期・ツール使用を実装する手順

この記事のポイント

・pip install ollama でフレームワーク不要のPythonクライアントを導入できる
・ollama.chat(stream=True) でストリーミング応答をジェネレータとして逐次処理できる
・AsyncClient と asyncio.gather() を組み合わせて複数リクエストを並列処理できる
・tools パラメータで Function Calling を OpenAI 互換フォーマットで実装できる

$ ollama list
NAME                               ID              SIZE    MODIFIED
llama3.3:70b-instruct-q4_0        abc1234def56    43 GB   2 hours ago

$ curl -s http://localhost:11434/api/tags | python3 -m json.tool | grep '"name"'
            "name": "llama3.3:70b-instruct-q4_0",

$ python3 -m venv ~/ollama-py-env
$ source ~/ollama-py-env/bin/activate
(ollama-py-env) $ pip install ollama
Collecting ollama
  Downloading ollama-0.3.0-py3-none-any.whl (11 kB)
Installing collected packages: ollama
Successfully installed ollama-0.3.0

(ollama-py-env) $ python3 -c "import ollama; print(ollama.__version__)"
0.3.0

import ollama

# basic synchronous chat call
response = ollama.chat(
    model='llama3.3:70b-instruct-q4_0',
    messages=[
        {'role': 'user', 'content': 'Linuxのinodeとは何ですか？50字以内で答えてください'}
    ]
)
print(response['message']['content'])

import ollama

response = ollama.chat(
    model='llama3.3:70b-instruct-q4_0',
    messages=[{'role': 'user', 'content': 'Hello'}]
)
print(f"回答: {response['message']['content']}")
print(f"入力トークン: {response['prompt_eval_count']}")
print(f"出力トークン: {response['eval_count']}")
print(f"処理時間: {response['total_duration'] / 1_000_000:.0f}ms")

response = ollama.chat(
    model='llama3.3:70b-instruct-q4_0',
    messages=[
        {
            'role': 'system',
            'content': 'あなたはLinuxサーバー管理の専門家です。技術的な質問に簡潔に答えてください。'
        },
        {
            'role': 'user',
            'content': 'systemctlとserviceコマンドの違いを教えてください'
        }
    ]
)
print(response['message']['content'])

import ollama

stream = ollama.chat(
    model='llama3.3:70b-instruct-q4_0',
    messages=[
        {'role': 'user', 'content': 'Linuxのファイルパーミッションについて詳しく説明してください'}
    ],
    stream=True
)

for chunk in stream:
    print(chunk['message']['content'], end='', flush=True)

print()  # 最後に改行

import ollama
import time

start = time.time()
stream = ollama.chat(
    model='llama3.3:70b-instruct-q4_0',
    messages=[{'role': 'user', 'content': 'bashのfor文の書き方をコード例付きで説明してください'}],
    stream=True
)

for chunk in stream:
    print(chunk['message']['content'], end='', flush=True)
    if chunk['done']:
        elapsed = time.time() - start
        print(f"\n\n--- 処理時間: {elapsed:.1f}秒 / 出力トークン: {chunk.get('eval_count', 'N/A')} ---")

import asyncio
from ollama import AsyncClient

async def ask(client: AsyncClient, question: str) -> str:
    response = await client.chat(
        model='llama3.3:70b-instruct-q4_0',
        messages=[{'role': 'user', 'content': question}]
    )
    return response['message']['content']

async def main():
    client = AsyncClient()
    questions = [
        'cronの書き方を教えてください',
        'rsyncの主なオプションを教えてください',
        'iptablesでポートを開ける方法を教えてください',
    ]
    results = await asyncio.gather(*[ask(client, q) for q in questions])
    for q, a in zip(questions, results):
        print(f"Q: {q}\nA: {a[:100]}...\n")

asyncio.run(main())

import asyncio
from ollama import AsyncClient

async def stream_response(prompt: str):
    client = AsyncClient()
    async_stream = await client.chat(
        model='llama3.3:70b-instruct-q4_0',
        messages=[{'role': 'user', 'content': prompt}],
        stream=True
    )
    async for chunk in async_stream:
        print(chunk['message']['content'], end='', flush=True)
    print()

asyncio.run(stream_response('DockerとPodmanの違いを教えてください'))

import ollama
import subprocess

tools = [
    {
        'type': 'function',
        'function': {
            'name': 'run_readonly_command',
            'description': '読み取り専用のLinuxコマンドを実行して結果を返す',
            'parameters': {
                'type': 'object',
                'properties': {
                    'command': {
                        'type': 'string',
                        'description': '実行するコマンド（例: df -h, free -m, uptime）'
                    }
                },
                'required': ['command']
            }
        }
    }
]

SAFE_COMMANDS = ('df ', 'free ', 'uptime', 'uname ', 'date', 'hostname')

def run_readonly_command(command: str) -> str:
    if not any(command.startswith(p) for p in SAFE_COMMANDS):
        return "エラー: 許可されていないコマンドです"
    result = subprocess.run(
        command.split(), capture_output=True, text=True, timeout=5
    )
    return result.stdout or result.stderr

messages = [
    {'role': 'user', 'content': 'このサーバーのディスク使用状況とメモリ使用状況を教えてください'}
]

response = ollama.chat(
    model='llama3.3:70b-instruct-q4_0',
    messages=messages,
    tools=tools
)

while response['message'].get('tool_calls'):
    messages.append(response['message'])
    for tool_call in response['message']['tool_calls']:
        func_name = tool_call['function']['name']
        func_args = tool_call['function']['arguments']
        if func_name == 'run_readonly_command':
            result = run_readonly_command(func_args['command'])
        messages.append({'role': 'tool', 'content': result})
    response = ollama.chat(
        model='llama3.3:70b-instruct-q4_0',
        messages=messages,
        tools=tools
    )

print(response['message']['content'])

import ollama

conversation = [
    {'role': 'system', 'content': 'あなたはLinuxの技術サポートAIです。'}
]

def chat(user_input: str) -> str:
    conversation.append({'role': 'user', 'content': user_input})
    response = ollama.chat(
        model='llama3.3:70b-instruct-q4_0',
        messages=conversation
    )
    answer = response['message']['content']
    conversation.append({'role': 'assistant', 'content': answer})
    return answer

print(chat('sshdの設定ファイルはどこにありますか？'))
print(chat('そのファイルでポート番号を変えるにはどうすればいいですか？'))
print(chat('変更後にどんなコマンドで反映させますか？'))

def trim_history(messages: list, max_pairs: int = 10) -> list:
    """systemプロンプトを保持しつつ最新N往復のみ残す"""
    system = [m for m in messages if m['role'] == 'system']
    non_system = [m for m in messages if m['role'] != 'system']
    return system + non_system[-(max_pairs * 2):]

# 50往復を超えたらトリミング
if len(conversation) > 105:
    conversation[:] = trim_history(conversation, max_pairs=10)
    print("会話履歴を最新10往復に圧縮しました")

from ollama import Client

# タイムアウトとコンテキストウィンドウを明示指定する例
client = Client(host='http://localhost:11434', timeout=300)
response = client.chat(
    model='llama3.3:70b-instruct-q4_0',
    messages=[{'role': 'user', 'content': '長い文書の要約をお願いします...'}],
    options={'num_ctx': 8192, 'temperature': 0.1}
)
print(response['message']['content'])

ローカルLLMのPython連携を2日間体験する

公式ライブラリの使い方を手を動かして習得したい方、ストリーミングや非同期処理をチームのプロジェクトに組み込みたい方向けに、「ローカルAIマスターセミナー」を開催しています。
少人数（最大8名）ZOOMハンズオン形式で実施しています。