Google Search no LangChain depois do fim do Custom Search

O GoogleSearchAPIWrapper e as tools google_search construídas sobre ele chamam a Custom Search JSON API do Google, que foi fechada para novos clientes e termina em 1º de janeiro de 2027. Nada no wrapper consegue contornar isso, então a solução é trocar o backend. Abaixo está uma tool do LangChain de 79 linhas, executada em 19 de outubro de 2026, que mantém os mesmos nomes de tool e os mesmos campos title, link e snippet, então suas chains e prompts de agente existentes continuam funcionando. Cada busca custa 5 créditos do FastAF.

O que exatamente para de funcionar no LangChain em 1º de janeiro de 2027?

A visão geral da Custom Search JSON API do Google diz que a API “está fechada para novos clientes” e que os clientes existentes “têm até 1º de janeiro de 2027 para migrar para uma solução alternativa” (conferido em 19 de outubro de 2026). A única alternativa que ela cita, o Vertex AI Search, é descrita como “uma alternativa favorável para buscar em até 50 domínios”, ou seja, busca dentro de um site, não busca na web.

No langchain-google-community 5.1.1 (instalado hoje em um venv limpo, com langchain 1.4.6 e langchain-core 1.6.12), o GoogleSearchAPIWrapper monta um cliente customsearch v1 a partir de GOOGLE_API_KEY e de um GOOGLE_CSE_ID, e toda chamada passa por cse().list(). A cópia mais antiga em langchain_community já está marcada como obsoleta. A página do Google Search no LangChain ainda documenta o wrapper sem nenhum aviso de encerramento, e o índice de tools lista dezenas de outras integrações de busca. Não consegui testar o comportamento depois da data de encerramento, então essa parte é uma declaração do Google, não uma medição minha.

Para o panorama maior, veja a API de busca oficial do Google contra as SERP APIs e busca na web para agentes de IA depois do Bing. Este post trata só da troca no LangChain.

Como substituir o GoogleSearchAPIWrapper por uma tool compatível?

O wrapper antigo tem duas chamadas das quais as chains dependem: run(query), que junta os snippets em uma única string, e results(query, num_results, search_params), que devolve dicts com title, link e snippet, ou [{"Result": "No good Google Search Result was found"}] quando vazio. As tools google_search e google_search_results_json ficam por cima. Este módulo reproduz tudo isso, síncrono e assíncrono, em cima do endpoint /v1/serp:

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),
    ]

Use onde você usava o par antigo:

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)

Um detalhe é medido, não suposto. Em 20 consultas variadas (192 resultados, 100 créditos, 19 de outubro de 2026), 38 resultados (20%) traziam um inline_snippet. Em 36 deles o campo snippet continha apenas rótulos de sitelinks, como “More results from www.reddit.com”, e a descrição de verdade estava em inline_snippet. O módulo prefere inline_snippet, que é o que o antigo snippet parecia. Toda consulta retornou 9 ou 10 resultados orgânicos.

O que eu de fato executei, e quanto custou?

Eu não tinha chave de API de LLM neste ambiente, então nenhum modelo real está envolvido. Executei as tools diretamente (invoke e ainvoke) e depois dentro de um loop create_agent conduzido por um modelo de chat falso com roteiro, que repete três chamadas de tool. Isso prova a ligação no runtime do LangChain, não o comportamento de um modelo.

Contra https://api.fastaf.co/v1 em 19 de outubro de 2026:

Execução	Latência	Créditos	Resultado
`google_search_results_json`, síncrono	3,20 s	5	5 resultados, chaves `link`, `snippet`, `title`
Mesma consulta de novo	0,0002 s	0	servida pelo cache
`google_search`, assíncrono	4,74 s	5	1.642 caracteres de snippets unidos
Agente com roteiro, 3 chamadas de tool, orçamento 10	5,8 s	10	terceira chamada recusada com uma mensagem

