K
AI Engineering & Trends·13 min read·2,563 words

How to Build a RAG Application with LangChain and Pinecone

Build a reproducible RAG application that ingests a document, stores its embeddings in Pinecone, tests retrieval, returns sources, and handles unsupported questions.

Krunal Kanojiya

Krunal Kanojiya

·Updated
Share:
#embeddings#langchain#llm#pinecone#python#rag#tutorial#vector-database
How to Build a RAG Application with LangChain and Pinecone

A RAG demo can produce a convincing answer even when its retrieval is poor. The language model smooths over missing context, so the final response looks better than the system underneath it.

This tutorial takes the opposite approach. You will build a small Retrieval-Augmented Generation (RAG) application, but you will test each stage before adding the next one:

  1. Load a known document.
  2. Split it into traceable chunks.
  3. Store those chunks in a Pinecone namespace.
  4. Inspect retrieval results for known questions.
  5. Generate an answer from the retrieved context.
  6. Return the source with the answer.
  7. Calibrate an explicit fallback for unsupported questions.
  8. Delete the tutorial index when you finish.

If you need the conceptual model first, read how RAG works. Here, the goal is implementation.

Verification status: The imports, dependency resolution, and local document/chunking code in this tutorial were checked on Python 3.12. The Pinecone and OpenAI calls follow current official documentation but were not executed because no API credentials were supplied. Run the included checkpoints in your own account before describing the application as tested end to end.

What you are building

The application has an offline ingestion path and an online question-answering path:

text
Offline ingestion
handbook.txt -> load -> split -> embed -> Pinecone namespace

Online query
question -> embed -> retrieve chunks -> prompt + context -> LLM
                                                   |
                                                   -> answer + sources

The example uses a short fictional employee handbook. A known fixture makes retrieval testable: you know which passage should answer each question before an LLM becomes involved.

This is a learning system, not a production template. A production RAG service also needs authorization, tenant isolation, incremental ingestion, monitoring, evaluation, recovery, and security controls.

Prerequisites and compatible versions

LangChain v1 requires Python 3.10 or later. This tutorial uses Python 3.12.

You also need:

  • A Pinecone account and API key
  • An OpenAI API key
  • A shell that can set environment variables
  • Basic Python familiarity

The dependency set below was resolved together on July 31, 2026:

text
langchain==1.3.14
langchain-openai==1.4.1
langchain-pinecone==0.2.13
langchain-text-splitters==1.1.2
pinecone==7.3.0
python-dotenv==1.2.2

Why pin Pinecone 7.3.0 instead of installing the newest standalone SDK? At the time of verification, langchain-pinecone==0.2.13 declared compatibility with pinecone>=6.0.0,<8.0.0. Installing Pinecone v9 beside it creates a dependency conflict. Recheck the current LangChain Pinecone integration before you upgrade.

API use can incur charges. OpenAI currently lists text-embedding-3-small at $0.02 per million input tokens. Pinecone currently lists a $0 Starter plan, but its pricing and database limits can change.

Create the project

Create this structure:

text
rag-langchain-pinecone/
├── .env
├── .gitignore
├── requirements.txt
├── data/
│   └── handbook.txt
├── ingest.py
├── evaluate.py
├── ask.py
└── cleanup.py

Create and activate a virtual environment:

bash
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
pip install -r requirements.txt

Put the pinned package list from the previous section in requirements.txt.

Add your keys to .env:

dotenv
OPENAI_API_KEY=replace-with-your-openai-key
PINECONE_API_KEY=replace-with-your-pinecone-key

Prevent Git from tracking the secrets:

gitignore
.env
.venv/
__pycache__/

Never paste real API keys into source code, screenshots, or a public repository.

Checkpoint: verify the environment

Run:

bash
python --version
pip check
pip freeze | grep -E "langchain|pinecone|python-dotenv"

pip check should report no broken requirements. If it reports a Pinecone version conflict, reinstall the versions from requirements.txt in a clean virtual environment.

Add a reproducible source document

Create data/handbook.txt:

text
Northstar Labs Employee Handbook

Onboarding
New employees complete security training during their first week. They receive
access to the source-code repositories only after their manager approves the
access request.

Remote work
Employees may work remotely up to three days per week. Fully remote arrangements
require written approval from the department director.

