LangChain ↗ is a framework for building applications with large language models. The langchain-cloudflare ↗ package provides CloudflareAISearchRetriever, a standard LangChain retriever backed by AI Search.
The retriever only searches. To create an instance and upload content, pair it with the Cloudflare Python SDK ↗. This guide uses the Python SDK to create an AI Search instance with hybrid search enabled and index a file, then uses the LangChain retriever to search it as a tool.
- Python ↗ 3.10 or later
- Your account ID
- An API token with both the AI Search:Edit and AI Search:Run permissions
To create the token, follow Create an API token and add both permissions. Edit provisions the instance and uploads files; Run performs the search.
Create a project directory and a virtual environment to isolate your dependencies.
mkdir ai-search-langchain && cd ai-search-langchain
python3 -m venv .venv
source .venv/bin/activateOn Windows, activate the virtual environment with .venv\Scripts\activate instead.
Install both packages:
pip install -U langchain-cloudflare cloudflareThe cloudflare SDK creates the instance and uploads files. The langchain-cloudflare package provides the retriever and the RAG and agent-tool helpers. Installing langchain-cloudflare also installs langchain-core, so you do not need to install langchain separately.
Export your account ID and API token. The Cloudflare SDK reads these automatically.
export CLOUDFLARE_ACCOUNT_ID="<ACCOUNT_ID>"
export CLOUDFLARE_API_TOKEN="<API_TOKEN>"Create a file named main.py. The following code creates an instance with hybrid search enabled by setting index_method to index both vectors and keywords. Because no data source is connected, the instance uses built-in storage.
Creating an instance that already exists fails, so the code checks for it first and creates it only if it is missing.
import os
from cloudflare import Cloudflare, NotFoundError
ACCOUNT_ID = os.environ["CLOUDFLARE_ACCOUNT_ID"]
API_TOKEN = os.environ["CLOUDFLARE_API_TOKEN"]
NAMESPACE = "default"
INSTANCE_ID = "knowledge-base"
# The SDK authenticates with this token; the account ID is passed on each call.
client = Cloudflare(api_token=API_TOKEN)
# create() fails if the instance already exists, so check for it with read()
# first and only create it when read() raises NotFoundError.
try:
client.aisearch.namespaces.instances.read(
INSTANCE_ID, account_id=ACCOUNT_ID, name=NAMESPACE
)
print(f"Instance '{INSTANCE_ID}' already exists.")
except NotFoundError:
client.aisearch.namespaces.instances.create(
name=NAMESPACE,
account_id=ACCOUNT_ID,
id=INSTANCE_ID,
# Index both vectors and keywords to enable hybrid search.
index_method={"vector": True, "keyword": True},
)
print(f"Created instance '{INSTANCE_ID}'.")The first positional argument to create() is the namespace name. If you created a vector-only instance earlier, enable hybrid search on it with client.aisearch.namespaces.instances.update(...) instead.
Add the following to main.py to upload a document to built-in storage. Setting wait_for_completion to True inside the file argument waits until the file is indexed before returning.
item = client.aisearch.namespaces.instances.items.upload(
id=INSTANCE_ID,
account_id=ACCOUNT_ID,
name=NAMESPACE,
file={
# (filename, file bytes, content type)
"file": (
"workers-ai.md",
b"To configure Workers AI, add an [ai] binding and call env.AI.run().",
"text/markdown",
),
# Block until indexing finishes so the file is searchable right away.
"wait_for_completion": True,
},
)
print(f"Uploaded '{item.key}' (status: {item.status}).")If indexing is still finishing, item.status may be running; the file continues indexing in the background and becomes searchable shortly after.
There are three ways to use your instance from LangChain. Pick the one that fits your application:
- Search directly returns the matching documents.
- Use AI Search as an agent tool gives an agent the ability to search your content.
- Build a RAG chain generates an answer from the retrieved documents.
All three use the same CloudflareAISearchRetriever, which the first option creates.
Point a CloudflareAISearchRetriever at the instance. Set retrieval_type to hybrid to use the vector and keyword indexes you enabled.
from langchain_cloudflare import CloudflareAISearchRetriever
retriever = CloudflareAISearchRetriever(
account_id=ACCOUNT_ID,
api_token=API_TOKEN,
instance_name=INSTANCE_ID,
namespace=NAMESPACE,
retrieval_type="hybrid", # query both the vector and keyword indexes
k=5, # maximum number of results to return (capped at 50)
)
# invoke() runs a search and returns standard LangChain Documents.
docs = retriever.invoke("How do I configure Workers AI?")
for doc in docs:
# Each document carries its relevance score and source filename in metadata.
print(doc.metadata["score"], doc.metadata["filename"])
print(doc.page_content)The k parameter sets the maximum number of results, mapped to max_num_results and capped at 50.
Wrap the retriever with create_retriever_tool to give an agent the ability to search your content. This is the recommended way to use AI Search from a LangChain agent.
from langchain_core.tools import create_retriever_tool
# create_retriever_tool wraps the retriever as a standard LangChain tool. The
# name and description are what the agent's model sees when deciding to call it.
search_tool = create_retriever_tool(
retriever,
name="cloudflare_ai_search",
description="Search the knowledge base for relevant passages.",
)
print(search_tool.invoke({"query": "How do I configure Workers AI?"}))search_tool is a standard LangChain tool. Pass it to any LangChain or LangGraph agent alongside your other tools.
To answer questions from the retrieved content, combine the retriever with a model. This example uses ChatCloudflareWorkersAI, which is included in the same package. Pass your account ID and token the same way you did for the retriever.
Because this option calls Workers AI to generate the answer, the token you pass here also needs the Workers AI permission. Add it to the token you already created, or pass a separate token.
from langchain_cloudflare import ChatCloudflareWorkersAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
llm = ChatCloudflareWorkersAI(
account_id=ACCOUNT_ID,
api_token=API_TOKEN,
model="@cf/zai-org/glm-5.2",
)
prompt = ChatPromptTemplate.from_template(
"Answer the question using only the context below.\n\n"
"Context:\n{context}\n\n"
"Question: {question}"
)
# Join the retrieved documents into a single context string for the prompt.
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
# LCEL chain: retrieve context and pass the question through, fill the prompt,
# call the model, then parse the response down to a plain string.
chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
print(chain.invoke("How do I configure Workers AI?"))Inside a Python Worker, pass a Worker binding instead of REST credentials. The binding path is asynchronous, so use ainvoke.
from workers import WorkerEntrypoint, Response
from langchain_cloudflare import CloudflareAISearchRetriever
class Default(WorkerEntrypoint):
async def fetch(self, request):
# Pass a Worker binding instead of REST credentials.
retriever = CloudflareAISearchRetriever(binding=self.env.MY_SEARCH)
# The binding path is asynchronous, so use ainvoke.
docs = await retriever.ainvoke("How do I configure Workers AI?")
return Response.json({"matches": [doc.page_content for doc in docs]})self.env.MY_SEARCH is a dedicated ai_search binding, not the Workers AI binding (self.env.AI). For a namespace binding, pass self.env.<NAMESPACE>.get("my-instance").