LangChain Google-Suche ersetzen: Custom Search wird eingestellt

GoogleSearchAPIWrapper und die darauf aufbauenden google_search-Tools rufen die Google Custom Search JSON API auf, die Google für Neukunden geschlossen hat und am 1. Januar 2027 beendet. Der Wrapper kann das nicht umgehen, die Lösung ist also, das Backend zu tauschen. Unten steht ein LangChain-Tool mit 79 Zeilen, ausgeführt am 19. Oktober 2026, das dieselben Tool-Namen und dieselben Felder title, link und snippet behält, damit bestehende Chains und Agent-Prompts weiterlaufen. Eine Suche kostet 5 FastAF-Credits.

Was genau hört in LangChain am 1. Januar 2027 auf zu funktionieren?

Die Übersicht zur Custom Search JSON API von Google sagt, die API „is closed to new customers” und bestehende Kunden „have until January 1, 2027 to transition to an alternative solution” (geprüft am 19. Oktober 2026). Die einzige genannte Alternative, Vertex AI Search, wird als „a favorable alternative for searching up to 50 domains” beschrieben, das ist Seitensuche, keine Websuche.

In langchain-google-community 5.1.1 (heute in einem frischen venv installiert, mit langchain 1.4.6 und langchain-core 1.6.12) baut GoogleSearchAPIWrapper aus GOOGLE_API_KEY und einer GOOGLE_CSE_ID einen customsearch-v1-Client, und jeder Aufruf läuft über cse().list(). Die ältere Kopie in langchain_community ist bereits als deprecated markiert. LangChains Seite zu Google Search dokumentiert den Wrapper weiterhin ohne Hinweis auf die Abschaltung, und der Tool-Index listet Dutzende weiterer Such-Integrationen. Das Verhalten nach dem Stichtag konnte ich nicht testen, dieser Teil ist also Googles Aussage, nicht meine Messung.

Das größere Bild findest du unter Googles offizielle Search API vs. SERP APIs und Websuche für KI-Agenten nach Bing. Dieser Beitrag behandelt nur den Austausch in LangChain.

Wie ersetze ich GoogleSearchAPIWrapper durch ein Drop-in-Tool?

Der alte Wrapper hat zwei Aufrufe, auf die Chains angewiesen sind: run(query), das Snippets zu einem String verbindet, und results(query, num_results, search_params), das Dicts mit title, link und snippet liefert, bei leerem Ergebnis [{"Result": "No good Google Search Result was found"}]. Darauf sitzen die Tools google_search und google_search_results_json. Dieses Modul bildet alles nach, synchron und asynchron, auf Basis des /v1/serp-Endpoints:

import os, time, httpx
from langchain_core.tools import StructuredTool, ToolException

BASE = "https://api.fastaf.co/v1"
NOTHING = [{"Result": "No good Google Search Result was found"}]
COST = 5  # credits per search

class FastAFSearch:
    """Same surface as GoogleSearchAPIWrapper: run(), results(), arun(), aresults()."""

    def __init__(self, k=10, max_credits=None, ttl=300, **defaults):
        self.k, self.max_credits, self.ttl, self.defaults = k, max_credits, ttl, defaults
        self.spent, self._cache = 0, {}
        self._headers = {"Authorization": f"Bearer {os.environ['FASTAF_API_KEY']}"}

    def _hit(self, key):
        hit = self._cache.get(key)
        return hit[1] if hit and time.time() - hit[0] < self.ttl else None

    def _params(self, query, extra):
        if self.max_credits is not None and self.spent + COST > self.max_credits:
            raise ToolException(f"Search budget of {self.max_credits} credits is used up.")
        self.spent += COST  # reserve before the request, so parallel async calls cannot both pass
        return {"query": query, **self.defaults, **(extra or {})}

    def _parse(self, key, resp, num):
        # An API error is never charged: give the reservation back. A 200 reports its real cost.
        self.spent += (int(resp.headers["X-FastAF-Credits-Used"]) if resp.status_code == 200 else 0) - COST
        if resp.status_code != 200:
            raise ToolException(f"Search failed: HTTP {resp.status_code}")
        out = [{"title": r["title"], "link": r["link"],
                "snippet": r.get("inline_snippet") or r.get("snippet", "")}
               for r in resp.json()["organic_results"][:min(num, 10)]]
        self._cache[key] = (time.time(), out or NOTHING)
        return out or NOTHING

    def results(self, query, num_results, search_params=None):
        key = (query, num_results, str(search_params))
        if (hit := self._hit(key)) is not None:
            return hit
        resp = httpx.get(f"{BASE}/serp", params=self._params(query, search_params),
                         headers=self._headers, timeout=60)
        return self._parse(key, resp, num_results)

    async def aresults(self, query, num_results, search_params=None):
        key = (query, num_results, str(search_params))
        if (hit := self._hit(key)) is not None:
            return hit
        async with httpx.AsyncClient(timeout=60) as c:
            resp = await c.get(f"{BASE}/serp", params=self._params(query, search_params),
                               headers=self._headers)
        return self._parse(key, resp, num_results)

    @staticmethod
    def _join(rs):
        return " ".join(r["snippet"] for r in rs if "snippet" in r) or NOTHING[0]["Result"]

    def run(self, query: str) -> str:
        return self._join(self.results(query, self.k))

    async def arun(self, query: str) -> str:
        return self._join(await self.aresults(query, self.k))

