Complete Cohere API Tutorial: Embeddings, Reranking, and RAG with Python

Cohere is an AI platform that provides large language models (LLMs) through an easy-to-use API. Unlike other AI providers, Cohere has distinct strengths in three key areas: high-quality text embeddings, reranking to improve search relevance, and Retrieval-Augmented Generation (RAG) with built-in support. This tutorial covers how to use the Cohere API comprehensively with Python, from basic setup to building a production-ready RAG system.

Why Cohere?

Before diving into implementation, here are several reasons why Cohere deserves consideration:

Installation and Setup

SDK Installation

pip install cohere

For a more complete project, install additional dependencies:

pip install cohere numpy scikit-learn pandas chromadb

Getting an API Key

export COHEREAPIKEY="your-api-key-here"

Initializing the Client

import cohere
import os

co = cohere.ClientV2(
    apikey=os.environ.get("COHEREAPIKEY")

)

response = co.chat(
    model="command-a-08-2025",
    messages=[
        {
            "role": "user",
            "content": "Explain the concept of machine learning in 3 sentences"
        }
    ]
)

print(response.message.content[0].text)

response = co.chat(
    model="command-a-08-2025",
    messages=[
        {
            "role": "system",
            "content": "You are an AI assistant specializing in data science. Answer clearly and concisely with practical examples."
        },
        {
            "role": "user",
            "content": "What is the difference between supervised and unsupervised learning?"
        }
    ]
)

print(response.message.content[0].text)

response = co.chatstream(
    model="command-a-08-2025",
    messages=[
        {
            "role": "user",
            "content": "Write a brief guide on feature engineering"
        }
    ]
)

for event in response:
    if event.type == "content-delta":
        print(event.delta.message.content.text, end="")

Structured Output with Response Format

Cohere supports structured output using JSON schema:

import json

response = co.chat(
    model="command-a-08-2025",
    messages=[
        {
            "role": "user",
            "content": "Give 3 Python library recommendations for data visualization"
        }
    ],
    responseformat={

        "type": "jsonobject",
        "jsonschema": {

            "type": "object",
            "properties": {
                "recommendations": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "name": {"type": "string"},
                            "usecase": {"type": "string"},
                            "difficulty": {"type": "string", "enum": ["beginner", "intermediate", "advanced"]}
                        },
                        "required": ["name", "usecase", "difficulty"]

                    }
                }
            },
            "required": ["recommendations"]
        }
    }
)

result = json.loads(response.message.content[0].text)
print(json.dumps(result, indent=2))

texts = [
    "Machine learning is a branch of artificial intelligence",
    "Deep learning uses multi-layered neural networks",
    "Natural language processing handles human language",
    "Computer vision understands images and video",
    "Reinforcement learning learns from environmental interaction"
]

response = co.embed(
    texts=texts,
    model="embed-v4.0",
    inputtype="searchdocument",

    embeddingtypes=["float"]
)

embeddings = response.embeddings.float


print(f"Number of embeddings: {len(embeddings)}")
print(f"Dimensions per embedding: {len(embeddings[0])}")

The inputtype parameter is crucial for embedding quality:

• searchdocument: For texts that will be searched (stored in a database)
• searchquery: For user search queries
• classification: For text classification tasks
• clustering: For clustering/grouping tasks

queryembedding = co.embed(
    texts=["What is deep learning?"],
    model="embed-v4.0",
    inputtype="searchquery",

    embeddingtypes=["float"]
)

docembeddings = co.embed(

    texts=texts,
    model="embed-v4.0",
    inputtype="searchdocument",

    embeddingtypes=["float"]
)

Semantic Search with Embeddings

import numpy as np

def cosinesimilarity(a, b):

    return np.dot(a, b) / (np.linalg.norm(a)  np.linalg.norm(b))


query = "How can computers understand images?"

queryemb = co.embed(
    texts=[query],
    model="embed-v4.0",
    inputtype="searchquery",
    embeddingtypes=["float"]

).embeddings.float[0]

docembs = co.embed(

    texts=texts,
    model="embed-v4.0",
    inputtype="searchdocument",

    embeddingtypes=["float"]
).embeddings.float


similarities = []
for i, docemb in enumerate(docembs):

    sim = cosinesimilarity(queryemb, docemb)
    similarities.append((i, sim, texts[i]))

similarities.sort(key=lambda x: x[1], reverse=True)

print(f"Query: {query}\n")
for idx, sim, text in similarities:
    print(f"  Score: {sim:.4f} | {text}")

