コンテンツにスキップ

ツール呼び出しエージェント

モデルがツールを呼び出し、その結果を見て、タスクが終わるまで続けるループを作ります。

  1. ツールを JSON Schema リストとして定義。
  2. messages とツールを送信。
  3. tool_calls が返ってきたら、ローカルで実行して結果を追加。
  4. 通常のメッセージが返るまで繰り返す。
import json
from openai import OpenAI
client = OpenAI(base_url="https://api.aiand.com/v1", api_key="sk-...")
def get_weather(city: str) -> str:
return json.dumps({"city": city, "temp_c": 22, "conditions": "sunny"})
def search_web(query: str) -> str:
return json.dumps({"results": [{"title": "Tokyo", "url": "https://en.wikipedia.org/wiki/Tokyo"}]})
TOOLS = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get current weather for a city.",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"],
},
},
},
{
"type": "function",
"function": {
"name": "search_web",
"description": "Search the web.",
"parameters": {
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"],
},
},
},
]
HANDLERS = {"get_weather": get_weather, "search_web": search_web}
def run(question: str) -> str:
messages = [{"role": "user", "content": question}]
for _ in range(10):
response = client.chat.completions.create(
model="deepseek-ai/deepseek-v4-flash",
messages=messages,
tools=TOOLS,
)
msg = response.choices[0].message
messages.append(msg.model_dump(exclude_none=True))
if not msg.tool_calls:
return msg.content
for call in msg.tool_calls:
args = json.loads(call.function.arguments)
result = HANDLERS[call.function.name](**args)
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": result,
})
raise RuntimeError("Agent exceeded max iterations")
print(run("What's the weather in Tokyo, and is it famous for anything?"))
  • 反復回数の上限 (上記では 10 回) を必ず設定 — 暴走を防ぎます。
  • tool_choice="auto" がデフォルト。特定ツールを強制するなら tool_choice={"type": "function", "function": {"name": "..."}}。
  • 並列ツール呼び出し (1 ターンで複数) は parallel_tool_calls: true がデフォルト。上記ループは複数の tool_calls を 1 イテレーション内で処理します。