Learning budget
Each employee receives an annual learning budget of $1,200. The budget may be used
for courses, books, and professional certifications. Conference travel requires
separate approval.

Leave
Full-time employees receive 20 days of paid annual leave. Unused annual leave does
not carry into the next calendar year.

This fixture gives you four facts with distinct wording. It also gives you an obvious unsupported question—such as “Does the company provide dental insurance?”—because the document says nothing about insurance.

Load and split the document

Create the first part of ingest.py:

python
import hashlib
import os
import time
from pathlib import Path

from dotenv import load_dotenv
from langchain_core.documents import Document
from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain_text_splitters import RecursiveCharacterTextSplitter
from pinecone import Pinecone, ServerlessSpec

load_dotenv()

INDEX_NAME = "rag-tutorial"
NAMESPACE = "northstar-handbook"
EMBEDDING_MODEL = "text-embedding-3-small"
EMBEDDING_DIMENSION = 1536
SOURCE_PATH = Path("data/handbook.txt")

if not SOURCE_PATH.exists():
    raise FileNotFoundError(f"Missing fixture: {SOURCE_PATH}")

documents = [
    Document(
        page_content=SOURCE_PATH.read_text(encoding="utf-8"),
        metadata={"source": SOURCE_PATH.name},
    )
]

splitter = RecursiveCharacterTextSplitter(
    chunk_size=350,
    chunk_overlap=40,
    separators=["\n\n", "\n", ". ", " ", ""],
)
chunks = splitter.split_documents(documents)

for position, chunk in enumerate(chunks):
    stable_input = (
        f"{NAMESPACE}:{chunk.metadata['source']}:"
        f"{position}:{chunk.page_content}"
    )
    chunk.metadata["chunk_id"] = hashlib.sha256(
        stable_input.encode("utf-8")
    ).hexdigest()

print(f"Loaded {len(documents)} document")
print(f"Created {len(chunks)} chunks")
for chunk in chunks:
    print(
        chunk.metadata["chunk_id"][:12],
        len(chunk.page_content),
        chunk.page_content[:60].replace("\n", " "),
    )

RecursiveCharacterTextSplitter measures chunk_size in characters unless you configure another length function. The values above are starting settings for this small fixture, not general RAG recommendations. For a larger corpus, use a labeled question set to choose chunk size and overlap.

The SHA-256 hash gives each chunk a deterministic ID. Running the same ingestion code again sends the same IDs rather than creating a new anonymous set.

Checkpoint: inspect the chunks

Run:

bash
python ingest.py

At this stage, the script has not contacted OpenAI or Pinecone. Confirm that:

  • one source document loads;
  • the chunk count is greater than zero;
  • every chunk prints a different ID prefix;
  • the text remains readable around each boundary.

If these checks fail, fix document loading or chunking before adding cloud services.

Create a compatible Pinecone index

Append this code to ingest.py:

python
required_keys = ["OPENAI_API_KEY", "PINECONE_API_KEY"]
missing_keys = [name for name in required_keys if not os.getenv(name)]
if missing_keys:
    raise RuntimeError(
        "Missing environment variables: " + ", ".join(missing_keys)
    )

embeddings = OpenAIEmbeddings(model=EMBEDDING_MODEL)
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])

if not pc.has_index(INDEX_NAME):
    pc.create_index(
        name=INDEX_NAME,
        dimension=EMBEDDING_DIMENSION,
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1"),
    )

while not pc.describe_index(INDEX_NAME).status["ready"]:
    time.sleep(1)

description = pc.describe_index(INDEX_NAME)
if description.dimension != EMBEDDING_DIMENSION:
    raise RuntimeError(
        f"Index {INDEX_NAME!r} has dimension {description.dimension}; "
        f"{EMBEDDING_MODEL!r} expects {EMBEDDING_DIMENSION}. "
        "Use a different index name or recreate the tutorial index."
    )

index = pc.Index(INDEX_NAME)
vector_store = PineconeVectorStore(
    index=index,
    embedding=embeddings,
    namespace=NAMESPACE,
)

text-embedding-3-small returns 1,536 dimensions by default, so the Pinecone index must use the same dimension. You can learn why that compatibility matters in this guide to embeddings in RAG.

Do not delete and recreate an existing index automatically when its dimension is wrong. The index may contain unrelated data. This tutorial stops with a specific error instead.

Pinecone’s Starter-plan region and index limits may change, so check the current Pinecone limits if index creation fails.

