ローカルLLMのOllamaとPythonで作るFunction Calling機能の実装方法

はじめに：Function Callingとは

Function Callingは、大規模言語モデル（LLM）に特定の関数の呼び出しを指示させる革新的な機能です。この機能により、AIと既存のシステムを簡単に統合できます。

環境構築：Ollamaのセットアップ

Dockerを使用したOllamaの導入

# Ollamaコンテナの起動（GPU対応） 
docker run -d --gpu all --name ollama -p 11434:11434 ollama/ollama 

# mistral-smallモデルのダウンロード 
docker exec -it ollama ollama pull mistral-small

実装解説：システムの仕組み

処理フローの詳細図解

コード解説：主要コンポーネント

get_weather関数

指定された場所の天気と温度を返す関数です。

def get_weather(location: str) -> str:
    """
    指定された場所の天気と温度を返す関数です。
    """
    weather_data = {
        "東京": {"weather": "晴れ", "temperature": 28},
        "大阪": {"weather": "曇り", "temperature": 25},
        "札幌": {"weather": "雪", "temperature": -5},
    }
    data = weather_data.get(
        location, {"weather": "情報なし", "temperature": "情報なし"}
    )
    return f"{location} の天気は {data['weather']} で、温度は {data['temperature']} 度です。"

create_payload関数

システムプロンプトとユーザーメッセージを含むリクエストペイロードを生成します。

def create_payload() -> dict:
    """
    システムプロンプトとユーザーメッセージを含むリクエストペイロードを生成します。
    """
    messages = [
        {
            "role": "system",
            "content": (
                "あなたは天気情報を提供するエージェントです。"
                "ユーザーからの質問に対しては、必ず下記の形式に沿った JSON 形式の function_call として回答してください。\n\n"
                "例:\n"
                "{\n"
                '  "function_call": {\n'
                '    "name": "get_weather",\n'
                '    "arguments": "{\\"location\\": \\"東京\\"}"\n'
                "  }\n"
                "}\n\n"
                "絶対に他の形式では回答しないでください。"
            ),
        },
        {"role": "user", "content": "大阪の天気と温度を教えてください。"},
    ]

    functions = [
        {
            "name": "get_weather",
            "description": "指定された場所の天気と温度を返す関数です。",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "天気情報を取得する場所の名前",
                    }
                },
                "required": ["location"],
            },
        }
    ]

    payload = {
        "model": MODEL_NAME,
        "messages": messages,
        "functions": functions,
        # 明示的に関数呼び出しを指定
        "function_call": {"name": "get_weather"},
    }
    return payload

send_request関数

Ollama サーバーへリクエストを送信し、JSON レスポンスを返します。

def send_request(payload: dict) -> dict:
    """
    Ollama サーバーへリクエストを送信し、JSON レスポンスを返します。
    """
    response = requests.post(OLLAMA_URL, json=payload)
    response.raise_for_status()
    return response.json()

extract_function_call_from_content関数

メッセージの content 内に含まれるコードブロックから JSON を抽出し、その中に含まれる function_call 情報を返します。

def extract_function_call_from_content(content: str) -> dict:
    """
    メッセージの content 内に含まれるコードブロックから JSON を抽出し、
    その中に含まれる function_call 情報を返します。
    """
    # 正規表現で ```json ... ``` ブロックを抽出
    pattern = r"```json\s*(\{.*?\})\s*```"
    match = re.search(pattern, content, re.DOTALL)
    if match:
        json_str = match.group(1)
        try:
            parsed = json.loads(json_str)
            if "function_call" in parsed:
                return parsed["function_call"]
        except Exception as e:
            print("コードブロック内の JSON パースに失敗:", e)
    else:
        # もしコードブロックがない場合、内容全体が JSON であれば試す
        try:
            parsed = json.loads(content)
            if "function_call" in parsed:
                return parsed["function_call"]
        except Exception:
            pass
    return {}

handle_response関数

LLM のレスポンスから function_call 情報を抽出し、対応するローカル関数を実行する処理を行います。

