用 GroqCloud Function Calling 实现受控工具调用循环

Groq 支持 OpenAI 风格的 tools / tool_calls。最小可用循环是：发消息 → 若返回 tool_calls 则本地执行 → 以 role=tool 回灌 → 再要最终回答。 这篇做天气查询演示（数据写死在字典里）和 mock 工具调用两轮对话。

先看清成品

• mock 代理跑通：用户问上海天气 → 调用 get_weather → 最终回答含 24/多云。
• 2 项 pytest 通过。

合规提醒：使用自己的 GroqCloud 账号与 API Key；语音内容需你有权处理。模型与价格以 groq.com/pricing 与控制台为准，本文不写价格。遵守各模型许可。

运行说明：没有可用的 Groq 密钥。真实运行过：流式聊天 / Whisper / tool-calling 客户端对本地 mock 的请求，以及对 https://api.groq.com 使用无效密钥得到 HTTP 401（2026-10-23，UTC+8）。未上传真实用户音频，未产生计费。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

94-groqcloud-C-src/
├── tool_agent.py
├── mock_tools.py
└── test_94c.py

第 1 步：Agent

#!/usr/bin/env python3
"""Minimal Groq tool-calling loop (official tools / tool_calls / role=tool)."""
from __future__ import annotations

import argparse
import json
import os
import sys
import urllib.error
import urllib.request


WEATHER = {
    "上海": {"temp": 24, "condition": "多云"},
    "北京": {"temp": 18, "condition": "晴"},
}

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询城市当前天气（演示数据）",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string", "description": "城市名"},
                    "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
                },
                "required": ["location"],
            },
        },
    }
]


def get_weather(location: str, unit: str = "celsius") -> dict:
    info = WEATHER.get(location) or {"temp": None, "condition": "未知"}
    temp = info["temp"]
    if temp is not None and unit == "fahrenheit":
        temp = round(temp * 9 / 5 + 32, 1)
    return {"location": location, "temperature": temp, "unit": unit, "condition": info["condition"]}


DISPATCH = {"get_weather": lambda **kw: get_weather(**kw)}


def _post(base_url, api_key, payload):
    url = f"{base_url.rstrip('/')}/openai/v1/chat/completions"
    req = urllib.request.Request(
        url,
        data=json.dumps(payload).encode(),
        headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            return json.loads(resp.read().decode())
    except urllib.error.HTTPError as e:
        raise SystemExit(f"HTTP {e.code}: {e.read().decode(errors='replace')}") from e


def run_agent(*, base_url, api_key, model, user_text, max_rounds=3) -> str:
    messages = [
        {"role": "system", "content": "你是天气助手。需要实时天气时调用 get_weather。"},
        {"role": "user", "content": user_text},
    ]
    for _ in range(max_rounds):
        resp = _post(base_url, api_key, {"model": model, "messages": messages, "tools": TOOLS})
        msg = resp["choices"][0]["message"]
        tool_calls = msg.get("tool_calls") or []
        if not tool_calls:
            return msg.get("content") or ""
        messages.append(msg)
        for tc in tool_calls:
            fn = tc["function"]["name"]
            args = json.loads(tc["function"]["arguments"] or "{}")
            if fn not in DISPATCH:
                result = {"error": f"unknown tool {fn}"}
            else:
                result = DISPATCH[fn](**args)
            messages.append({
                "role": "tool",
                "tool_call_id": tc["id"],
                "name": fn,
                "content": json.dumps(result, ensure_ascii=False),
            })
    return "超过最大轮数"


def main(argv=None) -> int:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("question", nargs="?", default="上海现在天气怎么样？")
    p.add_argument("--base", default=os.environ.get("GROQ_BASE_URL", "https://api.groq.com"))
    p.add_argument("--key", default=os.environ.get("GROQ_API_KEY", ""))
    p.add_argument("--model", default="llama-3.1-8b-instant")
    args = p.parse_args(argv)
    if not args.key:
        print("设置 GROQ_API_KEY", file=sys.stderr)
        return 2
    print(run_agent(base_url=args.base, api_key=args.key, model=args.model, user_text=args.question))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

第 2 步：Mock

#!/usr/bin/env python3
from __future__ import annotations
import json
from http.server import BaseHTTPRequestHandler, HTTPServer

API_KEY = "gsk-test"
STATE = {"calls": 0}


class H(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass

    def do_POST(self):
        if self.path != "/openai/v1/chat/completions":
            self.send_error(404); return
        if self.headers.get("Authorization") != f"Bearer {API_KEY}":
            self.send_response(401); self.end_headers(); return
        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        messages = body["messages"]
        tools = body.get("tools") or []
        # If last message is tool result, finish
        if messages and messages[-1].get("role") == "tool":
            content = f"根据工具结果：{messages[-1].get('content')}"
            resp = {"choices": [{"message": {"role": "assistant", "content": content}, "finish_reason": "stop"}]}
        elif tools and STATE["calls"] == 0:
            STATE["calls"] += 1
            args = json.dumps({"location": "上海", "unit": "celsius"}, ensure_ascii=False)
            resp = {"choices": [{"message": {
                "role": "assistant",
                "content": None,
                "tool_calls": [{"id": "call_1", "type": "function",
                                "function": {"name": "get_weather", "arguments": args}}],
            }, "finish_reason": "tool_calls"}]}
        else:
            resp = {"choices": [{"message": {"role": "assistant", "content": "不用工具也能聊"}, "finish_reason": "stop"}]}
        self.send_response(200); self.send_header("Content-Type", "application/json"); self.end_headers()
        self.wfile.write(json.dumps(resp, ensure_ascii=False).encode())


if __name__ == "__main__":
    import os
    HTTPServer(("127.0.0.1", int(os.environ.get("PORT", "3097"))), H).serve_forever()

第 3 步：测试

import threading
from http.server import HTTPServer
import mock_tools
from tool_agent import run_agent, get_weather

def serve():
    mock_tools.STATE["calls"] = 0
    httpd = HTTPServer(("127.0.0.1", 0), mock_tools.H)
    threading.Thread(target=httpd.serve_forever, daemon=True).start()
    return httpd, httpd.server_address[1]

def test_weather_fn():
    r = get_weather("上海")
    assert r["temperature"] == 24

def test_agent_loop():
    httpd, port = serve()
    try:
        out = run_agent(base_url=f"http://127.0.0.1:{port}", api_key=mock_tools.API_KEY,
                        model="llama-3.1-8b-instant", user_text="上海天气")
        assert "24" in out or "多云" in out or "工具结果" in out
    finally:
        httpd.shutdown()

第 4 步：运行

# 2026-10-23 12:37 UTC+8
根据工具结果：{"location": "上海", "temperature": 24, "unit": "celsius", "condition": "多云"}

$ python -m pytest -q
2 passed

怎么检查结果？

• 最终自然语言回答应含工具结果中的温度或状况。
• DISPATCH 白名单：不要 eval 任意函数名。

调试示例

只信任白名单函数名。 tool_calls[].function.name 来自模型，必须在 DISPATCH 内查找；未知名称直接报错，避免模型“调用”到危险函数。

常见问题

模板规则

• 工具白名单。
• 限制回合数。
• 工具副作用要可审计。

项目资源

工具调用的安全边界在你的 DISPATCH，不在模型保证里。