Ingest the chunks without multiplying records

Finish ingest.py:

python
ids = [chunk.metadata["chunk_id"] for chunk in chunks]
vector_store.add_documents(documents=chunks, ids=ids)

# Pinecone updates are eventually visible. Poll the namespace count briefly.
deadline = time.time() + 30
while time.time() < deadline:
    stats = index.describe_index_stats()
    namespace_stats = stats.namespaces.get(NAMESPACE)
    vector_count = namespace_stats.vector_count if namespace_stats else 0
    if vector_count >= len(chunks):
        break
    time.sleep(1)
else:
    raise TimeoutError(
        "The expected vectors were not visible within 30 seconds."
    )

print(
    f"Namespace {NAMESPACE!r} contains "
    f"{vector_count} vectors; expected {len(chunks)}."
)

A namespace keeps this tutorial’s records separate from other logical datasets in the index. Deterministic IDs make the script rerunnable: an upsert with an existing ID updates that record.

Run the script twice:

bash
python ingest.py
python ingest.py

The namespace count should equal the chunk count after both runs. If the count grows on the second run, stop and inspect how IDs and namespaces are being passed.

This behavior is enough for a tutorial. A production ingestion system must also detect changed and deleted source content rather than only upserting current chunks.

Test retrieval before adding an LLM

Create evaluate.py:

python
import os

from dotenv import load_dotenv
from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from pinecone import Pinecone

load_dotenv()

INDEX_NAME = "rag-tutorial"
NAMESPACE = "northstar-handbook"

pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
index = pc.Index(INDEX_NAME)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = PineconeVectorStore(
    index=index,
    embedding=embeddings,
    namespace=NAMESPACE,
)

tests = [
    {
        "question": "How much can an employee spend on courses each year?",
        "expected_text": "$1,200",
        "in_scope": True,
    },
    {
        "question": "How many remote days are allowed each week?",
        "expected_text": "three days",
        "in_scope": True,
    },
    {
        "question": "Does the company provide dental insurance?",
        "expected_text": None,
        "in_scope": False,
    },
]

observations = []

for test in tests:
    results = vector_store.similarity_search_with_relevance_scores(
        test["question"],
        k=3,
    )
    best_score = results[0][1] if results else None
    joined_text = "\n".join(doc.page_content for doc, _ in results)
    passage_found = (
        test["expected_text"] in joined_text
        if test["expected_text"]
        else None
    )
    observations.append(
        {
            "question": test["question"],
            "in_scope": test["in_scope"],
            "best_score": best_score,
            "passage_found": passage_found,
        }
    )

    print(f"\nQuestion: {test['question']}")
    for rank, (document, score) in enumerate(results, start=1):
        print(
            f"{rank}. score={score:.4f} "
            f"source={document.metadata.get('source')} "
            f"chunk={document.metadata.get('chunk_id', '')[:12]}"
        )
        print(document.page_content[:180].replace("\n", " "))

print("\nSummary")
for observation in observations:
    print(observation)

Run it:

bash
python evaluate.py

For the two in-scope questions, confirm that the expected passage appears in the retrieved text. Record the best relevance score for all three questions.

Do not choose a cutoff merely because another tutorial uses 0.75. Score behavior depends on the embedding model, vector-store wrapper, metric, corpus, and query set. A practical starting cutoff for this fixture must sit above the unsupported question’s score while retaining both supported questions. If the scores overlap, a single cutoff cannot separate these cases; improve retrieval or use a stronger support check.

Three questions do not constitute production evaluation. They make this tutorial falsifiable. Expand them into a representative dataset using the RAG evaluation workflow.

Build the answer pipeline with current LangChain primitives

LangChain v1 streamlined its main namespace, moving legacy chains to langchain-classic. The current migration guide explains that boundary. This tutorial uses ChatPromptTemplate, Runnables, and StrOutputParser; it does not import create_retrieval_chain from langchain.chains.

Create ask.py:

python
import os
import sys

from dotenv import load_dotenv
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from pinecone import Pinecone

load_dotenv()

INDEX_NAME = "rag-tutorial"
NAMESPACE = "northstar-handbook"

# Set this only after reviewing evaluate.py output.
raw_threshold = os.getenv("MIN_RELEVANCE_SCORE")
if raw_threshold is None:
    raise RuntimeError(
        "Set MIN_RELEVANCE_SCORE after running evaluate.py. "
        "Do not copy an uncalibrated threshold."
    )