def handle_response(result: dict):
    """
    LLM のレスポンスから function_call 情報を抽出し、
    対応するローカル関数を実行する処理を行います。
    """
    print("LLM からの応答:")
    print(json.dumps(result, indent=2, ensure_ascii=False))

    choices = result.get("choices", [])
    if not choices:
        print("応答形式が不正です。")
        return

    message = choices[0].get("message", {})
    # まず、直接 function_call フィールドがあるか確認
    function_call = message.get("function_call")
    if not function_call:
        # なければ、content 内のコードブロックから抽出を試みる
        content = message.get("content", "")
        function_call = extract_function_call_from_content(content)

    if function_call:
        function_name = function_call.get("name")
        raw_arguments = function_call.get("arguments", "{}")
        # 引数が余分な引用符で囲まれている場合、その除去を試みる
        if raw_arguments.startswith('"') and raw_arguments.endswith('"'):
            raw_arguments = raw_arguments[1:-1]

        try:
            arguments = json.loads(raw_arguments)
        except Exception as e:
            print("関数引数のパースに失敗:", e)
            arguments = {}

        print(f"\nLLM からの function_call: {function_name}({arguments})")
        if function_name == "get_weather":
            result_value = get_weather(**arguments)
            print("実行結果:", result_value)
        else:
            print(f"定義されていない関数が呼ばれました: {function_name}")
    else:
        print("LLM の応答内に function_call が見つかりませんでした。")
        print("返答内容:", message.get("content"))

main関数

プログラムのエントリーポイントであり、リクエストの送信と応答の処理を統括します。

def main():
    payload = create_payload()
    try:
        result = send_request(payload)
        handle_response(result)
    except Exception as e:
        print("リクエスト中にエラーが発生しました:", e)

システムプロンプトの設計

{
  "function_call": {
    "name": "get_weather",
    "arguments": "{\"location\": \"東京\"}"
  }
}

Function定義のベストプラクティス

functions = [
    {
        "name": "get_weather",
        "description": "指定された場所の天気と温度を返す関数です。",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "天気情報を取得する場所の名前",
                }
            },
            "required": ["location"],
        }
    }
]

実践：エラーハンドリング

実装された主要なエラー対策

1. ✅ JSONパースエラーの適切な処理
2. ✅ 未定義関数呼び出しの検出機能
3. ✅ API通信エラーのハンドリング
4. ✅ バリデーションチェック

動作確認：実行結果

実行例と出力結果

LLM からの応答:
{
  "id": "chatcmpl-448",
  "object": "chat.completion",
  "created": 1739539494,
  "model": "mistral-small",
  "system_fingerprint": "fp_ollama",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "```json\n{\n  \"function_call\": {\n    \"name\": \"get_weather\",\n    \"arguments\": \"{\\\"location\\\": \\\"大阪\\\"}\"\n  }\n}\n```"
      },
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 108,
    "completion_tokens": 36,
    "total_tokens": 144
  }
}

LLM からの function_call: get_weather({'location': '大阪'})
実行結果: 大阪 の天気は 曇り で、温度は 25 度です。

まとめと発展的な使い方

実装のポイント

• ✅ システムプロンプトの適切な設計
• ✅ 堅牢なエラーハンドリング
• ✅ 柔軟なレスポンス解析

今後の展開

• 🔄 他のLLMモデルへの応用
• 🔄 複数機能の連携
• 🔄 実用的なユースケース開発

コード全体

import json
import re
import requests

# 定数設定
OLLAMA_URL = "http://localhost:11434/v1/chat/completions"
MODEL_NAME = "mistral-small"


def get_weather(location: str) -> str:
    """
    ダミーデータを用いて、指定された場所の天気と温度を返す関数です。
    """
    weather_data = {
        "東京": {"weather": "晴れ", "temperature": 28},
        "大阪": {"weather": "曇り", "temperature": 25},
        "札幌": {"weather": "雪", "temperature": -5},
    }
    data = weather_data.get(
        location, {"weather": "情報なし", "temperature": "情報なし"}
    )
    return f"{location} の天気は {data['weather']} で、温度は {data['temperature']} 度です。"