def fastaf_search_tools(api: FastAFSearch, num_results: int = 4):
    """Two tools with the names and descriptions of GoogleSearchRun / GoogleSearchResults."""
    def as_json(query: str) -> str:
        return str(api.results(query, num_results))
    async def aas_json(query: str) -> str:
        return str(await api.aresults(query, num_results))
    desc = ("A wrapper around Google Search. Useful for when you need to answer questions "
            "about current events. Input should be a search query.")
    return [
        StructuredTool.from_function(func=api.run, coroutine=api.arun, name="google_search",
                                     description=desc, handle_tool_error=True),
        StructuredTool.from_function(func=as_json, coroutine=aas_json,
                                     name="google_search_results_json",
                                     description=desc + " Output is a JSON array of the query results",
                                     handle_tool_error=True),
    ]

Setz es dort ein, wo du das alte Paar benutzt hast:

from fastaf_search_tools import FastAFSearch, fastaf_search_tools

api = FastAFSearch(country="us", max_credits=50)
tools = fastaf_search_tools(api, num_results=5)   # drop into create_agent(model, tools=tools)

Ein Detail ist gemessen, nicht angenommen. Über 20 verschiedene Abfragen (192 Ergebnisse, 100 Credits, 19. Oktober 2026) trugen 38 Ergebnisse (20 %) ein inline_snippet. Bei 36 davon enthielt das Feld snippet nur Sitelink-Beschriftungen wie „More results from www.reddit.com”, und die eigentliche Beschreibung stand in inline_snippet. Das Modul bevorzugt inline_snippet, so sah auch das alte snippet aus. Jede Abfrage lieferte 9 oder 10 organische Ergebnisse.

Was habe ich tatsächlich ausgeführt, und was hat es gekostet?

In dieser Umgebung hatte ich keinen LLM-API-Key, es ist also kein echtes Modell beteiligt. Ich habe die Tools direkt ausgeführt (invoke und ainvoke) und dann in einer create_agent-Schleife, gesteuert von einem geskripteten Fake-Chatmodell, das drei Tool-Aufrufe abspielt. Das belegt die Verdrahtung in der LangChain-Runtime, nicht das Verhalten eines Modells.

Gegen https://api.fastaf.co/v1 am 19. Oktober 2026:

Lauf	Latenz	Credits	Ergebnis
`google_search_results_json`, sync	3,20 s	5	5 Ergebnisse, Schlüssel `link`, `snippet`, `title`
Dieselbe Abfrage nochmal	0,0002 s	0	aus dem Cache geliefert
`google_search`, async	4,74 s	5	1.642 Zeichen verbundene Snippets
Geskripteter Agent, 3 Tool-Aufrufe, Budget 10	5,8 s	10	dritter Aufruf mit einer Meldung abgelehnt

Frühere Läufe desselben Codes in Produktion brauchten 6,9 und 8,1 s pro Suche, plane also mit etwa 3 bis 8 Sekunden, nicht mit Millisekunden. Im Lauf mit 20 Abfragen bei Nebenläufigkeit 4 lag der Median bei 4,0 s und das Maximum bei 12,4 s. Fehlgeschlagene Requests werden nicht berechnet. Die Budgetprüfung wurde nach diesen Läufen verschärft (sie reserviert die Kosten jetzt vor dem Request) und nur gegen einen Stub erneut geprüft: Mit max_credits=6 lassen zwei parallele aresults-Aufrufe einen durch und lehnen den anderen ab.