multilingualtexts = [
    "Artificial intelligence is transforming industries",
    "Kecerdasan buatan mengubah industri",
    "L'intelligence artificielle transforme les industries",
    "人工知能は産業を変革している",
    "La inteligencia artificial está transformando las industrias"
]

response = co.embed(
    texts=multilingualtexts,
    model="embed-v4.0",
    inputtype="searchdocument",
    embeddingtypes=["float"]

)

embs = response.embeddings.float
for i in range(len(multilingualtexts)):

    for j in range(i + 1, len(multilingualtexts)):
        sim = cosinesimilarity(embs[i], embs[j])

        print(f"'{multilingualtexts[i][:40]}...' <-> '{multilingualtexts[j][:40]}...': {sim:.4f}")

query = "How to improve machine learning model performance?"

documents = [
    "Hyperparameter tuning finds the best parameter combinations for ML models",
    "Python is a popular programming language for data science",
    "Feature engineering can significantly improve model accuracy",
    "Transfer learning enables using pre-trained models for new tasks",
    "Docker is used for application containerization",
    "Cross-validation helps evaluate model performance more robustly",
    "Ensemble methods combine multiple models for better results",
    "Git is an important version control system for developers"
]

response = co.rerank(
    model="rerank-v3.5",
    query=query,
    documents=documents,
    topn=5
)

print(f"Query: {query}\n")
print("Reranked Results (Top 5):")
for result in response.results:
    print(f"  Score: {result.relevancescore:.4f} | {documents[result.index]}")

documentswithmetadata = [
    {"title": "Hyperparameter Tuning Guide", "text": "Hyperparameter tuning finds optimal parameter combinations using grid search, random search, or Bayesian optimization"},
    {"title": "Feature Engineering Handbook", "text": "Feature engineering creates new features from raw data to improve model predictions"},
    {"title": "Docker Basics", "text": "Docker provides a containerization platform for running applications in isolated environments"},
    {"title": "Ensemble Methods", "text": "Ensemble methods like Random Forest and Gradient Boosting combine predictions from multiple models"}
]

response = co.rerank(
    model="rerank-v3.5",
    query="how to improve prediction accuracy",
    documents=[d["text"] for d in documentswithmetadata],

    topn=3
)

print("Most Relevant Documents:")
for result in response.results:
    doc = documentswithmetadata[result.index]
    print(f"\n  Title: {doc['title']}")
    print(f"  Score: {result.relevancescore:.4f}")

    print(f"  Text: {doc['text']}")

import numpy as np

corpus = [
    "Neural networks consist of interconnected layers of neurons",
    "Random Forest is an ensemble method based on decision trees",
    "Gradient descent optimizes model weights iteratively",
    "Batch normalization stabilizes the neural network training process",
    "Dropout is a regularization technique to prevent overfitting",
    "Learning rate determines how much weights update at each iteration",
    "Data augmentation increases training data through transformations",
    "Backpropagation calculates error gradients for weight updates",
    "Convolutional layers extract spatial features from image input",
    "Attention mechanism allows models to focus on relevant input parts",
    "Tokenization breaks text into small units for model processing",
    "Cats are adorable household pets",
    "Fried rice recipes require rice, soy sauce, and eggs",
    "Football is the most popular sport in the world"
]

query = "How to prevent model overfitting?"

corpusembs = co.embed(
    texts=corpus,
    model="embed-v4.0",
    inputtype="searchdocument",
    embeddingtypes=["float"]

).embeddings.float

queryemb = co.embed(

    texts=[query],
    model="embed-v4.0",
    inputtype="searchquery",

    embeddingtypes=["float"]
).embeddings.float[0]


similarities = [cosinesimilarity(queryemb, docemb) for docemb in corpusembs]
topindices = np.argsort(similarities)[-8:][::-1]

candidates = [corpus[i] for i in topindices]

print("Stage 1 - Embedding Search (Top 8):")
for i, idx in enumerate(topindices):

    print(f"  {i+1}. [{similarities[idx]:.4f}] {corpus[idx]}")

rerankresponse = co.rerank(
    model="rerank-v3.5",
    query=query,
    documents=candidates,
    topn=3

)

print(f"\nStage 2 - Reranked (Top 3):")
for result in rerankresponse.results:
    print(f"  Score: {result.relevancescore:.4f} | {candidates[result.index]}")

