-
Notifications
You must be signed in to change notification settings - Fork 666
Expand file tree
/
Copy pathagentic_tools.py
More file actions
118 lines (104 loc) · 3.51 KB
/
Copy pathagentic_tools.py
File metadata and controls
118 lines (104 loc) · 3.51 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
"""
Example demonstrating the Python SDK's agentic tool callback system.
The Python SDK lets you register tool callbacks directly on the Runner.
When combined with max_tool_rounds on the request, the engine automatically
executes your callbacks and feeds results back to the model in a loop.
You can also set tool_dispatch_url on the request to POST unhandled tool
calls to an external HTTP endpoint for execution.
Usage:
python examples/python/agentic_tools.py
"""
import json
from mistralrs import (
Runner,
Which,
Architecture,
ChatCompletionRequest,
ToolChoice,
)
def tool_callback(name: str, args: dict) -> str:
"""Dispatch tool calls to local implementations."""
if name == "get_weather":
city = args.get("city", "unknown")
return json.dumps({"city": city, "temp": 22, "condition": "Sunny"})
if name == "calculate":
expression = args.get("expression", "0")
try:
result = eval(expression) # noqa: S307 — example only
except Exception as e:
result = str(e)
return json.dumps({"result": result})
return json.dumps({"error": f"Unknown tool: {name}"})
def main():
# Register tool callbacks at Runner level — these are available to all requests
runner = Runner(
which=Which.Plain(
model_id="Qwen/Qwen3-4B",
arch=Architecture.Qwen3,
),
tool_callbacks={
"get_weather": tool_callback,
"calculate": tool_callback,
},
)
tools = [
json.dumps(
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get the current weather for a city.",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "City name",
},
},
"required": ["city"],
},
"strict": True,
},
}
),
json.dumps(
{
"type": "function",
"function": {
"name": "calculate",
"description": "Evaluate a math expression.",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "Math expression to evaluate",
},
},
"required": ["expression"],
},
"strict": True,
},
}
),
]
# max_tool_rounds enables the agentic loop: model calls tools,
# engine executes callbacks, feeds results back, repeats.
request = ChatCompletionRequest(
messages=[
{
"role": "user",
"content": "What's the weather in Tokyo? Also calculate 42 * 17.",
}
],
model="default",
tool_schemas=tools,
tool_choice=ToolChoice.Auto,
max_tool_rounds=5,
)
response = runner.send_chat_completion_request(request)
for choice in response.choices:
print(choice.message.content)
if __name__ == "__main__":
main()