Wie werden die alten Parameter auf das neue Tool abgebildet?

GoogleSearchAPIWrapper	Dieses Tool
`GOOGLE_API_KEY` + `GOOGLE_CSE_ID`	nur `FASTAF_API_KEY`, keine Engine-ID
`k`, `num_results`	gleiche Namen; auf 10 pro Aufruf begrenzt, wie Googles `num`
`search_params={"gl": "de"}`	`{"country": "de"}`
`{"hl": "de"}`	`{"lang": "de"}`
`{"lr": ...}`, `{"cr": ...}`, `{"safe": "active"}`	gleiche Namen
`{"start": 11}` (ab 1 gezählt)	`{"page": 1}` (ab null gezählt; Seite 1 lieferte andere Ergebnisse als Seite 0)
`siterestrict=True`, `siteSearch`	`site:example.com` in die Abfrage schreiben (getestet)
`exactTerms`, `excludeTerms`	Anführungszeichen und `-term` in der Abfrage
`dateRestrict="w1"`	`{"tbs": "qdr:w"}`, ein Google-Filterstring (wird akzeptiert; die Datumsfilterung habe ich nicht geprüft)
Felder `image`, `thumbnail`	werden nicht geliefert
`searchType="image"`	von diesem Endpoint nicht abgedeckt

Der Unterschied im Verhalten wiegt schwerer. Custom Search hat eine Engine abgefragt, die du selbst konfiguriert hast. Dieses Tool liefert, was google.com für die Abfrage zurückgab, die Rankings weichen also von deinen alten Ergebnissen ab. Prüf jedes Eval-Set neu, das auf bestimmte URLs festgelegt war. Der Parameter results der API verkleinerte in meinen Tests die Seitengröße nicht (results=3 und results=20 lieferten beide 9), das Modul schneidet deshalb clientseitig zu.

Wie cache ich und begrenze die Ausgaben pro Lauf?

Beides steckt im Modul. ttl cacht nach Abfrage, Anzahl und Parametern, ein Agent, der eine Suche wiederholt, zahlt also nur einmal. max_credits reserviert die 5 Credits einer Suche vor dem Request, tauscht sie gegen den Header X-FastAF-Credits-Used der Antwort (ein API-Fehler gibt sie zurück) und wirft ToolException, sobald die nächste Suche nicht mehr passen würde; mit handle_tool_error=True erhält der Agent „Search budget of 10 credits is used up.” als normales Tool-Ergebnis und kann darüber nachdenken, wie in der Tabelle oben. Der Wrapper hält keinen Zustand über Prozesse hinweg, und der Cache liegt im Arbeitsspeicher.

Das Guthaben des Accounts ist die äußere Grenze: Ein aufgebrauchtes Guthaben liefert 402 und stoppt alles. Einem KI-Agenten Live-Websuche geben, mit hartem Ausgabenlimit behandelt die Ebenen drumherum, und SERP APIs für RAG zeigt, was du mit den Ergebnissen machst.

Kann ich statt eines eigenen Tools MCP nutzen?

Ja. FastAF betreibt einen gehosteten MCP-Server mit einem Tool search_web. Ich habe ihn mit langchain-mcp-adapters 0.3.2 geladen (MultiServerMCPClient, Transport streamable_http, Bearer-Header), bekam alle 15 Tools, und search_web lieferte 3 Ergebnisse für 5 Credits in 4,9 s. Zwei Haken: Es liefert url statt link, und es reicht snippet unverändert durch, du bekommst also die oben beschriebenen Sitelink-Beschriftungen. Für bestehende Chains, die link lesen, behalte das eigene Tool. Für einen neuen Agenten, der auch Seiten öffnen soll, nimm MCP.

Die Kurzfassung

Google Custom Search endet am 1. Januar 2027, und GoogleSearchAPIWrapper geht damit. Kopier das Modul, behalte die Tool-Namen, prüf deine rankingabhängigen Tests und setz max_credits, damit ein Agent in einer Schleife keine Rechnung auftürmt. Eine Suche kostet 5 Credits, je nach Paket 1,33 $ bis 2,50 $ pro 1.000, und das Free-Kontingent sind 500 Credits, etwa 100 Suchen. Hol dir einen kostenlosen Key und probier es mit deinen eigenen Abfragen aus.