knowledgebase = [
    {
        "title": "About the Company",
        "text": "DataTech Solutions was founded in 2020. The company operates in data consulting and artificial intelligence. Headquarters is located in Singapore."
    },
    {
        "title": "Core Services",
        "text": "Core services include: Data Engineering, Machine Learning Operations (MLOps), Business Intelligence, and AI Consulting. Each service is available in monthly or project-based packages."
    },
    {
        "title": "Team and Expertise",
        "text": "The team consists of 25 data engineers, 15 ML engineers, and 10 data analysts. Average team experience is 5 years in the technology industry."
    },
    {
        "title": "Technology Stack",
        "text": "The technology stack includes Python, Apache Spark, Airflow, dbt, MLflow, Kubernetes, and various cloud services from AWS, GCP, and Azure."
    },
    {
        "title": "Pricing",
        "text": "Consulting packages start from $1,000/month for Basic, $2,500 for Professional, and $5,000 for Enterprise. Custom pricing is available for large-scale projects."
    }
]

query = "How much does the consulting service cost?"

doctexts = [doc["text"] for doc in knowledgebase]
docembs = co.embed(

    texts=doctexts,
    model="embed-v4.0",
    inputtype="searchdocument",
    embeddingtypes=["float"]

).embeddings.float

queryemb = co.embed(

    texts=[query],
    model="embed-v4.0",
    inputtype="searchquery",

    embeddingtypes=["float"]
).embeddings.float[0]


similarities = [cosinesimilarity(queryemb, emb) for emb in docembs]
topindices = np.argsort(similarities)[-3:][::-1]

relevantdocs = [knowledgebase[i] for i in topindices]

rerankresponse = co.rerank(

    model="rerank-v3.5",
    query=query,
    documents=[doc["text"] for doc in relevantdocs],
    topn=2

)

contextdocs = []
for result in rerankresponse.results:

    doc = relevantdocs[result.index]
    contextdocs.append(doc)


documentsforrag = [

    {"id": str(i), "data": {"title": doc["title"], "text": doc["text"]}}
    for i, doc in enumerate(contextdocs)
]

response = co.chat(
    model="command-a-08-2025",
    messages=[
        {
            "role": "user",
            "content": query
        }
    ],
    documents=documentsforrag
)

print(f"Query: {query}\n")
print(f"Answer: {response.message.content[0].text}")

if response.message.citations:
    print("\nSources:")
    for citation in response.message.citations:
        print(f"  - {citation}")

import chromadb

chromaclient = chromadb.Client()

collection = chromaclient.createcollection(name="knowledgebase")

documents = [
    "Python is the most popular programming language for data science and machine learning.",
    "TensorFlow and PyTorch are the two main deep learning frameworks used in the industry.",
    "Pandas provides DataFrame data structures for tabular data manipulation and analysis.",
    "Scikit-learn provides easy-to-use implementations of classical machine learning algorithms.",
    "Apache Spark enables distributed processing of large datasets.",
    "MLflow is an open-source platform for managing the machine learning lifecycle.",
    "Docker containers ensure environment consistency between development and production.",
    "Kubernetes orchestrates container deployments at scale.",
    "FastAPI is a modern Python framework for building high-performance APIs.",
    "Airflow is a workflow orchestrator for data and ML pipelines."
]

docembeddings = co.embed(

    texts=documents,
    model="embed-v4.0",
    inputtype="searchdocument",

    embeddingtypes=["float"]
).embeddings.float


collection.add(
    documents=documents,
    embeddings=docembeddings,
    ids=[f"doc{i}" for i in range(len(documents))]

)

def ragquery(query: str, topk: int = 5) -> str:

    queryemb = co.embed(
        texts=[query],
        model="embed-v4.0",
        inputtype="searchquery",
        embeddingtypes=["float"]

    ).embeddings.float[0]

    results = collection.query(
        queryembeddings=[queryemb],
        nresults=topk
    )

    candidatedocs = results["documents"][0]


    rerankresponse = co.rerank(
        model="rerank-v3.5",
        query=query,
        documents=candidatedocs,

        topn=3
    )

    topdocs = [candidatedocs[r.index] for r in rerankresponse.results]

    documentsforrag = [

        {"id": str(i), "data": {"text": doc}}
        for i, doc in enumerate(topdocs)
    ]

    response = co.chat(
        model="command-a-08-2025",
        messages=[
            {
                "role": "system",
                "content": "Answer questions based on the provided document context. If the information is not available in the documents, say that you don't have that information."
            },
            {
                "role": "user",
                "content": query
            }
        ],
        documents=documentsforrag
    )

    return response.message.content[0].text

answer = ragquery("Which framework is suitable for deep learning?")

print(answer)

import json