Execuções anteriores do mesmo código em produção levaram 6,9 e 8,1 s por busca, então conte com algo entre 3 e 8 segundos, não milissegundos. Na execução de 20 consultas com concorrência 4, a mediana foi 4,0 s e o máximo 12,4 s. Requisições que falham não são cobradas. A checagem do orçamento foi reforçada depois dessas execuções (agora reserva o custo antes da requisição) e repetida apenas com um stub: com max_credits=6, duas chamadas paralelas a aresults deixam uma passar e recusam a outra.

Como os parâmetros antigos se mapeiam para a nova tool?

GoogleSearchAPIWrapper	Esta tool
`GOOGLE_API_KEY` + `GOOGLE_CSE_ID`	só `FASTAF_API_KEY`, sem ID de mecanismo
`k`, `num_results`	mesmos nomes; limitado a 10 por chamada, como o `num` do Google
`search_params={"gl": "de"}`	`{"country": "de"}`
`{"hl": "de"}`	`{"lang": "de"}`
`{"lr": ...}`, `{"cr": ...}`, `{"safe": "active"}`	mesmos nomes
`{"start": 11}` (base 1)	`{"page": 1}` (base zero; a página 1 retornou resultados diferentes da página 0)
`siterestrict=True`, `siteSearch`	coloque `site:example.com` na consulta (testado)
`exactTerms`, `excludeTerms`	aspas e `-termo` na consulta
`dateRestrict="w1"`	`{"tbs": "qdr:w"}`, uma string de filtro do Google (aceita; não verifiquei o filtro de datas)
campos `image`, `thumbnail`	não retornados
`searchType="image"`	não coberto por este endpoint

A diferença de comportamento é a maior. O Custom Search consultava um mecanismo que você configurava. Esta tool devolve o que o google.com devolveu para a consulta, então os rankings vão diferir dos seus resultados antigos. Reveja qualquer conjunto de avaliação fixado em URLs específicas. O parâmetro results da API não reduziu o tamanho da página nos meus testes (results=3 e results=20 retornaram 9), então o módulo corta no cliente.

Como fazer cache e limitar o gasto por execução?

Os dois estão no módulo. O ttl faz cache por consulta, quantidade e parâmetros, então um agente que repete uma busca paga uma vez. O max_credits reserva os 5 créditos de uma busca antes da requisição, troca-os pelo header X-FastAF-Credits-Used da resposta (um erro da API os devolve) e lança ToolException quando a próxima busca não caberia mais; com handle_tool_error=True, o agente recebe “Search budget of 10 credits is used up.” como um resultado normal de tool e pode raciocinar sobre isso, como na tabela acima. O wrapper não guarda estado entre processos, e o cache fica em memória.

O saldo da conta é o limite externo: um saldo esgotado devolve 402 e para tudo. Dê a um agente de IA busca na web ao vivo com um teto de gastos rígido cobre as camadas ao redor disso, e SERP APIs para RAG cobre o que fazer com os resultados.

Posso usar MCP em vez de uma tool customizada?

Sim. O FastAF roda um servidor MCP hospedado com uma tool search_web. Carreguei com langchain-mcp-adapters 0.3.2 (MultiServerMCPClient, transporte streamable_http, header Bearer), recebi as 15 tools, e search_web devolveu 3 resultados por 5 créditos em 4,9 s. Duas ressalvas: ele devolve url em vez de link, e repassa o snippet sem alteração, então você recebe os rótulos de sitelinks descritos acima. Para chains existentes que leem link, mantenha a tool customizada. Para um agente novo que também deva abrir páginas, use MCP.

Resumindo

O Google Custom Search termina em 1º de janeiro de 2027 e o GoogleSearchAPIWrapper para junto. Copie o módulo, mantenha os nomes das tools, confira os testes sensíveis a ranking e defina max_credits para que um agente em loop não gere uma conta alta. Uma busca custa 5 créditos, de $1,33 a $2,50 por 1.000 conforme o pacote, e o plano gratuito é de 500 créditos, cerca de 100 buscas. Pegue uma chave grátis e teste com as suas próprias consultas.