OllamaのAPIでマルチターン会話を実装する方法｜Pythonで会話履歴を管理してフレームワークなしでチャットAPIを本番品質に仕上げる手順

この記事のポイント

・Ollamaの/api/chatはmessages配列で会話全履歴を毎回送信する設計
・Pythonでマルチターン会話を実装するにはrequestsライブラリだけで十分
・deque(maxlen=N)でターン数を制限し、コンテキスト上限を超えないよう管理する
・タイムアウト・リトライ・ストリーミングの実装で本番品質に仕上げる

# /api/chat へのリクエスト例（会話履歴を2ターン含む）
$ curl -s http://localhost:11434/api/chat \
  -H "Content-Type: application/json" \
  -d '{
    "model": "llama3.3:8b-instruct-q4_K_M",
    "messages": [
      {"role": "system", "content": "あなたはLinuxの専門家です。"},
      {"role": "user", "content": "sudoとは何ですか？"},
      {"role": "assistant", "content": "sudoは一般ユーザーが特定のコマンドをroot権限で実行する仕組みです。"},
      {"role": "user", "content": "sudoersファイルはどこにありますか？"}
    ],
    "stream": false
  }' | python3 -c "import sys,json; print(json.load(sys.stdin)['message']['content'])"
/etc/sudoers です。visudo コマンドで安全に編集します。

# python3 と requests のバージョンを確認する
$ python3 --version
Python 3.11.9
$ pip3 show requests 2>/dev/null | grep Version || pip3 install requests
Version: 2.31.0

#!/usr/bin/env python3
import requests

OLLAMA_URL = "http://localhost:11434/api/chat"
MODEL = "llama3.3:8b-instruct-q4_K_M"
SYSTEM_PROMPT = "あなたはLinuxサーバー管理の専門家です。正確で実用的な回答をしてください。"

def chat(history: list, user_message: str) -> str:
    history.append({"role": "user", "content": user_message})
    payload = {
        "model": MODEL,
        "messages": history,
        "stream": False,
    }
    response = requests.post(OLLAMA_URL, json=payload, timeout=120)
    response.raise_for_status()
    assistant_message = response.json()["message"]["content"]
    history.append({"role": "assistant", "content": assistant_message})
    return assistant_message

def main():
    history = [{"role": "system", "content": SYSTEM_PROMPT}]
    print("チャットを開始します（終了は Ctrl+C）")
    while True:
        user_input = input("You: ").strip()
        if not user_input:
            continue
        reply = chat(history, user_input)
        print(f"AI: {reply}\n")

if __name__ == "__main__":
    main()

$ python3 chat.py
チャットを開始します（終了は Ctrl+C）
You: cronとは何ですか？
AI: cronはLinuxで定期的なタスクを自動実行するスケジューラです...

You: どこに設定ファイルがありますか？
AI: cronの設定ファイルは以下の場所にあります。/etc/crontab...

from collections import deque
import requests

class OllamaChat:
    def __init__(self, model: str, system_prompt: str, max_turns: int = 10):
        self.model = model
        self.system_prompt = system_prompt
        # (user発言, assistant発言) の組を max_turns 件保持する
        self._turns = deque(maxlen=max_turns)

    def _build_messages(self) -> list:
        messages = [{"role": "system", "content": self.system_prompt}]
        for user_msg, assistant_msg in self._turns:
            messages.append({"role": "user", "content": user_msg})
            messages.append({"role": "assistant", "content": assistant_msg})
        return messages

    def ask(self, user_message: str) -> str:
        messages = self._build_messages()
        messages.append({"role": "user", "content": user_message})
        payload = {
            "model": self.model,
            "messages": messages,
            "stream": False,
            "options": {"num_ctx": 4096},
        }
        resp = requests.post(
            "http://localhost:11434/api/chat",
            json=payload,
            timeout=120,
        )
        resp.raise_for_status()
        assistant_reply = resp.json()["message"]["content"]
        self._turns.append((user_message, assistant_reply))
        return assistant_reply

$ python3 -c "
from ollama_chat import OllamaChat
c = OllamaChat(
    model='llama3.3:8b-instruct-q4_K_M',
    system_prompt='Linuxの専門家です。',
    max_turns=5,
)
print(c.ask('SELinuxとは何ですか？'))
print(c.ask('Enforcingモードに設定するコマンドは？'))
"
SELinuxは強制アクセス制御（MAC）を提供するLinuxのセキュリティ機構です...
setenforce 1 で Enforcing モードに切り替えます...

# 現在のモデル設定を確認する
$ curl -s http://localhost:11434/api/show \
  -H "Content-Type: application/json" \
  -d '{"name": "llama3.3:8b-instruct-q4_K_M"}' \
  | python3 -c "