min_relevance_score = float(raw_threshold)

pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
index = pc.Index(INDEX_NAME)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = PineconeVectorStore(
    index=index,
    embedding=embeddings,
    namespace=NAMESPACE,
)

prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "Answer only from the supplied context. "
            "If the context does not support the answer, reply exactly: "
            "I don't have enough information in the handbook.\n\n"
            "Context:\n{context}",
        ),
        ("human", "{question}"),
    ]
)

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
answer_chain = prompt | model | StrOutputParser()


def answer_question(question: str) -> dict:
    scored_documents = (
        vector_store.similarity_search_with_relevance_scores(
            question,
            k=4,
        )
    )
    supported = [
        (document, score)
        for document, score in scored_documents
        if score >= min_relevance_score
    ]

    if not supported:
        return {
            "answer": "I don't have enough information in the handbook.",
            "sources": [],
        }

    context = "\n\n---\n\n".join(
        document.page_content for document, _ in supported
    )
    answer = answer_chain.invoke(
        {"context": context, "question": question}
    )

    sources = []
    seen = set()
    for document, score in supported:
        source = {
            "source": document.metadata.get("source"),
            "chunk_id": document.metadata.get("chunk_id"),
            "relevance_score": round(score, 4),
        }
        key = (source["source"], source["chunk_id"])
        if key not in seen:
            seen.add(key)
            sources.append(source)

    return {"answer": answer, "sources": sources}


question = " ".join(sys.argv[1:]).strip()
if not question:
    raise SystemExit('Usage: python ask.py "your question"')

result = answer_question(question)
print(result["answer"])
if result["sources"]:
    print("\nSources")
    for source in result["sources"]:
        print(
            f"- {source['source']} "
            f"(chunk {source['chunk_id'][:12]}, "
            f"score {source['relevance_score']})"
        )

Add the threshold you selected from evaluate.py to .env:

dotenv
MIN_RELEVANCE_SCORE=replace-with-your-calibrated-value

Then ask an in-scope question:

bash
python ask.py "What is the annual learning budget?"

The answer should state $1,200 and list at least one source chunk. Inspect that chunk in the evaluation output rather than trusting the generated sentence alone.

Now ask the unsupported question:

bash
python ask.py "Does the company provide dental insurance?"

The expected response is:

text
I don't have enough information in the handbook.

The grounding instruction and relevance cutoff reduce the chance of an unsupported answer; they do not guarantee factuality. Retrieval can still miss evidence, and an LLM can still misread supplied context. Use a broader evaluation set before relying on the behavior.

Stream the answer

Once the non-streaming path works, LCEL can stream model output:

python
for token in answer_chain.stream(
    {"context": context, "question": question}
):
    print(token, end="", flush=True)

Keep the ordinary .invoke() path while debugging. Streaming changes how output is delivered, not whether retrieval is correct.

Troubleshooting

ModuleNotFoundError: No module named 'langchain.chains'

The code is mixing pre-v1 chain imports with LangChain v1. Use the Runnable pipeline shown here. If you intentionally need legacy chains, install langchain-classic and follow its current retrieval reference.

Pinecone and langchain-pinecone dependency conflict

Recreate the virtual environment and install the mutually compatible versions from requirements.txt. Do not force Pinecone v9 beside langchain-pinecone==0.2.13.

Index dimension error

The existing index was created for a different embedding dimension. Use a new tutorial index name, or manually delete the tutorial index only after confirming it contains no valuable data.

The namespace count grows after every run

Check that add_documents receives the deterministic ids list and that the namespace is identical on every run.

Retrieval returns the wrong section

Inspect chunks before changing the LLM. Adjust boundaries, chunk size, overlap, query wording, or retrieval strategy, then rerun the labeled questions. The guide to why RAG fails covers the broader diagnostic process.

The unsupported question passes the cutoff

A single score threshold does not separate your in-scope and out-of-scope examples. Add more negative questions, improve the corpus or retrieval strategy, or introduce a separate support-verification step. Do not raise the threshold until it breaks valid questions.

Rate-limit or quota errors

Read the error details and check current OpenAI usage limits and Pinecone database limits. Avoid retry loops without a maximum attempt count and backoff.

Clean up the tutorial index