tools = [
    {
        "type": "function",
        "function": {
            "name": "getweather",
            "description": "Get weather information for a specific city",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "City name"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Temperature unit"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculatebmi",

            "description": "Calculate Body Mass Index",
            "parameters": {
                "type": "object",
                "properties": {
                    "weightkg": {
                        "type": "number",
                        "description": "Body weight in kilograms"
                    },
                    "heightm": {

                        "type": "number",
                        "description": "Height in meters"
                    }
                },
                "required": ["weightkg", "heightm"]

            }
        }
    }
]

def getweather(city: str, unit: str = "celsius") -> dict:
    weatherdata = {

        "New York": {"temp": 22, "condition": "Partly cloudy", "humidity": 65},
        "London": {"temp": 15, "condition": "Overcast", "humidity": 80},
        "Tokyo": {"temp": 28, "condition": "Sunny", "humidity": 70}
    }
    data = weatherdata.get(city, {"temp": 25, "condition": "Unknown", "humidity": 60})
    if unit == "fahrenheit":
        data["temp"] = data["temp"]  9/5 + 32
    return {data, "city": city, "unit": unit}


def calculatebmi(weightkg: float, heightm: float) -> dict:

    bmi = weightkg / (heightm  2)

    if bmi < 18.5:
        category = "Underweight"
    elif bmi < 25:
        category = "Normal"
    elif bmi < 30:
        category = "Overweight"
    else:
        category = "Obese"
    return {"bmi": round(bmi, 2), "category": category}

response = co.chat(
    model="command-a-08-2025",
    messages=[
        {"role": "user", "content": "What's the weather like in Tokyo today?"}
    ],
    tools=tools
)

if response.message.toolcalls:
    toolresults = []

    for toolcall in response.message.toolcalls:

        funcname = toolcall.function.name

        args = json.loads(toolcall.function.arguments)

        if funcname == "getweather":
            result = getweather(args)

        elif funcname == "calculatebmi":

            result = calculatebmi(args)

        toolresults.append({

            "call": toolcall,
            "outputs": [result]
        })

    followup = co.chat(

        model="command-a-08-2025",
        messages=[
            {"role": "user", "content": "What's the weather like in Tokyo today?"},
            {"role": "assistant", "toolcalls": response.message.toolcalls},

            {"role": "tool", "toolcallid": response.message.toolcalls[0].id, "content": json.dumps(result)}
        ],
        tools=tools
    )
    print(followup.message.content[0].text)

def batchembed(texts: list, batchsize: int = 96) -> list:
    allembeddings = []
    for i in range(0, len(texts), batchsize):

        batch = texts[i:i + batchsize]
        response = co.embed(
            texts=batch,
            model="embed-v4.0",
            inputtype="searchdocument",
            embeddingtypes=["float"]

        )
        allembeddings.extend(response.embeddings.float)

        print(f"Processed batch {i // batchsize + 1}/{(len(texts) - 1) // batchsize + 1}")

    return allembeddings

largecorpus = [f"Document number {i} contains information about topic {i % 10}" for i in range(500)]

embeddings = batchembed(largecorpus)

print(f"Total embeddings: {len(embeddings)}")

from sklearn.linearmodel import LogisticRegression
from sklearn.modelselection import traintestsplit

from sklearn.metrics import classificationreport

traintexts = [

    "Stock prices rose 2% today",
    "Inflation this month recorded at 3.5%",
    "Central bank raises benchmark interest rate",
    "Company revenue grew 15% YoY",
    "National team wins 3-0 against rivals",
    "Premier League enters week 20",
    "New player joins the football club",
    "The World Cup final was very exciting",
    "Researchers discover new diabetes medication",
    "New COVID vaccine begins distribution",
    "Hospital adds ICU capacity",
    "Government launches public health program"
]

trainlabels = [
    "economy", "economy", "economy", "economy",
    "sports", "sports", "sports", "sports",
    "health", "health", "health", "health"
]

trainembs = co.embed(

    texts=traintexts,
    model="embed-v4.0",
    inputtype="classification",

    embeddingtypes=["float"]
).embeddings.float


clf = LogisticRegression(maxiter=1000)
clf.fit(trainembs, trainlabels)

testtexts = [

    "Dollar weakens against the euro",
    "A beautiful goal scored in the final minute",
    "Bird flu pandemic draws WHO attention"
]

testembs = co.embed(
    texts=testtexts,

    model="embed-v4.0",
    inputtype="classification",
    embeddingtypes=["float"]

).embeddings.float

predictions = clf.predict(testembs)

