import json
import os
from openai import OpenAI
MAX_TOOL_ROUNDS = 4
ALLOWED_UNITS = {"celsius", "fahrenheit"}
TOOLS = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "读取指定城市的当前天气摘要",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string"},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
},
},
"required": ["city", "unit"],
"additionalProperties": False,
},
},
}
]
def dispatch_tool(name, arguments):
if name != "get_weather":
raise ValueError("不允许的工具")
if not isinstance(arguments, dict):
raise ValueError("参数必须是对象")
if set(arguments) != {"city", "unit"}:
raise ValueError("参数字段不匹配")
city = arguments["city"]
unit = arguments["unit"]
if not isinstance(city, str) or not city.strip() or len(city) > 80:
raise ValueError("city 无效")
if unit not in ALLOWED_UNITS:
raise ValueError("unit 无效")
# 固定演示数据：实际项目应在此调用受控后端。
return {
"ok": True,
"city": city.strip(),
"temperature": 26 if unit == "celsius" else 79,
"unit": unit,
"source": "demo_fixture",
}
client = OpenAI(
api_key=os.environ["MUXLLM_API_KEY"],
base_url="https://api.muxllm.com/v1",
)
messages = [
{"role": "user", "content": "查询上海天气，并用一句话建议穿着。"}
]
executed_call_ids = set()
for _ in range(MAX_TOOL_ROUNDS):
response = client.chat.completions.create(
model=os.environ["MUXLLM_MODEL"],
messages=messages,
tools=TOOLS,
tool_choice="auto",
)
assistant = response.choices[0].message
calls = assistant.tool_calls or []
assistant_message = {"role": "assistant"}
if assistant.content is not None:
assistant_message["content"] = assistant.content
if calls:
assistant_message["tool_calls"] = [
{
"id": call.id,
"type": call.type,
"function": {
"name": call.function.name,
"arguments": call.function.arguments,
},
}
for call in calls
]
messages.append(assistant_message)
if not calls:
print(assistant.content or "")
break
for call in calls:
if call.id in executed_call_ids:
result = {"ok": False, "error": "duplicate_tool_call"}
else:
executed_call_ids.add(call.id)
try:
arguments = json.loads(call.function.arguments)
result = dispatch_tool(call.function.name, arguments)
except (json.JSONDecodeError, KeyError, TypeError, ValueError):
result = {"ok": False, "error": "invalid_tool_request"}
messages.append(
{
"role": "tool",
"tool_call_id": call.id,
"content": json.dumps(result, ensure_ascii=False),
}
)
else:
raise RuntimeError("达到最大工具轮数，已停止")