def create_payload() -> dict:
    """
    システムプロンプトとユーザーメッセージを含むリクエストペイロードを生成します。
    ※システムプロンプトで、出力は必ず JSON 形式の function_call として返すように明示しています。
    """
    messages = [
        {
            "role": "system",
            "content": (
                "あなたは天気情報を提供するエージェントです。"
                "ユーザーからの質問に対しては、必ず下記の形式に沿った JSON 形式の function_call として回答してください。\n\n"
                "例:\n"
                "{\n"
                '  "function_call": {\n'
                '    "name": "get_weather",\n'
                '    "arguments": "{\\"location\\": \\"東京\\"}"\n'
                "  }\n"
                "}\n\n"
                "絶対に他の形式では回答しないでください。"
            ),
        },
        {"role": "user", "content": "大阪の天気と温度を教えてください。"},
    ]

    functions = [
        {
            "name": "get_weather",
            "description": "指定された場所の天気と温度を返す関数です。",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "天気情報を取得する場所の名前",
                    }
                },
                "required": ["location"],
            },
        }
    ]

    payload = {
        "model": MODEL_NAME,
        "messages": messages,
        "functions": functions,
        # 明示的に関数呼び出しを指定
        "function_call": {"name": "get_weather"},
    }
    return payload


def send_request(payload: dict) -> dict:
    """
    Ollama サーバーへリクエストを送信し、JSON レスポンスを返します。
    """
    response = requests.post(OLLAMA_URL, json=payload)
    response.raise_for_status()
    return response.json()


def extract_function_call_from_content(content: str) -> dict:
    """
    メッセージの content 内に含まれるコードブロックから JSON を抽出し、
    その中に含まれる function_call 情報を返します。
    """
    # 正規表現で ```json ... ``` ブロックを抽出
    pattern = r"```json\s*(\{.*?\})\s*```"
    match = re.search(pattern, content, re.DOTALL)
    if match:
        json_str = match.group(1)
        try:
            parsed = json.loads(json_str)
            if "function_call" in parsed:
                return parsed["function_call"]
        except Exception as e:
            print("コードブロック内の JSON パースに失敗:", e)
    else:
        # もしコードブロックがない場合、内容全体が JSON であれば試す
        try:
            parsed = json.loads(content)
            if "function_call" in parsed:
                return parsed["function_call"]
        except Exception:
            pass
    return {}


def handle_response(result: dict):
    """
    LLM のレスポンスから function_call 情報を抽出し、
    対応するローカル関数を実行する処理を行います。
    """
    print("LLM からの応答:")
    print(json.dumps(result, indent=2, ensure_ascii=False))

    choices = result.get("choices", [])
    if not choices:
        print("応答形式が不正です。")
        return

    message = choices[0].get("message", {})
    # まず、直接 function_call フィールドがあるか確認
    function_call = message.get("function_call")
    if not function_call:
        # なければ、content 内のコードブロックから抽出を試みる
        content = message.get("content", "")
        function_call = extract_function_call_from_content(content)

    if function_call:
        function_name = function_call.get("name")
        raw_arguments = function_call.get("arguments", "{}")
        # 引数が余分な引用符で囲まれている場合、その除去を試みる
        if raw_arguments.startswith('"') and raw_arguments.endswith('"'):
            raw_arguments = raw_arguments[1:-1]

        try:
            arguments = json.loads(raw_arguments)
        except Exception as e:
            print("関数引数のパースに失敗:", e)
            arguments = {}

        print(f"\nLLM からの function_call: {function_name}({arguments})")
        if function_name == "get_weather":
            result_value = get_weather(**arguments)
            print("実行結果:", result_value)
        else:
            print(f"定義されていない関数が呼ばれました: {function_name}")
    else:
        print("LLM の応答内に function_call が見つかりませんでした。")
        print("返答内容:", message.get("content"))


def main():
    payload = create_payload()
    try:
        result = send_request(payload)
        handle_response(result)
    except Exception as e:
        print("リクエスト中にエラーが発生しました:", e)


if __name__ == "__main__":
    main()

関連リンク

• Ollama公式ドキュメント
• Mistrial-small
• Function calling