GLM 5.2 函数调用：工具使用、并行调用与智能体工作流

from openai import OpenAI

client = OpenAI(
    api_key="YOUR_Z_AI_KEY",
    base_url="https://api.z.ai/v1",
)

MODEL = "glm-5.2"

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": (
                "Returns current weather conditions for a given city. "
                "Use this when the user asks about weather or temperature."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "City name, e.g. 'Tokyo' or 'San Francisco'",
                    },
                    "units": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Temperature unit. Defaults to celsius.",
                    },
                },
                "required": ["city"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "Fetches top search results for a query. Use for factual lookups.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "The search query string.",
                    },
                    "num_results": {
                        "type": "integer",
                        "description": "Number of results to return. Default 5.",
                    },
                },
                "required": ["query"],
            },
        },
    },
]

messages = [
    {"role": "user", "content": "What is the weather in Tokyo and Berlin right now?"},
]

response = client.chat.completions.create(
    model=MODEL,
    messages=messages,
    tools=tools,
    tool_choice="auto",
)

message = response.choices[0].message
print(message)

import json

def run_tool(name: str, arguments: dict) -> str:
    """Dispatch to the real implementation and return a string result."""
    if name == "get_weather":
        city = arguments["city"]
        units = arguments.get("units", "celsius")
        # Replace with a real weather API call
        return json.dumps({"city": city, "temp": 22, "units": units, "condition": "sunny"})
    elif name == "search_web":
        query = arguments["query"]
        num = arguments.get("num_results", 5)
        # Replace with a real search API call
        return json.dumps({"query": query, "results": [f"Result {i} for {query}" for i in range(num)]})
    else:
        return json.dumps({"error": f"Unknown tool: {name}"})


# Add the assistant's message to history first
messages.append(message)

# Execute each tool call and append a tool-role message for each
if message.tool_calls:
    for tool_call in message.tool_calls:
        name = tool_call.function.name
        args = json.loads(tool_call.function.arguments)
        result = run_tool(name, args)

        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": result,
        })

# The model may return something like:
# tool_calls = [
#   ToolCall(id="call_abc", function=Function(name="get_weather", arguments='{"city":"Tokyo"}')),
#   ToolCall(id="call_xyz", function=Function(name="get_weather", arguments='{"city":"Berlin"}')),
# ]

from concurrent.futures import ThreadPoolExecutor

def execute_tool_call(tc):
    name = tc.function.name
    args = json.loads(tc.function.arguments)
    result = run_tool(name, args)
    return tc.id, result

if message.tool_calls:
    with ThreadPoolExecutor() as executor:
        futures = {executor.submit(execute_tool_call, tc): tc for tc in message.tool_calls}
        for future in futures:
            call_id, result = future.result()
            messages.append({
                "role": "tool",
                "tool_call_id": call_id,
                "content": result,
            })

import json
from concurrent.futures import ThreadPoolExecutor
from openai import OpenAI

client = OpenAI(api_key="YOUR_Z_AI_KEY", base_url="https://api.z.ai/v1")
MODEL = "glm-5.2"

def agent_loop(user_message: str, tools: list, max_iterations: int = 10) -> str:
    messages = [{"role": "user", "content": user_message}]

    for iteration in range(max_iterations):
        response = client.chat.completions.create(
            model=MODEL,
            messages=messages,
            tools=tools,
            tool_choice="auto",
        )
        message = response.choices[0].message
        messages.append(message)

        # No tool calls → model is done
        if not message.tool_calls:
            return message.content

        # Execute tool calls (parallel)
        def execute(tc):
            name = tc.function.name
            args = json.loads(tc.function.arguments)
            return tc.id, run_tool(name, args)

        with ThreadPoolExecutor() as ex:
            for call_id, result in ex.map(execute, message.tool_calls):
                messages.append({
                    "role": "tool",
                    "tool_call_id": call_id,
                    "content": result,
                })

    raise RuntimeError(f"Agent did not finish within {max_iterations} iterations")


answer = agent_loop(
    "Compare the current weather in Tokyo and Berlin, then search for the best time to visit each city.",
    tools=tools,
)
print(answer)

# Instead of: return "The temperature is 22 degrees"
# Return:
return json.dumps({
    "city": "Tokyo",
    "temperature": 22,
    "units": "celsius",
    "humidity": 68,
    "condition": "partly cloudy",
    "wind_kph": 14,
})