import sys, json
d = json.load(sys.stdin)
params = d.get('parameters', '')
found = False
for line in params.splitlines():
    if 'num_ctx' in line:
        print('num_ctx設定:', line)
        found = True
        break
if not found:
    print('num_ctx: デフォルト（2048）')
"
num_ctx: デフォルト（2048）

$ python3 -c "
import requests, json
payload = {
    'model': 'llama3.3:8b-instruct-q4_K_M',
    'messages': [
        {'role': 'system', 'content': 'Linuxの専門家です。'},
        {'role': 'user', 'content': 'ファイルのパーミッションについて詳しく教えてください。'}
    ],
    'stream': False,
    'options': {'num_ctx': 4096}
}
resp = requests.post('http://localhost:11434/api/chat', json=payload, timeout=120)
d = resp.json()
print('プロンプトトークン数:', d.get('prompt_eval_count', 0))
print('応答トークン数:', d.get('eval_count', 0))
"
プロンプトトークン数: 42
応答トークン数: 318

import requests
import json

def ask_stream(messages: list, model: str) -> str:
    payload = {
        "model": model,
        "messages": messages,
        "stream": True,
        "options": {"num_ctx": 4096},
    }
    full_response = ""
    with requests.post(
        "http://localhost:11434/api/chat",
        json=payload,
        stream=True,
        timeout=180,
    ) as resp:
        resp.raise_for_status()
        for line in resp.iter_lines():
            if not line:
                continue
            chunk = json.loads(line)
            token = chunk.get("message", {}).get("content", "")
            print(token, end="", flush=True)
            full_response += token
            if chunk.get("done"):
                break
    print()
    return full_response

$ python3 -c "
import sys
sys.path.insert(0, '.')
from ask_stream import ask_stream
messages = [
    {'role': 'system', 'content': 'Linuxの専門家です。'},
    {'role': 'user', 'content': 'iptablesの基本的な使い方を教えてください。'}
]
result = ask_stream(messages, 'llama3.3:8b-instruct-q4_K_M')
print('--- 合計文字数:', len(result))
"
iptablesはLinuxのパケットフィルタリングツールです...（リアルタイム表示）
--- 合計文字数: 412

import time
import requests

def ask_with_retry(
    messages: list,
    model: str,
    max_retries: int = 3,
    retry_wait: float = 2.0,
) -> str:
    last_error = None
    for attempt in range(1, max_retries + 1):
        try:
            payload = {
                "model": model,
                "messages": messages,
                "stream": False,
                "options": {"num_ctx": 4096},
            }
            resp = requests.post(
                "http://localhost:11434/api/chat",
                json=payload,
                timeout=120,
            )
            resp.raise_for_status()
            return resp.json()["message"]["content"]
        except requests.exceptions.Timeout:
            last_error = f"タイムアウト（試行{attempt}回目）"
        except requests.exceptions.ConnectionError:
            last_error = f"接続エラー（試行{attempt}回目）"
        except requests.exceptions.HTTPError as e:
            last_error = f"HTTPエラー {e.response.status_code}（試行{attempt}回目）"
            if e.response.status_code < 500:
                break  # 4xx はリトライしない
        if attempt < max_retries:
            time.sleep(retry_wait * attempt)
    raise RuntimeError(f"応答取得に失敗しました: {last_error}")

$ python3 -c "
import requests
try:
    resp = requests.get('http://localhost:11434/api/tags', timeout=5)
    models = [m['name'] for m in resp.json()['models']]
    print('稼働中。利用可能モデル:', models)
except requests.exceptions.ConnectionError:
    print('Ollamaサーバーに接続できません。systemctl status ollama で確認してください。')
"
稼働中。利用可能モデル: ['llama3.3:8b-instruct-q4_K_M']

# messages配列の中身を確認するデバッグコマンド
$ python3 -c "
import json
messages = [
    {'role': 'system', 'content': 'Linuxの専門家'},
    {'role': 'user', 'content': '質問1'},
    {'role': 'assistant', 'content': '回答1'},
    {'role': 'user', 'content': '質問2'},
]
print('メッセージ数:', len(messages))
for m in messages:
    print(m['role'], ':', m['content'][:30])
"
メッセージ数: 4
system : Linuxの専門家
user : 質問1
assistant : 回答1
user : 質問2

# Ollamaが正常に稼働しているか確認する
$ curl -s http://localhost:11434/api/tags | python3 -c "
import sys, json
d = json.load(sys.stdin)
for m in d.get('models', []):
    print(m['name'])
"
llama3.3:8b-instruct-q4_K_M

ローカルLLMのPython連携を2日間のハンズオンで体験する

この記事で解説したマルチターン会話の実装は、実際に手を動かしながら試すことで理解が深まります。実機GPU環境で手を動かしながら習得したい方向けに、「ローカルAIマスターセミナー」を開催しています。
少人数（最大8名）ZOOMハンズオン形式で実施しています。