AIエージェントの作り方：ツール呼び出しの仕組みをPythonとOllamaで実装する

ollama show qwen2.5:0.5b --template | sed -n '/# Tools/,/<\/tool_call>/p'

# Tools

You may call one or more functions to assist with the user query.

You are provided with function signatures within <tools></tools> XML tags:
<tools>
{{- range .Tools }}
{"type": "function", "function": {{ .Function }}}
{{- end }}
</tools>

For each function call, return a json object with function name and arguments within <tool_call></tool_call> XML tags:
<tool_call>
{"name": <function-name>, "arguments": <args-json-object>}
</tool_call>

{
  "role": "assistant",
  "content": "",
  "tool_calls": [
    {
      "function": {
        "name": "get_ticket_status",
        "arguments": {
          "ticket_id": "INC-1042"
        }
      }
    }
  ]
}

ollama pull qwen2.5:0.5b
ollama show qwen2.5:0.5b | sed -n '/Capabilities/,/^$/p'
#   Capabilities
#     completion
#     tools

python3 -m venv .venv && . .venv/bin/activate
pip install ollama==0.6.2

import json
import sys

from ollama import Client

MODEL = sys.argv[1] if len(sys.argv) > 1 else "qwen2.5:0.5b"
MAX_STEPS = 5
client = Client()

# 例: 本来は社内システムのAPIを呼ぶ部分。ここでは固定データで代用する
TICKETS = {
    "INC-1042": {"status": "対応中", "owner": "佐藤", "updated": "2026-09-22 17:40"},
    "INC-1043": {"status": "完了", "owner": "鈴木", "updated": "2026-09-23 09:05"},
}
ACCOUNTS = {
    "tanaka": {"locked": True, "failed_logins": 7},
    "kato": {"locked": False, "failed_logins": 0},
}


def get_ticket_status(ticket_id: str) -> str:
    """問い合わせチケットの状態・担当者・最終更新日時を返す

    Args:
        ticket_id: チケット番号。例: INC-1042
    """
    t = TICKETS.get(ticket_id.strip().upper())
    if t is None:
        return json.dumps({"error": f"{ticket_id} は存在しません"}, ensure_ascii=False)
    return json.dumps(t, ensure_ascii=False)


def check_account_lock(user_id: str) -> str:
    """社内アカウントがロックされているかと、連続ログイン失敗回数を返す

    Args:
        user_id: 社員のユーザーID。例: tanaka
    """
    a = ACCOUNTS.get(user_id.strip().lower())
    if a is None:
        return json.dumps({"error": f"{user_id} は存在しません"}, ensure_ascii=False)
    return json.dumps(a, ensure_ascii=False)


TOOLS = {f.__name__: f for f in (get_ticket_status, check_account_lock)}

{
  "type": "function",
  "function": {
    "name": "get_ticket_status",
    "description": "問い合わせチケットの状態・担当者・最終更新日時を返す",
    "parameters": {
      "type": "object",
      "required": ["ticket_id"],
      "properties": {
        "ticket_id": {"type": "string", "description": "チケット番号。例: INC-1042"}
      }
    }
  }
}

def run(question: str) -> str:
    messages = [
        {"role": "system", "content": "あなたは社内ヘルプデスクの補助エージェントです。"
         "事実はツールの結果だけを根拠にし、ツールで確認できないことは「確認できません」と答えます。"},
        {"role": "user", "content": question},
    ]
    for step in range(1, MAX_STEPS + 1):
        resp = client.chat(model=MODEL, messages=messages, tools=list(TOOLS.values()),
                           options={"temperature": 0, "num_predict": 512})
        messages.append(resp.message)
        if not resp.message.tool_calls:
            return resp.message.content
        for call in resp.message.tool_calls:
            name, args = call.function.name, call.function.arguments
            fn = TOOLS.get(name)
            if fn is None:
                result = json.dumps({"error": f"未登録のツール: {name}"}, ensure_ascii=False)
            else:
                try:
                    result = fn(**args)
                except TypeError as e:  # 引数名の間違いなど
                    result = json.dumps({"error": str(e)}, ensure_ascii=False)
            print(f"[step {step}] {name}({json.dumps(args, ensure_ascii=False)}) -> {result}")
            messages.append({"role": "tool", "tool_name": name, "content": result})
    return "ステップ上限に達したため中断しました"


if __name__ == "__main__":
    for q in sys.argv[2:] or ["INC-1042の状況を教えてください"]:
        print("Q:", q)
        print("A:", run(q))
        print()

$ python agent.py qwen2.5:0.5b "INC-1042の状況を教えてください" "tanakaさんのアカウントはロックされていますか？"
Q: INC-1042の状況を教えてください
[step 1] get_ticket_status({"ticket_id": "INC-1042"}) -> {"status": "対応中", "owner": "佐藤", "updated": "2026-09-22 17:40"}
A: チケットの状態は対応中です。現在の担当者は佐藤です。チケットの最終更新日時は2026-09-22 17:40です。

Q: tanakaさんのアカウントはロックされていますか？
[step 1] check_account_lock({"user_id": "tanaka"}) -> {"locked": true, "failed_logins": 7}
A: 社内アカウントのtanakaさんはロックされています。連続ログイン失敗回数は7回です。

import re

TICKET_RE = re.compile(r"^INC-\d{4}$")


def validate(name: str, args: dict, question: str) -> str | None:
    """問題があればエラーメッセージを、なければ None を返す"""
    if name == "get_ticket_status":
        tid = str(args.get("ticket_id", "")).strip().upper()
        if not TICKET_RE.match(tid):
            return f"チケット番号の形式が不正です: {tid}"
    if name == "check_account_lock":
        uid = str(args.get("user_id", "")).strip().lower()
        if uid not in question.lower():  # 質問に出てこないIDは使わせない
            return f"質問に含まれないユーザーIDです: {uid}"
    return None

WRITE_TOOLS = {"unlock_account", "create_ticket"}


def confirm(name: str, args: dict) -> bool:
    print(f"実行してよいですか？ {name}({json.dumps(args, ensure_ascii=False)}) [y/N] ", end="")
    return input().strip().lower() == "y"

# run() の中で:
# if name in WRITE_TOOLS and not confirm(name, args):
#     result = json.dumps({"error": "利用者が実行を拒否しました"}, ensure_ascii=False)

# 1. モデルがツール呼び出しに対応しているか
ollama show qwen2.5:0.5b | grep -A3 Capabilities
# 期待: completion と tools が並ぶ

# 2. 単純な質問でツールが1回呼ばれるか
python agent.py qwen2.5:0.5b "INC-1042の状況を教えてください"
# 期待: [step 1] get_ticket_status({"ticket_id": "INC-1042"}) -> {...} の行と、状態・担当者を含む回答

# 3. 存在しないIDでもエラーにならず答えるか
python agent.py qwen3:0.6b "INC-9999の担当者は誰ですか？"
# 期待: ツール結果の error を受けて「存在しません」と答える（モデルにより揺れる）