LangChain

mkdir ai-search-langchain && cd ai-search-langchain
python3 -m venv .venv
source .venv/bin/activate

pip install -U langchain-cloudflare cloudflare

export CLOUDFLARE_ACCOUNT_ID="<ACCOUNT_ID>"
export CLOUDFLARE_API_TOKEN="<API_TOKEN>"

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}'.")

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}).")

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)

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?"}))

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?"))

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