Create cleanup.py:

python
import os

from dotenv import load_dotenv
from pinecone import Pinecone

load_dotenv()

INDEX_NAME = "rag-tutorial"

confirmation = input(
    f"Type {INDEX_NAME!r} to permanently delete that Pinecone index: "
)
if confirmation != INDEX_NAME:
    raise SystemExit("Index deletion cancelled.")

pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
if pc.has_index(INDEX_NAME):
    pc.delete_index(INDEX_NAME)
    print(f"Deleted index {INDEX_NAME!r}.")
else:
    print(f"Index {INDEX_NAME!r} does not exist.")

Run it only when you no longer need the tutorial data:

bash
python cleanup.py

The confirmation protects against an accidental deletion caused by a typo or copied index name. Still inspect the target in your Pinecone project before confirming: index deletion is destructive.

What must change for production

The pipeline you built contains the core indexing, retrieval, and generation stages. Production changes more than the number of vectors.

A production design usually needs:

  • authentication and authorization before retrieval;
  • tenant isolation that cannot be bypassed by user-supplied filters;
  • document-change and deletion tracking;
  • batch controls, retries, and dead-letter handling;
  • a representative retrieval and answer-quality evaluation set;
  • traces, latency metrics, failure alerts, and cost monitoring;
  • prompt-injection and untrusted-document defenses;
  • secret management instead of local .env files;
  • backups, retention rules, and recovery procedures;
  • deployment, scaling, and rollback controls.

If exact product codes, error messages, or names are frequently missed, test hybrid search for RAG. If the managed database itself is still an open decision, use the vector-database selection guide before extending this prototype.

Frequently Asked Questions

Do I need Pinecone to build a RAG application with LangChain?

No. LangChain supports multiple vector-store integrations. Pinecone is useful when you want managed vector search without operating the database infrastructure yourself, but local and self-hosted alternatives may be a better fit for development, privacy, existing infrastructure, or cost requirements. Compare those trade-offs before choosing a production database rather than treating the tutorial stack as a universal default.

Why does `from langchain.chains import create_retrieval_chain` fail?

LangChain v1 reduced the main `langchain` namespace and moved legacy chains to `langchain-classic`. Use the Runnable/LCEL implementation in this tutorial, or deliberately install `langchain-classic` and follow its [current retrieval-chain reference](https://reference.langchain.com/python/langchain-classic/chains/retrieval). Do not mix v1 installation instructions with pre-v1 imports.

How should I choose a relevance-score threshold?

Run labeled in-scope and out-of-scope questions through the same embedding model, vector store, metric, and corpus that your application uses. Choose a cutoff only when it rejects unsupported questions without removing evidence needed for supported ones. If the score ranges overlap, a single threshold is insufficient; improve retrieval or add another support-verification step.

Why does rerunning ingestion create duplicate vectors?

This usually happens when each run generates new record IDs or writes to a different namespace. Assign a deterministic ID derived from stable source and chunk information, pass those IDs explicitly to `add_documents`, and keep the namespace constant. Also remember that production ingestion must delete records for source content that has been removed.

How much does this LangChain and Pinecone RAG application cost?

There is no reliable universal monthly figure. Cost depends on document volume, embedding tokens, vector storage, reads, writes, LLM input and output tokens, and traffic. OpenAI currently lists `text-embedding-3-small` at [$0.02 per million input tokens](https://developers.openai.com/api/docs/models/text-embedding-3-small), while Pinecone currently offers a $0 Starter plan with quotas described on its [pricing](https://www.pinecone.io/pricing/) and [limits](https://docs.pinecone.io/reference/api/database-limits) pages. Recheck all three before publication or deployment.

Is this RAG application production-ready?

No. It demonstrates a verifiable indexing, retrieval, and generation path. Production use additionally requires authentication, authorization, tenant isolation, lifecycle-aware ingestion, larger evaluation sets, monitoring, security controls, recovery procedures, cost controls, and deployment safeguards.

Follow on Google

Add as a preferred source in Search & Discover

Add as preferred source
Appears in Google Discover
Krunal Kanojiya

Krunal Kanojiya

Technical Content Writer

I am a technical writer and former software developer from India. I publish practical tutorials and in-depth guides on AI engineering, data engineering, programming, algorithms, blockchain, and modern software development.

AboutEditorial standardsGitHubLinkedInX