for text, pred in zip(testtexts, predictions):
    print(f"  '{text}' -> {pred}")

Document Clustering

from sklearn.cluster import KMeans

documents = [
    "How to cook authentic beef rendang",
    "Traditional chicken soup recipe with coconut milk",
    "Tips for making delicious fried rice",
    "Python tutorial for beginners",
    "Learning machine learning from scratch",
    "Guide to using Git and GitHub",
    "Tips for maintaining heart health",
    "Benefits of regular daily exercise",
    "How to get quality sleep"
]

embeddings = co.embed(
    texts=documents,
    model="embed-v4.0",
    inputtype="clustering",

    embeddingtypes=["float"]
).embeddings.float


kmeans = KMeans(nclusters=3, randomstate=42, ninit=10)
clusters = kmeans.fitpredict(embeddings)


for clusterid in range(3):
    print(f"\nCluster {clusterid}:")

    for i, doc in enumerate(documents):
        if clusters[i] == clusterid:
            print(f"  - {doc}")

Best Practices

1. Choose the Right Model

| Use Case | Model | Notes |

|-----------|-------|-------|

| Text Generation | command-a-08-2025 | Latest and most capable model |

| Embeddings | embed-v4.0 | Top-ranked on MTEB benchmark |

| Reranking | rerank-v3.5 | Supports multi-aspect and long documents |

| Lightweight Generation | command-r | Faster, lower cost |

2. Embedding Optimization

• Always use the appropriate inputtype for your use case
• Process in batches for large volumes (maximum 96 texts per request)
• Cache embeddings for documents that rarely change
• Use embeddingtypes=["float"] for full precision

3. Effective Reranking Strategies

• Use two-stage retrieval: embeddings for recall, reranking for precision
• Set topn appropriately to save latency
• Reranking works best on 50-200 candidate documents
• For small datasets (under 50 documents), direct reranking without embeddings is sufficient

def productionrag(query: str, collection, topk: int = 10, reranktop: int = 3) -> dict:
    queryemb = co.embed(

        texts=[query],
        model="embed-v4.0",
        inputtype="searchquery",

        embeddingtypes=["float"]
    ).embeddings.float[0]


    results = collection.query(
        queryembeddings=[queryemb],

        nresults=topk

    )

    if not results["documents"][0]:
        return {"answer": "No relevant documents found.", "sources": []}

    candidatedocs = results["documents"][0]

    rerankresponse = co.rerank(

        model="rerank-v3.5",
        query=query,
        documents=candidatedocs,
        topn=reranktop
    )

    filtereddocs = [

        candidatedocs[r.index]
        for r in rerankresponse.results

        if r.relevancescore > 0.3
    ]

    if not filtereddocs:

        return {"answer": "No sufficiently relevant documents found.", "sources": []}

    documentsforrag = [

        {"id": str(i), "data": {"text": doc}}
        for i, doc in enumerate(filtereddocs)
    ]

    response = co.chat(
        model="command-a-08-2025",
        messages=[
            {
                "role": "system",
                "content": "Answer based on the available documents. If no relevant information exists, say so honestly."
            },
            {"role": "user", "content": query}
        ],
        documents=documentsforrag,
        temperature=0.3
    )

    return {
        "answer": response.message.content[0].text,
        "sources": filtereddocs,

        "citations": response.message.citations if response.message.citations else []
    }

import cohere
import time

def safeapicall(func, maxretries=3, basedelay=1):

    for attempt in range(maxretries):
        try:
            return func()
        except cohere.TooManyRequestsError:
            delay = basedelay  (2  attempt)

            print(f"Rate limited. Retrying in {delay}s...")
            time.sleep(delay)
        except cohere.BadRequestError as e:
            print(f"Bad request: {e}")
            raise
        except Exception as e:
            if attempt == maxretries - 1:
                raise
            print(f"Error: {e}. Retrying...")
            time.sleep(basedelay)


result = safeapicall(

    lambda: co.embed(
        texts=["test"],
        model="embed-v4.0",
        inputtype="searchquery",

        embeddingtypes=["float"]
    )
)

The two-stage retrieval approach (embedding + reranking) demonstrated in this tutorial is a proven pattern in production environments and has become an industry standard for AI-powered search systems. By mastering the Cohere API, you have a solid foundation for building various AI applications ranging from chatbots and search systems to knowledge management platforms.

For next steps, explore integrating Cohere with vector databases like Qdrant or Weaviate to build more scalable RAG systems, and consider using fine-tuning to customize models for your specific domain.