OpenAI function calling：60 行 Python 实现一个 Google 搜索工具

{
  "type": "function",
  "function": {
    "name": "google_search",
    "description": "Search Google for current information. Returns the top results with title, link and snippet. Use it for facts newer than your training data.",
    "parameters": {
      "type": "object",
      "properties": {
        "query": { "type": "string", "description": "The search query, as a person would type it." },
        "gl": { "type": "string", "description": "Two-letter country code of the results, e.g. us, gb, de." },
        "hl": { "type": "string", "description": "Language code, e.g. en, de, pt-BR." }
      },
      "required": ["query"]
    }
  }
}

"""Tool-calling loop: the model asks for searches, this code runs them on SerpCrypto."""
import json
import os

import requests
from openai import OpenAI

client = OpenAI()  # reads OPENAI_API_KEY
MODEL = os.environ["OPENAI_MODEL"]  # any OpenAI chat model with tool calling
API_URL = "https://serpcrypto.com/api/v1/search"

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "google_search",
            "description": "Search Google for current information. Returns the top results with title, link and snippet.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "The search query."},
                    "gl": {"type": "string", "description": "Two-letter country code, e.g. us."},
                    "hl": {"type": "string", "description": "Language code, e.g. en."},
                },
                "required": ["query"],
            },
        },
    }
]


def google_search(query, gl="us", hl="en"):
    r = requests.get(
        API_URL,
        params={"q": query, "gl": gl, "hl": hl},
        headers={"Authorization": "Bearer " + os.environ["SERPCRYPTO_API_KEY"]},
        timeout=90,
    )
    if r.status_code != 200:
        return f"Search failed with HTTP {r.status_code}."
    items = r.json().get("organic_results", [])[:5]
    return "\n\n".join(f"{x['title']}\n{x['link']}\n{x.get('snippet', '')}" for x in items) or "No results."


def ask(question, max_searches=3):
    messages = [{"role": "user", "content": question}]
    for _ in range(max_searches + 1):
        reply = client.chat.completions.create(model=MODEL, messages=messages, tools=TOOLS)
        msg = reply.choices[0].message
        if not msg.tool_calls:
            return msg.content
        messages.append(msg)
        for call in msg.tool_calls:
            args = json.loads(call.function.arguments)
            messages.append({"role": "tool", "tool_call_id": call.id, "content": google_search(**args)})
    return "Stopped: search limit reached."


print(ask("Which countries changed their data protection laws in 2026? Cite sources."))

{
  "name": "google_search",
  "description": "Search Google for current information. Returns the top results with title, link and snippet.",
  "input_schema": {
    "type": "object",
    "properties": {
      "query": { "type": "string", "description": "The search query." },
      "gl": { "type": "string", "description": "Two-letter country code, e.g. us." },
      "hl": { "type": "string", "description": "Language code, e.g. en." }
    },
    "required": ["query"]
  }
}